fix(tradein/tests): тесты авторизации проверяют настоящий guard + реальный рендер PDF #2541

Merged
lekss361 merged 1 commit from fix/tradein-audit-tests into main 2026-07-26 23:04:44 +00:00
6 changed files with 396 additions and 212 deletions

View file

@ -0,0 +1,136 @@
"""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 with no DB/lifespan/scheduler side effects
(only ``app.core.auth`` + ``app.core.config``, both side-effect-free at
import time beyond requiring ``DATABASE_URL`` in the environment for
``Settings()``). ``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.
"""
from __future__ import annotations
import logging
import re
import secrets
from collections.abc import Awaitable, Callable
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
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/")
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
# #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")
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 = request.headers.get("X-Authenticated-User")
if not username:
return JSONResponse(
status_code=401,
content={"detail": "no authenticated user (Caddy basic_auth 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"},
)
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: энфорсим roles.yaml scope (paths/deny) для ВСЕХ non-admin путей, а не
# только /admin/*. Иначе revoked (role=expired, paths:[] deny:/**) или узко-
# скоупленный аккаунт достаёт non-admin API (напр. POST /api/v1/search —
# экспорт листингов), который roles.yaml ему запрещает. Bootstrap-пути (/me,
# /brand) исключены выше по списку. roles.yaml globs внешние → восстанавливаем
# внешний путь (Caddy срезал /trade-in). На сбой парса — fail-open + громкий
# лог: не лочим платящего pilot из-за конфиг-бага (admin-гейт выше остаётся).
if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
external_path = _EXTERNAL_PREFIX + path
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)

View file

@ -8,15 +8,12 @@ from __future__ import annotations
import logging
import os
import re
import secrets
from collections.abc import AsyncGenerator, Awaitable, Callable
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
import sentry_sdk
from fastapi import FastAPI, Request
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse, Response
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.httpx import HttpxIntegration
from sentry_sdk.integrations.logging import LoggingIntegration
@ -35,11 +32,11 @@ from app.api.v1 import (
support,
trade_in,
)
from app.core.auth import get_role, is_path_allowed
from app.core.config import settings
from app.core.db import SessionLocal
from app.core.fdw import ensure_fdw_user_mapping
from app.core.ratelimit import RateLimitMiddleware
from app.core.rbac import rbac_guard
from app.core.request_audit import RequestAuditMiddleware
from app.observability.sentry_scrub import scrub_pii_event
@ -138,104 +135,9 @@ app = FastAPI(
# не видит» — 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/")
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
# #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")
@app.middleware("http")
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 = request.headers.get("X-Authenticated-User")
if not username:
return JSONResponse(
status_code=401,
content={"detail": "no authenticated user (Caddy basic_auth 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"},
)
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: энфорсим roles.yaml scope (paths/deny) для ВСЕХ non-admin путей, а не
# только /admin/*. Иначе revoked (role=expired, paths:[] deny:/**) или узко-
# скоупленный аккаунт достаёт non-admin API (напр. POST /api/v1/search —
# экспорт листингов), который roles.yaml ему запрещает. Bootstrap-пути (/me,
# /brand) исключены выше по списку. roles.yaml globs внешние → восстанавливаем
# внешний путь (Caddy срезал /trade-in). На сбой парса — fail-open + громкий
# лог: не лочим платящего pilot из-за конфиг-бага (admin-гейт выше остаётся).
if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
external_path = _EXTERNAL_PREFIX + path
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)
# Guard body живёт в app/core/rbac.py (без DB/lifespan side effects), чтобы
# тесты могли импортировать РЕАЛЬНЫЙ guard вместо hand-maintained копии.
app.middleware("http")(rbac_guard)
app.add_middleware(

View file

@ -0,0 +1,18 @@
"""Repo-wide test config for tradein-mvp/backend.
Currently only registers custom pytest markers so they don't emit
PytestUnknownMarkWarning when used (`--strict-markers` is not enabled in
pyproject.toml, so an unregistered marker would only warn, not fail this
just keeps output clean and documents intent in one place).
"""
from __future__ import annotations
def pytest_configure(config) -> None:
config.addinivalue_line(
"markers",
"pdf_render: real (non-mocked) WeasyPrint render — needs native "
"Pango/cairo/GObject libs, self-skips where unavailable (see "
"tests/test_pdf_real_render.py docstring for how to run it for real).",
)

View file

@ -2,13 +2,16 @@
Backend доверяет X-Authenticated-User (его ставит Caddy). На общей docker-сети
gendesign_shared любой контейнер мог бы отправить поддельный
`X-Authenticated-User: admin` напрямую на tradein-backend:8000. Общий секрет
X-Internal-Auth-Secret закрывает дыру: если TRADEIN_INTERNAL_AUTH_SECRET задан,
каждый запрос с X-Authenticated-User обязан нести валидный секрет (constant-time),
`X-Authenticated-User: admin` напрямую на tradein-backend:8000, минуя Caddy.
Общий секрет X-Internal-Auth-Secret закрывает дыру: если TRADEIN_INTERNAL_AUTH_SECRET
задан, каждый запрос с X-Authenticated-User обязан нести валидный секрет (constant-time),
иначе 401. Пусто = fail-open (backward-compat до провижининга).
MIRROR of rbac_guard из app/main.py включая #2213 secret-gate. Держим копию
здесь (как test_rbac.py), чтобы не тянуть тяжёлый app.main (lifespan/DB/scheduler).
Использует РЕАЛЬНЫЙ rbac_guard (app/core/rbac.py) тот же, что регистрирует
app/main.py в проде. Раньше здесь была hand-maintained копия ("MIRROR of
rbac_guard из app/main.py"); соседний test_rbac.py держал СВОЮ отдельную
копию, которая успела отстать (потеряла именно этот secret-gate) регрессия
в реальном guard'е могла бы пройти CI незамеченной. См. app/core/rbac.py.
"""
from __future__ import annotations
@ -17,20 +20,13 @@ import os
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import re
import secrets
from collections.abc import Awaitable, Callable
import pytest
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.core import auth as auth_mod
from app.core import config
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
from app.core.rbac import rbac_guard
@pytest.fixture(autouse=True)
@ -39,44 +35,9 @@ def _reset_auth_cache() -> None:
def _build_test_app() -> FastAPI:
"""Копия rbac_guard из app/main.py (с #2213 secret-gate)."""
"""Test app используя РЕАЛЬНЫЙ rbac_guard (с #2213 secret-gate)."""
app = FastAPI()
@app.middleware("http")
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 = request.headers.get("X-Authenticated-User")
if not username:
return JSONResponse(
status_code=401,
content={"detail": "no authenticated user (Caddy basic_auth required)"},
)
secret = config.settings.tradein_internal_auth_secret
if secret:
provided = request.headers.get("X-Internal-Auth-Secret", "")
if not secrets.compare_digest(provided, secret):
return JSONResponse(
status_code=401,
content={"detail": "invalid or missing internal auth secret"},
)
try:
role = auth_mod.get_role(username)
except KeyError:
return JSONResponse(
status_code=403,
content={"detail": "user not in roles config"},
)
if _ADMIN_API_RE.match(path) and role != "admin":
return JSONResponse(status_code=403, content={"detail": "admin only"})
return await call_next(request)
app.middleware("http")(rbac_guard)
@app.get("/api/v1/ping")
async def ping() -> dict:

View file

@ -0,0 +1,209 @@
"""Real (non-mocked) WeasyPrint render invariants for the Trade-In PDF report.
tests/test_pdf_security.py exercises only the HTML-builder layer with
WeasyPrint stubbed out (``sys.modules['weasyprint'] = MagicMock()``) by
design, so those tests run everywhere without needing WeasyPrint's native
GTK/Pango/cairo libs. That design has a blind spot: it can NEVER catch a real
WeasyPrint pagination regression e.g. the bug fixed in commit 42a50cf8
("running @page header/footer → ровно 4 страницы (без пустых)"), where the
persistent running header/footer produced an extra TRAILING BLANK 5th page
instead of the documented "Структура отчёта — 4 страницы" invariant (see
app/services/exporters/trade_in_pdf.py module docstring). A mocked
WeasyPrint can never lay out/paginate anything, so it structurally cannot see
that class of bug only a real render can.
Requires WeasyPrint's native dependencies (Pango/cairo/GObject). These are
NOT installed in this dev sandbox (Windows, no libgobject-2.0-0) and NOT
installed on the bare ``ubuntu-latest`` runner used by
.forgejo/workflows/ci-tradein.yml (no apt-get step there see that file).
This whole module self-skips via ``pytest.skip(..., allow_module_level=True)``
the moment the real `import weasyprint` fails for ANY reason (missing native
lib, etc.), so it is a harmless no-op everywhere it can't actually run.
How to run it for real:
- Inside the prod/dev docker image the `runner` stage of
tradein-mvp/backend/Dockerfile installs libcairo2 + libpango-1.0-0 +
libpangoft2-1.0-0 + fonts-dejavu-core, so WeasyPrint's native deps are
present there:
docker exec tradein-backend python -m pytest -q -m pdf_render \
tests/test_pdf_real_render.py
- Locally on a Linux box with WeasyPrint's system deps installed (see
https://doc.courtbouillon.org/weasyprint/stable/first_steps.html):
uv run pytest -q -m pdf_render tests/test_pdf_real_render.py
- Marked ``@pytest.mark.pdf_render`` (registered in tests/conftest.py) so it
can be explicitly selected/excluded once a CI runner with the native libs
exists; today ci-tradein.yml's `ubuntu-latest` runner doesn't have them,
so this module simply self-skips there nothing to deselect.
"""
from __future__ import annotations
import os
import sys
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
# test_pdf_security.py deliberately does `sys.modules['weasyprint'] = MagicMock()`
# for its own unit tests. sys.modules is a process-global cache: if that module
# was imported before this one in the same pytest run, `import weasyprint` below
# would silently return the Mock instead of the real package. Purge any
# existing stub before attempting the real import — independent of collection
# order across test modules.
for _name in [n for n in sys.modules if n == "weasyprint" or n.startswith("weasyprint.")]:
del sys.modules[_name]
import pytest # noqa: E402
try:
import weasyprint
except Exception as exc: # pragma: no cover - env-dependent (native GTK/Pango/cairo libs)
pytest.skip(
f"WeasyPrint native deps unavailable, skipping real-render tests: {exc}",
allow_module_level=True,
)
pytestmark = pytest.mark.pdf_render
import unittest.mock as _mock # noqa: E402
from datetime import UTC, datetime, timedelta # noqa: E402
from uuid import uuid4 # noqa: E402
from app.schemas.trade_in import AggregatedEstimate # noqa: E402
from app.services.brand import Brand # noqa: E402
from app.services.exporters import trade_in_pdf as mod # noqa: E402
_GENERIC = Brand(
slug="generic",
name="Trade-In",
logo_url=None,
primary_color="#1d4ed8",
accent_color="#f59e0b",
footer_text=None,
pdf_disclaimer=None,
)
_SNAPSHOT = {
"address": "Екатеринбург, ул. Ленина, 1",
"area_m2": 50.0,
"rooms": 2,
"floor": 3,
"total_floors": 9,
"year_built": 2010,
"house_type": "panel",
"repair_state": "standard",
"has_balcony": True,
}
def _estimate(**overrides) -> AggregatedEstimate:
base = dict(
estimate_id=uuid4(),
median_price_rub=10_000_000,
range_low_rub=9_000_000,
range_high_rub=11_000_000,
median_price_per_m2=200_000,
confidence="high",
n_analogs=15,
period_months=24,
analogs=[],
actual_deals=[],
expires_at=datetime.now(UTC) + timedelta(days=30),
)
base.update(overrides)
return AggregatedEstimate(**base)
def _zero_estimate(**overrides) -> AggregatedEstimate:
"""Оценка с median=0 — insufficient_data=True (одностраничный empty-state)."""
base = dict(
estimate_id=uuid4(),
median_price_rub=0,
range_low_rub=0,
range_high_rub=0,
median_price_per_m2=0,
confidence="low",
n_analogs=0,
period_months=24,
analogs=[],
actual_deals=[],
expires_at=datetime.now(UTC) + timedelta(days=30),
)
base.update(overrides)
return AggregatedEstimate(**base)
def _render_real_document(estimate, snapshot, brand):
"""Call the REAL generate_trade_in_pdf (no mocking of WeasyPrint), capturing
the actual weasyprint.HTML/CSS instances + html string it constructs via
thin capturing subclasses so we can additionally call .render() on the
exact same HTML instance to get a weasyprint.Document (whose .pages is the
public, documented page-count API), without duplicating trade_in_pdf.py's
own html_str/css_str assembly logic here (that would drift out of sync
with the real function, defeating the point of this test).
Returns (document, pdf_bytes, html_string).
"""
html_instances: list[weasyprint.HTML] = []
css_instances: list[weasyprint.CSS] = []
html_strings: list[str] = []
class _CapturingHTML(weasyprint.HTML):
def __init__(self, *a, **kw):
html_strings.append(kw.get("string") if "string" in kw else (a[0] if a else None))
super().__init__(*a, **kw)
html_instances.append(self)
class _CapturingCSS(weasyprint.CSS):
def __init__(self, *a, **kw):
super().__init__(*a, **kw)
css_instances.append(self)
with (
_mock.patch.object(weasyprint, "HTML", _CapturingHTML),
_mock.patch.object(weasyprint, "CSS", _CapturingCSS),
):
pdf_bytes = mod.generate_trade_in_pdf(estimate, snapshot, brand=brand)
assert html_instances, "HTML() was never constructed by generate_trade_in_pdf"
assert css_instances, "CSS() was never constructed by generate_trade_in_pdf"
document = html_instances[-1].render(stylesheets=[css_instances[-1]])
return document, pdf_bytes, html_strings[-1]
def test_real_pdf_has_exactly_4_pages_no_trailing_blank() -> None:
"""Regression for commit 42a50cf8: the running @page header/footer must
NOT produce a 5th trailing blank page. trade_in_pdf.py's own module
docstring documents "Структура отчёта — 4 страницы" as the fixed
cover/listings/deals/offer composition that count IS the authoritative
"no empty pages" invariant for this report (any blank page shows up as an
extra page beyond these 4 named sections)."""
est = _estimate(n_analogs=12, sources_used=["avito"])
document, pdf_bytes, _html_str = _render_real_document(est, _SNAPSHOT, _GENERIC)
assert len(document.pages) == 4, (
f"expected exactly 4 pages (cover/listings/deals/offer), got "
f"{len(document.pages)} — likely a trailing/leading blank-page regression"
)
assert pdf_bytes.startswith(b"%PDF-"), "write_pdf must produce a real PDF"
def test_real_pdf_insufficient_data_is_single_page() -> None:
"""insufficient_data=True → one-page empty-state, not the 4-page report."""
est = _zero_estimate()
document, _pdf_bytes, html_str = _render_real_document(est, _SNAPSHOT, _GENERIC)
assert len(document.pages) == 1
assert "Недостаточно данных" in html_str
def test_real_pdf_key_blocks_present_exactly_once() -> None:
"""Key section headings for each of the 4 pages are present in the actual
composed HTML that WeasyPrint rendered (integration-level check the
mocked tests in test_pdf_security.py already verify each builder function
in isolation; this catches a composition bug where wiring them together
in generate_trade_in_pdf silently drops or duplicates a section)."""
est = _estimate(n_analogs=12, sources_used=["avito"])
document, _pdf_bytes, html_str = _render_real_document(est, _SNAPSHOT, _GENERIC)
assert len(document.pages) == 4
markers = ["РЫНОК КВАРТИР", "ФОРМИРОВАНИЕ ВЫКУПНОЙ СТОИМОСТИ"]
for marker in markers:
found = html_str.count(marker)
assert found == 1, f"expected exactly one {marker!r} block, found {found}"

View file

@ -1,7 +1,5 @@
"""RBAC unit + integration tests for tradein backend.
MIRROR of backend/tests/test_rbac.py kept in sync manually.
Coverage:
- get_role / get_user_scope happy + error paths
- is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**)
@ -14,22 +12,29 @@ Coverage:
Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не
тянуть тяжёлые модули из app.main (lifespan task + DB session + scheduler).
RBAC middleware и /me router импортируются напрямую это даёт нам ровно
тот же поведение что в проде.
Guard больше НЕ мокается/копируется вручную импортируется тот же
``app.core.rbac.rbac_guard``, что регистрирует app/main.py в проде. Раньше
здесь была hand-maintained "MIRROR of app.main" копия, которая незаметно
отстала (не имела #2213 X-Internal-Auth-Secret check) — правки реального
guard'а тесты бы не заметили. См. app/core/rbac.py.
"""
from __future__ import annotations
import re
from collections.abc import Awaitable, Callable
import os
# Settings (app.core.config, импортируемый через app.core.rbac) требует
# DATABASE_URL на конструирование — stub перед любым app-импортом (тот же
# паттерн что и в остальных tests/*.py).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
import pytest
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, Response
from fastapi import FastAPI
from fastapi.testclient import TestClient
from app.api.v1 import me as me_router
from app.core import auth as auth_mod
from app.core.rbac import rbac_guard
@pytest.fixture(autouse=True)
@ -38,60 +43,13 @@ def _reset_auth_cache() -> None:
# ---------------------------------------------------------------------------
# Test app — копия rbac_guard из app/main.py.
# Test app — использует РЕАЛЬНЫЙ rbac_guard (app/core/rbac.py), а не копию.
# ---------------------------------------------------------------------------
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
# MIRROR of app.main (#R2-H3): keep in sync with the real rbac_guard.
_EXTERNAL_PREFIX = "/trade-in"
_RBAC_BOOTSTRAP_EXEMPT = ("/api/v1/me", "/api/v1/brand")
def _build_test_app() -> FastAPI:
app = FastAPI()
@app.middleware("http")
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 = request.headers.get("X-Authenticated-User")
if not username:
return JSONResponse(
status_code=401,
content={"detail": "no authenticated user (Caddy basic_auth required)"},
)
try:
role = auth_mod.get_role(username)
except KeyError:
return JSONResponse(
status_code=403,
content={"detail": "user not in roles config"},
)
if _ADMIN_API_RE.match(path) and role != "admin":
return JSONResponse(
status_code=403,
content={"detail": "admin only"},
)
# #R2-H3 mirror: enforce roles.yaml scope on non-bootstrap paths.
if not path.startswith(_RBAC_BOOTSTRAP_EXEMPT):
external_path = _EXTERNAL_PREFIX + path
try:
allowed = auth_mod.is_path_allowed(role, external_path)
except Exception:
allowed = True
if not allowed:
return JSONResponse(
status_code=403,
content={"detail": "forbidden for role"},
)
return await call_next(request)
app.middleware("http")(rbac_guard)
app.include_router(me_router.router, prefix="/api/v1", tags=["me"])