8 changed files with 10 additions and 2720 deletions
--- a/backend/app/core/auth_db.py
+++ b/backend/app/core/auth_db.py
@ -1,269 +0,0 @@
 """Engine + session-factory для БД `auth` — общего реестра людей (эпик «единый вход»).
 Отдельный модуль, а не ещё пара строк в `app.core.db`, ровно по одной причине:
 `app.core.db` создаёт engine НА ИМПОРТЕ (`create_engine(settings.database_url)` в
 теле модуля, db.py:8). Сделай мы так же для БД `auth` — приложение начало бы
 падать на старте везде, где реестр не сконфигурирован: локально, в pytest и на
 любом стенде, где переменных AUTH_* нет. Здесь engine создаётся ЛЕНИВО, при
 первом реальном обращении.
 Контракт (⚠️ после мержа прод обязан работать ТОЧНО как сейчас — Caddy basic_auth
 ещё стоит и снимается последним PR эпика):
  * `AUTH_MODE=legacy` (ДЕФОЛТ; `settings.auth_session_enabled is False`) — в этот
    модуль не заходит никто: `app.main.rbac_guard` в этом режиме куку не читает
    вовсе. Пустая конфигурация БД `auth` при этом не ошибка ни на импорте, ни в
    рантайме; ни одно соединение с БД `auth` не открывается.
  * Режим включён (`dual`/`db_only`) + не сконфигурированный реестр — обращение поднимает
    `AuthDatabaseNotConfiguredError` с внятным текстом. Именно исключение, а НЕ
    тихий возврат «сессия не найдена»: молчаливая деградация означала бы, что все
    владельцы валидных кук выглядят как анонимы, то есть массовый отказ доступа
    под видом «просто не залогинен» — либо, если guard в этот момент откатывается
    на trusted-header, наоборот, раздача прав в обход реестра (включая аккаунты с
    access_state 'disabled'). Оба исхода обязаны быть громкими.
 «Птица» реестр только ЧИТАЕТ: сессии выдаёт и отзывает единственная форма входа —
 у «Меры». Здесь нет и не должно появиться ни create-, ни revoke-пути.
 Сам DSN этот модуль НЕ выбирает и НЕ склеивает — берёт готовый у
 `settings.resolved_auth_database_url` (явный `AUTH_DATABASE_URL`, иначе сборка из
 `AUTH_DB_PASSWORD` + частей хоста/порта/базы/пользователя, иначе пусто).
 ⚠️ В DSN — пароль роли `auth_app`. Он не логируется и не попадает в текст
 исключений НИ В ОДНОЙ ветке этого модуля: сообщения ниже — константы, а ошибку
 разбора URL от SQLAlchemy (её текст содержит исходную строку) мы перехватываем и
 заменяем своей, обрывая цепочку `from None`, чтобы исходник не всплыл в traceback.
 Добавляешь сюда `logger`/`raise ... {dsn}` — не добавляй.
 `create_engine` сам по себе к серверу не ходит (пул коннектов ленивый) — то есть
 одна лишь сборка engine доказывает только «DSN не пуст и парсится». Поэтому
 `require_auth_db_configured` (fail-fast старта) дополнительно ОТКРЫВАЕТ соединение
 и делает `SELECT 1`: неверный пароль, опечатка в хосте, отсутствующая БД и
 отозванная роль обязаны ронять деплой, а не превращаться в «ни у кого нет сессии».
 Зеркало по подходу: tradein-mvp/backend/app/core/auth_db.py («Мера»). Синхронизация
 руками — стеки разные, общего кода между ними нет и заводить его этот эпик не
 собирается.
 """
 from __future__ import annotations
 import threading
 from collections.abc import Iterator
 from contextlib import contextmanager
 from sqlalchemy import Engine, create_engine, text
 from sqlalchemy.exc import ArgumentError
 from sqlalchemy.orm import Session, sessionmaker
 from app.core.config import settings
 class AuthDatabaseNotConfiguredError(RuntimeError):
    """`AUTH_MODE` не `legacy`, а DSN БД `auth` не задан/не разобрался."""
 class AuthDatabaseUnreachableError(RuntimeError):
    """DSN синтаксически корректен, но соединиться по нему не удалось (старт приложения)."""
 _NOT_CONFIGURED_MSG = (
    "Приём сессионной куки включён (AUTH_MODE=dual|db_only), но реестр людей "
    "(БД `auth`) не сконфигурирован: пусты и AUTH_DB_PASSWORD, и AUTH_DATABASE_URL — "
    "подключаться не к чему. Задай в backend/.env.runtime AUTH_DB_PASSWORD (пароль "
    "роли auth_app; остальные части DSN — AUTH_DB_HOST/AUTH_DB_PORT/AUTH_DB_NAME/"
    "AUTH_DB_USER — имеют прод-дефолты), либо целиком AUTH_DATABASE_URL, либо верни "
    "AUTH_MODE=legacy (сегодняшнее поведение: Caddy basic_auth + заголовок "
    "X-Authenticated-User)."
 )
 _UNREACHABLE_MSG = (
    "Приём сессионной куки включён (AUTH_MODE=dual|db_only), DSN разобрался, но "
    "соединиться с БД `auth` не удалось (см. причину ниже: хост/порт/база/роль/пароль "
    "или сеть). Старт прерван намеренно: иначе сломанная конфигурация выглядела бы как "
    "«ни у кого нет сессии» — сутками, при живом приложении и 200-х в ответах. Проверь "
    "AUTH_DB_* в backend/.env.runtime и пароль роли auth_app (data/sql/auth/002), либо "
    "верни AUTH_MODE=legacy."
 )
 # Текст для нечитаемого DSN. БЕЗ подстановки самого DSN — там пароль; исходную
 # ошибку SQLAlchemy (она цитирует строку целиком) гасим `from None`.
 _MALFORMED_DSN_MSG = (
    "DSN БД `auth` не разобрался SQLAlchemy. Проверь AUTH_DATABASE_URL (если задан "
    "явно) либо части AUTH_DB_HOST/AUTH_DB_PORT/AUTH_DB_NAME/AUTH_DB_USER. Схема "
    "обязана быть postgresql+psycopg:// (psycopg v3). Сам DSN сюда намеренно НЕ "
    "подставлен: в нём пароль роли auth_app."
 )
 # Кеш engine/factory + защита от гонки: rbac_guard будет резолвить сессию на каждом
 # non-public запросе, а uvicorn обслуживает их из нескольких потоков (sync-роуты
 # уходят в threadpool). Без лока два одновременных первых запроса создали бы два
 # engine — то есть два независимых пула коннектов, один из которых потеряется.
 _LOCK = threading.Lock()
 _engine: Engine | None = None
 _session_factory: sessionmaker[Session] | None = None
 def _build() -> tuple[Engine, sessionmaker[Session]]:
    """Создаёт engine + session-factory по текущему DSN. Нет DSN → явная ошибка.
    DSN резолвит `settings` (явный AUTH_DATABASE_URL или сборка из AUTH_DB_*) —
    здесь только «пусто или нет» и создание engine.
    `pool_size`/`max_overflow` не переопределяем: дефолтов SQLAlchemy (5+10) хватает
    с запасом — на запрос приходится один короткий SELECT, а раз в 5 минут ещё и
    UPDATE sliding-refresh.
    А вот таймауты переопределяем, и это не тюнинг, а требование: реестр — НЕ
    критический путь «Птицы», его сбой обязан деградировать за секунды, а не за
    минуты (в dual-режиме деградация — уход на легаси-заголовок, в db_only — 401).
      * `connect_timeout=3` (libpq, секунды). Без него дропнутые SYN (хост поднят, но
        недоступен по сети / фаервол молча глотает пакеты) держат попытку соединения
        до TCP-таймаута ОС — на Linux порядка 130 с. `pool_pre_ping=True` делает такую
        попытку на КАЖДОМ checkout'е.
      * `statement_timeout=3000` (мс, серверный). Ограничивает уже установленное
        соединение: залипший SELECT/UPDATE в auth-пути не имеет права висеть дольше.
      * `pool_timeout=3` — ожидание свободного коннекта в пуле. Дефолтные 30 с в
        auth-пути не нужны никогда: лучше быстро сдаться.
    Резолв сессии в rbac_guard уходит в threadpool (`run_in_threadpool`), так что эти
    ожидания не блокируют event loop, — но они всё равно держат worker-поток и время
    ответа, поэтому короткие.
    """
    dsn = settings.resolved_auth_database_url
    if not dsn:
        raise AuthDatabaseNotConfiguredError(_NOT_CONFIGURED_MSG)
    try:
        engine = create_engine(
            dsn,
            pool_pre_ping=True,
            future=True,
            pool_timeout=3,
            connect_args={"connect_timeout": 3, "options": "-c statement_timeout=3000"},
        )
    except (ArgumentError, ValueError):
        # ValueError — не паранойя: на «почти URL» разбор SQLAlchemy доходит до
        # `int(port)` и падает с `invalid literal for int() with base 10: 'w'`, где
        # 'w' — КУСОК ПАРОЛЯ, съехавший на позицию порта. `from None` обязателен: он
        # гасит цепочку, иначе исходная ошибка (а с ней и этот кусок) печатается в
        # traceback как «During handling of...».
        raise AuthDatabaseNotConfiguredError(_MALFORMED_DSN_MSG) from None
    factory = sessionmaker(autocommit=False, autoflush=False, bind=engine, expire_on_commit=False)
    return engine, factory
 def _ensure_built() -> tuple[Engine, sessionmaker[Session]]:
    global _engine, _session_factory
    # Быстрый путь читает глобалы РОВНО ОДИН раз, в локальные переменные. Читать их
    # второй раз в `return` нельзя: между проверкой и возвратом может вклиниться
    # `reset_auth_db()` (обнуляет оба под локом) — и функция вернула бы (None, None),
    # то есть вызывающий упал бы на `factory()` → `TypeError: 'NoneType' object is not
    # callable` прямо в auth-пути.
    engine, factory = _engine, _session_factory
    if engine is not None and factory is not None:
        return engine, factory
    with _LOCK:
        if _engine is None or _session_factory is None:
            _engine, _session_factory = _build()
        return _engine, _session_factory
 def get_auth_engine() -> Engine:
    """Engine БД `auth` (создаётся при первом вызове).
    Raises:
        AuthDatabaseNotConfiguredError: реестр не сконфигурирован (нет ни
            AUTH_DATABASE_URL, ни AUTH_DB_PASSWORD) либо DSN не разобрался.
    """
    engine, _ = _ensure_built()
    return engine
 def get_auth_session_factory() -> sessionmaker[Session]:
    """Session-factory БД `auth` (создаётся при первом вызове).
    Raises:
        AuthDatabaseNotConfiguredError: реестр не сконфигурирован (нет ни
            AUTH_DATABASE_URL, ни AUTH_DB_PASSWORD) либо DSN не разобрался.
    """
    _, factory = _ensure_built()
    return factory
@contextmanager
 def auth_session() -> Iterator[Session]:
    """Сессия к БД `auth`, закрывается на выходе из блока.
    Это НЕ `app.core.db.get_db`: там продуктовая БД gendesign, где таблиц
    `users`/`sessions` реестра нет. Прямой вызов из роутов не предполагается —
    ходи через `app.services.auth_session.resolve_session_token()`.
    """
    factory = get_auth_session_factory()
    with factory() as db:
        yield db
 def _probe_connection(engine: Engine) -> None:
    """Открывает соединение и делает `SELECT 1`. Вынесено функцией ради тестов.
    Отдельная функция, а не две строки в `require_auth_db_configured`: тестам нужна
    точка подмены, чтобы проверять ветвление старта, не поднимая Postgres.
    """
    with engine.connect() as conn:
        conn.execute(text("SELECT 1"))
 def require_auth_db_configured() -> None:
    """Fail-fast для старта приложения: включённый режим обязан иметь РАБОЧИЙ реестр.
    Вызывается из `lifespan` (`app/main.py:111`). Смысл проверки именно на старте: если
    сломанная конфигурация обнаружится только в rbac_guard, там её поймает общий
    `except` вокруг резолва сессии, и она будет выглядеть как «ни у кого нет сессии» —
    сутками, потому что продуктовая БД жива и приложение работоспособно, а сигнал
    остаётся только в логах. Дешевле не стартовать.
    Проверяется ИМЕННО СОЕДИНЕНИЕ, а не только синтаксис DSN. `create_engine` к серверу
    не ходит вовсе (пул ленивый), поэтому одна лишь сборка engine отлавливала бы ровно
    два случая — «DSN пуст» и «DSN не парсится», — а весь класс вероятных ошибок
    (неверный AUTH_DB_PASSWORD, опечатка в хосте, не созданная БД `auth`, отозванная
    роль auth_app, нет сетевой связности) проходил бы мимо и материализовался как та
    самая тихая деградация, ради которой эта функция и заведена. Проба короткая:
    `connect_timeout=3` в `_build`.
    Цена — контейнер не поднимется, пока БД `auth` недоступна. Это осознанно: реестр
    живёт на ТОМ ЖЕ сервере, что и продуктовая БД (сервис `postgres` корневого
    docker-compose.prod.yml, см. `app/core/config.py`), так что «реестр недоступен, а
    продукт работоспособен» — состояние вырожденное, а `restart: unless-stopped`
    поднимет контейнер, как только Postgres вернётся.
    Режим `legacy` (ДЕФОЛТ) → no-op: ни проверки DSN, ни создания engine, ни коннекта.
    Дефолтное поведение обязано оставаться ровно сегодняшним.
    Raises:
        AuthDatabaseNotConfiguredError: режим не `legacy`, но DSN пуст или не разобрался.
        AuthDatabaseUnreachableError: DSN разобрался, но соединиться не удалось.
    """
    if not settings.auth_session_enabled:
        return
    engine, _ = _ensure_built()
    try:
        _probe_connection(engine)
    except Exception as exc:
        # Исходную ошибку СОХРАНЯЕМ в цепочке (`from exc`): в ней хост/порт/роль и
        # причина отказа — то, ради чего проверка и делается. Пароля libpq в тексте
        # ошибок не печатает, а наш DSN сюда не подставляется (см. модульный докстринг).
        raise AuthDatabaseUnreachableError(_UNREACHABLE_MSG) from exc
 def reset_auth_db() -> None:
    """Сбрасывает закешированные engine/factory (смена DSN в рантайме, тесты).
    Старый engine `dispose()`-ится вне лока: закрытие пула может блокировать, а
    держать в это время лок незачем — ссылки на него уже сняты.
    """
    global _engine, _session_factory
    with _LOCK:
        stale = _engine
        _engine = None
        _session_factory = None
    if stale is not None:
        stale.dispose()
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@ -1,46 +1,10 @@
 import os
 import warnings
-from typing import Annotated, Literal
+from typing import Annotated
 from urllib.parse import quote
-from pydantic import SecretStr, field_validator, model_validator
+from pydantic import field_validator, model_validator
 from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
 # ── Дефолтные части DSN БД `auth` (общий реестр людей, эпик «единый вход») ─────
 # Вынесены константами, потому что используются ДВАЖДЫ: как дефолт поля и как
 # запасное значение, если переменная окружения задана ПУСТОЙ строкой
 # (`AUTH_DB_HOST=` в .env.runtime не должен давать DSN вида `...@:5432/auth`).
 #
 # ⚠️ ХОСТ — главная ловушка, и для «Птицы» она ЗЕРКАЛЬНА ловушке «Меры».
 # У «Меры» (tradein-mvp/backend/app/core/config.py:27) дефолт — `gendesign-postgres`,
 # потому что внутри ЕЁ стека имя `postgres` резолвится в её собственный контейнер
 # (tradein-mvp/docker-compose.prod.yml:143 собирает им продуктовый DATABASE_URL
 # `...@postgres:5432/tradein`), и БД `auth` там нет.
 #
 # У «Птицы» ровно наоборот: её стек и есть главный. Сервис `postgres` в корневом
 # docker-compose.prod.yml:22 (postgis/postgis:16-3.4) — это И ЕСТЬ тот сервер, где
 # живёт БД `auth`: bootstrap и миграции data/sql/auth/*.sql применяет к нему шаг
 # «Apply DB migrations» в .forgejo/workflows/deploy.yml:339-375. Соседи по тому же
 # compose-проекту так к нему и обращаются — `@postgres:5432` (docker-compose.prod.yml:232
 # и :265, DATABASE_URL сервисов glitchtip).
 #
 # Алиас `gendesign-postgres` (docker-compose.prod.yml:43-45) навешен ТОЛЬКО в внешней
 # сети `shared` (gendesign_shared) и заведён ради ЧУЖИХ стеков — им и пользуется
 # «Мера». Ставить его дефолтом здесь нельзя: в сети `shared` состоят лишь backend и
 # worker (`networks: [default, shared]`, строки 152 и 199), а `beat` (строки 201-217)
 # сетей не объявляет вовсе — он только в `default`, и `gendesign-postgres` из него
 # просто не разрезолвится. `postgres` резолвится из всех трёх.
 #
 # Порт 5432 — ВНУТРИСЕТЕВОЙ порт контейнера. Публикация `127.0.0.1:5432:5432`
 # (docker-compose.prod.yml:31-32) существует только ради SSH-туннеля с хоста и к
 # этому пути отношения не имеет.
 _AUTH_DB_DEFAULT_HOST = "postgres"
 _AUTH_DB_DEFAULT_PORT = 5432
 _AUTH_DB_DEFAULT_NAME = "auth"
 # Роль приложения из data/sql/auth/002_auth_app_role.sql (least privilege: SELECT/
 # INSERT/UPDATE/DELETE на sessions, SELECT + column-level UPDATE на users).
 _AUTH_DB_DEFAULT_USER = "auth_app"
 class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
@ -407,201 +371,5 @@ class Settings(BaseSettings):
    # на недоступном сервисе. ENV: DADATA_TIMEOUT_S.
    dadata_timeout_s: float = 8.0
    # ── Эпик «единый вход»: «Птица» ПРИНИМАЕТ сессию общего реестра ────────────
    # Форма входа во всём продукте одна и живёт у «Меры» (/trade-in/login): она
    # проверяет пароль и выдаёт сессию в auth.sessions. «Птица» сессии НЕ выдаёт и
    # НЕ отзывает — только читает куку и резолвит её в человека. Кука host-only на
    # gendsgn.ru с path="/" (tradein-mvp/backend/app/api/v1/auth.py:173-181),
    # поэтому браузер шлёт её на оба продукта одного домена.
    #
    # Режим — ТРЁХЗНАЧНЫЙ, а не булев флаг, и это сделано ради последнего PR эпика:
    #   legacy  (ДЕФОЛТ) — сегодняшнее поведение бит-в-бит: кука не читается вовсе,
    #                      личность берётся из X-Authenticated-User (Caddy basic_auth);
    #                      engine БД `auth` не создаётся, соединение не открывается,
    #                      отсутствие AUTH_* в окружении не роняет старт;
    #   dual            — сначала кука общего реестра, при её отсутствии/сбое реестра
    #                      деградация на легаси-заголовок (переходный режим: popup
    #                      Caddy ещё стоит и прикрывает заголовок от подделки);
    #   db_only         — легаси-ветка НЕДОСТИЖИМА: нет валидной сессии → 401, даже
    #                      если X-Authenticated-User присутствует.
    #
    # Почему именно так, а не `AUTH_SESSION_ENABLED=true/false`. В dual-режиме сбой
    # реестра (или просто отсутствие куки) уводит запрос на trusted-header. Пока
    # popup стоит, это безопасно: заголовок на `/api/*` перезаписывает Caddy из
    # basic_auth (Caddyfile:178-182), клиент подставить его не может. Ровно в тот
    # момент, когда последний PR эпика снимет `basic_auth` + `header_up`, заголовок
    # станет полностью клиентским — и та же деградация превратится в ПОЛНЫЙ обход
    # аутентификации (`curl -H 'X-Authenticated-User: admin'`). Булев флаг оставлял бы
    # это на память мейнтейнера («не забыть выпилить фолбэк»); режим делает переход
    # сменой ОДНОГО значения (`AUTH_MODE=db_only`), а недостижимость легаси-ветки в
    # нём закреплена тестами (tests/test_auth_session_guard.py, секция db_only).
    # Зеркало «Меры»: tradein-mvp/backend/app/core/config.py:91 (`auth_mode`); там
    # значений два — легаси-режима у неё уже нет, она на реестре с #2552.
    #
    # ⚠️ ДЕФОЛТ `legacy` — ЧАСТЬ КОНТРАКТА PR, А НЕ ЗАГЛУШКА: после мержа прод обязан
    # работать ровно как сегодня (popup Caddy снимается последним PR эпика).
    # Читатели режима: `app.main.rbac_guard` (какой источник личности и есть ли
    # фолбэк), `app.services.auth_session.resolve_session_token` и
    # `app.core.auth_db.require_auth_db_configured` — через производное свойство
    # `auth_session_enabled` ниже.
    #
    # Включение на проде = одна переменная: AUTH_DB_PASSWORD в backend/.env.runtime
    # уже есть (её пишет ops и читает .forgejo/workflows/deploy.yml:381-386, чтобы
    # сделать ALTER ROLE auth_app), остальные части DSN имеют прод-дефолты.
    # ENV: AUTH_MODE.
    auth_mode: Literal["legacy", "dual", "db_only"] = "legacy"
    # DSN БД `auth` целиком. Пусто по умолчанию — задавать руками не обязательно:
    # см. `resolved_auth_database_url` ниже, при пустом значении DSN собирается из
    # AUTH_DB_PASSWORD + частей. Явное значение, если оно есть, выигрывает всегда
    # (аварийный обход: другой хост, sslmode, байпас пула). ENV: AUTH_DATABASE_URL.
    auth_database_url: str = ""
    # Пароль роли auth_app. Живёт в ОДНОМ месте — этой переменной: требовать вдобавок
    # целиковый AUTH_DATABASE_URL значило бы держать один секрет в двух местах
    # (сменили пароль роли, забыли переписать DSN → вход ложится молча и целиком).
    #
    # SecretStr, а не str как у соседних секретов файла: `repr(settings)` и
    # `settings.model_dump()` печатают обычные str-поля ДОСЛОВНО. Сегодня их никто не
    # рендерит, но появиться такой рендер может тихо — с SecretStr он напечатает
    # `SecretStr('**********')`. Значение достаётся ровно в одном месте —
    # `.get_secret_value()` в резолвере ниже. Соседи (openai_api_key, dadata_api_secret,
    # database_url) остались str — это предсуществующее положение, а не «там безопасно».
    # ENV: AUTH_DB_PASSWORD.
    auth_db_password: SecretStr = SecretStr("")
    # Остальные части — с дефолтами, верными для ЭТОГО стека (см. константы выше и
    # разбор ловушки хоста). Переопределяются через ENV для локального запуска (напр.
    # AUTH_DB_HOST=localhost + AUTH_DB_PORT=15432 поверх SSH-туннеля).
    # ENV: AUTH_DB_HOST, AUTH_DB_PORT, AUTH_DB_NAME, AUTH_DB_USER.
    auth_db_host: str = _AUTH_DB_DEFAULT_HOST
    auth_db_port: int = _AUTH_DB_DEFAULT_PORT
    auth_db_name: str = _AUTH_DB_DEFAULT_NAME
    auth_db_user: str = _AUTH_DB_DEFAULT_USER
    # Имя cookie сессии. ОБЯЗАНО совпадать с тем, которым пользуется «Мера»
    # (tradein-mvp/backend/app/core/config.py:84-86) — иначе браузер шлёт куку, а
    # «Птица» её не узнаёт и молча остаётся без сессии.
    #
    # ⚠️ Имя ИСТОРИЧЕСКОЕ: оно родилось в trade-in до того, как реестр стал общим, и
    # «tradein_» в нём теперь ни о чём не говорит. Переименование разлогинивает ВСЕХ
    # и СРАЗУ в обоих продуктах (старую куку никто больше не читает), поэтому меняется
    # только отдельным решением — синхронно в обоих стеках и с обдуманным моментом.
    # ENV: SESSION_COOKIE_NAME.
    session_cookie_name: str = "tradein_session"
    # TTL сессии в часах (720 = 30 дней) — тот же дефолт, что у «Меры»
    # (tradein-mvp/backend/app/core/config.py:88). «Птица» сессии не выдаёт, поэтому
    # значение используется ЕДИНСТВЕННЫМ образом: на сколько sliding-refresh отодвигает
    # expires_at (app/services/auth_session.py). Держать его РАВНЫМ значению «Меры»
    # обязательно — иначе срок жизни сессии начнёт зависеть от того, в каком продукте
    # человек кликнул последним. ENV: SESSION_TTL_HOURS.
    session_ttl_hours: int = 720
    @field_validator("auth_mode", mode="before")
    @classmethod
    def _blank_auth_mode_means_legacy(cls, value: object) -> object:
        """`AUTH_MODE=` (пустая строка) → `legacy`, а не ValidationError на импорте.
        Та же ловушка, что у `AUTH_DB_PORT` ниже: `settings = Settings()` выполняется на
        уровне модуля, поэтому невалидное значение роняет ИМПОРТ конфига и уводит
        контейнер в restart-loop. Сценарий тот же — ops копирует блок AUTH_* в
        .env.runtime и заполняет только пароль. Пустое значение обязано означать
        «оставили как было», то есть сегодняшнее поведение.
        Регистр и обрамляющие пробелы нормализуются: `AUTH_MODE=DB_ONLY ` — очевидная
        опечатка со смыслом, а не запрос на падение. Непустой мусор (`AUTH_MODE=off`)
        по-прежнему валится, и правильно: молча трактовать его как `legacy` значило бы
        тихо оставить продукт на trusted-header после снятия popup'а.
        """
        if isinstance(value, str):
            normalized = value.strip().lower()
            return normalized or "legacy"
        return value
    @property
    def auth_session_enabled(self) -> bool:
        """Читает ли «Птица» сессионную куку общего реестра (то есть режим не `legacy`).
        Производное от `auth_mode`, а не отдельное поле: два независимых переключателя
        рано или поздно разъезжаются, и получилось бы состояние «куку читаем, но режим
        легаси» (или наоборот), которого нет ни в одном настоящем сценарии.
        Держит инвариант «`legacy` = ни одного коннекта к реестру»: по этому свойству
        закорачиваются `app.services.auth_session.resolve_session_token` и
        `app.core.auth_db.require_auth_db_configured`. Разница между `dual` и `db_only`
        свойству не видна и не должна быть — она касается только фолбэка на
        легаси-заголовок и живёт в `app.main.rbac_guard`.
        """
        return self.auth_mode != "legacy"
    @field_validator("auth_db_port", mode="before")
    @classmethod
    def _blank_auth_db_port_means_default(cls, value: object) -> object:
        """`AUTH_DB_PORT=` (пустая строка) → прод-дефолт, а не падение на импорте.
        Симметрия с host/name/user, у которых пустое значение переменной падает
        обратно на дефолт в резолвере. Для порта того же добиться нельзя: он
        типизирован `int` и валидируется pydantic'ом ДО всякой нашей логики, а
        `settings = Settings()` выполняется на уровне модуля — то есть `AUTH_DB_PORT=`
        в .env.runtime роняло бы ValidationError на импорте конфига и уводило контейнер
        в restart-loop. Причём В ЛЮБОМ режиме, включая дефолтный (флаг выключен), где к
        БД `auth` не идёт ни одного обращения — ровно тот инвариант «дефолт не трогаем»,
        который держит весь этот PR.
        Сценарий не гипотетический: ops копирует блок AUTH_DB_* в .env.runtime и
        заполняет только пароль — остальные строки остаются пустыми намеренно.
        `mode="before"` — потому что вмешаться надо ДО приведения к int. Непустой мусор
        (`AUTH_DB_PORT=abc`) по-прежнему валится, и правильно: это опечатка со смыслом,
        а не «оставил пустым».
        """
        if isinstance(value, str) and not value.strip():
            return _AUTH_DB_DEFAULT_PORT
        return value
    @property
    def resolved_auth_database_url(self) -> str:
        """DSN БД `auth` — единственный источник правды для `app.core.auth_db`.
        Приоритет:
          1. `AUTH_DATABASE_URL`, если задан — выигрывает всегда.
          2. Иначе, если задан `AUTH_DB_PASSWORD` — DSN собирается из частей.
          3. Иначе — пустая строка, то есть «не сконфигурировано». Это НЕ ошибка сама
             по себе: при `AUTH_MODE=legacy` (дефолт) сюда не заходит никто.
             Ошибку — явную, а не тихий фолбэк — поднимает `app.core.auth_db`, и только
             когда реестр реально понадобился.
        ⚠️ Возвращаемое значение СОДЕРЖИТ ПАРОЛЬ: не логировать, не класть в текст
        исключений, не отдавать наружу (`/health`, `/docs`, метрики).
        Пароль экранируется `quote(..., safe="")`: спецсимвол (`@`, `:`, `/`, `?`, `#`,
        `%`) внутри пароля иначе порвал бы URL по своей грамматике — `@` сдвинул бы
        границу host, `/` открыл бы path. Разбор дал бы либо ошибку, либо, что хуже,
        МОЛЧА другой хост/базу. По той же причине экранируется имя пользователя.
        А вот имя БД и хост — НЕ экранируются, и это не забывчивость: SQLAlchemy
        раскодирует обратно только userinfo (user/password), а path отдаёт как есть.
        Прогони мы имя БД через `quote`, в сервер уехало бы литеральное `c%2Fd` вместо
        `c/d`. Хосту %-кодирование тоже только мешает — оно поломало бы IPv6-скобки.
        """
        explicit = self.auth_database_url.strip()
        if explicit:
            return explicit
        # `.strip()` только для ПРОВЕРКИ «задан ли»: пробельная строка в .env — это
        # опечатка, а не пароль. В сам DSN идёт значение КАК ЕСТЬ (не стриппится):
        # ведущий/хвостовой пробел может быть частью настоящего пароля.
        password = self.auth_db_password.get_secret_value()
        if not password.strip():
            return ""
        user = quote(self.auth_db_user.strip() or _AUTH_DB_DEFAULT_USER, safe="")
        secret = quote(password, safe="")
        host = self.auth_db_host.strip() or _AUTH_DB_DEFAULT_HOST
        port = self.auth_db_port
        name = self.auth_db_name.strip() or _AUTH_DB_DEFAULT_NAME
        # Схема — ровно та же, что у продуктового database_url (psycopg v3;
        # `postgresql://` без суффикса увёл бы SQLAlchemy на psycopg2, которого в
        # зависимостях нет).
        return f"postgresql+psycopg://{user}:{secret}@{host}:{port}/{name}"
 settings = Settings()
--- a/backend/app/main.py
+++ b/backend/app/main.py
@ -3,14 +3,11 @@
 import logging
 import os
 import re
 import threading
 import time
 from collections.abc import AsyncIterator, Awaitable, Callable
 from contextlib import asynccontextmanager
 import sentry_sdk
 from fastapi import FastAPI, Request
 from fastapi.concurrency import run_in_threadpool
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, Response
 from sentry_sdk.integrations.celery import CeleryIntegration
@ -44,12 +41,10 @@ from app.api.v1 import (
    trade_in,
    users,
 )
 from app.core import auth_db
 from app.core.audit_middleware import audit_log_middleware
 from app.core.auth import get_role
 from app.core.config import settings
 from app.observability.sentry_scrub import scrub_sensitive_query
 from app.services.auth_session import resolve_session_token
 logger = logging.getLogger(__name__)
@ -102,18 +97,6 @@ if settings.glitchtip_dsn:
@asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    # Эпик «единый вход», fail-fast: AUTH_MODE=dual|db_only обязан иметь РАБОЧИЙ
    # реестр — проверяется не только разбор DSN, но и живое соединение (`SELECT 1`,
    # app/core/auth_db.py). Не соединились → контейнер НЕ стартует. Режим `legacy`
    # (ДЕФОЛТ) → no-op: ни проверки DSN, ни создания engine, ни коннекта.
    #
    # Почему именно на старте, а не «разберёмся в рантайме»: неверный пароль, опечатка
    # в хосте, не созданная БД `auth` иначе ловились бы `except`'ом вокруг резолва
    # сессии в rbac_guard, и сломанная конфигурация выглядела бы как «ни у кого нет
    # сессии» — СУТКАМИ, потому что продуктовая БД жива, приложение отвечает 200, а
    # сигнал остаётся только в логах. Дешевле не стартовать: деплой падает сразу и
    # громко.
    auth_db.require_auth_db_configured()
    yield
@ -139,212 +122,10 @@ app.middleware("http")(audit_log_middleware)
 #   3) /api/v1/admin/* — только role=admin, иначе 403.
 # Public paths без auth (/health, /docs, /openapi.json) пропускаем без проверки —
 # X-Authenticated-User там просто не приходит из Caddy.
 #
 # Эпик «единый вход»: к правилу 1 добавляется ПЕРВЫЙ источник личности —
 # сессионная кука общего реестра (БД `auth`). Выдаёт её единственная форма входа, у
 # «Меры» (/trade-in/login); «Птица» сессии только читает. Кука host-only на
 # gendsgn.ru с path="/" → браузер шлёт её и сюда. Порядок: кука → легаси-заголовок.
 # Дальше — ВСЁ как раньше: роль из auth/roles.yaml, admin-гейт по _ADMIN_API_RE.
 # Реестр отвечает на вопрос «кто ты», roles.yaml — «что тебе можно»; продуктовые
 # роли реестра (auth.users.role) в «Птицу» намеренно не протаскиваются.
 #
 # ⚠️ AUTH_MODE=legacy ПО УМОЛЧАНИЮ — popup Caddy basic_auth ещё стоит и снимается
 # ПОСЛЕДНИМ PR эпика. Пока режим legacy, этот файл ведёт себя бит-в-бит как до эпика:
 # кука не читается, БД `auth` не открывается. `dual` — переходный режим (кука, при её
 # отсутствии/сбое реестра фолбэк на заголовок), `db_only` — фолбэка нет вовсе.
 #
 # ⚠️ ДОЛГ, КОТОРЫЙ ОБЯЗАН БЫТЬ ЗАКРЫТ ДО СНЯТИЯ POPUP'А (не решается этим PR).
 # Guard проверяет ровно две вещи: есть ли username в auth/roles.yaml (get_role) и
 # admin-гейт по _ADMIN_API_RE. Списки `paths`/`deny` из roles.yaml на бэкенде НЕ
 # применяются — это зафиксировано в самом auth/roles.yaml:33-35 («path-level
 # enforcement делает frontend RouteGuard»). Следствие: в момент включения режима
 # «Птицу» получает КАЖДЫЙ аккаунт реестра, чей username совпадает с записью в
 # roles.yaml, — включая роль `expired` (user2: paths: [], deny: "/**"), которую
 # сегодня останавливает только фронт. Это не регрессия (те же люди сегодня в
 # caddy/users.caddy.snippet и добираются туда же через basic_auth), но эпик делает
 # её несущей: (а) до снятия popup'а отзыв доступа имеет ДВА рубильника —
 # caddy-snippet и access_state в реестре, их надо держать синхронными; (б) после
 # снятия roles.yaml остаётся ЕДИНСТВЕННЫМ гейтом, и `expired` в нём станет чисто
 # фронтовой фикцией. Перед включением: сверить `auth.users.username` на проде с
 # `users:` в roles.yaml и решить — применять `paths`/`deny` на бэкенде или убрать
 # `expired` как вводящий в заблуждение.
 _ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
 _PUBLIC_PATHS = frozenset({"/health", "/api/v1/ping", "/docs", "/redoc", "/openapi.json"})
 def _propagate_authenticated_user(request: Request, username: str) -> None:
    """Инжектит `X-Authenticated-User` в ASGI-scope — ПЕРЕЗАПИСЫВАЯ, а не дополняя.
    🔴 Перезапись, а не «поставить, если отсутствует» — это требование безопасности,
    а не стилистика. В бэкенде «Птицы» ОДИННАДЦАТЬ мест читают этот заголовок НАПРЯМУЮ,
    мимо guard'а, и решают по нему, кто автор/кому принадлежат данные:
      • app/core/audit_middleware.py:169     — атрибуция строки аудита;
      • app/api/v1/me.py:30                  — чей scope отдать (роль + фильтры);
      • app/api/v1/insights.py:74/124/138    — created_by + _require_user (POST/PUT/DELETE);
      • app/api/v1/own_projects.py:69/115/131 — created_by + _require_user (POST/PUT/DELETE);
      • app/api/v1/parcels.py:1481           — GET /{cad_num}/forecast;
      • app/api/v1/parcels.py:1902           — POST /{cad_num}/analyze (created_by рана,
        parcels.py:4212, и 3-й аргумент forecast_site_finder_report.delay, :4226);
      • сам rbac_guard ниже                  — легаси-ветка.
    Ни одно из них не знает про сессию: для них истина — сырой заголовок. Оставь мы
    skip-if-present — клиент с ВАЛИДНОЙ кукой прошёл бы guard как он сам, а во все эти
    места уехал бы его собственный подставленный `X-Authenticated-User: <кто угодно>`
    (Caddy шлёт этот заголовок на каждый прод-запрос, так что «просто добавить» его
    было бы некуда). Ровно этот баг ловили у «Меры» — #2552 post-review, CRITICAL.
    Резолвнутая сессия ОБЯЗАНА быть единственным источником личности.
    Механизм: `request.scope` — один и тот же dict, прокинутый ПО ССЫЛКЕ через весь
    ASGI-стек (Starlette не копирует scope между слоями). Мутация здесь видна:
      • всей downstream-цепочке — мы мутируем ДО вызова call_next();
      • audit-middleware — он ВНУТРЕННИЙ относительно rbac_guard (см. комментарий у
        app.middleware("http")(audit_log_middleware) выше: LIFO-регистрация даёт
        порядок rbac_guard → audit → router), т.е. его Request строится уже после
        мутации. У «Меры» этот слой, наоборот, внешний, и там мутация до него
        доезжает только потому, что читается ПОСЛЕ call_next.
    Имена заголовков в ASGI — по спеке всегда lowercase bytes, и uvicorn/TestClient
    её соблюдают. Фильтр всё равно нормализует ключ сам (`k.lower()`), а не полагается
    на спеку: попади в scope запись `b"X-Authenticated-User"` (другой ASGI-сервер,
    самодельный слой, тест-харнесс) — точное сравнение оставило бы её в списке рядом с
    нашей. Читатели при этом видели бы правильное значение (`Headers.get` лоуэркейсит
    искомый ключ, но не хранимый, так что смешанный регистр не матчится никогда), то
    есть дыры нет — но состояние «две записи с одним именем» в scope не должно
    существовать: оно ложное по построению и ломает любой обход списка глазами.
    `errors="replace"` в encode: латиницей логины реестра не ограничены, а падать
    UnicodeEncodeError в auth-пути нельзя.
    NB: `request.headers` САМОГО этого Request уже закеширован (мы читали cookies) и
    останется старым. Это не мешает: в session-ветке guard больше не читает заголовок,
    а нижележащие слои строят свой Request поверх обновлённого scope.
    """
    request.scope["headers"] = [
        (k, v) for k, v in request.scope.get("headers", []) if k.lower() != b"x-authenticated-user"
    ] + [(b"x-authenticated-user", username.encode("latin-1", "replace"))]
 # Троттлинг алерта «реестр не отвечает». Резолв сессии идёт на КАЖДОМ non-public
 # запросе с кукой, а `logger.exception` уровня ERROR уезжает событием в GlitchTip
 # (LoggingIntegration event_level=ERROR, см. sentry_sdk.init выше) — то есть лежащий
 # реестр давал бы поток событий, пропорциональный трафику: квота/rate-limit выгорают
 # за минуты, и настоящие ошибки этого же периода теряются. Полный traceback печатаем
 # не чаще раза в минуту (с числом подавленных за окно), остальное — WARNING без
 # exc_info, чтобы факт продолжающегося сбоя всё равно был виден в логах.
 # Лок нужен по-настоящему: функция исполняется в threadpool'е, то есть параллельно.
 _REGISTRY_FAILURE_ALERT_INTERVAL_S = 60.0
 _REGISTRY_FAILURE_LOCK = threading.Lock()
 _registry_failure_last_alert = 0.0
 _registry_failure_suppressed = 0
 def _reset_registry_failure_throttle() -> None:
    """Сбрасывает окно троттлинга. Для тестов: состояние модульное и живёт между ними."""
    global _registry_failure_last_alert, _registry_failure_suppressed
    with _REGISTRY_FAILURE_LOCK:
        _registry_failure_last_alert = 0.0
        _registry_failure_suppressed = 0
 def _log_registry_failure(path: str) -> None:
    """Логирует сбой резолва: раз в окно — ERROR с traceback, иначе WARNING.
    Зовётся ТОЛЬКО из `except`-блока: `logger.exception` берёт traceback из текущего
    sys.exc_info().
    """
    global _registry_failure_last_alert, _registry_failure_suppressed
    now = time.monotonic()
    with _REGISTRY_FAILURE_LOCK:
        alert = (now - _registry_failure_last_alert) >= _REGISTRY_FAILURE_ALERT_INTERVAL_S
        if alert:
            suppressed = _registry_failure_suppressed
            _registry_failure_last_alert = now
            _registry_failure_suppressed = 0
        else:
            suppressed = 0
            _registry_failure_suppressed += 1
    if alert:
        logger.exception(
            "RBAC: резолв сессии не удался на %s — эти запросы обслуживаются по "
            "легаси-пути (Caddy basic_auth + X-Authenticated-User); подавлено таких же "
            "за предыдущее окно: %d",
            path,
            suppressed,
        )
    else:
        logger.warning(
            "RBAC: резолв сессии не удался на %s (traceback подавлен троттлингом, "
            "следующий — не раньше чем через %.0f с)",
            path,
            _REGISTRY_FAILURE_ALERT_INTERVAL_S,
        )
 def _resolve_session_username(token: str | None, path: str) -> str | None:
    """Логин из сессионной куки, либо None, если личность по куке не установлена.
    🔴 СИНХРОННАЯ и вызывается ТОЛЬКО через `run_in_threadpool` (см. rbac_guard):
    внутри — psycopg-I/O (checkout из пула + SELECT, раз в 5 минут ещё UPDATE и
    commit). Позови её напрямую из корутины guard'а — и весь API «Птицы»
    сериализуется за один round-trip к БД `auth` на каждый запрос, а недоступный
    реестр (или исчерпанный пул) заморозит event loop целиком, включая /health. Ровно
    этот инцидент уже был на соседнем middleware — #1202, см. комментарий в
    app/core/audit_middleware.py:175-181, там он и починен через `run_in_threadpool`.
    Токен принимается ГОТОВЫМ (а не `Request`) именно поэтому: разбор Cookie-заголовка
    дёшев и делается на loop'е, в поток уезжает только строка.
    None означает ровно одно — «личность по куке не установлена», и вызывающий обязан
    трактовать это одинаково во всех трёх случаях: куки нет, кука невалидна (нет
    строки / истекла / access_state не active), резолв УПАЛ.
    Поведение при сбое БД `auth` (осознанный выбор, а не «поймали и забыли»): логируем
    ERROR с traceback — он уезжает событием в GlitchTip (LoggingIntegration
    event_level=ERROR, см. sentry_sdk.init выше), т.е. это алерт, а не строчка, которую
    никто не увидит (частота ограничена окном, `_log_registry_failure`), — и в режиме
    `dual` деградируем к легаси-ветке, то есть к сегодняшнему поведению: Caddy
    basic_auth + X-Authenticated-User. В режиме `db_only` деградации нет: guard
    отвечает 401.
    Почему НЕ 503/500. Пока идёт переходный период, popup basic_auth стоит перед
    бэкендом, и легаси-ветка защищена ровно тем же, чем защищён весь продукт сегодня, —
    множество людей, способных вообще достучаться, не расширяется. Отдавать же 503
    значит класть «Птицу» целиком из-за проблемы, которую basic_auth уже покрывает
    (отозванный пароль роли auth_app, пересозданная БД `auth`, исчерпанный пул её
    engine — всё это не мешает продуктовой БД gendesign работать).
    Почему это не «тихий фолбэк на легаси». Опасный сценарий — не «реестр упал», а
    «реестр не сконфигурирован»: тогда права раздавались бы из roles.yaml в обход
    реестра (включая аккаунты с access_state disabled/trial_expired) бессрочно и молча.
    Этот сценарий сюда НЕ доходит: конфигурацию проверяет lifespan, причём НЕ на глазок —
    `require_auth_db_configured` открывает соединение и делает `SELECT 1`, так что мимо
    него не проходят ни пустой/битый DSN, ни неверный пароль, ни опечатка в хосте, ни
    отозванная роль (app/core/auth_db.py). Здесь остаётся только второй рубеж — реестр,
    отвалившийся ПОСЛЕ успешного старта.
    ⚠️ Отдельно про отзыв доступа: пароли Caddy basic_auth (caddy/users.caddy.snippet)
    и `auth.users.access_state` — РАЗНЫЕ списки. Человек, которому в реестре поставили
    disabled/trial_expired, свой basic_auth-пароль не теряет, поэтому на время
    недоступности реестра деградация возвращает его в строй. То есть отзыв тут не
    «строже сегодняшнего», а откатывается к состоянию ДО отзыва — при включении режима
    caddy-snippet надо прополоть под список активных аккаунтов реестра.
    ⚠️ Когда последний PR эпика снимет popup, эта деградация обязана уйти вместе с ним:
    без basic_auth впереди фолбэк на легаси-заголовок превращается в дыру — заголовок
    станет полностью клиентским. Механика перехода уже готова: `AUTH_MODE=db_only`
    (см. app/core/config.py), в нём легаси-ветка недостижима и этот возврат None
    означает 401, а не «попробуем заголовок».
    """
    if not token:
        # Нет куки — ни одного обращения к БД `auth`. Это весь сегодняшний трафик.
        return None
    try:
        session_user = resolve_session_token(token)
    except Exception:
        _log_registry_failure(path)
        return None
    if session_user is None:
        return None
    return session_user.username
@app.middleware("http")
 async def rbac_guard(
    request: Request,
@ -353,17 +134,6 @@ async def rbac_guard(
    # Test-mode bypass: pytest бьёт по app мимо Caddy → нет X-Authenticated-User.
    # СТРОГО gated на settings.testing (default False) — прод RBAC не затронут.
    # RBAC-логика покрыта отдельно в tests/test_rbac.py (своя копия middleware).
    #
    # ⚠️ Он ОТКЛЮЧАЕТ ВЕСЬ guard целиком, включая session-ветку ниже, — и это сказано
    # здесь явно, чтобы не выглядело недосмотром. Следствие для тестов: сессионный путь
    # НЕЛЬЗЯ проверять запросом к настоящему `app` через TestClient (conftest ставит
    # settings.testing=True глобально, guard просто не отработает, тест «прошёл бы» ни о
    # чём). Он и проверяется иначе: tests/test_auth_session_guard.py зовёт ЭТУ САМУЮ
    # функцию напрямую, сняв settings.testing через monkeypatch, — то есть прод-код, а
    # не копию. Копия guard'а в tests/test_rbac.py про куку намеренно НЕ знает и
    # покрывает только режим legacy (там об этом написано). Сдвигать session-ветку ВЫШЕ
    # bypass'а нельзя: получился бы полуработающий guard (личность резолвится, а 401/403
    # не применяются) — состояние, которого нет ни в одном настоящем режиме.
    if settings.testing:
        return await call_next(request)
@ -371,67 +141,22 @@ async def rbac_guard(
    if path in _PUBLIC_PATHS:
        return await call_next(request)
-    # Внешний `if` по режиму — не дубль проверки внутри resolve_session_token(), а
+    username = request.headers.get("X-Authenticated-User")
-    # гарантия инварианта «legacy = поведение не меняется ни на байт»: в нём не
+    if not username:
-    # трогается даже request.cookies (разбор Cookie-заголовка).
+        # Любой non-public path без auth-header → 401. Локальный curl мимо Caddy
-    token = (
+        # или прокси-фронт без header_up. 401 точнее чем 403 — "сначала
-        request.cookies.get(settings.session_cookie_name) if settings.auth_session_enabled else None
+        # аутентифицируйся".
    )
    # 🔴 Резолв — В THREADPOOL. Внутри синхронный psycopg-I/O, а мы в корутине: прямой
    # вызов блокировал бы event loop на каждом запросе с кукой (инцидент #1202, тот же
    # класс, что чинили в app/core/audit_middleware.py:175-183). `if token` перед
    # хопом — не микрооптимизация: без куки резолвить нечего, и весь сегодняшний
    # трафик не платит ни за поток, ни за коннект.
    session_username = (
        await run_in_threadpool(_resolve_session_username, token, path) if token else None
    )
    if session_username is not None:
        username = session_username
        # 🔴 До call_next и до всего остального: личность из сессии обязана вытеснить
        # клиентский заголовок для одиннадцати прямых читателей (см. функцию).
        _propagate_authenticated_user(request, username)
    elif settings.auth_mode == "db_only":
        # Легаси-ветка ОТКЛЮЧЕНА: нет валидной сессии → отказ, даже если
        # X-Authenticated-User присутствует. Это конечное состояние эпика — режим
        # включается тем же PR, который снимает `basic_auth` + `header_up` из Caddy и
        # тем самым делает заголовок полностью клиентским. Отдельный текст ответа:
        # «no authenticated user» ниже говорит про basic_auth, которого в этот момент
        # уже нет.
        return JSONResponse(
            status_code=401,
-            content={"detail": "valid session required"},
+            content={"detail": "no authenticated user (Caddy basic_auth required)"},
        )
    else:
        # ---- легаси trusted-header путь — БИТ-В-БИТ как до эпика ----
        header_user = request.headers.get("X-Authenticated-User")
        if not header_user:
            # Любой non-public path без auth-header → 401. Локальный curl мимо Caddy
            # или прокси-фронт без header_up. 401 точнее чем 403 — "сначала
            # аутентифицируйся".
            return JSONResponse(
                status_code=401,
                content={"detail": "no authenticated user (Caddy basic_auth required)"},
            )
        username = header_user
    try:
        role = get_role(username)
    except KeyError:
        # Юзер в Caddy basic_auth, но не в roles.yaml → 403 на ВСЁ.
        # Decided 2026-05-25: «человек без ролей вообще ничего не видит».
-        if session_username is not None:
+        logger.warning("RBAC: unknown user %r tried %s", username, path)
            # Тот же отказ, но отдельным сообщением: «есть в реестре, нет в roles.yaml» —
            # это рассинхрон двух списков (типовой при заведении нового аккаунта), а не
            # подделка заголовка, и чинится он в другом месте.
            logger.warning(
                "RBAC: сессия резолвлена в %r, но юзера нет в auth/roles.yaml — отказ на %s",
                username,
                path,
            )
        else:
            logger.warning("RBAC: unknown user %r tried %s", username, path)
        return JSONResponse(
            status_code=403,
            content={"detail": "user not in roles config"},
--- a/backend/app/services/auth_session.py
+++ b/backend/app/services/auth_session.py
@ -1,256 +0,0 @@
 """Резолв сессионной куки общего реестра (БД `auth`) — сторона «Птицы».
 Эпик «единый вход»: вместо браузерного popup'а Caddy basic_auth у продукта одна
 нейтральная форма входа. Живёт она у «Меры» (`/trade-in/login`): та проверяет
 пароль, пишет строку в `auth.sessions` и ставит куку host-only на gendsgn.ru с
 `path="/"` — поэтому браузер шлёт её и на `/site-finder/**` тоже.
 «Птица» эту куку ТОЛЬКО ЧИТАЕТ. Здесь нет и не должно появиться `create_session` /
 `revoke_session`: выдача и отзыв — исключительная ответственность единственной
 формы входа, второй эмитент сессий означал бы два места, где решается «кого
 пускать», и расходящиеся правила блокировки.
 Что модуль отдаёт вызывающему: `resolve_session_token(token)` → `SessionUser`
 (username + состояние доступа) либо None. Что делать с username дальше — дело
 guard'а: авторизация «Птицы» (какие пути кому видны) по-прежнему живёт в
 `auth/roles.yaml` (`app.core.auth.get_role`), продуктовые роли реестра
 (`auth.users.role` — admin/manager/employee, миграция data/sql/auth/004) сюда
 намеренно НЕ протаскиваются: это другая ролевая модель, и её отображение на
 roles.yaml — отдельное решение стадии 2, а не побочный эффект резолва сессии.
 Токены опаковые (`secrets.token_urlsafe` на стороне «Меры») — не JWT, не подписаны:
 валидность проверяется исключительно наличием строки в БД + `expires_at` +
 состоянием доступа юзера. Никакого разделяемого секрета между стеками для этого
 не нужно — только доступ к одной БД.
 Имена таблиц (`users`, `sessions`) и колонок — литералы из data/sql/auth/001 и 004;
 снаружи в SQL-строку не попадает ничего, значения идут bind-параметрами.
 Зеркало по подходу: tradein-mvp/backend/app/services/auth_session.py («Мера»). Там
 модуль дополнительно умеет две схемы (переходный `identity_store`) и выдачу сессий —
 здесь этого нет за ненадобностью.
 """
 from __future__ import annotations
 import logging
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from enum import StrEnum
 from sqlalchemy import text
 from sqlalchemy.orm import Session
 from app.core import auth_db
 from app.core.config import settings
 logger = logging.getLogger(__name__)
 # Sliding-window refresh: last_seen_at/expires_at продлеваются НЕ чаще раза в 5
 # минут — иначе каждый API-запрос авторизованного юзера бил бы в БД лишним UPDATE
 # (guard резолвит сессию на КАЖДЫЙ non-public запрос). Значение и механика — те же,
 # что у «Меры» (tradein-mvp/.../auth_session.py:51): сессия общая, и продлевать её
 # два продукта обязаны одинаково.
 _SLIDING_REFRESH_INTERVAL = timedelta(minutes=5)
 class AccessState(StrEnum):
    """Состояние доступа аккаунта — значения дословно из `auth.users.access_state`.
    CHECK-констрейнт `users_access_state_ck`, миграция data/sql/auth/004; семантика
    оттуда же (решение владельца от 2026-07-31):
      active        — доступ есть;
      trial_expired — пароль верный, но пробный период истёк;
      disabled      — доступ закрыт владельцем.
    Для «Птицы» все три состояния делятся надвое (`can_sign_in`): отдельный экран
    «пробный доступ закончился» — сюжет формы входа, то есть «Меры»; сюда приходит
    уже вошедший человек, и всё, что не `active`, для него значит одно — сессии нет.
    """
    ACTIVE = "active"
    TRIAL_EXPIRED = "trial_expired"
    DISABLED = "disabled"
    @property
    def can_sign_in(self) -> bool:
        """True только для `active` — единственная проверка «пускать ли».
        Вынесена в свойство, чтобы вызывающий не писал `state == "active"`: добавится
        четвёртое состояние — оно по умолчанию окажется «не пускать», а не «пускать,
        потому что не disabled».
        """
        return self is AccessState.ACTIVE
 def to_access_state(value: object) -> AccessState:
    """Приводит значение колонки `users.access_state` к `AccessState`.
    Fail-closed: неизвестная строка, NULL и любой неожиданный тип → `disabled` +
    WARNING. Обратный выбор (пускать всё, что не `disabled`) означал бы, что новое
    состояние, добавленное миграцией раньше кода, молча раздаёт доступ — а миграции
    БД `auth` применяются деплоем «Птицы» (.forgejo/workflows/deploy.yml), то есть
    опередить код они могут запросто.
    """
    if isinstance(value, str):
        try:
            return AccessState(value)
        except ValueError:
            logger.warning(
                "auth_session: неизвестное состояние доступа %r → трактую как disabled", value
            )
            return AccessState.DISABLED
    logger.warning(
        "auth_session: состояние доступа %r неожиданного типа %s → трактую как disabled",
        value,
        type(value).__name__,
    )
    return AccessState.DISABLED
@dataclass(frozen=True, slots=True)
 class SessionUser:
    """Кто стоит за валидной сессионной кукой.
    Attributes:
        username: логин из реестра. Именно он, а не значение куки, дальше едет в
            RBAC «Птицы» (`app.core.auth.get_role`).
        access_state: всегда `AccessState.ACTIVE` — не-active сюда не доходит
            (см. `get_session_user`). Поле оставлено явным, чтобы состояние доступа
            во всём коде называлось и выражалось одинаково, а не превращалось в
            неявное «раз объект вернулся, значит active».
    """
    username: str
    access_state: AccessState
 def get_session_user(db: Session, token: str) -> SessionUser | None:
    """Резолвит сессионный токен в пользователя, или None если сессия невалидна.
    Невалидна = не найдена / истекла / состояние доступа юзера не `active`.
    Состояние доступа: пропускается ТОЛЬКО `AccessState.ACTIVE`. Любое другое
    (`disabled`, `trial_expired`, а также нераспознанное — `to_access_state`
    fail-closed'ит его в `disabled`) делает уже выданную сессию недействительной
    НЕМЕДЛЕННО, не дожидаясь `expires_at`. Иначе заблокированный человек продолжал
    бы работать до истечения TTL (до 30 дней), а sliding-refresh продлевал бы ему
    сессию бесконечно — то есть блокировка в реестре не блокировала бы ничего.
    Sliding refresh: если с последнего `last_seen_at` прошло >= 5 минут — продлевает
    `last_seen_at`/`expires_at` ОДНИМ UPDATE (ровно как «Мера»: тот же интервал, тот
    же одиночный UPDATE обеих колонок, тот же best-effort). Продлевать обе колонки
    обязательно: обновляй «Птица» только `last_seen_at`, человек, работающий весь
    день в ней одной, был бы разлогинен по `expires_at` несмотря на активность.
    Сбой refresh (напр. read-only реплика) логируется и НЕ мешает вернуть валидного
    юзера — это best-effort продление, а не часть решения «валидна ли сессия».
    Принимает уже открытую сессию БД `auth` (не открывает сам) — так модуль остаётся
    тривиально unit-тестируемым. Обычный вызывающий берёт `resolve_session_token`.
    ⚠️ `db` ОБЯЗАНА быть сессией БД `auth` (`app.core.auth_db.auth_session()`), а не
    `app.core.db.get_db`: в продуктовой БД gendesign таблиц `users`/`sessions` нет.
    Исключения БД наружу НЕ глушатся (кроме best-effort refresh): сбой реестра —
    часть auth-решения, и вызывающий обязан его увидеть, чтобы закрыться, а не
    трактовать как «сессии нет».
    """
    if not token:
        return None
    row = db.execute(
        text(
            """
            SELECT s.expires_at, s.last_seen_at, u.username, u.access_state
            FROM sessions s
            JOIN users u ON u.id = s.user_id
            WHERE s.token = :token
              AND s.expires_at > now()
            """
        ),
        {"token": token},
    ).fetchone()
    if row is None:
        return None
    now = datetime.now(UTC)
    # Второй пояс к `AND s.expires_at > now()` в SELECT'е выше. Первый пояс — часами
    # БД, и это принципиально: строку продлевает UPDATE ниже, где `expires_at =
    # now() + interval` считает СЕРВЕР. Реши мы срок годности только часами процесса
    # (`datetime.now(UTC)`), отставание этих часов давало бы не «сессия проживёт на
    # дельту дольше», а НЕОБРАТИМОЕ воскрешение: строку, которую БД уже считает
    # мёртвой, Python пропустил бы, тут же сработал бы sliding-refresh и отодвинул
    # expires_at на полный TTL от серверного now(). Секунда расхождения → +30 дней.
    # Обе стороны сравнения обязаны брать время из одного источника.
    #
    # Проверку на None оставляем первой: `expires_at` объявлен NOT NULL
    # (data/sql/auth/001), но если колонку когда-нибудь ослабят, это дешевле
    # разбирательства, почему сравнение с None упало TypeError'ом в auth-пути.
    if row.expires_at is None or row.expires_at <= now:
        return None
    access_state = to_access_state(row.access_state)
    if not access_state.can_sign_in:
        return None
    if row.last_seen_at is None or (now - row.last_seen_at) >= _SLIDING_REFRESH_INTERVAL:
        try:
            db.execute(
                text(
                    """
                    UPDATE sessions
                    SET last_seen_at = now(),
                        expires_at = now() + make_interval(hours => CAST(:ttl_hours AS integer))
                    WHERE token = :token
                    """
                ),
                {"ttl_hours": settings.session_ttl_hours, "token": token},
            )
            db.commit()
        except Exception:
            # Без username в сообщении: строка лога — не место для связки
            # «кто именно» + «в какой момент», а разбор всё равно идёт по времени.
            logger.warning("auth_session: sliding refresh failed", exc_info=True)
            try:
                db.rollback()
            except Exception:
                # Причина сбоя UPDATE'а может быть оборванным соединением — тогда и
                # rollback бросит. Без этого except «best-effort продление» переставало
                # бы быть best-effort: валидный юзер, чью сессию не удалось продлить,
                # получал бы не доступ, а исключение наружу (и в guard'е — деградацию
                # на легаси-заголовок, а в db_only — отказ).
                logger.warning("auth_session: rollback after failed refresh failed", exc_info=True)
    return SessionUser(username=row.username, access_state=access_state)
 def resolve_session_token(token: str | None) -> SessionUser | None:
    """Резолвит токен сессионной куки, сам открывая соединение с БД `auth`.
    Точка входа для `rbac_guard` (`app/main.py`), который зовёт её в threadpool —
    внутри синхронный psycopg-I/O, а guard живёт на event loop'е. Возвращает None,
    если сессии нет или она недействительна.
    Режим `legacy` (`AUTH_MODE=legacy`, ДЕФОЛТ) → None СРАЗУ, без единого
    обращения к БД: инвариант «выключенный флаг = ни одного коннекта к реестру»
    держится этим модулем, а не соглашением с вызывающим. Тихий None здесь безопасен,
    потому что направлен в сторону fail-closed — он означает ровно «session-auth не
    используется», то есть сегодняшнее поведение (Caddy basic_auth + trusted-header),
    и никому ничего не открывает.
    Исключения НЕ глушатся — ни `AuthDatabaseNotConfiguredError` (флаг включён, DSN
    пуст/битый), ни ошибки соединения. Решение «что делать со сломанным реестром»
    принимает guard, и оно неочевидно: молча откатиться на trusted-header значит
    раздавать права из roles.yaml в обход реестра, включая заблокированные аккаунты.
    Прятать такое внутри резолвера нельзя.
    Raises:
        AuthDatabaseNotConfiguredError: флаг включён, а DSN БД `auth` пуст или не
            разобрался (см. `app.core.auth_db`).
    """
    if not settings.auth_session_enabled:
        return None
    if not token:
        return None
    with auth_db.auth_session() as db:
        return get_session_user(db, token)
--- a/backend/tests/test_auth_db.py
+++ b/backend/tests/test_auth_db.py
@ -1,497 +0,0 @@
 """DSN и ленивый engine БД `auth` — `app/core/config.py` + `app/core/auth_db.py`.
 Эпик «единый вход», стадия 3. Три группы:
  1. Дефолты. Они ЧАСТЬ КОНТРАКТА PR, а не декорация: пока Caddy basic_auth стоит,
     прод обязан вести себя ровно как до эпика — флаг выключен, DSN не сконфигурирован,
     engine не создаётся, отсутствие AUTH_* в окружении не роняет старт.
  2. Сборка DSN из частей: приоритет явного URL, экранирование секрета, пустые
     значения переменных → прод-дефолты (а не мусорный DSN и не падение на импорте).
  3. `auth_db`: ленивость, кеш, внятная ошибка вместо утечки пароля.
 Сеть здесь не нужна: `create_engine` пул создаёт лениво и к серверу не ходит.
 """
 from __future__ import annotations
 import os
 import subprocess
 import sys
 from collections.abc import Iterator
 from pathlib import Path
 import pytest
 from pydantic import SecretStr, ValidationError
 from sqlalchemy.engine import make_url
 from app.core import auth_db
 from app.core.auth_db import AuthDatabaseNotConfiguredError, AuthDatabaseUnreachableError
 from app.core.config import Settings, settings
 _AUTH_ENV_VARS = (
    "AUTH_MODE",
    "AUTH_DATABASE_URL",
    "AUTH_DB_PASSWORD",
    "AUTH_DB_HOST",
    "AUTH_DB_PORT",
    "AUTH_DB_NAME",
    "AUTH_DB_USER",
    "SESSION_COOKIE_NAME",
    "SESSION_TTL_HOURS",
 )
 # Заведомо синтаксически корректный DSN на несуществующий хост: engine по нему
 # создаётся, но соединение не открывается (пул ленивый), поэтому тесты офлайновы.
 _OFFLINE_DSN = "postgresql+psycopg://auth_app:pw@127.0.0.1:1/auth"
@pytest.fixture
 def clean_env(monkeypatch: pytest.MonkeyPatch) -> None:
    """Ни одной AUTH_*/SESSION_* переменной — тест дефолтов не зависит от машины."""
    for name in _AUTH_ENV_VARS:
        monkeypatch.delenv(name, raising=False)
@pytest.fixture(autouse=True)
 def _reset_engine_cache() -> Iterator[None]:
    """Ни один тест не оставляет за собой закешированный engine БД `auth`."""
    auth_db.reset_auth_db()
    yield
    auth_db.reset_auth_db()
 # ---------------------------------------------------------------------------
 # 1. Дефолты
 # ---------------------------------------------------------------------------
 def test_session_mode_is_off_and_unconfigured_by_default(clean_env: None) -> None:
    """Свежие настройки без AUTH_*: режим legacy, DSN пуст — и это НЕ ошибка."""
    fresh = Settings()
    assert fresh.auth_mode == "legacy"
    assert fresh.auth_session_enabled is False
    assert fresh.resolved_auth_database_url == ""
 def test_live_settings_singleton_is_off() -> None:
    """Тот же инвариант на настоящем синглтоне, которым пользуется приложение."""
    assert settings.auth_mode == "legacy"
    assert settings.auth_session_enabled is False
@pytest.mark.parametrize("raw", ["", "   ", "LEGACY", " legacy "])
 def test_blank_or_odd_case_auth_mode_falls_back_to_legacy(
    clean_env: None, monkeypatch: pytest.MonkeyPatch, raw: str
 ) -> None:
    """`AUTH_MODE=` (или регистр/пробелы) → legacy, а не ValidationError на импорте.
    `settings = Settings()` выполняется на уровне модуля: невалидное значение уронило бы
    ИМПОРТ конфига и увело контейнер в restart-loop. Сценарий бытовой — ops копирует
    блок AUTH_* в .env.runtime и заполняет только пароль.
    """
    monkeypatch.setenv("AUTH_MODE", raw)
    assert Settings().auth_mode == "legacy"
 def test_meaningful_garbage_in_auth_mode_still_fails(
    clean_env: None, monkeypatch: pytest.MonkeyPatch
 ) -> None:
    """`AUTH_MODE=off` — опечатка со смыслом, и она обязана падать.
    Молча трактовать её как legacy значило бы тихо оставить продукт на trusted-header
    после того, как последний PR эпика снимет popup.
    """
    monkeypatch.setenv("AUTH_MODE", "off")
    with pytest.raises(ValidationError):
        Settings()
@pytest.mark.parametrize(
    ("mode", "enabled"), [("legacy", False), ("dual", True), ("db_only", True)]
 )
 def test_auth_session_enabled_is_derived_from_mode(
    clean_env: None, monkeypatch: pytest.MonkeyPatch, mode: str, enabled: bool
 ) -> None:
    """Свойство `auth_session_enabled` = «режим не legacy» — единый выключатель реестра.
    Оно и закорачивает `resolve_session_token` / `require_auth_db_configured`; разница
    dual vs db_only ему не видна и не должна быть (она про фолбэк в rbac_guard).
    """
    monkeypatch.setenv("AUTH_MODE", mode)
    assert Settings().auth_session_enabled is enabled
 def test_default_host_is_this_stacks_postgres(clean_env: None) -> None:
    """🪤 Дефолт хоста — `postgres`, и это ЗЕРКАЛЬНО «Мере», а не копия с неё.
    У «Меры» дефолт `gendesign-postgres`, потому что внутри её стека имя `postgres`
    занято её собственным контейнером. У «Птицы» наоборот: её стек главный, сервис
    `postgres` корневого docker-compose.prod.yml и есть сервер с БД `auth`. Алиас
    `gendesign-postgres` живёт только во внешней сети `shared`, куда входят не все
    сервисы (beat — нет), поэтому дефолтом он быть не может.
    """
    fresh = Settings()
    assert fresh.auth_db_host == "postgres"
    assert fresh.auth_db_host != "gendesign-postgres"
    assert fresh.auth_db_port == 5432
    assert fresh.auth_db_name == "auth"
    assert fresh.auth_db_user == "auth_app"
 def test_cookie_defaults_match_the_other_product(clean_env: None) -> None:
    """Имя куки и TTL обязаны совпадать с «Мерой» — иначе общая сессия не общая.
    Имя историческое («tradein_» уже ни о чём не говорит); переименование
    разлогинивает всех сразу в обоих продуктах, поэтому оно закреплено тестом.
    """
    fresh = Settings()
    assert fresh.session_cookie_name == "tradein_session"
    assert fresh.session_ttl_hours == 720
 # ---------------------------------------------------------------------------
 # 2. Сборка DSN
 # ---------------------------------------------------------------------------
 def test_explicit_url_wins_over_parts(clean_env: None) -> None:
    """AUTH_DATABASE_URL — аварийный обход (другой хост, sslmode): выигрывает всегда."""
    fresh = Settings(
        auth_database_url="  postgresql+psycopg://u:p@elsewhere:6432/auth?sslmode=require  ",
        auth_db_password=SecretStr("ignored"),
        auth_db_host="postgres",
    )
    assert (
        fresh.resolved_auth_database_url
        == "postgresql+psycopg://u:p@elsewhere:6432/auth?sslmode=require"
    )
 def test_dsn_built_from_password_and_defaults(clean_env: None) -> None:
    """Включение на проде = одна переменная: пароль + прод-дефолты остальных частей."""
    fresh = Settings(auth_db_password=SecretStr("s3cret"))
    assert (
        fresh.resolved_auth_database_url
        == "postgresql+psycopg://auth_app:s3cret@postgres:5432/auth"
    )
 def test_password_special_chars_survive_round_trip(clean_env: None) -> None:
    """Пароль экранируется: `@`/`/`/`:`/`#`/`%` иначе порвали бы URL по его грамматике.
    Проверяем не наличие процентов в строке, а РАЗБОР обратно: важно, что SQLAlchemy
    видит тот же пароль и, главное, тот же хост/базу. Незакавыченный `@` молча увёл бы
    подключение на другой хост.
    """
    raw = "p@ss:w/rd#1%zz?x"
    url = make_url(Settings(auth_db_password=SecretStr(raw)).resolved_auth_database_url)
    assert url.password == raw
    assert url.host == "postgres"
    assert url.port == 5432
    assert url.database == "auth"
    assert url.username == "auth_app"
 def test_username_is_quoted_too(clean_env: None) -> None:
    url = make_url(
        Settings(
            auth_db_password=SecretStr("pw"), auth_db_user="odd:user@name"
        ).resolved_auth_database_url
    )
    assert url.username == "odd:user@name"
    assert url.host == "postgres"
 def test_password_whitespace_is_preserved_not_stripped(clean_env: None) -> None:
    """Ведущий/хвостовой пробел может быть частью настоящего пароля — не режем."""
    url = make_url(Settings(auth_db_password=SecretStr(" pw ")).resolved_auth_database_url)
    assert url.password == " pw "
 def test_blank_password_means_not_configured(clean_env: None) -> None:
    """Пробельная строка — опечатка в .env, а не пароль: «не сконфигурировано»."""
    assert Settings(auth_db_password=SecretStr("   ")).resolved_auth_database_url == ""
    assert Settings(auth_db_password=SecretStr("")).resolved_auth_database_url == ""
 def test_blank_parts_fall_back_to_defaults(clean_env: None) -> None:
    """`AUTH_DB_HOST=` в .env.runtime не должен давать DSN вида `...@:5432/auth`.
    Сценарий бытовой: ops копирует блок AUTH_DB_* целиком и заполняет только пароль.
    """
    url = make_url(
        Settings(
            auth_db_password=SecretStr("pw"),
            auth_db_host="   ",
            auth_db_name="",
            auth_db_user=" ",
        ).resolved_auth_database_url
    )
    assert (url.host, url.database, url.username) == ("postgres", "auth", "auth_app")
@pytest.mark.parametrize("blank", ["", "   "])
 def test_blank_port_does_not_break_import(clean_env: None, blank: str) -> None:
    """`AUTH_DB_PORT=` → дефолт, а НЕ ValidationError.
    `settings = Settings()` выполняется на уровне модуля: падение здесь уводило бы
    контейнер в restart-loop — причём в дефолтном режиме, где к БД `auth` не идёт ни
    одного обращения.
    """
    assert Settings(auth_db_port=blank).auth_db_port == 5432
 def test_non_blank_garbage_port_still_fails(clean_env: None) -> None:
    """`AUTH_DB_PORT=abc` — опечатка со смыслом, её глушить нельзя."""
    with pytest.raises(ValueError):
        Settings(auth_db_port="abc")
 def test_password_is_not_printed_by_repr_or_dump(clean_env: None) -> None:
    """SecretStr: пароль не утекает в `repr(settings)` / `model_dump()`.
    Сегодня их никто не рендерит, но появиться такой рендер (лог старта, /debug) может
    тихо — а рядом с обычным str-полем это была бы утечка секрета в открытый лог.
    """
    fresh = Settings(auth_db_password=SecretStr("s3cret"))
    assert "s3cret" not in repr(fresh)
    assert "s3cret" not in str(fresh.model_dump())
    assert fresh.auth_db_password.get_secret_value() == "s3cret"
 # ---------------------------------------------------------------------------
 # 3. auth_db: ленивость, кеш, ошибки
 # ---------------------------------------------------------------------------
 def test_import_without_any_auth_env_does_not_build_engine() -> None:
    """Импорт в чистом окружении: ошибки нет, engine не создан, DSN пуст.
    Проверяется отдельным процессом, потому что в текущем модуль импортирован давно и
    любое утверждение про «на импорте» было бы про уже случившийся импорт. Это отличие
    от `app.core.db`, где engine создаётся в теле модуля: сделай мы так же, приложение
    падало бы на старте везде, где реестр не сконфигурирован — локально, в pytest, на
    любом стенде. Ровно тот контракт, который держит дефолтное поведение прода.
    """
    env = {k: v for k, v in os.environ.items() if k not in _AUTH_ENV_VARS}
    code = (
        "from app.core import auth_db\n"
        "from app.core.config import settings\n"
        "print(auth_db._engine, repr(settings.resolved_auth_database_url), "
        "settings.auth_session_enabled)\n"
    )
    proc = subprocess.run(
        [sys.executable, "-c", code],
        cwd=Path(__file__).resolve().parents[1],
        env=env,
        capture_output=True,
        text=True,
    )
    assert proc.returncode == 0, proc.stderr
    assert proc.stdout.strip() == "None '' False"
 def test_unconfigured_registry_raises_with_actionable_message(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """Пустой DSN → явная ошибка с инструкцией, а не «сессия не найдена»."""
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))
    with pytest.raises(AuthDatabaseNotConfiguredError) as excinfo:
        auth_db.get_auth_engine()
    assert "AUTH_MODE" in str(excinfo.value)
    assert "AUTH_DB_PASSWORD" in str(excinfo.value)
@pytest.mark.parametrize(
    "broken",
    [
        "not-a-dsn-at-all",
        # «Почти URL»: разбор доходит до int(port) и падает, унося в текст ошибки
        # кусок пароля, съехавший на позицию порта.
        "postgresql+psycopg://u:pa@ss@host:wo/auth",
    ],
 )
 def test_malformed_dsn_does_not_leak_into_the_error(
    monkeypatch: pytest.MonkeyPatch, broken: str
 ) -> None:
    """Битый DSN → своя ошибка БЕЗ самого DSN и без исходного traceback.
    Текст ошибки SQLAlchemy цитирует строку целиком, а в ней пароль роли auth_app.
    `from None` обязателен: без него исходная ошибка печаталась бы в traceback как
    «During handling of the above exception...» — то есть пароль всё равно оказался бы
    в логе.
    """
    monkeypatch.setattr(settings, "auth_database_url", broken)
    with pytest.raises(AuthDatabaseNotConfiguredError) as excinfo:
        auth_db.get_auth_engine()
    assert broken not in str(excinfo.value)
    assert "pa@ss" not in str(excinfo.value)
    assert excinfo.value.__suppress_context__ is True
    assert excinfo.value.__cause__ is None
 def test_engine_is_built_once_and_reused(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    first = auth_db.get_auth_engine()
    second = auth_db.get_auth_engine()
    assert first is second
    assert auth_db.get_auth_session_factory().kw["bind"] is first
    assert first.url.database == "auth"
 def test_reset_drops_the_cached_engine(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    first = auth_db.get_auth_engine()
    auth_db.reset_auth_db()
    assert auth_db._engine is None
    assert auth_db.get_auth_engine() is not first
 # ---------------------------------------------------------------------------
 # require_auth_db_configured — fail-fast на старте (lifespan)
 # ---------------------------------------------------------------------------
 def test_startup_check_is_noop_while_flag_is_off(monkeypatch: pytest.MonkeyPatch) -> None:
    """🔒 Дефолт: пустой DSN на старте — не ошибка, и engine не создаётся.
    Ровно то, что произойдёт на проде сразу после мержа этого PR.
    """
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))
    auth_db.require_auth_db_configured()
    assert auth_db._engine is None
 def test_startup_check_fails_fast_when_enabled_without_dsn(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """Флаг включили, DSN не задали → контейнер не стартует.
    Иначе пустой DSN ловил бы `except` в guard'е, и сломанная конфигурация выглядела бы
    как «ни у кого нет сессии» — сутками, при живом приложении и 200-х в ответах.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))
    with pytest.raises(AuthDatabaseNotConfiguredError):
        auth_db.require_auth_db_configured()
 def test_startup_check_builds_engine_and_probes_connection(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """Режим включён и реестр отвечает → engine готов ещё до первого запроса.
    Проба соединения подменена: поднимать Postgres ради этого теста незачем, важно, что
    она вызывается ИМЕННО на том engine, который останется закешированным.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    probed: list[object] = []
    monkeypatch.setattr(auth_db, "_probe_connection", probed.append)
    auth_db.require_auth_db_configured()
    assert auth_db._engine is not None
    assert probed == [auth_db._engine]
 def test_startup_check_fails_when_dsn_parses_but_connection_does_not(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """🔴 Смысл проверки: DSN разобрался — это ещё НЕ «реестр рабочий».
    `create_engine` к серверу не ходит, поэтому одна лишь сборка engine отлавливала бы
    ровно два случая (DSN пуст / не парсится). Весь вероятный класс ошибок — неверный
    AUTH_DB_PASSWORD, опечатка в хосте, не созданная БД `auth`, отозванная роль
    auth_app, нет сети — проходил бы мимо, контейнер стартовал бы зелёным, `/health`
    отвечал бы 200, а каждый запрос с кукой молча деградировал бы на легаси-заголовок.
    Сутками. Ровно то, что комментарий в app/main.py обещает НЕ допускать.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    def _refused(_engine: object) -> None:
        raise OSError("connection to server at 127.0.0.1, port 1 failed: Connection refused")
    monkeypatch.setattr(auth_db, "_probe_connection", _refused)
    with pytest.raises(AuthDatabaseUnreachableError) as excinfo:
        auth_db.require_auth_db_configured()
    # Причина сохранена в цепочке — ради неё проверка и делается; DSN (в нём пароль) в
    # наш текст не подставляется.
    assert isinstance(excinfo.value.__cause__, OSError)
    assert "AUTH_MODE" in str(excinfo.value)
    assert _OFFLINE_DSN not in str(excinfo.value)
 def test_startup_check_does_not_probe_while_flag_is_off(monkeypatch: pytest.MonkeyPatch) -> None:
    """🔒 legacy: ни коннекта, ни пробы — даже если DSN задан и валиден."""
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    def _boom(_engine: object) -> None:
        raise AssertionError("в режиме legacy соединение с реестром недопустимо")
    monkeypatch.setattr(auth_db, "_probe_connection", _boom)
    auth_db.require_auth_db_configured()
    assert auth_db._engine is None
 def test_engine_has_short_timeouts(monkeypatch: pytest.MonkeyPatch) -> None:
    """Реестр — не критический путь: его сбой обязан деградировать за секунды.
    Без `connect_timeout` дропнутые SYN (фаервол молча глотает пакеты) держали бы
    попытку до TCP-таймаута ОС — на Linux ~130 с, и так на КАЖДОМ checkout'е, потому
    что включён `pool_pre_ping`. `pool_timeout` по дефолту 30 с — в auth-пути столько
    ждать свободный коннект незачем.
    """
    monkeypatch.setattr(settings, "auth_database_url", _OFFLINE_DSN)
    captured: dict[str, object] = {}
    real_create_engine = auth_db.create_engine
    def _spy(dsn: str, **kwargs: object) -> object:
        captured.update(kwargs)
        return real_create_engine(dsn, **kwargs)  # type: ignore[arg-type]
    monkeypatch.setattr(auth_db, "create_engine", _spy)
    auth_db.get_auth_engine()
    assert captured["pool_timeout"] == 3
    assert captured["pool_pre_ping"] is True
    connect_args = captured["connect_args"]
    assert isinstance(connect_args, dict)
    assert connect_args["connect_timeout"] == 3
    assert "statement_timeout=3000" in connect_args["options"]
--- a/backend/tests/test_auth_session_guard.py
+++ b/backend/tests/test_auth_session_guard.py
@ -1,861 +0,0 @@
 """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
--- a/backend/tests/test_auth_session_service.py
+++ b/backend/tests/test_auth_session_service.py
@ -1,305 +0,0 @@
 """Резолв сессии общего реестра — `app/services/auth_session.py` (эпик «единый вход»).
 Слой ниже guard'а: «что считать валидной сессией» и «когда продлевать». Через guard
 эти правила проверяются end-to-end в `tests/test_auth_session_guard.py`; здесь —
 поштучно, включая ветки, до которых из guard'а дотянуться дорого (sliding refresh,
 сбой продления, исключения БД).
 Дублёр сессии БД (`FakeAuthDb`) намеренно ОДИН на оба файла и живёт в guard-тестах:
 разъехавшиеся двойники — типовой способ получить два зелёных теста при одном сломанном
 поведении. Прецедент кросс-импорта внутри пакета tests — `tests/integration/*`.
 """
 from __future__ import annotations
 import logging
 from datetime import UTC, datetime, timedelta
 import pytest
 from pydantic import SecretStr
 from app.core import auth_db
 from app.core.auth_db import AuthDatabaseNotConfiguredError
 from app.core.config import settings
 from app.services import auth_session as svc
 from app.services.auth_session import AccessState, SessionUser, get_session_user, to_access_state
 from tests.test_auth_session_guard import FakeAuthDb, _install_auth_db, _Row
 _TOKEN = "tok-1"
 _USER = "user1"
 def _row(
    *,
    expires_in: timedelta = timedelta(days=7),
    last_seen_ago: timedelta | None = timedelta(seconds=30),
    username: str = _USER,
    access_state: str = "active",
 ) -> _Row:
    now = datetime.now(UTC)
    return _Row(
        expires_at=now + expires_in,
        last_seen_at=None if last_seen_ago is None else now - last_seen_ago,
        username=username,
        access_state=access_state,
    )
 # ---------------------------------------------------------------------------
 # AccessState / to_access_state — fail-closed
 # ---------------------------------------------------------------------------
 def test_only_active_can_sign_in() -> None:
    assert AccessState.ACTIVE.can_sign_in is True
    assert AccessState.TRIAL_EXPIRED.can_sign_in is False
    assert AccessState.DISABLED.can_sign_in is False
 def test_to_access_state_known_values() -> None:
    assert to_access_state("active") is AccessState.ACTIVE
    assert to_access_state("trial_expired") is AccessState.TRIAL_EXPIRED
    assert to_access_state("disabled") is AccessState.DISABLED
@pytest.mark.parametrize("value", ["frozen", "", None, 42])
 def test_to_access_state_unknown_is_disabled_with_warning(
    value: object, caplog: pytest.LogCaptureFixture
 ) -> None:
    """Неизвестное/NULL/чужой тип → disabled + WARNING.
    Миграции БД `auth` применяет деплой «Птицы», то есть новое состояние может
    появиться в базе раньше, чем код о нём узнает. Обратный выбор («не disabled =
    пускаем») означал бы, что такая миграция молча раздаёт доступ.
    """
    with caplog.at_level(logging.WARNING, logger="app.services.auth_session"):
        assert to_access_state(value) is AccessState.DISABLED
    assert caplog.records
 # ---------------------------------------------------------------------------
 # get_session_user — что считается валидной сессией
 # ---------------------------------------------------------------------------
 def test_valid_session_resolves_to_user() -> None:
    db = FakeAuthDb({_TOKEN: _row()})
    assert get_session_user(db, _TOKEN) == SessionUser(
        username=_USER, access_state=AccessState.ACTIVE
    )
    assert db.select_tokens == [_TOKEN]
 def test_empty_token_short_circuits_without_query() -> None:
    db = FakeAuthDb({_TOKEN: _row()})
    assert get_session_user(db, "") is None
    assert db.select_tokens == []
 def test_unknown_token_returns_none() -> None:
    db = FakeAuthDb()
    assert get_session_user(db, "never-issued") is None
    assert db.updates == []
 def test_expired_session_returns_none_and_is_not_refreshed() -> None:
    """Истёкшая сессия не воскресает sliding-refresh'ем — иначе TTL был бы вечным."""
    db = FakeAuthDb(
        {_TOKEN: _row(expires_in=timedelta(seconds=-1), last_seen_ago=timedelta(days=1))}
    )
    assert get_session_user(db, _TOKEN) is None
    assert db.updates == []
 def test_null_expires_at_returns_none() -> None:
    """`expires_at IS NULL` (колонку ослабили) → сессии нет, а не TypeError в auth-пути."""
    row = _row()
    row.expires_at = None
    db = FakeAuthDb({_TOKEN: row})
    assert get_session_user(db, _TOKEN) is None
@pytest.mark.parametrize("access_state", ["disabled", "trial_expired", "unheard_of"])
 def test_non_active_user_returns_none_immediately(access_state: str) -> None:
    """Блокировка в реестре бьёт сразу, не дожидаясь expires_at (иначе до 30 дней)."""
    db = FakeAuthDb({_TOKEN: _row(access_state=access_state)})
    assert get_session_user(db, _TOKEN) is None
    assert db.updates == []
 # ---------------------------------------------------------------------------
 # Sliding refresh
 # ---------------------------------------------------------------------------
 def test_refresh_skipped_for_recent_session() -> None:
    """Свежий last_seen_at → ни одного UPDATE: иначе каждый API-запрос бил бы в БД."""
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(seconds=30))})
    assert get_session_user(db, _TOKEN) is not None
    assert db.updates == []
    assert db.commits == 0
@pytest.mark.parametrize(
    "last_seen_ago", [timedelta(minutes=5), timedelta(hours=3), None], ids=["at-5m", "3h", "null"]
 )
 def test_refresh_extends_after_interval(last_seen_ago: timedelta | None) -> None:
    """>= 5 минут (и NULL) → один UPDATE на обе колонки + commit.
    TTL берётся из настроек и обязан совпадать с «Мерой»: продлевает сессию тот
    продукт, в котором кликнули последним, и срок жизни не должен от этого зависеть.
    """
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=last_seen_ago)})
    assert get_session_user(db, _TOKEN) is not None
    assert db.updates == [{"ttl_hours": settings.session_ttl_hours, "token": _TOKEN}]
    assert db.commits == 1
 def test_refresh_uses_configured_ttl(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "session_ttl_hours", 12)
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(hours=1))})
    get_session_user(db, _TOKEN)
    assert db.updates == [{"ttl_hours": 12, "token": _TOKEN}]
 def test_refresh_failure_does_not_block_valid_session(caplog: pytest.LogCaptureFixture) -> None:
    """Продление — best-effort: сбой логируется и откатывается, юзер всё равно валиден.
    Иначе read-only реплика или блокировка строки разлогинивала бы всех, у кого
    сессия старше пяти минут.
    """
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(hours=1))}, fail_refresh=True)
    with caplog.at_level(logging.WARNING, logger="app.services.auth_session"):
        user = get_session_user(db, _TOKEN)
    assert user == SessionUser(username=_USER, access_state=AccessState.ACTIVE)
    assert db.rollbacks == 1
    assert any("sliding refresh failed" in r.getMessage() for r in caplog.records)
    # В сообщении не должно быть ни username, ни токена: лог — не место для связки
    # «кто именно» + «когда», а разбор идёт по времени.
    assert not any(_USER in r.getMessage() or _TOKEN in r.getMessage() for r in caplog.records)
 def test_select_failure_is_not_swallowed() -> None:
    """Сбой SELECT'а летит наружу: решение «что делать со сломанным реестром» — не здесь.
    Проглоти резолвер ошибку — вызывающий получил бы «сессии нет», то есть отказ
    выглядел бы как «просто не залогинен», а откат на trusted-header — как норма.
    """
    db = FakeAuthDb()
    def _boom(*_a: object, **_k: object) -> None:
        raise RuntimeError("auth registry is down")
    db.execute = _boom  # type: ignore[method-assign]
    with pytest.raises(RuntimeError, match="auth registry is down"):
        get_session_user(db, _TOKEN)
 # ---------------------------------------------------------------------------
 # resolve_session_token — точка входа guard'а
 # ---------------------------------------------------------------------------
 def test_resolve_returns_none_without_touching_db_when_flag_off(
    monkeypatch: pytest.MonkeyPatch,
 ) -> None:
    """🔒 Инвариант «выключенный флаг = ни одного коннекта» держится этим модулем.
    Он не полагается на то, что вызывающий сам не позовёт резолв: даже с валидным
    токеном соединение не открывается.
    """
    monkeypatch.setattr(settings, "auth_mode", "legacy")
    _install_auth_db(monkeypatch, None)  # открытие сессии → RuntimeError
    assert svc.resolve_session_token(_TOKEN) is None
@pytest.mark.parametrize("token", [None, ""])
 def test_resolve_returns_none_for_empty_token(
    monkeypatch: pytest.MonkeyPatch, token: str | None
 ) -> None:
    monkeypatch.setattr(settings, "auth_mode", "dual")
    _install_auth_db(monkeypatch, None)
    assert svc.resolve_session_token(token) is None
 def test_resolve_opens_registry_session_when_flag_on(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(settings, "auth_mode", "dual")
    db = FakeAuthDb({_TOKEN: _row()})
    _install_auth_db(monkeypatch, db)
    assert svc.resolve_session_token(_TOKEN) == SessionUser(
        username=_USER, access_state=AccessState.ACTIVE
    )
    assert db.select_tokens == [_TOKEN]
 def test_resolve_propagates_not_configured_error(monkeypatch: pytest.MonkeyPatch) -> None:
    """Флаг включён, DSN пуст → исключение наружу, а не «сессия не найдена».
    Тихий None здесь означал бы либо массовый отказ доступа под видом «не залогинен»,
    либо (в guard'е) бессрочную раздачу прав в обход реестра. Настоящий `auth_db` не
    подменяется — проверяется именно связка сервис ↔ конфигурация.
    """
    monkeypatch.setattr(settings, "auth_mode", "dual")
    monkeypatch.setattr(settings, "auth_database_url", "")
    monkeypatch.setattr(settings, "auth_db_password", SecretStr(""))
    auth_db.reset_auth_db()
    try:
        with pytest.raises(AuthDatabaseNotConfiguredError):
            svc.resolve_session_token(_TOKEN)
    finally:
        auth_db.reset_auth_db()
 def test_expiry_is_also_filtered_by_db_clock() -> None:
    """Срок годности отсекается ЧАСАМИ БД, а не только часами процесса.
    Асимметрия, которую это закрывает: решение «жива ли сессия» принимал Python
    (`datetime.now(UTC)`), а продление писало `expires_at = now() + interval` часами
    СЕРВЕРА. Отставание часов приложения давало бы не «сессия проживёт на дельту
    дольше», а необратимое воскрешение: строку, которую БД уже считает мёртвой, Python
    пропускал бы, тут же срабатывал sliding-refresh и отодвигал expires_at на полный
    TTL от серверного now(). Секунда расхождения → +30 дней жизни.
    Форма запроса проверяется дублёром (`_assert_select_shape`), поэтому здесь
    достаточно одного прохода: потеряется `AND s.expires_at > now()` — тест покраснеет.
    """
    db = FakeAuthDb({_TOKEN: _row()})
    assert get_session_user(db, _TOKEN) is not None
 def test_rollback_failure_does_not_break_the_resolve(caplog: pytest.LogCaptureFixture) -> None:
    """Сбой самого rollback'а (оборванный коннект) не отменяет валидную сессию.
    Иначе «best-effort продление» переставало быть best-effort: исключение улетало бы
    из get_session_user наружу, и валидный юзер получал бы вместо доступа ERROR в
    GlitchTip и деградацию на легаси-заголовок (а в db_only — отказ).
    """
    db = FakeAuthDb({_TOKEN: _row(last_seen_ago=timedelta(hours=1))}, fail_refresh=True)
    def _dead_connection() -> None:
        raise RuntimeError("server closed the connection unexpectedly")
    db.rollback = _dead_connection  # type: ignore[method-assign]
    with caplog.at_level(logging.WARNING, logger="app.services.auth_session"):
        user = get_session_user(db, _TOKEN)
    assert user == SessionUser(username=_USER, access_state=AccessState.ACTIVE)
    assert any("rollback" in r.getMessage() for r in caplog.records)
--- a/backend/tests/test_rbac.py
+++ b/backend/tests/test_rbac.py
@ -42,26 +42,11 @@ def _reset_auth_cache() -> None:
 # Test app — копия rbac_guard из app/main.py, чтобы не подтягивать тяжёлые
 # импорты (weasyprint, celery worker, ...). Если поведение middleware меняется
 # в проде — синхронизируй здесь.
 #
 # NB: копия воспроизводит ЛЕГАСИ-ВЕТКУ принятия решения (trusted-header) и намеренно
 # не знает про сессионную куку общего реестра, добавленную эпиком «единый вход»:
 # при AUTH_MODE=legacy (дефолт) прод-guard принимает решение ровно так же, и тесты
 # ниже проверяют именно тот режим. Режимы dual/db_only (кука → заголовок, приоритет
 # куки над подделанным заголовком, 401/403, публичные пути) покрыты в
 # tests/test_auth_session_guard.py — там вызывается НАСТОЯЩИЙ app.main.rbac_guard,
 # без копии.
 #
 # «Ровно так же» — про ЛОГИКУ, не про списки: `_PUBLIC_PATHS` ниже держится
 # синхронным с прод-версией руками (расхождение уже случалось — в копии не было
 # /api/v1/ping), и никакой механики, которая бы это гарантировала, нет. Прод-список
 # параметризован в test_auth_session_guard.py, поэтому его расширение хотя бы там
 # проверяется автоматически.
 # ---------------------------------------------------------------------------
 _ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
-# Синхронно с app.main._PUBLIC_PATHS (там же и /api/v1/ping — он был потерян здесь).
+_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
 _PUBLIC_PATHS = frozenset({"/health", "/api/v1/ping", "/docs", "/redoc", "/openapi.json"})
 def _build_test_app() -> FastAPI: