feat(sitefinder): «Птица» принимает сессию общего реестра [PR-2c/6] #2635

Merged
lekss361 merged 1 commit from feat/auth-ptica-session-guard into main 2026-08-02 14:18:24 +00:00
8 changed files with 2720 additions and 10 deletions

269
backend/app/core/auth_db.py Normal file
View file

@ -0,0 +1,269 @@
"""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()

View file

@ -1,10 +1,46 @@
import os
import warnings
from typing import Annotated
from typing import Annotated, Literal
from urllib.parse import quote
from pydantic import field_validator, model_validator
from pydantic import SecretStr, 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")
@ -371,5 +407,201 @@ 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()

View file

@ -3,11 +3,14 @@
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
@ -41,10 +44,12 @@ 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__)
@ -97,6 +102,18 @@ 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
@ -122,10 +139,212 @@ 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,
@ -134,6 +353,17 @@ 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)
@ -141,22 +371,67 @@ async def rbac_guard(
if path in _PUBLIC_PATHS:
return await call_next(request)
username = request.headers.get("X-Authenticated-User")
if not username:
# Любой non-public path без auth-header → 401. Локальный curl мимо Caddy
# или прокси-фронт без header_up. 401 точнее чем 403 — "сначала
# аутентифицируйся".
# Внешний `if` по режиму — не дубль проверки внутри resolve_session_token(), а
# гарантия инварианта «legacy = поведение не меняется ни на байт»: в нём не
# трогается даже request.cookies (разбор Cookie-заголовка).
token = (
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": "no authenticated user (Caddy basic_auth required)"},
content={"detail": "valid session 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: «человек без ролей вообще ничего не видит».
logger.warning("RBAC: unknown user %r tried %s", username, path)
if session_username is not None:
# Тот же отказ, но отдельным сообщением: «есть в реестре, нет в 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"},

View file

@ -0,0 +1,256 @@
"""Резолв сессионной куки общего реестра (БД `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)

View file

@ -0,0 +1,497 @@
"""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"]

View file

@ -0,0 +1,861 @@
"""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

View file

@ -0,0 +1,305 @@
"""Резолв сессии общего реестра — `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)

View file

@ -42,11 +42,26 @@ 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/")
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
# Синхронно с app.main._PUBLIC_PATHS (там же и /api/v1/ping — он был потерян здесь).
_PUBLIC_PATHS = frozenset({"/health", "/api/v1/ping", "/docs", "/redoc", "/openapi.json"})
def _build_test_app() -> FastAPI: