gendesign/tradein-mvp/backend/app/core/rbac.py

"""RBAC guard middleware — extracted from ``app/main.py``.

Historically ``rbac_guard`` lived inline in ``app/main.py`` and the test suite
(``tests/test_rbac.py``, ``tests/test_internal_auth_secret.py``) kept a
hand-maintained *copy* of it, labelled "MIRROR of app.main — keep in sync
manually". The copy drifted: it was missing the #2213
``X-Internal-Auth-Secret`` defense-in-depth check that the real guard has,
so a regression in that check would NOT have failed CI.

This module holds the real guard. Historically it had "no DB/lifespan/scheduler
side effects" beyond ``app.core.auth``/``app.core.config`` (both side-effect-free
at import time). #2552 (dual-mode DB-session auth) adds a conditional per-request
DB round trip via ``app.services.identity_store.identity_session`` — но ТОЛЬКО
когда запрос реально несёт session-cookie
(``request.cookies.get(settings.session_cookie_name)``); без cookie (весь
существующий тестовый трафик, legacy Caddy trusted-header запросы) ветка не
выполняется — ноль новых DB-побочных эффектов для старых путей. ``app/main.py``
and the test apps both import THIS module, so tests exercise the exact
production code path instead of a copy that can silently fall out of sync.

Сессия открывается через ``identity_session()``, а не через
``app.core.db.SessionLocal`` напрямую: guard — middleware, FastAPI-DI здесь нет,
а реестр людей при ``IDENTITY_STORE=auth`` лежит в другой БД. В дефолтном режиме
``identity_session()`` открывает ровно ``app.core.db.SessionLocal()`` — тот же
коннект-пул и то же поведение, что до эпика «единый вход».
"""

from __future__ import annotations

import logging
import re
import secrets
from collections.abc import Awaitable, Callable
from typing import Any

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

from app.core.auth import get_role, is_path_allowed
from app.core.config import settings
from app.services.auth_session import get_db_role_scope, get_session_user
from app.services.identity_store import identity_session

logger = logging.getLogger(__name__)

# RBAC: defense-in-depth поверх Caddy basic_auth + X-Authenticated-User
# (см. app/core/auth.py + auth/roles.yaml). Правила:
#   1) Любой non-public path требует X-Authenticated-User — иначе 401.
#   2) Юзер должен быть в roles.yaml — иначе 403 («неизвестный юзер ничего
#      не видит» — decided 2026-05-25).
#   3) /api/v1/admin/* (= внешний /trade-in/api/v1/admin/* после Caddy
#      `uri strip_prefix /trade-in`) — только role=admin, иначе 403.
# Public paths без auth (/health, /docs, /openapi.json) пропускаем —
# X-Authenticated-User там не приходит из Caddy.
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
# #2552: /api/v1/auth/login + /logout — по определению вызываются ДО того, как
# клиент аутентифицирован (login) или могут вызываться с уже протухшей/отсутствующей
# сессией (logout — должен уметь чистить stale cookie без валидной auth). Свой
# rate-limit у /login отдельный (app.api.v1.auth._LOGIN_LIMITER), RateLimitMiddleware
# на /api/* всё равно применяется — это ослабляет ТОЛЬКО rbac_guard'овский
# auth-required gate, не остальные защиты.
#
# Инцидент 2026-07-31: /api/v1/trade-in/support/anon/* — по той же логике. Единственным
# каналом в поддержку был чат ЗА логином, а типовая причина писать в поддержку —
# «не могу войти» (в тот день так и вышло: «Практика» билась в форму весь день и
# достучаться из продукта не могла). Ветка НЕ трогает авторизованные
# /api/v1/trade-in/support/* — те по-прежнему требуют identity; у анонимной свой,
# заведомо более узкий бюджет (per-token + per-IP, см. app.api.v1.support) и своя
# идентичность из httpOnly-куки, которая структурно не может совпасть с чьим-то
# логином.
_PUBLIC_PATHS = frozenset(
    {
        "/health",
        # Публичен ЗДЕСЬ и не публичен снаружи — это два разных периметра.
        # Снимает `/metrics` агент Alloy изнутри docker-сети, где заголовка
        # `X-Authenticated-User` нет ни у кого; без записи в этом множестве
        # скрейп получал бы 401 и метрик не было бы вовсе. Наружу путь не
        # открывается: у `gendsgn.ru` бэкенду «Меры» отдаётся только
        # `/trade-in/api/*`, у `meraocenka.ru` работает белый список, и в обоих
        # блоках на `/metrics` стоит явный `respond 404`.
        "/metrics",
        "/docs",
        "/redoc",
        "/openapi.json",
        "/api/v1/auth/login",
        "/api/v1/auth/logout",
        # NB: префикс — /api/v1/trade-in (app/main.py include_router), а Caddy
        # срезает ВНЕШНИЙ /trade-in ещё раньше. Т.е. снаружи это
        # /trade-in/api/v1/trade-in/support/anon/*, сюда приходит вот такое.
        "/api/v1/trade-in/support/anon/messages",
        "/api/v1/trade-in/support/anon/unread",
        "/api/v1/trade-in/support/anon/read",
        # Версионирование (VERSION-файл + build-args, см. app/core/version.py):
        # не секрет, читает только process env — быстрая справка для клиента/
        # поддержки/смоук-теста, не должна требовать сессию.
        "/api/v1/trade-in/version",
        # GlitchTip webhook → Telegram (app/api/v1/glitchtip.py): вызывается
        # ИЗ glitchtip-worker (docker-сеть gendesign_shared), не может нести
        # X-Authenticated-User/сессию. Путь фиксированный и не секрет — секрет
        # это query-параметр `secret`, который проверяет сам хендлер
        # (secrets.compare_digest против TRADEIN_INTERNAL_AUTH_SECRET).
        "/api/v1/trade-in/ops/glitchtip-webhook",
        # Публичный B2C-периметр МЕРЫ (meraocenka.ru): у посетителя лендинга
        # идентичности нет и не будет — Caddy на этом домене вообще без
        # basic_auth. Обе ручки только читают (SELECT/прокси автокомплита) и не
        # пишут в БД адрес физлица, поэтому открываются ДО контура согласия
        # 152-ФЗ (#2895), а платный расчёт — только вместе с ним. Разбор
        # периметра и бюджетов — app/api/public/mera.py.
        #
        # Строки ТОЧНЫЕ, префиксной ветки здесь нет: множество проверяется как
        # `path in _PUBLIC_PATHS`, и превращать его в префиксный матчер ради
        # двух путей значит расширить механизм, от которого зависит auth-гейт
        # всего бэкенда. Гарантия «под /api/public/ не лежит ничего закрытого»
        # держится на структуре пакета app/api/public/, а не на матчере.
        "/api/public/mera/suggest",
        "/api/public/mera/coverage",
    }
)
# #R2-H3: Caddy срезает внешний префикс /trade-in (uri strip_prefix) перед
# tradein-backend, а globs в roles.yaml — ВНЕШНИЕ (/trade-in/api/v1/**). Для
# scope-проверки восстанавливаем внешний путь.
_EXTERNAL_PREFIX = "/trade-in"
# Bootstrap-пути, доступные ЛЮБОМУ известному юзеру независимо от роли: /me отдаёт
# роль (expired → trial-экран), /brand/* — брендинг login/trial-экрана. Без них
# expired (roles.yaml paths:[] deny:/**) не получил бы роль и не увидел trial-экран.
_RBAC_BOOTSTRAP_EXEMPT = ("/api/v1/me", "/api/v1/brand")


def _db_glob_match(pattern: str, path: str) -> bool:
    """Мини-матчер для фиксированного набора DB-role паттернов
    (``app.services.auth_session.DB_ROLE_PATHS`` — только формы ``/**`` и
    ``<prefix>/**``, не нужна полная semantics ``app.core.auth._glob_to_regex``
    — тот модуль private и MIRROR'ится вручную с основным бэкендом, лишний
    импорт private-символа оттуда увеличивал бы drift-риск)."""
    if pattern == "/**":
        return True
    if pattern.endswith("/**"):
        prefix = pattern[: -len("/**")]
        return path == prefix or path.startswith(prefix + "/")
    return path == pattern


def _db_role_path_allowed(role: str, path: str) -> bool:
    paths, deny = get_db_role_scope(role)
    if any(_db_glob_match(p, path) for p in deny):
        return False
    return any(_db_glob_match(p, path) for p in paths)


def _propagate_authenticated_user(request: Request, username: str) -> None:
    """Инжектит ``X-Authenticated-User`` в ASGI scope — ПЕРЕЗАПИСЫВАЯ, а не
    только добавляя при отсутствии, — чтобы ``RateLimitMiddleware``/
    ``RequestAuditMiddleware`` (оба читают сырой заголовок напрямую,
    #2213/#2550) и downstream route-хендлеры (читающие его через FastAPI
    ``Header()``) видели РЕЗОЛВЛЕННОГО ИЗ СЕССИИ юзера — без правок в каждом
    из этих мест по отдельности (минимально инвазивный способ).

    #2552 post-review fix (CRITICAL): раньше это была skip-if-present
    мутация (``if request.headers.get(...): return``) — сессия резолвилась
    ПЕРВОЙ (см. rbac_guard), но клиент-контролируемый ``X-Authenticated-User``
    (который Caddy шлёт на КАЖДЫЙ прод-запрос) выигрывал у неё для ВСЕГО
    downstream-трафика: атакующий с валидной cookie юзера ``alice`` мог
    подделать заголовок ``X-Authenticated-User: victim`` и получить доступ к
    данным victim в ~15 роутах, читающих заголовок напрямую
    (``_assert_estimate_access*``, ``account_quota``, ``/trade-in/history``,
    ``support.py``) — работало в ОБОИХ auth_mode (dual и db_only), т.к. эти
    хендлеры не знают про rbac_guard'овский ``from_session`` флаг, только про
    сырой заголовок. Session-identity ДОЛЖНА быть источником истины, если
    сессия резолвлена — полная перезапись, не skip.

    Механизм: ``request.scope`` — ОДИН и тот же dict-объект, прокинутый по
    ссылке через весь ASGI call chain (Starlette не копирует scope между
    слоями middleware). Мутация ``scope["headers"]`` ЗДЕСЬ видна:
      - downstream call_next() цепочке (ExceptionMiddleware → Router →
        endpoint) — т.к. rbac_guard мутирует scope ДО вызова call_next();
      - ``RequestAuditMiddleware`` — он внешний относительно rbac_guard
        (см. app/main.py: последний ``add_middleware`` оборачивает
        предыдущие) и читает ``request.headers`` уже ПОСЛЕ ``call_next()``
        отработал весь внутренний стек, включая эту мутацию.

    ASGI header-имена — всегда lowercase bytes (см. ASGI spec), поэтому
    фильтр по ``b"x-authenticated-user"`` ловит заголовок независимо от
    регистра, в котором его прислал клиент (Starlette уже нормализует).

    Известное ограничение: ``RateLimitMiddleware`` тоже внешний относительно
    rbac_guard, но читает заголовок ДО вызова call_next() (до того, как этот
    guard успевает отработать) — для ЭТОГО конкретного запроса сессионный
    юзер лимитируется по IP, а не по username (per-user множитель не
    применяется). Не регрессия (IP-лимит применялся бы и раньше — до
    добавления session-auth такие запросы вообще были 401), просто более
    строгий бюджет специфично для session-cookie-запросов; при необходимости
    точного per-user квотинга для DB-юзеров — переносить резолв сессии выше
    RateLimit в app/main.py отдельным issue.
    """
    request.scope["headers"] = [
        (k, v) for k, v in request.scope.get("headers", []) if k != b"x-authenticated-user"
    ] + [(b"x-authenticated-user", username.encode("latin-1", "replace"))]


async def rbac_guard(
    request: Request,
    call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
    path = request.url.path
    if path in _PUBLIC_PATHS:
        return await call_next(request)

    username: str | None = None
    role: str | None = None
    from_session = False

    # #2552: session-cookie резолвится ПЕРВЫМ. Если cookie нет вообще —
    # request.cookies.get() возвращает None без единого похода в БД (ноль
    # side-effects для всего существующего трафика без cookie).
    token = request.cookies.get(settings.session_cookie_name)
    if token:
        session_user: dict[str, Any] | None = None
        try:
            with identity_session() as db:
                session_user = get_session_user(db, token)
        except Exception:
            # Сюда попадает и AuthDatabaseNotConfiguredError (IDENTITY_STORE=auth
            # без AUTH_DATABASE_URL): резолв сессии не состоялся, дальше работает
            # тот же путь, что и при любом сбое БД, — auth_mode решает, пускать ли
            # legacy trusted-header.
            #
            # ⚠️ Этот except НЕ должен быть тем, что ловит сломанный DSN: молча
            # деградировать в legacy trusted-header означало бы раздавать права
            # из roles.yaml в обход реестра (включая аккаунты с access_state
            # 'disabled'/'trial_expired'), причём сутками — продуктовая БД жива,
            # приложение работоспособно, сигнал только в логах. Поэтому
            # конфигурацию проверяет lifespan (app/main.py): при
            # IDENTITY_STORE=auth пустой DSN роняет СТАРТ. Здесь остаётся второй
            # рубеж — реестр, отвалившийся уже после успешного старта, не имеет
            # права отдавать 500.
            logger.exception("RBAC: session lookup failed for %s", path)
        if session_user is not None:
            username = session_user["username"]
            role = session_user["role"]
            from_session = True
            _propagate_authenticated_user(request, username)

    if not from_session:
        # auth_mode == "db_only" — легаси trusted-header путь ПОЛНОСТЬЮ
        # отключён, даже если валидный X-Authenticated-User присутствует.
        if settings.auth_mode != "dual":
            return JSONResponse(
                status_code=401,
                content={"detail": "valid session required"},
            )

        # ---- legacy trusted-header path — BIT-FOR-BIT как было до #2552 ----
        username = request.headers.get("X-Authenticated-User")
        if not username:
            return JSONResponse(
                status_code=401,
                content={"detail": "no authenticated user (valid session required)"},
            )

        # #2213 defense-in-depth: если общий секрет задан — запрос с X-Authenticated-User
        # ОБЯЗАН нести валидный X-Internal-Auth-Secret (его добавляет Caddy из env).
        # Иначе это подделка заголовка мимо Caddy (напр. изнутри gendesign_shared) → 401.
        # Constant-time compare против timing-атак. Пусто = защита не активна (fail-open).
        secret = settings.tradein_internal_auth_secret
        if secret:
            provided = request.headers.get("X-Internal-Auth-Secret", "")
            if not secrets.compare_digest(provided, secret):
                logger.warning(
                    "RBAC: X-Authenticated-User=%r без валидного X-Internal-Auth-Secret "
                    "на %s — возможная подделка заголовка мимо Caddy",
                    username,
                    path,
                )
                return JSONResponse(
                    status_code=401,
                    content={"detail": "invalid or missing internal auth secret"},
                )

        try:
            role = get_role(username)
        except KeyError:
            logger.warning("RBAC: unknown user %r tried %s", username, path)
            return JSONResponse(
                status_code=403,
                content={"detail": "user not in roles config"},
            )

    assert username is not None
    assert role is not None

    if _ADMIN_API_RE.match(path) and role != "admin":
        logger.info("RBAC: blocked %s (role=%s) from %s", username, role, path)
        return JSONResponse(
            status_code=403,
            content={"detail": "admin only"},
        )

    # #R2-H3: энфорсим scope (paths/deny) для ВСЕХ non-admin путей, а не
    # только /admin/*. Bootstrap-пути (/me, /brand) исключены — иначе revoked/
    # scope-narrowed юзер не смог бы получить свою роль вовсе.
    if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
        external_path = _EXTERNAL_PREFIX + path
        if from_session:
            allowed = _db_role_path_allowed(role, external_path)
        else:
            try:
                allowed = is_path_allowed(role, external_path)
            except Exception:
                logger.exception(
                    "RBAC scope-check raised for %s %s (ext=%s) — fail-open",
                    username,
                    path,
                    external_path,
                )
                allowed = True
        if not allowed:
            logger.info(
                "RBAC: scope-blocked %s (role=%s) from %s (ext=%s)",
                username,
                role,
                path,
                external_path,
            )
            return JSONResponse(
                status_code=403,
                content={"detail": "forbidden for role"},
            )

    return await call_next(request)