gendesign/tradein-mvp/backend/tests/test_auth_api.py

"""Integration tests for #2552 auth-core: POST /login, /logout, dual-mode GET /me,
and rbac_guard session-cookie resolution.

Uses the REAL `rbac_guard` (app.core.rbac) + REAL `auth.router` / `me.router` wired
into an isolated FastAPI test app (same pattern as tests/test_rbac.py), with an
in-memory fake DB standing in for the identity registry:
  - сессия РЕЕСТРА подменяется на самом низком уровне — `identity_store.SessionLocal`
    и `auth_db.auth_session` (см. `tests.support.identity_modes.patch_identity_sessions`),
    так что и `identity_session()` (rbac_guard — middleware, FastAPI-DI там нет), и
    `Depends(get_identity_db)` (auth.py / me.py) выполняются РЕАЛЬНЫЕ, вместе со своим
    ветвлением по `settings.identity_store`;
  - `app.core.db.get_db` переопределён через `app.dependency_overrides` — это
    продуктовая БД (в дефолтном режиме она же и реестр).

Все они смотрят в ОДИН `_Store` на тест, поэтому сессия, созданная POST /login,
сразу видна собственному DB-раунд-трипу rbac_guard'а на следующем запросе.

⚠️ ДВА РЕЖИМА РЕЕСТРА И ЛОВУШКА FAKE-DB. `_FakeDB` диспатчит по ТЕКСТУ SQL, а
эпик «единый вход» переименовывает таблицы (`tradein_users`/`tradein_sessions` →
`users`/`sessions`) и меняет тип колонки состояния доступа. Литерал
«tradein_users» в диспатчере означал бы, что при `IDENTITY_STORE=auth` ветка
молча перестаёт матчиться, fake отдаёт пустоту, а тест остаётся ЗЕЛЁНЫМ на
сломанном коде. Поэтому имена берутся из `sql_names()` (= `identity_schema()`,
тот же словарь, что у продакшн-кода), а непонятый SQL падает `AssertionError`,
а не возвращает пустой результат.

Дефолт (`identity_store="tradein"`) — сегодняшний прод; тесты без фикстуры
`auth_store` идут именно в нём. Тесты про режим `auth` (в т.ч. про состояние
`trial_expired`, невыразимое булевым `is_active`) — в конце файла.
"""

from __future__ import annotations

import os
import re
import time
from datetime import UTC, datetime, timedelta
from types import SimpleNamespace
from typing import Annotated, Any

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest
from fastapi import FastAPI, Header
from fastapi.testclient import TestClient

from app.api.v1 import auth as auth_router
from app.api.v1 import me as me_router
from app.core import auth as auth_mod
from app.core import auth_db, config
from app.core.db import get_db
from app.core.password import hash_password
from app.core.rbac import rbac_guard
from app.services.identity_store import AccessState
from tests.support.identity_modes import (
    assert_reads_access_state,
    column_value,
    patch_identity_sessions,
    sql_names,
    use_identity_mode,
)

# ---------------------------------------------------------------------------
# Fake DB backing the identity registry (users/sessions таблицы текущего режима)
# ---------------------------------------------------------------------------


class _Store:
    def __init__(self) -> None:
        self.users: dict[str, dict[str, Any]] = {}
        self.sessions: dict[str, dict[str, Any]] = {}
        self.sql_log: list[str] = []  # весь SQL, доехавший до «БД» — см. тесты режимов
        self._next_id = 1

    def add_user(
        self,
        username: str,
        password_hash: str | None,
        *,
        role: str = "employee",
        access_state: AccessState = AccessState.ACTIVE,
        display_name: str | None = "Alice A.",
        org_name: str | None = "Org LLC",
        email: str | None = "alice@example.com",
    ) -> int:
        uid = self._next_id
        self._next_id += 1
        self.users[username] = {
            "id": uid,
            "username": username,
            "password_hash": password_hash,
            "role": role,
            # СЫРОЕ значение колонки текущего режима (boolean либо text) — ровно
            # то, что вернул бы драйвер; в AccessState его превращает код.
            "access_state": column_value(access_state),
            "display_name": display_name,
            "org_name": org_name,
            "email": email,
        }
        return uid

    def set_access_state(self, username: str, state: AccessState) -> None:
        """Меняет состояние доступа уже заведённого юзера (как сделал бы админ/миграция)."""
        self.users[username]["access_state"] = column_value(state)

    def user_by_id(self, uid: int) -> dict[str, Any] | None:
        for u in self.users.values():
            if u["id"] == uid:
                return u
        return None

    def add_expired_session(self, token: str, user_id: int) -> None:
        now = datetime.now(UTC)
        self.sessions[token] = {
            "user_id": user_id,
            "expires_at": now - timedelta(minutes=1),
            "last_seen_at": now - timedelta(minutes=1),
        }


class _FakeDB:
    """Minimal Session stand-in dispatching on SQL text — see module docstring."""

    def __init__(self, store: _Store) -> None:
        self.store = store

    def __enter__(self) -> _FakeDB:
        return self

    def __exit__(self, *exc: object) -> bool:
        return False

    def close(self) -> None:
        pass

    def commit(self) -> None:
        pass

    def rollback(self) -> None:
        pass

    def execute(self, stmt: object, params: dict[str, Any] | None = None) -> SimpleNamespace:
        sql = str(stmt)
        p = params or {}
        # Имена таблиц берутся ИЗ КОДА (identity_schema), а не из литералов —
        # см. «ЛОВУШКА FAKE-DB» в модульном docstring.
        names = sql_names()
        self.store.sql_log.append(sql)

        if f"INSERT INTO {names.sessions}" in sql:
            now = datetime.now(UTC)
            self.store.sessions[p["token"]] = {
                "user_id": p["user_id"],
                "expires_at": now + timedelta(hours=p["ttl_hours"]),
                "last_seen_at": now,
            }
            return SimpleNamespace(fetchone=lambda: None)

        if f"UPDATE {names.sessions}" in sql and "SET last_seen_at" in sql:
            sess = self.store.sessions.get(p["token"])
            if sess is not None:
                now = datetime.now(UTC)
                sess["last_seen_at"] = now
                sess["expires_at"] = now + timedelta(hours=p["ttl_hours"])
            return SimpleNamespace(fetchone=lambda: None)

        if f"DELETE FROM {names.sessions} WHERE token" in sql:
            self.store.sessions.pop(p["token"], None)
            return SimpleNamespace(fetchone=lambda: None)

        if f"DELETE FROM {names.sessions} WHERE user_id" in sql:
            uid = p["user_id"]
            for tok in [t for t, s in self.store.sessions.items() if s["user_id"] == uid]:
                del self.store.sessions[tok]
            return SimpleNamespace(fetchone=lambda: None)

        if f"FROM {names.sessions} s" in sql and f"JOIN {names.users} u" in sql:
            assert_reads_access_state(sql, names)
            sess = self.store.sessions.get(p["token"])
            if sess is None:
                return SimpleNamespace(fetchone=lambda: None)
            user = self.store.user_by_id(sess["user_id"])
            if user is None:
                return SimpleNamespace(fetchone=lambda: None)
            # Колонка состояния приезжает под алиасом `access_state` в ОБОИХ
            # режимах (`u.<колонка> AS access_state` в реальном SELECT'е);
            # значение — сырое, типа своей схемы.
            row = SimpleNamespace(
                user_id=sess["user_id"],
                expires_at=sess["expires_at"],
                last_seen_at=sess["last_seen_at"],
                username=user["username"],
                role=user["role"],
                display_name=user["display_name"],
                org_name=user["org_name"],
                email=user["email"],
                access_state=user["access_state"],
            )
            return SimpleNamespace(fetchone=lambda: row)

        if f"FROM {names.users}" in sql and "WHERE username = :username" in sql:
            assert_reads_access_state(sql, names)
            user = self.store.users.get(p["username"])
            if user is None:
                return SimpleNamespace(fetchone=lambda: None)
            row = SimpleNamespace(**user)
            return SimpleNamespace(fetchone=lambda: row)

        raise AssertionError(f"unhandled fake SQL in test_auth_api: {sql!r}")


# ---------------------------------------------------------------------------
# Test app
# ---------------------------------------------------------------------------


def _build_test_app(store: _Store) -> FastAPI:
    app = FastAPI()
    app.middleware("http")(rbac_guard)
    app.include_router(auth_router.router, prefix="/api/v1/auth", tags=["auth"])
    app.include_router(me_router.router, prefix="/api/v1", tags=["me"])

    @app.get("/api/v1/trade-in/dummy")
    async def tradein_dummy() -> dict:
        return {"ok": True}

    @app.get("/api/v1/trade-in/whoami")
    async def tradein_whoami(
        x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
    ) -> dict:
        """Echoes the X-Authenticated-User header exactly as a downstream handler
        (`_assert_estimate_access*`, `account_quota`, etc.) would see it — used to
        assert session-identity wins over a client-forged header (#2552 spoof fix)."""
        return {"user": x_authenticated_user}

    # Внутренние инструменты, закрытые от клиентских DB-ролей 2026-07-31
    # («Доля в продаже» / «Кэш»). Ручки настоящие (не заглушки rbac_guard'а),
    # чтобы 403 приходил именно от scope-чека, а не от отсутствия роута.
    @app.get("/api/v1/buildings/sale-share")
    async def buildings_sale_share() -> dict:
        return {"ok": True}

    @app.get("/api/v1/trade-in/cache-stats")
    async def tradein_cache_stats() -> dict:
        return {"ok": True}

    def _override_get_db():  # generator dependency — matches app.core.db.get_db shape
        yield _FakeDB(store)

    app.dependency_overrides[get_db] = _override_get_db
    return app


@pytest.fixture(autouse=True)
def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None:
    auth_mod.reset_cache_for_tests()
    auth_router._LOGIN_LIMITER._hits.clear()
    auth_router._USERNAME_FAIL_LIMITER._hits.clear()
    monkeypatch.setattr(config.settings, "auth_mode", "dual")
    # Каждый тест стартует в ДЕФОЛТНОМ режиме реестра (сегодняшний прод), даже
    # если предыдущий переключался на `auth`.
    use_identity_mode(monkeypatch, "tradein")


@pytest.fixture
def store() -> _Store:
    return _Store()


@pytest.fixture
def auth_store(store: _Store, monkeypatch: pytest.MonkeyPatch) -> _Store:
    """Тот же `store`, но реестр — БД `auth` (`users`/`sessions`, text-состояние).

    Запрашивай ПЕРЕД `client` в списке аргументов теста: `client` строится уже с
    учётом режима (`_build_test_app` читает его лениво, но `store.add_user`
    сохраняет значение колонки по режиму НА МОМЕНТ ВЫЗОВА).
    """
    use_identity_mode(monkeypatch, "auth")
    return store


@pytest.fixture
def client(store: _Store, monkeypatch: pytest.MonkeyPatch) -> TestClient:
    # Подменяем сессию РЕЕСТРА на обоих её источниках сразу, а не ветвление по
    # режиму: `identity_session()` / `get_identity_db()` остаются настоящими.
    patch_identity_sessions(monkeypatch, lambda: _FakeDB(store))
    # base_url=https:// — login sets the session cookie with Secure=True (real prod
    # behaviour, not weakened for tests); httpx's cookie jar silently drops Secure
    # cookies on a plain-http connection, so a plain http://testserver client would
    # never resend the cookie on subsequent requests within the same test.
    return TestClient(_build_test_app(store), base_url="https://testserver")


# ---------------------------------------------------------------------------
# POST /login
# ---------------------------------------------------------------------------


def test_login_happy_path_sets_cookie(client: TestClient, store: _Store) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")

    resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    assert resp.status_code == 200, resp.text
    assert resp.json() == {"ok": True}
    cookie_name = config.settings.session_cookie_name
    assert cookie_name in resp.cookies
    assert resp.cookies[cookie_name]
    # Сессия реально создана в сторе под этим токеном.
    assert resp.cookies[cookie_name] in store.sessions


def test_login_wrong_password_401_and_logs_failed_event(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    events: list[dict[str, Any]] = []
    monkeypatch.setattr(auth_router, "schedule_event", lambda **kw: events.append(kw))

    resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "wrong"})

    assert resp.status_code == 401
    assert "detail" in resp.json()
    assert config.settings.session_cookie_name not in resp.cookies
    assert any(e["event_type"] == "login_failed" for e in events)
    failed = next(e for e in events if e["event_type"] == "login_failed")
    assert failed["username"] == "alice"
    # Raw-пароль никогда не попадает в событие.
    assert "wrong" not in str(failed)


def test_login_unknown_username_401_generic_message(client: TestClient) -> None:
    resp = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "whatever"})
    assert resp.status_code == 401
    # НЕ раскрываем, что юзера не существует — то же сообщение, что и wrong-password.
    body_ghost = resp.json()["detail"]

    resp2 = client.post("/api/v1/auth/login", json={"username": "ghost2", "password": "x"})
    assert resp2.json()["detail"] == body_ghost


def test_login_inactive_user_401(client: TestClient, store: _Store) -> None:
    store.add_user(
        "bob", hash_password("Secret123!"), role="employee", access_state=AccessState.DISABLED
    )
    resp = client.post("/api/v1/auth/login", json={"username": "bob", "password": "Secret123!"})
    assert resp.status_code == 401


def test_login_null_password_hash_401(client: TestClient, store: _Store) -> None:
    store.add_user("carol", None, role="employee")
    resp = client.post("/api/v1/auth/login", json={"username": "carol", "password": "anything"})
    assert resp.status_code == 401


def test_login_always_calls_verify_password_timing_oracle_guard(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    """#2552 post-review Medium 2: `verify_password` должен выполняться ровно
    один раз на КАЖДУЮ попытку логина — включая неизвестный username и NULL
    password_hash — иначе короткое замыкание даёт наблюдаемый timing-oracle
    для user-enumeration. Тест не измеряет тайминг (флейки в CI), а проверяет
    сам факт + аргумент вызова через monkeypatch-счётчик."""
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    store.add_user("nullhash", None, role="employee")

    calls: list[str] = []
    real_verify = auth_router.verify_password

    def _counting_verify(plain: str, hashed: str) -> bool:
        calls.append(hashed)
        return real_verify(plain, hashed)

    monkeypatch.setattr(auth_router, "verify_password", _counting_verify)

    resp_unknown = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "x"})
    assert resp_unknown.status_code == 401

    resp_null_hash = client.post(
        "/api/v1/auth/login", json={"username": "nullhash", "password": "x"}
    )
    assert resp_null_hash.status_code == 401

    resp_wrong_pw = client.post(
        "/api/v1/auth/login", json={"username": "alice", "password": "wrong"}
    )
    assert resp_wrong_pw.status_code == 401

    assert len(calls) == 3
    # Unknown user / NULL hash — сверяется против dummy-хеша, не против NULL.
    assert calls[0] == auth_router._DUMMY_PASSWORD_HASH
    assert calls[1] == auth_router._DUMMY_PASSWORD_HASH
    # Реальный юзер с реальным hash — НЕ dummy.
    assert calls[2] != auth_router._DUMMY_PASSWORD_HASH


def test_login_rate_limit_429(client: TestClient, store: _Store) -> None:
    store.add_user("dave", hash_password("Secret123!"), role="employee")
    limit = config.settings.login_rate_limit

    for _ in range(limit):
        resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
        assert resp.status_code == 401

    resp = client.post("/api/v1/auth/login", json={"username": "dave", "password": "wrong"})
    assert resp.status_code == 429
    assert "Retry-After" in resp.headers


# ---------------------------------------------------------------------------
# #2571 — глобальный потолок попыток НА ИМЯ (распределённый перебор)
# ---------------------------------------------------------------------------


def _throttle_settings(
    monkeypatch: pytest.MonkeyPatch, *, threshold: int, max_delay_s: float
) -> None:
    """Порог и потолок задержки под тест: реальные 20/час и 8с ждать некогда."""
    monkeypatch.setattr(config.settings, "login_username_fail_threshold", threshold)
    monkeypatch.setattr(config.settings, "login_username_throttle_max_delay_s", max_delay_s)


def _capture_events(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
    events: list[dict[str, Any]] = []
    monkeypatch.setattr(auth_router, "schedule_event", lambda **kw: events.append(kw))
    return events


def _failed_login_delays(events: list[dict[str, Any]]) -> list[float]:
    return [e["payload"]["throttle_delay_s"] for e in events if e["event_type"] == "login_failed"]


def test_throttle_delay_grows_and_caps(monkeypatch: pytest.MonkeyPatch) -> None:
    """Задержка = 0 до порога, дальше удваивается от 1с и упирается в потолок.

    Потолок обязателен: без него растущая задержка вырождается в ту самую
    блокировку учётки, которой #2571 сознательно избегает (DoS против человека).
    """
    _throttle_settings(monkeypatch, threshold=3, max_delay_s=4.0)

    assert auth_router._throttle_delay_s(1) == 0.0
    assert auth_router._throttle_delay_s(3) == 0.0  # ровно порог — ещё не перебор
    assert auth_router._throttle_delay_s(4) == 1.0
    assert auth_router._throttle_delay_s(5) == 2.0
    assert auth_router._throttle_delay_s(6) == 4.0
    assert auth_router._throttle_delay_s(7) == 4.0  # потолок
    assert auth_router._throttle_delay_s(1000) == 4.0


def test_distributed_bruteforce_one_username_many_ips_hits_global_ceiling(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    """DoD 1: перебор одного имени с РАЗНЫХ адресов упирается в потолок на имя.

    Каждая попытка приходит со своего IP, поэтому per-(username, IP) лимит
    (`_LOGIN_LIMITER`, 5/300с) не срабатывает НИ РАЗУ — ровно та дыра, из-за
    которой issue заведена. Ловит её только глобальный счётчик на имя.
    """
    store.add_user("victim", hash_password("Secret123!"), role="employee")
    _throttle_settings(monkeypatch, threshold=3, max_delay_s=0.01)
    events = _capture_events(monkeypatch)

    for i in range(6):
        resp = client.post(
            "/api/v1/auth/login",
            json={"username": "victim", "password": f"guess{i}"},
            headers={"x-forwarded-for": f"10.0.0.{i}"},
        )
        # per-IP бюджет не исчерпан ни на одном адресе — никаких 429.
        assert resp.status_code == 401, resp.text

    delays = _failed_login_delays(events)
    assert len(delays) == 6
    # Первые три — в пределах порога, дальше замедление.
    assert delays[:3] == [0.0, 0.0, 0.0]
    assert all(d > 0 for d in delays[3:])
    # Счётчик глобальный: считает попытки со ВСЕХ адресов, а не по каждому свой.
    fails = [e["payload"]["username_fails_in_window"] for e in events]
    assert fails == [1, 2, 3, 4, 5, 6]


def test_throttle_actually_delays_the_response(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Задержка именно ЖДЁТСЯ, а не только пишется в аудит.

    Отдельный тест, потому что все остальные читают величину задержки из
    события: без этой проверки реализация, которая посчитала задержку и забыла
    `await`, оставила бы весь набор зелёным.
    """
    store.add_user("slowpoke", hash_password("Secret123!"), role="employee")
    # threshold=0 → тормозим уже первую неудачу; потолок 1с — цена одного теста.
    _throttle_settings(monkeypatch, threshold=0, max_delay_s=1.0)

    started = time.monotonic()
    resp = client.post("/api/v1/auth/login", json={"username": "slowpoke", "password": "wrong"})
    elapsed = time.monotonic() - started

    assert resp.status_code == 401
    assert elapsed >= 1.0


def test_typo_does_not_throttle_and_correct_password_still_works(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    """DoD 2: живой человек с опечаткой не наказан.

    Порог штатный (20/час), три опечатки до него не дотягивают — задержки нет,
    а верный пароль пускает сразу же. Замедление применяется ТОЛЬКО к неудаче,
    поэтому даже перебранный порог не мешает владельцу пароля войти.
    """
    store.add_user("erin", hash_password("Secret123!"), role="employee")
    events = _capture_events(monkeypatch)

    for _ in range(3):
        resp = client.post("/api/v1/auth/login", json={"username": "erin", "password": "Secret123"})
        assert resp.status_code == 401

    assert _failed_login_delays(events) == [0.0, 0.0, 0.0]

    ok = client.post("/api/v1/auth/login", json={"username": "erin", "password": "Secret123!"})
    assert ok.status_code == 200, ok.text
    assert config.settings.session_cookie_name in ok.cookies


def test_throttle_identical_for_existing_and_unknown_username(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Замедление не должно стать оракулом существования учётки.

    Если тормозить только существующие имена, злоумышленник перечисляет живые
    логины по времени ответа, не зная ни одного пароля — та же user-enumeration,
    от которой защищают одинаковый 401 и безусловный bcrypt (`_DUMMY_PASSWORD_HASH`),
    только по другому каналу. Счётчик ведётся по присланному имени БЕЗ проверки
    в реестре, поэтому обе последовательности обязаны совпасть.
    """
    store.add_user("real", hash_password("Secret123!"), role="employee")
    _throttle_settings(monkeypatch, threshold=2, max_delay_s=0.01)
    events = _capture_events(monkeypatch)

    real_responses = []
    for i in range(4):
        real_responses.append(
            client.post(
                "/api/v1/auth/login",
                json={"username": "real", "password": "wrong"},
                headers={"x-forwarded-for": f"10.1.0.{i}"},
            )
        )
    real_delays = _failed_login_delays(events)

    events.clear()
    ghost_responses = []
    for i in range(4):
        ghost_responses.append(
            client.post(
                "/api/v1/auth/login",
                json={"username": "ghost", "password": "wrong"},
                headers={"x-forwarded-for": f"10.2.0.{i}"},
            )
        )
    ghost_delays = _failed_login_delays(events)

    assert real_delays == ghost_delays
    assert real_delays[:2] == [0.0, 0.0] and all(d > 0 for d in real_delays[2:])
    # И «наблюдаемая» часть ответа тоже неотличима.
    assert [r.status_code for r in real_responses] == [r.status_code for r in ghost_responses]
    assert {r.json()["detail"] for r in real_responses} == {
        r.json()["detail"] for r in ghost_responses
    }


def test_disabled_account_throttled_like_wrong_password(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    """Отказ по «доступ закрыт» идёт тем же хвостом: счётчик, аудит, задержка.

    Иначе быстрый 401 у заблокированной учётки отличал бы её от несуществующей.
    """
    store.add_user(
        "blocked", hash_password("Secret123!"), role="employee", access_state=AccessState.DISABLED
    )
    _throttle_settings(monkeypatch, threshold=1, max_delay_s=0.01)
    events = _capture_events(monkeypatch)

    for i in range(2):
        # Пароль ВЕРНЫЙ — отказ приходит именно из ветки состояния доступа.
        resp = client.post(
            "/api/v1/auth/login",
            json={"username": "blocked", "password": "Secret123!"},
            headers={"x-forwarded-for": f"10.3.0.{i}"},
        )
        assert resp.status_code == 401

    assert _failed_login_delays(events) == [0.0, pytest.approx(0.01)]


def test_failed_login_events_reach_audit_with_counter_state(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    """DoD 3: неудачные входы видны в аудите — с ip/ua и состоянием счётчика.

    Именно счётчик отличает распределённый перебор от россыпи случайных
    одиночных неудач: в `user_events` иначе видно только «по одной неудаче с
    каждого адреса», что само по себе не тревожно.
    """
    store.add_user("audited", hash_password("Secret123!"), role="employee")
    events = _capture_events(monkeypatch)

    resp = client.post(
        "/api/v1/auth/login",
        json={"username": "audited", "password": "s3cret-typo"},
        # Левый хоп подделан клиентом, правый добавлен Caddy — в аудит обязан
        # попасть ПРАВЫЙ (`ratelimit._client_ip`, допущение «ровно один прокси»).
        # Иначе злоумышленник переписывал бы себе ip в журнале одним заголовком.
        headers={
            "x-forwarded-for": "198.51.100.9, 203.0.113.7",
            "user-agent": "probe/1.0",
        },
    )

    assert resp.status_code == 401
    (failed,) = (e for e in events if e["event_type"] == "login_failed")
    assert failed["username"] == "audited"
    assert failed["ip"] == "203.0.113.7"
    assert failed["user_agent"] == "probe/1.0"
    assert failed["path"] == "/api/v1/auth/login"
    assert failed["payload"]["username_fails_in_window"] == 1
    # Raw-пароль не утекает ни в поля события, ни в payload.
    assert "s3cret-typo" not in str(failed)


# ---------------------------------------------------------------------------
# POST /logout
# ---------------------------------------------------------------------------


def test_logout_revokes_session_and_clears_cookie(client: TestClient, store: _Store) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    login_resp = client.post(
        "/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"}
    )
    token = login_resp.cookies[config.settings.session_cookie_name]
    assert token in store.sessions

    logout_resp = client.post("/api/v1/auth/logout")
    assert logout_resp.status_code == 200
    assert logout_resp.json() == {"ok": True}
    assert token not in store.sessions


# ---------------------------------------------------------------------------
# GET /me — session-first + dual-mode legacy fallback
# ---------------------------------------------------------------------------


def test_me_with_session_cookie_returns_db_role(client: TestClient, store: _Store) -> None:
    store.add_user(
        "alice",
        hash_password("Secret123!"),
        role="employee",
        display_name="Алиса",
        org_name="ООО Ромашка",
        email="alice@romashka.ru",
    )
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    resp = client.get("/api/v1/me")
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["username"] == "alice"
    assert body["role"] == "employee"
    assert "/trade-in/**" in body["allowed_paths"]
    assert "/admin/**" in body["deny_paths"]
    # Пункты меню «Доля в продаже» / «Кэш» прячет Topbar, фильтруя NAV_ITEMS по
    # deny_paths ИЗ /me — т.е. видимость держится на ЭТОМ выводе, а не только на
    # DB_ROLE_PATHS. Сборка dict-а в app/api/v1/me.py может регрессировать
    # независимо от get_db_role_scope, поэтому пиним её здесь.
    assert "/trade-in/sale-share/**" in body["deny_paths"]
    assert "/trade-in/cache/**" in body["deny_paths"]
    assert body["display_name"] == "Алиса"
    assert body["org"] == "ООО Ромашка"
    assert body["email"] == "alice@romashka.ru"


def test_me_manager_role_gets_team_path(client: TestClient, store: _Store) -> None:
    store.add_user("mgr", hash_password("Secret123!"), role="manager")
    client.post("/api/v1/auth/login", json={"username": "mgr", "password": "Secret123!"})

    resp = client.get("/api/v1/me")
    assert resp.status_code == 200
    body = resp.json()
    assert "/api/v1/team/**" in body["allowed_paths"]
    # Тот же пин, что и для employee: manager (роль praktika) не должен получать
    # из /me deny-список без внутренних разделов — иначе пункты вернутся в топбар.
    assert "/trade-in/sale-share/**" in body["deny_paths"]
    assert "/trade-in/cache/**" in body["deny_paths"]


def test_me_without_cookie_dual_mode_legacy_still_works(client: TestClient) -> None:
    """Regression guard: без сессии, auth_mode=dual — legacy X-Authenticated-User
    путь через roles.yaml работает БЕЗ ИЗМЕНЕНИЙ."""
    resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"})
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["username"] == "admin"
    assert body["role"] == "admin"
    assert body["allowed_paths"] == ["/**"]


def test_me_no_cookie_no_header_401(client: TestClient) -> None:
    resp = client.get("/api/v1/me")
    assert resp.status_code == 401


def test_expired_session_falls_back_to_no_auth(client: TestClient, store: _Store) -> None:
    """Истёкшая сессия трактуется как отсутствие cookie: без legacy-заголовка — 401."""
    uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
    store.add_expired_session("expired-tok", uid)

    client.cookies.set(config.settings.session_cookie_name, "expired-tok")
    resp = client.get("/api/v1/trade-in/dummy")
    assert resp.status_code == 401


def test_expired_session_falls_back_to_legacy_header_in_dual_mode(
    client: TestClient, store: _Store
) -> None:
    """Истёкшая сессия + валидный legacy header в dual-mode — header отрабатывает."""
    uid = store.add_user("alice", hash_password("Secret123!"), role="employee")
    store.add_expired_session("expired-tok", uid)

    client.cookies.set(config.settings.session_cookie_name, "expired-tok")
    resp = client.get(
        "/api/v1/trade-in/dummy",
        headers={"X-Authenticated-User": "kopylov"},
    )
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


# ---------------------------------------------------------------------------
# rbac_guard dual vs db_only
# ---------------------------------------------------------------------------


def test_db_only_mode_rejects_legacy_header_without_session(
    client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setattr(config.settings, "auth_mode", "db_only")

    resp = client.get(
        "/api/v1/trade-in/dummy",
        headers={"X-Authenticated-User": "admin"},
    )
    assert resp.status_code == 401
    assert "session" in resp.json()["detail"].lower()


def test_db_only_mode_accepts_valid_session(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    monkeypatch.setattr(config.settings, "auth_mode", "db_only")

    resp = client.get("/api/v1/trade-in/dummy")
    assert resp.status_code == 200
    assert resp.json() == {"ok": True}


def test_session_user_can_reach_tradein_but_not_admin(client: TestClient, store: _Store) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    ok = client.get("/api/v1/trade-in/dummy")
    assert ok.status_code == 200

    denied = client.get("/api/v1/admin/dummy")
    # rbac_guard's admin-gate matches the path regex BEFORE routing even happens
    # (route isn't registered on this test app) — role=employee != admin -> 403,
    # never a 404 (a bare "any non-2xx" assertion would mask a rbac_guard typo).
    assert denied.status_code == 403


# ---------------------------------------------------------------------------
# 2026-07-31: внутренние разделы («Доля в продаже» / «Кэш») закрыты от клиентов —
# СКВОЗЬ РЕАЛЬНЫЙ rbac_guard по SESSION-ветке (from_session=True).
#
# Тесты в tests/test_rbac.py проверяют матчеры напрямую + guard по ЛЕГАСИ
# trusted-header ветке (is_path_allowed / roles.yaml). Но в проде клиентские
# аккаунты (praktika и прочие DB-юзеры) ходят именно session-cookie'ой, где
# scope считает ДРУГАЯ ветка — `_db_role_path_allowed(role, external_path)`.
# Без тестов ниже её можно было сломать, не уронив ни одного теста.
#
# Пути тут — ВНУТРЕННИЕ (Caddy срезает внешний /trade-in), rbac_guard
# восстанавливает внешний как '/trade-in' + path.
# ---------------------------------------------------------------------------

_INTERNAL_TOOL_API = ("/api/v1/buildings/sale-share", "/api/v1/trade-in/cache-stats")


def test_session_manager_denied_on_internal_tool_api(client: TestClient, store: _Store) -> None:
    store.add_user("mgr", hash_password("Secret123!"), role="manager")
    client.post("/api/v1/auth/login", json={"username": "mgr", "password": "Secret123!"})

    for path in _INTERNAL_TOOL_API:
        resp = client.get(path)
        assert resp.status_code == 403, f"manager {path}: {resp.status_code} {resp.text}"
        assert "forbidden for role" in resp.json()["detail"].lower()

    # ...и при этом основной продукт для той же сессии открыт (иначе тест выше
    # проходил бы и на «сломали scope целиком»).
    ok = client.get("/api/v1/trade-in/dummy")
    assert ok.status_code == 200, ok.text


def test_session_employee_denied_on_internal_tool_api(client: TestClient, store: _Store) -> None:
    store.add_user("emp", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "emp", "password": "Secret123!"})

    for path in _INTERNAL_TOOL_API:
        resp = client.get(path)
        assert resp.status_code == 403, f"employee {path}: {resp.status_code} {resp.text}"
        assert "forbidden for role" in resp.json()["detail"].lower()

    ok = client.get("/api/v1/trade-in/dummy")
    assert ok.status_code == 200, ok.text


def test_session_admin_keeps_internal_tool_api(client: TestClient, store: _Store) -> None:
    """Контрольная группа: DB-роль admin ('/**') разделы по-прежнему видит."""
    store.add_user("root", hash_password("Secret123!"), role="admin")
    client.post("/api/v1/auth/login", json={"username": "root", "password": "Secret123!"})

    for path in _INTERNAL_TOOL_API:
        resp = client.get(path)
        assert resp.status_code == 200, f"admin {path}: {resp.text}"


# ---------------------------------------------------------------------------
# #2552 post-review CRITICAL fix: session identity must win over a spoofed
# client-sent X-Authenticated-User header (was a skip-if-present bug — the
# forged header used to override the session for every downstream reader of
# the raw header: _assert_estimate_access*, account_quota, /trade-in/history,
# support.py — in BOTH auth_mode=dual and db_only).
# ---------------------------------------------------------------------------


def test_session_identity_wins_over_spoofed_header_dual_mode(
    client: TestClient, store: _Store
) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    store.add_user("victim", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    resp = client.get(
        "/api/v1/trade-in/whoami",
        headers={"X-Authenticated-User": "victim"},
    )
    assert resp.status_code == 200
    assert resp.json()["user"] == "alice"


def test_session_identity_wins_over_spoofed_header_db_only_mode(
    client: TestClient, store: _Store, monkeypatch: pytest.MonkeyPatch
) -> None:
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    store.add_user("victim", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    monkeypatch.setattr(config.settings, "auth_mode", "db_only")

    resp = client.get(
        "/api/v1/trade-in/whoami",
        headers={"X-Authenticated-User": "victim"},
    )
    assert resp.status_code == 200
    assert resp.json()["user"] == "alice"


def test_cyrillic_username_session_propagation_does_not_500(
    client: TestClient, store: _Store
) -> None:
    """#2552 post-review Medium 1: `.encode("latin-1")` без errors="replace" на
    кириллическом username крашил бы КАЖДЫЙ запрос такого юзера с 500."""
    store.add_user("алиса", hash_password("Secret123!"), role="employee")
    login_resp = client.post(
        "/api/v1/auth/login", json={"username": "алиса", "password": "Secret123!"}
    )
    assert login_resp.status_code == 200, login_resp.text

    resp = client.get("/api/v1/trade-in/whoami")
    assert resp.status_code == 200, resp.text
    # latin-1 "replace" гарантированно не крашит — точное значение (что именно
    # получится из non-latin1 байт) не является контрактом, важно отсутствие 500.
    assert resp.json()["user"] is not None


# ---------------------------------------------------------------------------
# Эпик «единый вход»: режим IDENTITY_STORE=auth (общий реестр в БД `auth`).
#
# Всё выше идёт в ДЕФОЛТНОМ режиме — он же прод — и служит регрессионным
# доказательством «после мержа работает точно как сейчас». Ниже — поведение,
# которое появляется ТОЛЬКО после переезда: трёхзначное состояние доступа
# (`active` / `trial_expired` / `disabled`) вместо булева `is_active`.
# ---------------------------------------------------------------------------


def test_default_mode_talks_to_tradein_tables_only(client: TestClient, store: _Store) -> None:
    """Дефолт трогает РОВНО сегодняшние таблицы — и ни одной таблицы реестра `auth`.

    Пин на случай, если флаг когда-нибудь начнёт «протекать» (например, дефолт
    поменяют или ветвление уедет не туда): расхождение здесь означало бы, что
    прод после мержа пошёл в другую БД.
    """
    store.add_user("alice", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
    assert client.get("/api/v1/me").status_code == 200

    joined = "\n".join(store.sql_log)
    assert "tradein_users" in joined
    assert "tradein_sessions" in joined
    # Ни один запрос не адресован таблицам общего реестра.
    assert not re.search(r"\b(FROM|INTO|UPDATE|JOIN)\s+users\b", joined)
    assert not re.search(r"\b(FROM|INTO|UPDATE|JOIN)\s+sessions\b", joined)
    # И engine БД `auth` даже не создавался (AUTH_DATABASE_URL на проде пуст —
    # ленивое построение обязано не случиться, иначе запрос упал бы).
    assert auth_db._engine is None


def test_auth_mode_talks_to_shared_registry_tables(auth_store: _Store, client: TestClient) -> None:
    """Зеркало предыдущего: при IDENTITY_STORE=auth запросы уходят в users/sessions."""
    auth_store.add_user("alice", hash_password("Secret123!"), role="employee")
    resp = client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
    assert resp.status_code == 200, resp.text
    assert client.get("/api/v1/me").status_code == 200

    joined = "\n".join(auth_store.sql_log)
    assert "tradein_users" not in joined
    assert "tradein_sessions" not in joined
    assert re.search(r"FROM\s+users\b", joined)
    assert re.search(r"INSERT INTO\s+sessions\b", joined)


def test_login_trial_expired_403_with_code_and_no_session(
    auth_store: _Store, client: TestClient, monkeypatch: pytest.MonkeyPatch
) -> None:
    """ВЕРНЫЙ пароль + `trial_expired` → 403 с машиночитаемым кодом, сессии НЕТ.

    Единственный не-generic ответ логина: аккаунт существует и владелец это уже
    доказал паролем, так что осмысленный текст постороннему ничего не выдаёт.
    """
    auth_store.add_user(
        "trialguy",
        hash_password("Secret123!"),
        role="employee",
        access_state=AccessState.TRIAL_EXPIRED,
    )
    events: list[dict[str, Any]] = []
    monkeypatch.setattr(auth_router, "schedule_event", lambda **kw: events.append(kw))

    resp = client.post(
        "/api/v1/auth/login", json={"username": "trialguy", "password": "Secret123!"}
    )

    assert resp.status_code == 403, resp.text
    detail = resp.json()["detail"]
    # Контракт для фронта — `code`, а не текст сообщения.
    assert detail["code"] == "access_expired"
    assert detail["message"]
    # Сессия не выдана: ни куки, ни строки в реестре.
    assert config.settings.session_cookie_name not in resp.cookies
    assert auth_store.sessions == {}
    assert [e["event_type"] for e in events] == ["login_blocked_expired"]


def test_login_wrong_password_on_trial_expired_is_generic_401(
    auth_store: _Store, client: TestClient
) -> None:
    """НЕверный пароль на `trial_expired` → тот же generic 401, что у чужого логина.

    Иначе отдельный 403 превращается в оракул существования аккаунта: перебором
    можно было бы перечислить логины, не зная ни одного пароля.
    """
    auth_store.add_user(
        "trialguy",
        hash_password("Secret123!"),
        role="employee",
        access_state=AccessState.TRIAL_EXPIRED,
    )

    wrong_pw = client.post("/api/v1/auth/login", json={"username": "trialguy", "password": "nope"})
    ghost = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "nope"})

    assert wrong_pw.status_code == 401
    # Побайтово тот же ответ, что и на несуществующий логин.
    assert wrong_pw.json() == ghost.json()
    assert auth_store.sessions == {}


def test_login_disabled_is_generic_401_not_403(auth_store: _Store, client: TestClient) -> None:
    """`disabled` + верный пароль → generic 401, НЕ 403: заблокированный аккаунт
    для пользователя неотличим от несуществующего (в отличие от `trial_expired`,
    у которого есть свой экран)."""
    auth_store.add_user(
        "blocked",
        hash_password("Secret123!"),
        role="employee",
        access_state=AccessState.DISABLED,
    )

    blocked = client.post(
        "/api/v1/auth/login", json={"username": "blocked", "password": "Secret123!"}
    )
    ghost = client.post("/api/v1/auth/login", json={"username": "ghost", "password": "x"})

    assert blocked.status_code == 401
    assert blocked.json() == ghost.json()
    assert auth_store.sessions == {}


def test_unknown_access_state_is_fail_closed_401(auth_store: _Store, client: TestClient) -> None:
    """Состояние, которого код не знает (миграция уехала вперёд кода), НЕ пускает."""
    auth_store.add_user("newbie", hash_password("Secret123!"), role="employee")
    auth_store.users["newbie"]["access_state"] = "pending_review"

    resp = client.post("/api/v1/auth/login", json={"username": "newbie", "password": "Secret123!"})

    assert resp.status_code == 401
    assert auth_store.sessions == {}


@pytest.mark.parametrize("state", [AccessState.TRIAL_EXPIRED, AccessState.DISABLED])
def test_live_session_dies_when_access_state_leaves_active(
    auth_store: _Store, client: TestClient, state: AccessState
) -> None:
    """Уже выданная сессия перестаёт работать СРАЗУ, как только состояние != active.

    Без этого sliding-refresh (`get_session_user` продлевает expires_at на каждом
    запросе) держал бы сессию истёкшего/заблокированного бесконечно долго.
    """
    auth_store.add_user("alice", hash_password("Secret123!"), role="employee")
    login = client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})
    assert login.status_code == 200
    assert client.get("/api/v1/trade-in/dummy").status_code == 200

    auth_store.set_access_state("alice", state)

    # auth_mode=dual, но legacy-заголовка нет → сессия больше не резолвится → 401.
    assert client.get("/api/v1/trade-in/dummy").status_code == 401
    assert client.get("/api/v1/me").status_code == 401


def test_session_identity_wins_over_spoofed_header_auth_store(
    auth_store: _Store, client: TestClient
) -> None:
    """Перезапись X-Authenticated-User в ASGI-scope работает и на общем реестре.

    Тот же CRITICAL, что и в дефолтном режиме (см. выше): подделанный клиентом
    заголовок не должен выигрывать у резолвленной сессии ни в одном режиме — эти
    ~15 downstream-хендлеров читают сырой заголовок и про режим ничего не знают.
    """
    auth_store.add_user("alice", hash_password("Secret123!"), role="employee")
    auth_store.add_user("victim", hash_password("Secret123!"), role="employee")
    client.post("/api/v1/auth/login", json={"username": "alice", "password": "Secret123!"})

    resp = client.get("/api/v1/trade-in/whoami", headers={"X-Authenticated-User": "victim"})

    assert resp.status_code == 200
    assert resp.json()["user"] == "alice"


def test_auth_mode_role_scope_and_logout(auth_store: _Store, client: TestClient) -> None:
    """Роль/скоуп и logout на общем реестре ведут себя как в дефолтном режиме."""
    auth_store.add_user("mgr", hash_password("Secret123!"), role="manager")
    login = client.post("/api/v1/auth/login", json={"username": "mgr", "password": "Secret123!"})
    token = login.cookies[config.settings.session_cookie_name]
    assert token in auth_store.sessions

    body = client.get("/api/v1/me").json()
    assert body["role"] == "manager"
    assert "/api/v1/team/**" in body["allowed_paths"]
    assert "/trade-in/sale-share/**" in body["deny_paths"]

    assert client.post("/api/v1/auth/logout").status_code == 200
    assert token not in auth_store.sessions