gendesign/backend/tests/test_auth_session_guard.py

"""Dual-mode `rbac_guard` «Птицы» — эпик «единый вход», стадия 3 (тесты).

Что здесь проверяется и почему именно так.

ТЕСТИРУЕТСЯ НАСТОЯЩИЙ `app.main.rbac_guard`, а не его копия. `app.middleware("http")`
у Starlette возвращает саму функцию (декоратор регистрирует dispatch и отдаёт `func`),
поэтому middleware вызывается напрямую: `await rbac_guard(request, call_next)`. Это
принципиально — в отличие от `tests/test_rbac.py`, где живёт РУЧНАЯ КОПИЯ guard'а
(она заведена, чтобы не тянуть тяжёлые импорты, и ценой этого расходится с прод-кодом
при каждой правке). Главный тест этого файла — про подделку заголовка, то есть про
безопасность; проверять безопасность на копии нельзя, копия не деплоится.

Почему не через `TestClient(app)`: `rbac_guard` первой строкой уходит в
test-mode bypass при `settings.testing=True`, а conftest.py ставит этот флаг
глобально (иначе весь остальной сьют получал бы 401). Прямой вызов middleware
позволяет снять именно этот флаг (monkeypatch, см. `_no_test_bypass`) и получить
прод-поведение guard'а целиком: и session-ветку, и легаси-ветку, и 401/403.

Как проверяется «downstream видит нужного юзера». `_propagate_authenticated_user`
перезаписывает заголовок в `request.scope["headers"]`; scope прокинут по ссылке через
весь ASGI-стек, и следующий слой (audit-middleware, роутер) строит поверх него СВОЙ
`Request`. Дублёр `_Downstream` делает ровно это — `Request(request.scope)` — то есть
видит заголовок так же, как одиннадцать мест бэкенда, читающих его напрямую мимо
guard'а (перечислены в докстринге `_propagate_authenticated_user`).

БД `auth` здесь не поднимается: подменяется `app.core.auth_db.auth_session` (сам резолв
сессии — `get_session_user` — прогоняется НАСТОЯЩИЙ, чтобы «истекла»/«не active»
проверялись кодом, а не заглушкой). Юнит-тесты самого резолва — в
`tests/test_auth_session_service.py`, конфигурация DSN — в `tests/test_auth_db.py`.
"""

from __future__ import annotations

import logging
from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any

import pytest
from fastapi import Request
from fastapi.responses import JSONResponse, Response

import app.main as app_main
from app.core import auth as auth_mod
from app.core import auth_db
from app.core.config import settings
from app.main import rbac_guard

# Логины из auth/roles.yaml (см. tests/test_rbac.py::test_get_role_known_users):
_ADMIN_LOGIN = "admin"  # role=admin
_PILOT_LOGIN = "user1"  # role=pilot
_NOT_IN_ROLES_YAML = "ghost"  # роли нет вообще → 403 на всё

_VALID_TOKEN = "tok-valid"
_EXPIRED_TOKEN = "tok-expired"
_UNKNOWN_TOKEN = "tok-never-issued"


# ---------------------------------------------------------------------------
# Дублёры
# ---------------------------------------------------------------------------


@dataclass
class _Row:
    """Строка ответа SELECT'а из `app/services/auth_session.py` (4 колонки)."""

    expires_at: datetime | None
    last_seen_at: datetime | None
    username: str
    access_state: str


class _FetchOne:
    def __init__(self, row: _Row | None) -> None:
        self._row = row

    def fetchone(self) -> _Row | None:
        return self._row


# Форма запросов к реестру. Без этих проверок дублёр диспетчеризует по одному лишь
# `startswith`, и тела SQL не покрыты ВООБЩЕ: мутационный прогон показал, что
# `sessions`→`sessionz`, `users`→`userz`, `s.token`→`s.tokenX`, `last_seen_at`→
# `last_seen_atX` не роняли ни одного теста. Настоящего Postgres в сьюте нет, а цена
# опечатки/дрейфа схемы здесь высокая: не 500, а «ни у кого нет сессии» с тихим
# откатом на легаси-заголовок (после снятия popup'а — локаут всех).
_SELECT_MUST_CONTAIN = (
    "FROM sessions s",
    "JOIN users u ON u.id = s.user_id",
    "WHERE s.token = :token",
    # Срок годности отсекается часами БД — теми же, которыми UPDATE ниже пишет
    # expires_at. Питоновская проверка остаётся вторым поясом.
    "AND s.expires_at > now()",
    "s.expires_at",
    "s.last_seen_at",
    "u.username",
    "u.access_state",
)
_UPDATE_MUST_CONTAIN = (
    "UPDATE sessions",
    "last_seen_at = now()",
    # Обе колонки одним UPDATE: продлевай «Птица» только last_seen_at — человек,
    # работающий весь день в ней одной, был бы разлогинен по expires_at.
    "expires_at = now() + make_interval(hours => CAST(:ttl_hours AS integer))",
    "WHERE token = :token",
)


def _assert_select_shape(sql: str) -> None:
    for fragment in _SELECT_MUST_CONTAIN:
        assert fragment in sql, f"SELECT к БД auth потерял {fragment!r}: {sql}"


def _assert_update_shape(sql: str) -> None:
    for fragment in _UPDATE_MUST_CONTAIN:
        assert fragment in sql, f"UPDATE к БД auth потерял {fragment!r}: {sql}"


class FakeAuthDb:
    """Дублёр сессии SQLAlchemy к БД `auth`: понимает ровно два запроса модуля.

    Считает обращения (`select_tokens`, `updates`) — по ним тесты доказывают не только
    результат, но и что запрос вообще был/не был сделан.
    """

    def __init__(self, rows: dict[str, _Row] | None = None, *, fail_refresh: bool = False) -> None:
        self.rows: dict[str, _Row] = dict(rows or {})
        self.select_tokens: list[str] = []
        self.updates: list[dict[str, Any]] = []
        self.commits = 0
        self.rollbacks = 0
        self.closed = False
        self.fail_refresh = fail_refresh

    def execute(self, clause: Any, params: dict[str, Any]) -> _FetchOne:
        sql = " ".join(str(clause).split())
        if sql.startswith("SELECT"):
            _assert_select_shape(sql)
            self.select_tokens.append(params["token"])
            return _FetchOne(self.rows.get(params["token"]))
        if sql.startswith("UPDATE sessions"):
            _assert_update_shape(sql)
            if self.fail_refresh:
                raise RuntimeError("sessions is read-only on this replica")
            self.updates.append(dict(params))
            return _FetchOne(None)
        raise AssertionError(f"неожиданный SQL к БД auth: {sql}")

    def commit(self) -> None:
        self.commits += 1

    def rollback(self) -> None:
        self.rollbacks += 1


def _install_auth_db(monkeypatch: pytest.MonkeyPatch, db: FakeAuthDb | None) -> None:
    """Подменяет `auth_db.auth_session`. `db=None` → любое обращение к БД падает."""

    @contextmanager
    def _fake_session() -> Iterator[FakeAuthDb]:
        if db is None:
            raise RuntimeError("connection to auth registry refused")
        yield db

    monkeypatch.setattr(auth_db, "auth_session", _fake_session)


class _Downstream:
    """`call_next`: запоминает, каким юзером запрос выглядит для следующего слоя."""

    def __init__(self) -> None:
        self.calls = 0
        self.seen_users: list[str | None] = []
        self.seen_header_counts: list[int] = []

    async def __call__(self, request: Request) -> Response:
        # Именно так заголовок видят 11 прямых читателей: свой Request поверх того же
        # scope, который guard уже успел переписать.
        downstream = Request(request.scope)
        self.calls += 1
        self.seen_users.append(downstream.headers.get("X-Authenticated-User"))
        self.seen_header_counts.append(
            sum(1 for k, _ in request.scope["headers"] if k == b"x-authenticated-user")
        )
        return JSONResponse({"ok": True})


def _make_request(
    path: str,
    *,
    cookie_token: str | None = None,
    header_user: str | None = None,
    cookie_name: str | None = None,
    header_name: bytes = b"x-authenticated-user",
) -> Request:
    """ASGI-scope запроса. Имена заголовков lowercase — как их отдаёт любой сервер.

    `header_name` позволяет подсунуть имя в НЕканоническом регистре: спека ASGI требует
    lowercase, но полагаться на неё в фильтре `_propagate_authenticated_user` мы не
    хотим (чужой ASGI-слой/харнесс может её нарушить).
    """
    headers: list[tuple[bytes, bytes]] = [(b"host", b"gendsgn.ru")]
    if cookie_token is not None:
        name = cookie_name or settings.session_cookie_name
        headers.append((b"cookie", f"{name}={cookie_token}".encode()))
    if header_user is not None:
        headers.append((header_name, header_user.encode("latin-1")))
    return Request(
        {
            "type": "http",
            "asgi": {"version": "3.0", "spec_version": "2.3"},
            "http_version": "1.1",
            "method": "GET",
            "scheme": "https",
            "server": ("gendsgn.ru", 443),
            "client": ("203.0.113.7", 51234),
            "root_path": "",
            "path": path,
            "raw_path": path.encode(),
            "query_string": b"",
            "headers": headers,
        }
    )


async def _run_guard(request: Request) -> tuple[Response, _Downstream]:
    downstream = _Downstream()
    response = await rbac_guard(request, downstream)
    return response, downstream


def _valid_session(username: str, *, access_state: str = "active") -> _Row:
    now = datetime.now(UTC)
    return _Row(
        expires_at=now + timedelta(days=7),
        last_seen_at=now - timedelta(seconds=30),  # свежее 5 минут → без UPDATE
        username=username,
        access_state=access_state,
    )


# ---------------------------------------------------------------------------
# Фикстуры
# ---------------------------------------------------------------------------


@pytest.fixture(autouse=True)
def _reset_auth_cache() -> None:
    """Свежий YAML-кэш ролей на каждый тест (как в tests/test_rbac.py)."""
    auth_mod.reset_cache_for_tests()


@pytest.fixture(autouse=True)
def _reset_registry_throttle() -> Iterator[None]:
    """Окно троттлинга алерта «реестр не отвечает» — модульное состояние app.main.

    Без сброса первый же тест, поймавший сбой реестра, глушил бы ERROR у всех
    следующих в течение минуты, и они краснели/зеленели бы в зависимости от порядка
    и скорости прогона.
    """
    app_main._reset_registry_failure_throttle()
    yield
    app_main._reset_registry_failure_throttle()


@pytest.fixture(autouse=True)
def _no_test_bypass(monkeypatch: pytest.MonkeyPatch) -> None:
    """Снимает test-mode bypass: без этого guard возвращает call_next первой строкой.

    conftest.py ставит `settings.testing = True` глобально; monkeypatch вернёт его
    обратно после каждого теста, так что остальной сьют не затронут.
    """
    monkeypatch.setattr(settings, "testing", False)


@pytest.fixture
def resolve_spy(monkeypatch: pytest.MonkeyPatch) -> list[str | None]:
    """Считает вызовы `resolve_session_token` из app.main, не подменяя его логику.

    Нужен, чтобы доказывать НЕ-обращения: «флаг выключен → к реестру не ходим»,
    «публичный путь → к реестру не ходим».
    """
    calls: list[str | None] = []
    real = app_main.resolve_session_token

    def _spy(token: str | None) -> Any:
        calls.append(token)
        return real(token)

    monkeypatch.setattr(app_main, "resolve_session_token", _spy)
    return calls


@pytest.fixture
def no_engine_build(monkeypatch: pytest.MonkeyPatch) -> None:
    """Ломает создание engine БД `auth`: тест покраснеет, если его вообще строят."""

    def _boom() -> tuple[Any, Any]:
        raise AssertionError("engine БД `auth` не должен создаваться в этом сценарии")

    monkeypatch.setattr(auth_db, "_build", _boom)


# ---------------------------------------------------------------------------
# ФЛАГ ВЫКЛЮЧЕН (дефолт) — прод обязан вести себя ровно как до эпика
# ---------------------------------------------------------------------------


async def test_flag_is_off_by_default() -> None:
    """Дефолт синглтона settings — режим legacy. Весь файл ниже опирается на это."""
    assert settings.auth_mode == "legacy"
    assert settings.auth_session_enabled is False


async def test_flag_off_legacy_header_still_works(
    monkeypatch: pytest.MonkeyPatch, resolve_spy: list[str | None], no_engine_build: None
) -> None:
    """Сегодняшний путь (Caddy basic_auth → X-Authenticated-User) не изменился."""
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    _install_auth_db(monkeypatch, None)

    response, downstream = await _run_guard(_make_request("/api/v1/me", header_user=_ADMIN_LOGIN))

    assert response.status_code == 200
    assert downstream.seen_users == [_ADMIN_LOGIN]
    assert resolve_spy == []


async def test_flag_off_ignores_session_cookie_and_never_touches_registry(
    monkeypatch: pytest.MonkeyPatch, resolve_spy: list[str | None], no_engine_build: None
) -> None:
    """🔒 Инвариант PR: при выключенном флаге кука не читается, к БД `auth` не идём.

    Валидная кука + нет легаси-заголовка → 401, как сегодня у любого запроса мимо
    Caddy. `resolve_spy`/`no_engine_build` доказывают, что дело не в «не нашли
    сессию», а в том, что резолв вообще не запускался и engine не строился.
    """
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    _install_auth_db(monkeypatch, FakeAuthDb({_VALID_TOKEN: _valid_session(_ADMIN_LOGIN)}))

    response, downstream = await _run_guard(_make_request("/api/v1/me", cookie_token=_VALID_TOKEN))

    assert response.status_code == 401
    assert downstream.calls == 0
    assert resolve_spy == []


async def test_flag_off_unknown_user_still_403(monkeypatch: pytest.MonkeyPatch) -> None:
    """Легаси-ветка целиком: юзер не в roles.yaml → 403 «user not in roles config»."""
    monkeypatch.setattr(settings, "auth_mode", "legacy")

    response, downstream = await _run_guard(
        _make_request("/api/v1/me", header_user=_NOT_IN_ROLES_YAML)
    )

    assert response.status_code == 403
    assert downstream.calls == 0


# ---------------------------------------------------------------------------
# ФЛАГ ВКЛЮЧЁН — сессионная кука как источник личности
# ---------------------------------------------------------------------------


@pytest.fixture
def session_on(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_mode", "dual")


async def test_valid_cookie_grants_access_without_any_header(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """Валидная кука пускает — легаси-заголовка при этом нет вовсе."""
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_PILOT_LOGIN)})
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(_make_request("/api/v1/me", cookie_token=_VALID_TOKEN))

    assert response.status_code == 200
    assert downstream.seen_users == [_PILOT_LOGIN]
    assert db.select_tokens == [_VALID_TOKEN]


async def test_expired_session_cookie_does_not_grant_access(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """Истёкшая сессия = сессии нет: без легаси-заголовка это 401."""
    now = datetime.now(UTC)
    db = FakeAuthDb(
        {
            _EXPIRED_TOKEN: _Row(
                expires_at=now - timedelta(seconds=1),
                last_seen_at=now - timedelta(days=1),
                username=_ADMIN_LOGIN,
                access_state="active",
            )
        }
    )
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(
        _make_request("/api/v1/me", cookie_token=_EXPIRED_TOKEN)
    )

    assert response.status_code == 401
    assert downstream.calls == 0
    # Истёкшая сессия не продлевается sliding-refresh'ем — иначе она была бы вечной.
    assert db.updates == []


@pytest.mark.parametrize("access_state", ["disabled", "trial_expired", "some_future_state"])
async def test_non_active_access_state_does_not_grant_access(
    monkeypatch: pytest.MonkeyPatch, session_on: None, access_state: str
) -> None:
    """Блокировка в реестре действует НЕМЕДЛЕННО, не дожидаясь expires_at.

    `some_future_state` — состояние, добавленное миграцией раньше кода: fail-closed
    (`to_access_state` → disabled), а не «раз не disabled, значит пускаем».
    """
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_ADMIN_LOGIN, access_state=access_state)})
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(_make_request("/api/v1/me", cookie_token=_VALID_TOKEN))

    assert response.status_code == 401
    assert downstream.calls == 0
    assert db.updates == []


async def test_session_user_missing_from_roles_yaml_is_403(
    monkeypatch: pytest.MonkeyPatch, session_on: None, caplog: pytest.LogCaptureFixture
) -> None:
    """Есть в реестре, нет в auth/roles.yaml → 403 + отдельное сообщение о рассинхроне.

    Реестр отвечает «кто ты», roles.yaml — «что тебе можно»; человек, заведённый только
    в реестре, не получает доступ по умолчанию.
    """
    db = FakeAuthDb({_VALID_TOKEN: _valid_session("brand_new_hire")})
    _install_auth_db(monkeypatch, db)

    with caplog.at_level(logging.WARNING, logger="app.main"):
        response, downstream = await _run_guard(
            _make_request("/api/v1/me", cookie_token=_VALID_TOKEN)
        )

    assert response.status_code == 403
    assert downstream.calls == 0
    assert any("roles.yaml" in r.getMessage() for r in caplog.records)


# ---------------------------------------------------------------------------
# 🔴 ГЛАВНОЕ: подделка X-Authenticated-User при валидной куке
# ---------------------------------------------------------------------------


async def test_valid_cookie_overrides_client_supplied_header(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """🔴 Кука выигрывает у присланного клиентом заголовка — downstream видит ВЛАДЕЛЬЦА КУКИ.

    Сценарий: у человека есть валидная сессия (`user1`, pilot), и он вручную добавляет
    к запросу `X-Authenticated-User: admin`. На проде Caddy шлёт этот заголовок на
    каждый запрос, так что «поставить только если отсутствует» здесь не сработало бы:
    заголовок присутствует ВСЕГДА, и любой из одиннадцати прямых читателей (аудит,
    /me, created_by в insights/own-projects, forecast/analyze) увидел бы подделку.

    Проверяем оба следствия перезаписи: значение — владелец куки, и заголовок в scope
    РОВНО ОДИН (append без фильтра оставил бы два, а `headers.get` вернул бы первый —
    то есть подделанный).
    """
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_PILOT_LOGIN)})
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(
        _make_request("/api/v1/me", cookie_token=_VALID_TOKEN, header_user=_ADMIN_LOGIN)
    )

    assert response.status_code == 200
    assert downstream.seen_users == [_PILOT_LOGIN], "downstream увидел подделанный заголовок"
    assert downstream.seen_header_counts == [1], "в scope осталось два X-Authenticated-User"


async def test_forged_admin_header_cannot_escalate_to_admin_api(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """🔴 Та же подделка на admin-эндпоинте: роль берётся от владельца куки → 403.

    Это тест на ЭСКАЛАЦИЮ ПРИВИЛЕГИЙ, а не на атрибуцию, и он ловит другую поломку,
    чем тест выше. Проверено мутацией: подмена перезаписи заголовка на append его НЕ
    красит — guard решает по локальной переменной `username`, уже взятой из сессии.
    Покраснеет он от поломки ПОРЯДКА: «сначала заголовок, потом кука» или повторное
    чтение `request.headers` после резолва — тогда pilot с подделанным `admin` вошёл
    бы в /api/v1/admin/*. Оба теста нужны: один держит downstream, другой — сам guard.
    """
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_PILOT_LOGIN)})
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(
        _make_request(
            "/api/v1/admin/scrape/status", cookie_token=_VALID_TOKEN, header_user=_ADMIN_LOGIN
        )
    )

    assert response.status_code == 403
    assert response.body == b'{"detail":"admin only"}'
    assert downstream.calls == 0


async def test_cookie_owner_wins_even_when_forged_header_is_unknown_user(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """Обратная сторона: мусор в заголовке не мешает владельцу валидной куки войти.

    Пинует порядок «кука → заголовок»: если бы заголовок проверялся первым, `ghost`
    дал бы 403 человеку с законной сессией.
    """
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_ADMIN_LOGIN)})
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(
        _make_request(
            "/api/v1/admin/scrape/status",
            cookie_token=_VALID_TOKEN,
            header_user=_NOT_IN_ROLES_YAML,
        )
    )

    assert response.status_code == 200
    assert downstream.seen_users == [_ADMIN_LOGIN]


# ---------------------------------------------------------------------------
# Dual-mode: нет куки / кука не резолвится → легаси-заголовок
# ---------------------------------------------------------------------------


async def test_no_cookie_falls_back_to_legacy_header(
    monkeypatch: pytest.MonkeyPatch, session_on: None, resolve_spy: list[str | None]
) -> None:
    """Флаг включён, куки нет — работает заголовок, и в БД `auth` не идёт ни запроса."""
    _install_auth_db(monkeypatch, None)  # любое обращение к реестру → RuntimeError

    response, downstream = await _run_guard(_make_request("/api/v1/me", header_user=_ADMIN_LOGIN))

    assert response.status_code == 200
    assert downstream.seen_users == [_ADMIN_LOGIN]
    assert resolve_spy == [], "куки нет — резолвить нечего, коннект открывать незачем"


async def test_unknown_token_falls_back_to_legacy_header(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """Кука есть, сессии в реестре нет (протухла/отозвана) → легаси-путь, не отказ.

    Пока стоит popup, это ровно тот же уровень доступа, что и сегодня; отказывать
    здесь значило бы ломать вход людям со старой кукой в браузере.
    """
    db = FakeAuthDb()  # пусто: токен не найден
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(
        _make_request("/api/v1/me", cookie_token=_UNKNOWN_TOKEN, header_user=_ADMIN_LOGIN)
    )

    assert response.status_code == 200
    assert downstream.seen_users == [_ADMIN_LOGIN]
    assert db.select_tokens == [_UNKNOWN_TOKEN]


async def test_foreign_cookie_name_is_not_a_session(
    monkeypatch: pytest.MonkeyPatch, session_on: None, resolve_spy: list[str | None]
) -> None:
    """Чужая кука (другое имя) сессией не считается — читаем только session_cookie_name."""
    _install_auth_db(monkeypatch, None)

    response, downstream = await _run_guard(
        _make_request("/api/v1/me", cookie_token="whatever", cookie_name="ym_uid")
    )

    assert response.status_code == 401
    assert downstream.calls == 0
    assert resolve_spy == []


async def test_no_cookie_no_header_is_401(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """Ни куки, ни заголовка → 401 с прежним текстом (его читает фронт)."""
    _install_auth_db(monkeypatch, None)

    response, downstream = await _run_guard(_make_request("/api/v1/me"))

    assert response.status_code == 401
    assert b"no authenticated user" in response.body
    assert downstream.calls == 0


# ---------------------------------------------------------------------------
# Публичные пути
# ---------------------------------------------------------------------------


@pytest.mark.parametrize("path", sorted(app_main._PUBLIC_PATHS))
async def test_public_paths_need_nothing_and_touch_no_registry(
    monkeypatch: pytest.MonkeyPatch,
    session_on: None,
    resolve_spy: list[str | None],
    no_engine_build: None,
    path: str,
) -> None:
    """/health и прочие публичные пути — без куки, без заголовка, без коннекта к `auth`.

    Параметризация по самому `_PUBLIC_PATHS`: добавят путь в список — он проверится.
    """
    _install_auth_db(monkeypatch, None)

    response, downstream = await _run_guard(_make_request(path))

    assert response.status_code == 200
    assert downstream.calls == 1
    assert resolve_spy == []


async def test_public_path_with_cookie_still_skips_registry(
    monkeypatch: pytest.MonkeyPatch, session_on: None, resolve_spy: list[str | None]
) -> None:
    """Публичный путь + кука в браузере → всё равно ни одного запроса к реестру."""
    _install_auth_db(monkeypatch, FakeAuthDb({_VALID_TOKEN: _valid_session(_ADMIN_LOGIN)}))

    response, _ = await _run_guard(_make_request("/health", cookie_token=_VALID_TOKEN))

    assert response.status_code == 200
    assert resolve_spy == []


# ---------------------------------------------------------------------------
# Сбой БД `auth` при резолве
# ---------------------------------------------------------------------------


async def test_registry_failure_does_not_silently_admit_cookie_owner(
    monkeypatch: pytest.MonkeyPatch, session_on: None, caplog: pytest.LogCaptureFixture
) -> None:
    """Реестр упал → 401 (нет легаси-заголовка) + ERROR с traceback, а не тихий проход.

    Два независимых требования:
      1) владелец куки НЕ входит «на всякий случай» — упавший резолв не даёт личности;
      2) событие громкое: `logger.exception` уровня ERROR уезжает в GlitchTip
         (LoggingIntegration event_level=ERROR), т.е. это алерт, а не строка в логе.
    """
    _install_auth_db(monkeypatch, None)

    with caplog.at_level(logging.ERROR, logger="app.main"):
        response, downstream = await _run_guard(
            _make_request("/api/v1/me", cookie_token=_VALID_TOKEN)
        )

    assert response.status_code == 401
    assert downstream.calls == 0
    errors = [r for r in caplog.records if r.levelno >= logging.ERROR]
    assert len(errors) == 1, "сбой реестра обязан быть ровно одним ERROR-событием"
    assert errors[0].exc_info is not None, "нужен traceback: без него алерт бесполезен"


async def test_registry_failure_degrades_to_legacy_while_popup_is_up(
    monkeypatch: pytest.MonkeyPatch, session_on: None, caplog: pytest.LogCaptureFixture
) -> None:
    """Осознанная переходная деградация: сломанный реестр → сегодняшний путь + ERROR.

    Пока Caddy basic_auth стоит перед бэкендом, легаси-заголовок защищён ровно тем же,
    чем защищён весь продукт сегодня, и класть «Птицу» целиком (503) из-за проблемы
    реестра незачем.

    ⚠️ Этот тест — маркер долга, а не одобрение поведения навсегда. Последний PR эпика
    снимает popup; вместе с ним деградация обязана уйти (у «Меры» это auth_mode=db_only),
    иначе заголовок станет полностью клиентским. Тест тогда переписывается на отказ.
    """
    _install_auth_db(monkeypatch, None)

    with caplog.at_level(logging.ERROR, logger="app.main"):
        response, downstream = await _run_guard(
            _make_request("/api/v1/me", cookie_token=_VALID_TOKEN, header_user=_ADMIN_LOGIN)
        )

    assert response.status_code == 200
    assert downstream.seen_users == [_ADMIN_LOGIN]
    assert [r for r in caplog.records if r.levelno >= logging.ERROR]


# ---------------------------------------------------------------------------
# Test-mode bypass остаётся выключателем ВСЕГО guard'а
# ---------------------------------------------------------------------------


async def test_testing_bypass_disables_session_branch_too(
    monkeypatch: pytest.MonkeyPatch, session_on: None, resolve_spy: list[str | None]
) -> None:
    """`settings.testing=True` отключает и session-ветку — сознательно, не по недосмотру.

    Промежуточного состояния «личность резолвим, а 401/403 не применяем» нет ни в одном
    реальном режиме; поэтому весь остальной сьют (conftest ставит testing=True) не
    начинает вдруг ходить в БД `auth`.
    """
    monkeypatch.setattr(settings, "testing", True)
    _install_auth_db(monkeypatch, None)

    response, downstream = await _run_guard(
        _make_request("/api/v1/me", cookie_token=_VALID_TOKEN, header_user=_ADMIN_LOGIN)
    )

    assert response.status_code == 200
    assert downstream.seen_users == [_ADMIN_LOGIN], "bypass не должен переписывать заголовок"
    assert resolve_spy == []


async def test_forged_header_in_mixed_case_is_replaced_not_duplicated(
    monkeypatch: pytest.MonkeyPatch, session_on: None
) -> None:
    """Заголовок в НЕканоническом регистре тоже вытесняется, а не остаётся вторым.

    По спеке ASGI имена заголовков в scope всегда lowercase, и uvicorn её соблюдает —
    но `_propagate_authenticated_user` на это не полагается. Если бы фильтр сравнивал
    сырые байты, в scope осталась бы ВТОРАЯ запись `X-Authenticated-User: admin` рядом
    с нашей. Эксплуатируемой дыры это не давало (`Headers.get` лоуэркейсит искомый
    ключ, но не хранимый, поэтому смешанный регистр не матчится никогда), но состояние
    «две записи с одним именем» ложное по построению — и в чужом ASGI-слое, который
    регистр нормализует, оно стало бы подделкой.
    """
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_PILOT_LOGIN)})
    _install_auth_db(monkeypatch, db)
    request = _make_request(
        "/api/v1/me",
        cookie_token=_VALID_TOKEN,
        header_user=_ADMIN_LOGIN,
        header_name=b"X-Authenticated-User",
    )

    response, downstream = await _run_guard(request)

    assert response.status_code == 200
    assert downstream.seen_users == [_PILOT_LOGIN]
    names = [k for k, _ in request.scope["headers"] if k.lower() == b"x-authenticated-user"]
    assert names == [b"x-authenticated-user"], "подделка осталась в scope вторым заголовком"


async def test_registry_failure_alert_is_throttled(
    monkeypatch: pytest.MonkeyPatch, session_on: None, caplog: pytest.LogCaptureFixture
) -> None:
    """Лежащий реестр даёт ОДИН ERROR на окно, остальное — WARNING без traceback.

    Guard резолвит сессию на каждом non-public запросе с кукой, а ERROR уезжает
    событием в GlitchTip (LoggingIntegration event_level=ERROR). Без троттлинга сбой
    реестра выжигал бы квоту за минуты — и настоящие ошибки этого же периода терялись
    бы вместе с ней. Факт продолжающегося сбоя при этом остаётся видимым в логах.
    """
    _install_auth_db(monkeypatch, None)

    with caplog.at_level(logging.WARNING, logger="app.main"):
        for _ in range(3):
            response, _ = await _run_guard(_make_request("/api/v1/me", cookie_token=_VALID_TOKEN))
            assert response.status_code == 401

    errors = [r for r in caplog.records if r.levelno >= logging.ERROR]
    warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
    assert len(errors) == 1, "второй и третий сбой обязаны быть подавлены троттлингом"
    assert errors[0].exc_info is not None
    assert len(warnings) == 2, "подавленные сбои всё равно обязаны быть видны в логе"
    assert all(w.exc_info is None for w in warnings)


# ---------------------------------------------------------------------------
# AUTH_MODE=db_only — конечное состояние эпика: легаси-ветка НЕДОСТИЖИМА
# ---------------------------------------------------------------------------


@pytest.fixture
def db_only(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_mode", "db_only")


async def test_db_only_ignores_legacy_header_completely(
    monkeypatch: pytest.MonkeyPatch, db_only: None, resolve_spy: list[str | None]
) -> None:
    """🔴 Ради этого режим и заведён: `X-Authenticated-User` больше не пускает никого.

    Этот режим включается тем же PR, который снимает `basic_auth` + `header_up` из
    Caddy, то есть делает заголовок полностью клиентским. Пройди `curl -H
    'X-Authenticated-User: admin'` здесь — это был бы полный обход аутентификации.
    """
    _install_auth_db(monkeypatch, None)

    response, downstream = await _run_guard(_make_request("/api/v1/me", header_user=_ADMIN_LOGIN))

    assert response.status_code == 401
    assert b"valid session required" in response.body
    assert downstream.calls == 0
    assert resolve_spy == [], "куки нет — резолвить нечего"


async def test_db_only_rejects_when_registry_is_down(
    monkeypatch: pytest.MonkeyPatch, db_only: None, caplog: pytest.LogCaptureFixture
) -> None:
    """Сбой реестра в db_only = отказ, а не деградация на заголовок.

    Тот же вход, что в `test_registry_failure_degrades_to_legacy_while_popup_is_up`
    (кука + заголовок + лежащий реестр), но исход противоположный. Пара тестов и есть
    механическая защита: удалить легаси-фолбэк забудут — этот тест покраснеет, если
    db_only начнёт вести себя как dual.
    """
    _install_auth_db(monkeypatch, None)

    with caplog.at_level(logging.ERROR, logger="app.main"):
        response, downstream = await _run_guard(
            _make_request("/api/v1/me", cookie_token=_VALID_TOKEN, header_user=_ADMIN_LOGIN)
        )

    assert response.status_code == 401
    assert downstream.calls == 0
    assert [r for r in caplog.records if r.levelno >= logging.ERROR]


@pytest.mark.parametrize("access_state", ["disabled", "trial_expired"])
async def test_db_only_blocked_account_cannot_fall_back_to_header(
    monkeypatch: pytest.MonkeyPatch, db_only: None, access_state: str
) -> None:
    """Заблокированный в реестре не добирает доступ подделанным заголовком."""
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_ADMIN_LOGIN, access_state=access_state)})
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(
        _make_request("/api/v1/me", cookie_token=_VALID_TOKEN, header_user=_ADMIN_LOGIN)
    )

    assert response.status_code == 401
    assert downstream.calls == 0


async def test_db_only_admits_valid_session(monkeypatch: pytest.MonkeyPatch, db_only: None) -> None:
    """Валидная сессия работает и в db_only — режим убирает фолбэк, а не вход."""
    db = FakeAuthDb({_VALID_TOKEN: _valid_session(_PILOT_LOGIN)})
    _install_auth_db(monkeypatch, db)

    response, downstream = await _run_guard(_make_request("/api/v1/me", cookie_token=_VALID_TOKEN))

    assert response.status_code == 200
    assert downstream.seen_users == [_PILOT_LOGIN]


async def test_db_only_keeps_public_paths_open(
    monkeypatch: pytest.MonkeyPatch, db_only: None, no_engine_build: None
) -> None:
    """/health и прочие публичные пути остаются публичными — иначе упадёт healthcheck."""
    _install_auth_db(monkeypatch, None)

    response, downstream = await _run_guard(_make_request("/health"))

    assert response.status_code == 200
    assert downstream.calls == 1