"""Authentication policy, password/passphrase parity and session revocation."""

from __future__ import annotations

import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock

import httpx
import pytest
import pytest_asyncio
from fastapi import FastAPI
from sqlalchemy import event, select, update
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.gateway import deps, langgraph_auth
from app.gateway.auth import config as auth_config
from app.gateway.auth import password as password_utils
from app.gateway.auth.disabled_mode import is_auth_disabled
from app.gateway.auth.jwt import create_access_token, decode_token
from app.gateway.auth.local_provider import LocalAuthProvider
from app.gateway.auth.models import User
from app.gateway.auth.password import hash_password, verify_password
from app.gateway.auth.repositories.base import CredentialConflictError
from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
from app.gateway.auth_middleware import AuthMiddleware
from app.gateway.csrf_middleware import CSRFMiddleware
from app.gateway.routers import auth, parallel_agents, service_tokens, system_settings
from deerflow.config import system_settings as settings
from deerflow.config.login_config import AuthLoginConfig
from deerflow.persistence.user.model import AuthInitializationRow, AuthUsernameRow, UserRow

PASSWORD = "Account-s3cret!"
NEW_PASSWORD = "Changed-s3cret!"


@pytest_asyncio.fixture
async def ctx(monkeypatch, tmp_path):
    engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'users.db'}")
    async with engine.begin() as connection:
        await connection.run_sync(lambda conn: UserRow.metadata.create_all(conn, tables=[UserRow.__table__, AuthInitializationRow.__table__, AuthUsernameRow.__table__]))
    sf = async_sessionmaker(engine, expire_on_commit=False)
    repository = SQLiteUserRepository(sf)
    provider = LocalAuthProvider(repository)
    monkeypatch.setattr(settings, "runtime_home", lambda: tmp_path)
    monkeypatch.setattr(auth, "get_local_provider", lambda: provider)
    monkeypatch.setattr(service_tokens, "get_local_provider", lambda: provider)
    monkeypatch.setattr(deps, "get_local_provider", lambda: provider)
    monkeypatch.setattr(langgraph_auth, "get_local_provider", lambda: provider)
    monkeypatch.setattr(auth_config, "_auth_config", auth_config.AuthConfig(jwt_secret="test-auth-key-with-sufficient-entropy-0123456789"))
    login_config = AuthLoginConfig()
    monkeypatch.setattr(auth, "get_app_config", lambda: SimpleNamespace(auth_login=login_config))
    monkeypatch.setattr(password_utils, "get_app_config", lambda: SimpleNamespace(auth_login=login_config))
    settings.save_system_settings(settings.SystemSettings(authentication=settings.AuthenticationSettings(password_login_enabled=True)))
    monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False)
    monkeypatch.delenv("DEER_FLOW_USERNAME_LOGIN_DOMAIN", raising=False)
    monkeypatch.setattr(auth, "_login_attempts", {})
    monkeypatch.setattr(auth, "_SETUP_STATUS_COOLDOWN", {})
    app = FastAPI()
    app.include_router(auth.router)
    app.include_router(service_tokens.router)
    app.include_router(system_settings.router)
    app.include_router(parallel_agents.router)
    app.add_middleware(CSRFMiddleware)
    app.add_middleware(AuthMiddleware)
    counter = SimpleNamespace(count_owned=AsyncMock(return_value=0))
    app.state.skill_store = counter
    app.state.agent_store = counter
    app.dependency_overrides[deps.get_skill_store] = lambda: counter
    app.dependency_overrides[deps.get_agent_store] = lambda: counter
    monkeypatch.setattr(auth, "get_run_store", lambda _: SimpleNamespace(count_by_user=AsyncMock(return_value={})))
    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
        yield SimpleNamespace(client=client, app=app, provider=provider, repository=repository, sf=sf, login_config=login_config)
    await engine.dispose()


@pytest.mark.asyncio
async def test_service_exchange_requires_shared_token(ctx):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)

    denied = await ctx.client.post(
        "/api/v1/auth/service/exchange",
        json={"username": "alice", "scope": ["llmwiki:sync"]},
    )
    assert denied.status_code == 401
    assert denied.json()["detail"] == "Invalid service token"

    wrong = await ctx.client.post(
        "/api/v1/auth/service/exchange",
        json={"username": "alice", "scope": ["llmwiki:sync"]},
        headers={"X-Service-Token": "wrong-token"},
    )
    assert wrong.status_code == 401

    response = await ctx.client.post(
        "/api/v1/auth/service/exchange",
        json={"username": "alice", "scope": ["llmwiki:sync"]},
        headers={"X-Service-Token": service_tokens._SERVICE_TOKEN_SECRET},
    )
    assert response.status_code == 200, response.text
    token = decode_token(response.json()["access_token"])
    assert token.sub == str(user.id)
    assert token.auth_method == "service"
    assert token.client_id is None
    assert token.scope == ["llmwiki:sync"]
    assert response.json()["expires_in"] == 300

    repeated = await ctx.client.post(
        "/api/v1/auth/service/exchange",
        json={"username": "alice", "scope": ["llmwiki:sync"]},
        headers={"X-Service-Token": service_tokens._SERVICE_TOKEN_SECRET},
    )
    assert repeated.status_code == 200


@pytest.mark.asyncio
async def test_service_exchange_rejects_unsupported_scope(ctx):
    response = await ctx.client.post(
        "/api/v1/auth/service/exchange",
        json={"username": "alice", "scope": ["admin"]},
        headers={"X-Service-Token": service_tokens._SERVICE_TOKEN_SECRET},
    )
    assert response.status_code == 403


@pytest.mark.asyncio
async def test_user_owner_lookup_batches_ids(ctx):
    first = await ctx.provider.create_user("first-owner@cm.com", password=PASSWORD)
    second = await ctx.provider.create_user("second-owner@cm.com", password=PASSWORD)

    users = await ctx.provider.get_users_by_ids([
        str(first.id), str(second.id), str(first.id), "missing-user",
    ])

    assert {str(user.id) for user in users} == {str(first.id), str(second.id)}


@pytest.mark.asyncio
async def test_username_login_uses_exact_indexed_key_and_backfills_legacy_rows(ctx):
    user = await ctx.provider.create_user("quick-login@legacy-domain.com", password=PASSWORD)
    statements: list[str] = []
    engine = ctx.sf.kw["bind"]

    def capture_sql(_conn, _cursor, statement, _parameters, _context, _executemany):
        if statement.lstrip().upper().startswith("SELECT") and "users" in statement.lower():
            statements.append(statement.lower())

    event.listen(engine.sync_engine, "before_cursor_execute", capture_sql)
    try:
        found = await ctx.provider.find_user_by_email_local_part("quick-login")
    finally:
        event.remove(engine.sync_engine, "before_cursor_execute", capture_sql)
    assert found is not None and found.id == user.id
    assert statements and "username_key =" in statements[0]
    assert "lower(users.email)" not in statements[0]

    async with ctx.sf() as session:
        await session.execute(update(UserRow).where(UserRow.id == str(user.id)).values(username_key=None))
        await session.commit()
    found_after_legacy_fallback = await ctx.provider.find_user_by_email_local_part("QUICK-LOGIN")
    assert found_after_legacy_fallback is not None and found_after_legacy_fallback.id == user.id
    async with ctx.sf() as session:
        saved_key = await session.scalar(select(UserRow.username_key).where(UserRow.id == str(user.id)))
    assert saved_key == "quick-login"
@pytest.mark.parametrize(
    ("method", "path", "allowed"),
    [
        ("GET", "/api/llmwiki/knowledge-bases", True),
        ("POST", "/api/llmwiki/knowledge-bases/kb/documents", True),
        ("POST", "/api/llmwiki/knowledge-bases/kb/documents/manual", True),
        ("POST", "/api/llmwiki/search", False),
        ("POST", "/api/threads/thread/runs", False),
        ("GET", "/api/v1/auth/me", False),
    ],
)
def test_service_scope_path_allowlist(method, path, allowed):
    from app.gateway.auth_middleware import _is_llmwiki_sync_request

    assert _is_llmwiki_sync_request(method, path) is allowed


@pytest.mark.parametrize(
    ("scopes", "method", "path", "all_requested", "allowed"),
    [
        (["recommended-questions:read"], "GET", "/api/recommended-questions", False, True),
        (["recommended-questions:read"], "GET", "/api/recommended-questions", True, False),
        (["recommended-questions:read"], "POST", "/api/recommended-questions", False, False),
        (["recommended-questions:read"], "GET", "/api/recommended-questions-reorder", False, False),
        (["recommended-questions:read"], "GET", "/api/llmwiki/knowledge-bases", False, False),
        (["llmwiki:sync"], "GET", "/api/recommended-questions", False, False),
    ],
)
def test_recommended_question_service_scope_is_read_only(scopes, method, path, all_requested, allowed):
    from app.gateway.auth_middleware import _is_service_scope_request

    assert _is_service_scope_request(scopes, method, path, all_requested=all_requested) is allowed


@pytest.mark.asyncio
async def test_service_token_is_rejected_outside_llmwiki_sync(ctx):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    token = create_access_token(
        str(user.id),
        expires_delta=service_tokens.timedelta(seconds=300),
        token_version=user.token_version,
        auth_method="service",
        client_id="surfsense",
        scope=["llmwiki:sync"],
    )
    response = await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})
    assert response.status_code == 403


def enable_registration():
    current = settings.load_system_settings()
    settings.save_system_settings(current.model_copy(update={"authentication": settings.AuthenticationSettings(password_login_enabled=True, registration_enabled=True)}))


def bearer(user, *, auth_method="password"):
    return {"Authorization": f"Bearer {create_access_token(str(user.id), token_version=user.token_version, auth_method=auth_method)}"}


async def make_admin(ctx, *, password=PASSWORD):
    return await ctx.provider.create_user("operator@cm.com", password=password, system_role="admin")


@pytest.mark.asyncio
async def test_login_identity_and_user_management_do_not_use_checkpointer(ctx, monkeypatch):
    def forbidden(*args, **kwargs):
        raise AssertionError("Authentication accessed the chat checkpoint backend")

    monkeypatch.setattr(deps, "get_checkpointer", forbidden)
    ctx.app.state.checkpointer = SimpleNamespace(aget_tuple=forbidden, alist=forbidden, aput=forbidden)
    user = await make_admin(ctx)
    ctx.login_config.token_login.enabled = True
    ctx.login_config.token_login.mock_tokens = {"audit-token": "operator"}
    assert (await ctx.client.get("/api/v1/auth/options")).status_code == 200
    for route, kwargs in [
        ("local", {"data": {"username": "operator", "password": PASSWORD}}),
        ("username", {"json": {"username": "operator"}, "params": {"password": PASSWORD}}),
        ("token", {"params": {"token": "audit-token"}}),
    ]:
        response = await ctx.client.post(f"/api/v1/auth/login/{route}", **kwargs)
        assert response.status_code == 200, response.text
        headers = {"Authorization": f"Bearer {response.json()['access_token']}"}
        assert (await ctx.client.get("/api/v1/auth/me", headers=headers)).json()["id"] == str(user.id)
        assert (await ctx.client.get("/api/v1/auth/users", headers=headers)).status_code == 200


@pytest.mark.asyncio
async def test_list_users_paged_shape_and_search(ctx):
    """传 limit → UserListPage（items/total/approved_count）；不传 → 旧全量列表。"""
    admin = await make_admin(ctx)
    for name in ("bob@cm.com", "carol@cm.com"):
        await ctx.provider.create_user(name, password=PASSWORD)
    await ctx.provider.create_user("dave@cm.com", password=PASSWORD)  # 将被取消放行
    dave = await ctx.provider.get_user_by_email("dave@cm.com")
    dave.approved = False
    await ctx.provider.update_user(dave)
    headers = bearer(admin)

    legacy = await ctx.client.get("/api/v1/auth/users", headers=headers)
    assert legacy.status_code == 200
    assert isinstance(legacy.json(), list) and len(legacy.json()) == 4

    paged = await ctx.client.get("/api/v1/auth/users", headers=headers, params={"limit": 2, "offset": 0, "include_counts": "false"})
    assert paged.status_code == 200, paged.text
    body = paged.json()
    assert body["total"] == 4 and body["approved_count"] == 3  # admin+bob+carol 可登录，dave 被拦
    assert body["limit"] == 2 and body["offset"] == 0
    assert len(body["items"]) == 2
    assert all(item["qa_count"] == 0 and item["skill_count"] == 0 for item in body["items"])

    page2 = await ctx.client.get("/api/v1/auth/users", headers=headers, params={"limit": 2, "offset": 2})
    assert page2.status_code == 200 and len(page2.json()["items"]) == 2

    searched = await ctx.client.get("/api/v1/auth/users", headers=headers, params={"limit": 10, "q": "dave"})
    assert searched.status_code == 200
    assert searched.json()["total"] == 1 and searched.json()["items"][0]["email"] == "dave@cm.com"
    assert searched.json()["approved_count"] == 0  # 搜索范围内唯一用户已被取消放行

    miss = await ctx.client.get("/api/v1/auth/users", headers=headers, params={"limit": 10, "offset": 999})
    assert miss.status_code == 200 and miss.json()["items"] == [] and miss.json()["total"] == 4


@pytest.mark.asyncio
async def test_anonymous_options_defaults_and_closed_registration(ctx):
    settings.save_system_settings(settings.SystemSettings())
    options = await ctx.client.get("/api/v1/auth/options")
    assert options.status_code == 200
    assert options.json() == {"password_login_enabled": False, "registration_enabled": False, "sso_enabled": True, "single_session_enabled": False, "needs_setup": True}
    assert options.headers["cache-control"] == "no-store"
    response = await ctx.client.post("/api/v1/auth/register", json={"username": "alice", "password": PASSWORD})
    assert response.status_code == 403
    assert await ctx.provider.count_users() == 0


@pytest.mark.asyncio
async def test_register_login_and_sso_share_password_and_user_role(ctx):
    enable_registration()
    response = await ctx.client.post("/api/v1/auth/register", json={"username": "b1admin", "password": PASSWORD})
    assert response.status_code == 201, response.text
    assert response.json()["system_role"] == "user"
    assert response.json()["access_token"]
    assert decode_token(response.json()["access_token"]).auth_method == "password"
    user = await ctx.provider.get_user(response.json()["user_id"])
    assert user.password_hash == user.login_password
    assert user.login_password != PASSWORD
    assert verify_password(PASSWORD, user.password_hash)
    for username in ("b1admin", "B1ADMIN@CM.COM"):
        login = await ctx.client.post("/api/v1/auth/login/local", data={"username": username, "password": PASSWORD})
        assert login.status_code == 200
        assert login.json()["user_id"] == str(user.id)
    denied = await ctx.client.post("/api/v1/auth/login/username", json={"username": "b1admin"})
    assert denied.status_code == 401
    sso = await ctx.client.post("/api/v1/auth/login/username", params={"password": PASSWORD}, json={"username": "b1admin"})
    assert sso.status_code == 200
    assert decode_token(sso.json()["access_token"]).auth_method == "passphrase"
    assert sso.json()["system_role"] == "user"
    me = await ctx.client.get("/api/v1/auth/me", headers=bearer(user))
    assert me.json()["has_password"] is True
    assert "password_hash" not in me.json()


@pytest.mark.asyncio
@pytest.mark.parametrize("username", ["alice", "ALICE", "alice@other.org"])
async def test_registration_rejects_legacy_username_collisions(ctx, username):
    enable_registration()
    legacy = await ctx.provider.create_user("alice@legacy.com", password=PASSWORD)
    response = await ctx.client.post("/api/v1/auth/register", json={"username": username, "password": NEW_PASSWORD})
    assert response.status_code == 409
    assert await ctx.provider.count_users() == 1
    assert verify_password(PASSWORD, (await ctx.provider.get_user(str(legacy.id))).password_hash)


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "body",
    [
        {"username": "alice", "password": "password123"},
        {"username": "alice", "password": "short"},
        {"username": "alice", "password": "x" * 256},
        {"username": "alice", "password": " Account-s3cret!"},
        {"username": "bad name", "password": PASSWORD},
        {"username": "alice", "password": PASSWORD, "system_role": "admin"},
        {"username": "alice", "password": PASSWORD, "email": "different@example.com"},
    ],
)
async def test_register_validates_credentials_and_rejects_role_injection(ctx, body):
    enable_registration()
    response = await ctx.client.post("/api/v1/auth/register", json=body)
    assert response.status_code == 422
    assert await ctx.provider.count_users() == 0


@pytest.mark.asyncio
async def test_policy_admin_only_and_lockout_protection(ctx):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    policy = {"password_login_enabled": True, "registration_enabled": True, "sso_enabled": False}
    assert (await ctx.client.get("/api/system-settings/authentication")).status_code == 401
    assert (await ctx.client.put("/api/system-settings/authentication", headers=bearer(user), json=policy)).status_code == 403
    admin = await make_admin(ctx, password=None)
    # The existing shared/default passphrase remains usable without a personal password.
    assert (await ctx.client.put("/api/system-settings/authentication", headers=bearer(admin), json=policy)).status_code == 200
    for invalid in [dict(policy, password_login_enabled=False), dict(policy, registration_enabled=False, password_login_enabled=False)]:
        assert (await ctx.client.put("/api/system-settings/authentication", headers=bearer(admin), json=invalid)).status_code == 400
    changed = await ctx.client.post("/api/v1/auth/change-password", headers=bearer(admin), json={"current_password": "123ewq", "new_password": PASSWORD})
    assert changed.status_code == 200
    assert decode_token(changed.json()["access_token"]).auth_method == "password"
    admin = await ctx.provider.get_user(str(admin.id))
    assert (await ctx.client.put("/api/system-settings/authentication", headers=bearer(admin), json=policy)).status_code == 200
    for path, kwargs in [
        ("/api/v1/auth/login/username", {"json": {"username": "newbie"}}),
        ("/api/v1/auth/login/token", {"params": {"token": "upstream"}}),
    ]:
        assert (await ctx.client.post(path, **kwargs)).status_code == 403
    # Detail integrations use the legacy token endpoint as an unconditional
    # preflight. Policy rejection returns a non-JWT compatibility value.
    compat = await ctx.client.post("/api/parallel-agents/auth/token", json={"username": "newbie"})
    assert compat.status_code == 200
    assert compat.json()["access_token"] == "force-situation-detail-compat-newbie"
    assert await ctx.provider.count_users() == 2
    assert (await ctx.client.get("/api/v1/auth/oauth/github", headers=bearer(admin))).status_code == 403


@pytest.mark.asyncio
async def test_disabling_password_login_blocks_login_and_registration(ctx):
    admin = await make_admin(ctx)
    response = await ctx.client.put("/api/system-settings/authentication", headers=bearer(admin), json={"password_login_enabled": False, "registration_enabled": False, "sso_enabled": True})
    assert response.status_code == 200
    assert (await ctx.client.post("/api/v1/auth/login/local", data={"username": "operator", "password": PASSWORD})).status_code == 403
    assert (await ctx.client.post("/api/v1/auth/register", json={"username": "alice", "password": PASSWORD})).status_code == 403


@pytest.mark.asyncio
async def test_change_password_revokes_old_sessions_and_keeps_current_bearer(ctx):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    old_headers = bearer(user)
    wrong = await ctx.client.post("/api/v1/auth/change-password", headers=old_headers, json={"current_password": "wrong", "new_password": NEW_PASSWORD})
    assert wrong.status_code == 400
    changed = await ctx.client.post("/api/v1/auth/change-password", headers=old_headers, json={"current_password": PASSWORD, "new_password": NEW_PASSWORD})
    assert changed.status_code == 200
    assert (await ctx.client.get("/api/v1/auth/me", headers=old_headers)).status_code == 401
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": "Bearer " + changed.json()["access_token"]})).status_code == 200
    saved = await ctx.provider.get_user(str(user.id))
    assert saved.token_version == 1
    assert saved.login_password == saved.password_hash
    for password, expected in [(PASSWORD, 401), (NEW_PASSWORD, 200)]:
        assert (await ctx.client.post("/api/v1/auth/login/local", data={"username": "alice", "password": password})).status_code == expected
        assert (await ctx.client.post("/api/v1/auth/login/username", params={"password": password}, json={"username": "alice"})).status_code == expected


@pytest.mark.asyncio
async def test_admin_reset_unified_password_no_hash_leak_and_self_session(ctx):
    admin = await make_admin(ctx)
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    old_headers = bearer(user)
    endpoint = f"/api/v1/auth/users/{user.id}/reset-password"
    assert (await ctx.client.put(endpoint, headers=old_headers, json={"password": NEW_PASSWORD})).status_code == 403
    response = await ctx.client.put(endpoint, headers=bearer(admin), json={"password": NEW_PASSWORD})
    assert response.status_code == 200
    assert response.json()["login_password"] is None
    assert response.json()["has_login_password"] is True
    assert (await ctx.client.get("/api/v1/auth/me", headers=old_headers)).status_code == 401
    listing = await ctx.client.get("/api/v1/auth/users", headers=bearer(admin))
    assert listing.status_code == 200
    assert "$dfv" not in listing.text and NEW_PASSWORD not in listing.text
    assert (await ctx.client.put(endpoint, headers=bearer(admin), json={"password": None})).status_code == 400
    old_admin = bearer(admin)
    self_reset = await ctx.client.put(f"/api/v1/auth/users/{admin.id}/login-password", headers=old_admin, json={"password": NEW_PASSWORD})
    assert self_reset.status_code == 200
    assert self_reset.json()["access_token"]
    assert decode_token(self_reset.json()["access_token"]).auth_method == "password"
    assert (await ctx.client.get("/api/v1/auth/me", headers=old_admin)).status_code == 401
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": "Bearer " + self_reset.json()["access_token"]})).status_code == 200


@pytest.mark.asyncio
@pytest.mark.parametrize("legacy_password", ["old-code", "$dfv-legacy-plaintext"])
async def test_historical_plaintext_passphrase_wins_and_changes_without_old_hash(ctx, legacy_password):
    user = User(email="legacy@cm.com", password_hash=hash_password("different-old-password"), login_password=legacy_password)
    await ctx.repository.create_user(user)
    login = await ctx.client.post("/api/v1/auth/login/local", data={"username": "legacy", "password": legacy_password})
    assert login.status_code == 200
    assert (await ctx.client.post("/api/v1/auth/login/local", data={"username": "legacy", "password": "different-old-password"})).status_code == 401
    change = await ctx.client.post("/api/v1/auth/change-password", headers=bearer(user), json={"current_password": legacy_password, "new_password": NEW_PASSWORD})
    assert change.status_code == 200
    saved = await ctx.provider.get_user(str(user.id))
    assert saved.login_password == saved.password_hash
    assert verify_password(NEW_PASSWORD, saved.login_password)


@pytest.mark.asyncio
async def test_parallel_token_endpoint_returns_non_jwt_compat_value_when_login_is_rejected(ctx):
    await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    endpoint = "/api/parallel-agents/auth/token"
    fallback = await ctx.client.post(endpoint, json={"username": "alice"})
    assert fallback.status_code == 200
    assert fallback.json()["access_token"] == "force-situation-detail-compat-alice"
    assert (await ctx.client.post(endpoint, params={"password": PASSWORD}, json={"username": "alice"})).status_code == 200


@pytest.mark.asyncio
async def test_parallel_token_endpoint_returns_compat_value_for_any_rejected_detail_preflight(ctx):
    """Every old detail client passes its preflight without a real login session."""
    from app.gateway.auth import TokenError, decode_token

    endpoint = "/api/parallel-agents/auth/token"
    for username in ("fs-collect", "fs-detail-read", "fs-detail-submit", "fs-detail-update", "task-detail-agent"):
        response = await ctx.client.post(endpoint, json={"username": username})
        assert response.status_code == 200
        payload = response.json()
        assert payload["access_token"] == f"force-situation-detail-compat-{username}"
        assert payload["user_id"] == f"external:{username}"
        # The compatibility value cannot authenticate unrelated APIs.
        assert isinstance(decode_token(payload["access_token"]), TokenError)


@pytest.mark.asyncio
async def test_approval_gate_applies_to_password_login(ctx):
    await ctx.provider.create_user("blocked@cm.com", password=PASSWORD, approved=False)
    current = settings.load_system_settings()
    settings.save_system_settings(current.model_copy(update={"login_whitelist": settings.LoginWhitelistSettings(enabled=True)}))
    response = await ctx.client.post("/api/v1/auth/login/local", data={"username": "blocked", "password": PASSWORD})
    assert response.status_code == 403
    assert response.json()["detail"]["code"] == "PENDING_APPROVAL"


@pytest.mark.asyncio
async def test_stale_profile_and_concurrent_password_updates_cannot_resurrect_credentials(ctx):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    stale = user.model_copy(deep=True)
    await ctx.repository.update_credentials(user, hash_password(NEW_PASSWORD))
    stale.approved = False
    with pytest.raises(CredentialConflictError):
        await ctx.repository.update_user(stale)
    with pytest.raises(CredentialConflictError):
        await ctx.repository.update_credentials(stale, hash_password("Concurrent-s3cret!"))
    saved = await ctx.provider.get_user(str(user.id))
    assert saved.token_version == 1
    assert verify_password(NEW_PASSWORD, saved.login_password)


@pytest.mark.asyncio
async def test_initialize_first_admin_is_atomic_and_returns_login_credentials(ctx):
    results = await asyncio.gather(*[ctx.client.post("/api/v1/auth/initialize", json={"email": f"initial{index}@cm.com", "password": PASSWORD}) for index in range(2)])
    assert sorted(response.status_code for response in results) == [201, 409]
    assert await ctx.provider.count_admin_users() == 1
    success = next(response for response in results if response.status_code == 201)
    assert success.json()["access_token"]
    assert decode_token(success.json()["access_token"]).auth_method == "password"
    assert success.json()["system_role"] == "admin"
    options = await ctx.client.get("/api/v1/auth/options")
    assert options.json()["needs_setup"] is False


@pytest.mark.asyncio
async def test_sql_credential_rotation_updates_both_columns_without_touching_approval(ctx):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    async with ctx.sf() as session:
        await session.execute(update(UserRow).where(UserRow.id == str(user.id)).values(approved=False))
        await session.commit()
    saved = await ctx.provider.update_credentials(user, hash_password(NEW_PASSWORD))
    persisted = await ctx.provider.get_user(str(saved.id))
    assert persisted.approved is False
    assert persisted.password_hash == persisted.login_password
    assert persisted.token_version == 1


@pytest.mark.asyncio
async def test_concurrent_registration_email_domains_cannot_claim_same_username(ctx):
    enable_registration()
    results = await asyncio.gather(*[ctx.client.post("/api/v1/auth/register", json={"email": f"alice@domain{index}.com", "password": PASSWORD}) for index in range(2)])
    assert sorted(response.status_code for response in results) == [201, 409]
    assert await ctx.provider.count_users() == 1


@pytest.mark.asyncio
async def test_username_login_accepts_form_body(ctx):
    response = await ctx.client.post(
        "/api/v1/auth/login/username",
        data={"username": "firefox-legacy"},
        params={"password": "123ewq"},
    )
    assert response.status_code == 200, response.text
    assert response.json()["username"] == "firefox-legacy"
    blank = await ctx.client.post("/api/v1/auth/login/username", data={"username": ""})
    assert blank.status_code == 422


@pytest.mark.asyncio
async def test_username_sso_rechecks_password_if_registration_won_a_race(ctx, monkeypatch):
    await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    monkeypatch.setattr(auth, "_lookup_user_by_username", AsyncMock(return_value=None))
    response = await ctx.client.post("/api/v1/auth/login/username", json={"username": "alice"})
    assert response.status_code == 401


@pytest.mark.asyncio
@pytest.mark.parametrize("configured,expected_password", [("", "123ewq"), ("Original-shared-code", "Original-shared-code")])
async def test_accounts_without_personal_password_use_existing_passphrase(ctx, configured, expected_password):
    ctx.login_config.username_login.password = configured
    user = await ctx.provider.create_user("legacy@cm.com")
    assert not ctx.login_config.username_login.require_password
    assert await ctx.provider.authenticate({"email": user.email, "password": "wrong"}) is None
    assert (await ctx.provider.authenticate({"email": user.email, "password": expected_password})).id == user.id
    for path, kwargs in [
        ("/api/v1/auth/login/local", {"data": {"username": "legacy", "password": expected_password}}),
        ("/api/v1/auth/login/username", {"json": {"username": "legacy"}, "params": {"password": expected_password}}),
        ("/api/parallel-agents/auth/token", {"json": {"username": "legacy"}, "params": {"password": expected_password}}),
    ]:
        response = await ctx.client.post(path, **kwargs)
        assert response.status_code == 200, response.text
        assert decode_token(response.json()["access_token"]).auth_method in {"password", "passphrase"}
    me = await ctx.client.get("/api/v1/auth/me", headers=bearer(user))
    assert me.json()["has_password"] is False
    assert me.json()["uses_default_password"] is True
    assert me.json()["requires_current_password"] is True
    if configured:
        denied = await ctx.client.post("/api/v1/auth/login/local", data={"username": "legacy", "password": "123ewq"})
        assert denied.status_code == 401


@pytest.mark.asyncio
async def test_first_personal_password_requires_current_shared_passphrase(ctx):
    user = await ctx.provider.create_user("legacy@cm.com")
    old_headers = bearer(user)
    for wrong in ("", "incorrect"):
        response = await ctx.client.post("/api/v1/auth/change-password", headers=old_headers, json={"current_password": wrong, "new_password": NEW_PASSWORD})
        assert response.status_code == 400
    changed = await ctx.client.post("/api/v1/auth/change-password", headers=old_headers, json={"current_password": "123ewq", "new_password": NEW_PASSWORD})
    assert changed.status_code == 200
    for path, kwargs in [
        ("/api/v1/auth/login/local", {"data": {"username": "legacy", "password": "123ewq"}}),
        ("/api/v1/auth/login/username", {"json": {"username": "legacy"}, "params": {"password": "123ewq"}}),
    ]:
        assert (await ctx.client.post(path, **kwargs)).status_code == 401
    updated = await ctx.provider.get_user(str(user.id))
    assert await ctx.provider.authenticate({"email": user.email, "password": "123ewq"}) is None
    me = await ctx.client.get("/api/v1/auth/me", headers=bearer(updated))
    assert me.json()["has_password"] is True
    assert me.json()["uses_default_password"] is False


@pytest.mark.asyncio
async def test_enabled_password_login_blocks_bare_username_and_parallel_logins(ctx):
    await ctx.provider.create_user("legacy@cm.com")
    for username in ("legacy", "not-created"):
        assert (await ctx.client.post("/api/v1/auth/login/username", json={"username": username})).status_code == 401
        compat = await ctx.client.post("/api/parallel-agents/auth/token", json={"username": username})
        assert compat.status_code == 200
        assert compat.json()["access_token"] == f"force-situation-detail-compat-{username}"
    assert await ctx.provider.count_users() == 1


def langgraph_request(token):
    return SimpleNamespace(method="GET", headers={"authorization": f"Bearer {token}"}, cookies={})


@pytest.mark.asyncio
async def test_gateway_and_langgraph_reject_legacy_session_after_login_enabled(ctx):
    user = await ctx.provider.create_user("legacy@cm.com")
    legacy = create_access_token(str(user.id), token_version=user.token_version)
    assert decode_token(legacy).auth_method is None
    for cookie_only in (False, True):
        ctx.client.cookies.clear()
        if cookie_only:
            ctx.client.cookies.set("access_token", legacy)
        headers = {} if cookie_only else {"Authorization": f"Bearer {legacy}"}
        assert (await ctx.client.get("/api/v1/auth/me", headers=headers)).status_code == 401
    with pytest.raises(Exception) as rejected:
        await langgraph_auth.authenticate(langgraph_request(legacy))
    assert rejected.value.status_code == 401
    login = await ctx.client.post("/api/v1/auth/login/local", data={"username": "legacy", "password": "123ewq"})
    assert login.status_code == 200
    fresh = login.json()["access_token"]
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {fresh}"})).status_code == 200
    assert await langgraph_auth.authenticate(langgraph_request(fresh)) == str(user.id)


@pytest.mark.asyncio
async def test_passwordless_compatibility_token_cannot_survive_enabling_password_login(ctx):
    settings.save_system_settings(settings.SystemSettings())
    response = await ctx.client.post("/api/v1/auth/login/username", json={"username": "legacy"})
    assert response.status_code == 200
    token = response.json()["access_token"]
    assert decode_token(token).auth_method is None
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})).status_code == 200
    assert await langgraph_auth.authenticate(langgraph_request(token)) == response.json()["user_id"]
    settings.save_system_settings(settings.SystemSettings(authentication=settings.AuthenticationSettings(password_login_enabled=True)))
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})).status_code == 401
    with pytest.raises(Exception) as rejected:
        await langgraph_auth.authenticate(langgraph_request(token))
    assert rejected.value.status_code == 401


@pytest.mark.asyncio
async def test_verified_upstream_sso_remains_independently_controlled(ctx):
    ctx.login_config.token_login.enabled = True
    ctx.login_config.token_login.mock_tokens = {"trusted-upstream-token": "sso-user"}
    response = await ctx.client.post("/api/v1/auth/login/token", params={"token": "trusted-upstream-token"})
    assert response.status_code == 200
    token = response.json()["access_token"]
    assert decode_token(token).auth_method == "sso"
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})).status_code == 200
    assert await langgraph_auth.authenticate(langgraph_request(token)) == response.json()["user_id"]
    settings.save_system_settings(settings.SystemSettings(authentication=settings.AuthenticationSettings(password_login_enabled=True, sso_enabled=False)))
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})).status_code == 401
    with pytest.raises(Exception) as rejected:
        await langgraph_auth.authenticate(langgraph_request(token))
    assert rejected.value.status_code == 401


@pytest.mark.asyncio
@pytest.mark.parametrize("policy_field", ["password_login_enabled", "single_session_enabled"])
async def test_development_bypass_does_not_override_enabled_login(ctx, monkeypatch, policy_field):
    settings.save_system_settings(settings.SystemSettings(authentication=settings.AuthenticationSettings(**{policy_field: True})))
    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "true")
    assert is_auth_disabled() is False
    assert (await ctx.client.get("/api/v1/auth/me")).status_code == 401
    with pytest.raises(Exception) as rejected:
        await langgraph_auth.authenticate(langgraph_request(""))
    assert rejected.value.status_code == 401
    settings.save_system_settings(settings.SystemSettings())
    assert is_auth_disabled() is True


def set_single_session(enabled):
    current = settings.load_system_settings()
    policy = current.authentication.model_copy(update={"single_session_enabled": enabled})
    settings.save_system_settings(current.model_copy(update={"authentication": policy}))


@pytest.mark.asyncio
@pytest.mark.parametrize("enabled", [False, True])
@pytest.mark.parametrize("method", ["local", "username", "token", "parallel"])
async def test_login_single_session_policy_across_all_entry_points(ctx, enabled, method):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    set_single_session(enabled)
    ctx.login_config.token_login.enabled = True
    ctx.login_config.token_login.mock_tokens = {"upstream-alice": "alice"}

    async def login():
        if method == "local":
            return await ctx.client.post("/api/v1/auth/login/local", data={"username": "alice", "password": PASSWORD})
        if method == "token":
            return await ctx.client.post("/api/v1/auth/login/token", params={"token": "upstream-alice"})
        endpoint = "/api/parallel-agents/auth/token" if method == "parallel" else "/api/v1/auth/login/username"
        return await ctx.client.post(endpoint, json={"username": "alice"}, params={"password": PASSWORD})

    first, second = await login(), await login()
    assert first.status_code == second.status_code == 200
    first_token, second_token = first.json()["access_token"], second.json()["access_token"]
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {first_token}"})).status_code == (401 if enabled else 200)
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {second_token}"})).status_code == 200
    if enabled:
        assert decode_token(second_token).ver == decode_token(first_token).ver + 1
        with pytest.raises(Exception) as rejected:
            await langgraph_auth.authenticate(langgraph_request(first_token))
        assert rejected.value.status_code == 401
    else:
        assert decode_token(first_token).ver == decode_token(second_token).ver == 0
        assert await langgraph_auth.authenticate(langgraph_request(first_token)) == str(user.id)
    assert await langgraph_auth.authenticate(langgraph_request(second_token)) == str(user.id)


@pytest.mark.asyncio
async def test_failed_login_does_not_kick_out_existing_session(ctx):
    await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    set_single_session(True)
    first = await ctx.client.post("/api/v1/auth/login/local", data={"username": "alice", "password": PASSWORD})
    token = first.json()["access_token"]
    failed = await ctx.client.post("/api/v1/auth/login/local", data={"username": "alice", "password": "incorrect"})
    assert failed.status_code == 401
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"})).status_code == 200


@pytest.mark.asyncio
async def test_concurrent_logins_have_exactly_one_surviving_session(ctx):
    await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    set_single_session(True)
    responses = await asyncio.gather(*[ctx.client.post("/api/v1/auth/login/local", data={"username": "alice", "password": PASSWORD}) for _ in range(2)])
    assert all(response.status_code == 200 for response in responses)
    tokens = [response.json()["access_token"] for response in responses]
    assert sorted(decode_token(token).ver for token in tokens) == [1, 2]
    checks = [await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {token}"}) for token in tokens]
    assert sorted(response.status_code for response in checks) == [200, 401]


@pytest.mark.asyncio
async def test_single_session_cannot_reissue_after_concurrent_password_reset(ctx):
    user = await ctx.provider.create_user("alice@cm.com", password=PASSWORD)
    stale = user.model_copy(deep=True)
    await ctx.repository.update_credentials(user, hash_password(NEW_PASSWORD))
    with pytest.raises(CredentialConflictError):
        await ctx.repository.start_login_session(stale)
    persisted = await ctx.provider.get_user(str(user.id))
    assert persisted.token_version == 1
    assert verify_password(NEW_PASSWORD, persisted.password_hash)


@pytest.mark.asyncio
@pytest.mark.parametrize("method", ["register", "initialize"])
async def test_first_account_login_obeys_single_session_setting(ctx, method):
    enable_registration()
    set_single_session(True)
    body = {"password": PASSWORD, "username" if method == "register" else "email": "first" if method == "register" else "first@cm.com"}
    first = await ctx.client.post(f"/api/v1/auth/{method}", json=body)
    assert first.status_code == 201
    old_token = first.json()["access_token"]
    assert decode_token(old_token).ver == 1
    second = await ctx.client.post("/api/v1/auth/login/local", data={"username": "first", "password": PASSWORD})
    assert second.status_code == 200
    assert decode_token(second.json()["access_token"]).ver == 2
    assert (await ctx.client.get("/api/v1/auth/me", headers={"Authorization": f"Bearer {old_token}"})).status_code == 401


@pytest.mark.asyncio
async def test_whitelist_is_independent_of_statistics_stores(ctx, monkeypatch):
    admin = await make_admin(ctx)
    def forbidden(*args, **kwargs):
        raise AssertionError("Whitelist must not resolve statistics stores")
    for name in ("get_skill_store", "get_agent_store", "get_run_store"):
        monkeypatch.setattr(auth, name, forbidden)
    response = await ctx.client.get("/api/v1/auth/users", headers=bearer(admin), params={"limit": 20, "include_counts": "false"})
    assert response.status_code == 200, response.text
    assert response.json()["total"] == 1
    assert response.json()["items"][0]["login_password"] is None


@pytest.mark.asyncio
@pytest.mark.parametrize("method", ["list_users_page", "count_users_filtered"])
async def test_whitelist_timeout_cancels_account_queries(ctx, monkeypatch, method):
    admin = await make_admin(ctx)
    cancelled = asyncio.Event()
    async def stalled(*args, **kwargs):
        try:
            await asyncio.Event().wait()
        finally:
            cancelled.set()
    monkeypatch.setattr(ctx.provider, method, stalled)
    monkeypatch.setattr(auth, "_USER_LIST_TIMEOUT_SECONDS", 0.05)
    response = await asyncio.wait_for(ctx.client.get("/api/v1/auth/users", headers=bearer(admin), params={"limit": 20, "include_counts": "false"}), 5)
    assert response.status_code == 503
    assert "超时" in response.json()["detail"]
    assert cancelled.is_set()


@pytest.mark.asyncio
async def test_whitelist_paged_is_admin_only(ctx):
    user = await ctx.provider.create_user("normal-whitelist@cm.com", password=PASSWORD)
    params = {"limit": 20, "include_counts": "false"}
    assert (await ctx.client.get("/api/v1/auth/users", headers=bearer(user), params=params)).status_code == 403
    assert (await ctx.client.get("/api/v1/auth/users", params=params)).status_code == 401


@pytest.mark.asyncio
async def test_whitelist_loads_when_resource_stores_unavailable(ctx):
    admin = await make_admin(ctx)
    del ctx.app.state.skill_store
    del ctx.app.state.agent_store
    response = await ctx.client.get("/api/v1/auth/users", headers=bearer(admin), params={"limit": 20, "include_counts": "false"})
    assert response.status_code == 200
    assert response.json()["total"] == 1
    # The full statistics endpoint still requires the resource stores.
    assert (await ctx.client.get("/api/v1/auth/users", headers=bearer(admin))).status_code == 503