fix(tradein/tests): stop mirroring rbac_guard, add real WeasyPrint page-count regression test
All checks were successful
CI Trade-In / changes (pull_request) Successful in 11s
CI / changes (pull_request) Successful in 12s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 5m27s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
All checks were successful
CI Trade-In / changes (pull_request) Successful in 11s
CI / changes (pull_request) Successful in 12s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 5m27s
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
test_rbac.py and test_internal_auth_secret.py each kept a hand-maintained
"MIRROR of app.main" copy of rbac_guard. The copies had already drifted:
test_rbac.py's copy was missing the #2213 X-Internal-Auth-Secret
defense-in-depth check entirely, so a regression in the real guard would not
have failed CI. Extracted rbac_guard (+ its constants) into app/core/rbac.py
(no DB/lifespan side effects) so app/main.py and both test files import and
exercise the exact same production code path instead of copies.
Verified the fix actually catches regressions: temporarily neutered the
secret-gate check in app/core/rbac.py, confirmed
test_internal_auth_secret.py went red (2 failures), then reverted — suite
back to green (31/31).
Also added tests/test_pdf_real_render.py: test_pdf_security.py stubs
WeasyPrint entirely, so it structurally cannot catch a real pagination
regression like the one just fixed in commit 42a50cf8 (extra trailing blank
5th page from the running @page header/footer). The new module renders the
REAL PDF via generate_trade_in_pdf and asserts the documented "4 pages, no
blank" invariant directly. It needs WeasyPrint's native Pango/cairo/GObject
libs (absent on this Windows sandbox and on ci-tradein.yml's bare
ubuntu-latest runner), so it self-skips via
pytest.skip(allow_module_level=True) wherever those aren't present, and runs
for real only where they are (e.g. `docker exec tradein-backend python -m
pytest -m pdf_render tests/test_pdf_real_render.py`) — see its docstring.
Existing mocked security tests are untouched (they check different things).
tests/conftest.py added (didn't exist before) to register the pdf_render
marker cleanly.
This commit is contained in:
parent
a0647a53a9
commit
64f3f8769f
6 changed files with 396 additions and 212 deletions
136
tradein-mvp/backend/app/core/rbac.py
Normal file
136
tradein-mvp/backend/app/core/rbac.py
Normal 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)
|
||||||
|
|
@ -8,15 +8,12 @@ from __future__ import annotations
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import re
|
from collections.abc import AsyncGenerator
|
||||||
import secrets
|
|
||||||
from collections.abc import AsyncGenerator, Awaitable, Callable
|
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
|
|
||||||
import sentry_sdk
|
import sentry_sdk
|
||||||
from fastapi import FastAPI, Request
|
from fastapi import FastAPI
|
||||||
from fastapi.middleware.cors import CORSMiddleware
|
from fastapi.middleware.cors import CORSMiddleware
|
||||||
from fastapi.responses import JSONResponse, Response
|
|
||||||
from sentry_sdk.integrations.fastapi import FastApiIntegration
|
from sentry_sdk.integrations.fastapi import FastApiIntegration
|
||||||
from sentry_sdk.integrations.httpx import HttpxIntegration
|
from sentry_sdk.integrations.httpx import HttpxIntegration
|
||||||
from sentry_sdk.integrations.logging import LoggingIntegration
|
from sentry_sdk.integrations.logging import LoggingIntegration
|
||||||
|
|
@ -35,11 +32,11 @@ from app.api.v1 import (
|
||||||
support,
|
support,
|
||||||
trade_in,
|
trade_in,
|
||||||
)
|
)
|
||||||
from app.core.auth import get_role, is_path_allowed
|
|
||||||
from app.core.config import settings
|
from app.core.config import settings
|
||||||
from app.core.db import SessionLocal
|
from app.core.db import SessionLocal
|
||||||
from app.core.fdw import ensure_fdw_user_mapping
|
from app.core.fdw import ensure_fdw_user_mapping
|
||||||
from app.core.ratelimit import RateLimitMiddleware
|
from app.core.ratelimit import RateLimitMiddleware
|
||||||
|
from app.core.rbac import rbac_guard
|
||||||
from app.core.request_audit import RequestAuditMiddleware
|
from app.core.request_audit import RequestAuditMiddleware
|
||||||
from app.observability.sentry_scrub import scrub_pii_event
|
from app.observability.sentry_scrub import scrub_pii_event
|
||||||
|
|
||||||
|
|
@ -138,104 +135,9 @@ app = FastAPI(
|
||||||
# не видит» — decided 2026-05-25).
|
# не видит» — decided 2026-05-25).
|
||||||
# 3) /api/v1/admin/* (= внешний /trade-in/api/v1/admin/* после Caddy
|
# 3) /api/v1/admin/* (= внешний /trade-in/api/v1/admin/* после Caddy
|
||||||
# `uri strip_prefix /trade-in`) — только role=admin, иначе 403.
|
# `uri strip_prefix /trade-in`) — только role=admin, иначе 403.
|
||||||
# Public paths без auth (/health, /docs, /openapi.json) пропускаем —
|
# Guard body живёт в app/core/rbac.py (без DB/lifespan side effects), чтобы
|
||||||
# X-Authenticated-User там не приходит из Caddy.
|
# тесты могли импортировать РЕАЛЬНЫЙ guard вместо hand-maintained копии.
|
||||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
app.middleware("http")(rbac_guard)
|
||||||
_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)
|
|
||||||
|
|
||||||
|
|
||||||
app.add_middleware(
|
app.add_middleware(
|
||||||
|
|
|
||||||
18
tradein-mvp/backend/tests/conftest.py
Normal file
18
tradein-mvp/backend/tests/conftest.py
Normal 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).",
|
||||||
|
)
|
||||||
|
|
@ -2,13 +2,16 @@
|
||||||
|
|
||||||
Backend доверяет X-Authenticated-User (его ставит Caddy). На общей docker-сети
|
Backend доверяет X-Authenticated-User (его ставит Caddy). На общей docker-сети
|
||||||
gendesign_shared любой контейнер мог бы отправить поддельный
|
gendesign_shared любой контейнер мог бы отправить поддельный
|
||||||
`X-Authenticated-User: admin` напрямую на tradein-backend:8000. Общий секрет
|
`X-Authenticated-User: admin` напрямую на tradein-backend:8000, минуя Caddy.
|
||||||
X-Internal-Auth-Secret закрывает дыру: если TRADEIN_INTERNAL_AUTH_SECRET задан,
|
Общий секрет X-Internal-Auth-Secret закрывает дыру: если TRADEIN_INTERNAL_AUTH_SECRET
|
||||||
каждый запрос с X-Authenticated-User обязан нести валидный секрет (constant-time),
|
задан, каждый запрос с X-Authenticated-User обязан нести валидный секрет (constant-time),
|
||||||
иначе 401. Пусто = fail-open (backward-compat до провижининга).
|
иначе 401. Пусто = fail-open (backward-compat до провижининга).
|
||||||
|
|
||||||
MIRROR of rbac_guard из app/main.py — включая #2213 secret-gate. Держим копию
|
Использует РЕАЛЬНЫЙ rbac_guard (app/core/rbac.py) — тот же, что регистрирует
|
||||||
здесь (как test_rbac.py), чтобы не тянуть тяжёлый app.main (lifespan/DB/scheduler).
|
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
|
from __future__ import annotations
|
||||||
|
|
@ -17,20 +20,13 @@ import os
|
||||||
|
|
||||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||||
|
|
||||||
import re
|
|
||||||
import secrets
|
|
||||||
from collections.abc import Awaitable, Callable
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from fastapi import FastAPI, Request
|
from fastapi import FastAPI
|
||||||
from fastapi.responses import JSONResponse, Response
|
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
from app.core import auth as auth_mod
|
from app.core import auth as auth_mod
|
||||||
from app.core import config
|
from app.core import config
|
||||||
|
from app.core.rbac import rbac_guard
|
||||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
|
||||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@pytest.fixture(autouse=True)
|
||||||
|
|
@ -39,44 +35,9 @@ def _reset_auth_cache() -> None:
|
||||||
|
|
||||||
|
|
||||||
def _build_test_app() -> FastAPI:
|
def _build_test_app() -> FastAPI:
|
||||||
"""Копия rbac_guard из app/main.py (с #2213 secret-gate)."""
|
"""Test app используя РЕАЛЬНЫЙ rbac_guard (с #2213 secret-gate)."""
|
||||||
app = FastAPI()
|
app = FastAPI()
|
||||||
|
app.middleware("http")(rbac_guard)
|
||||||
@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.get("/api/v1/ping")
|
@app.get("/api/v1/ping")
|
||||||
async def ping() -> dict:
|
async def ping() -> dict:
|
||||||
|
|
|
||||||
209
tradein-mvp/backend/tests/test_pdf_real_render.py
Normal file
209
tradein-mvp/backend/tests/test_pdf_real_render.py
Normal 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}"
|
||||||
|
|
@ -1,7 +1,5 @@
|
||||||
"""RBAC unit + integration tests for tradein backend.
|
"""RBAC unit + integration tests for tradein backend.
|
||||||
|
|
||||||
MIRROR of backend/tests/test_rbac.py — kept in sync manually.
|
|
||||||
|
|
||||||
Coverage:
|
Coverage:
|
||||||
- get_role / get_user_scope happy + error paths
|
- get_role / get_user_scope happy + error paths
|
||||||
- is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**)
|
- is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**)
|
||||||
|
|
@ -14,22 +12,29 @@ Coverage:
|
||||||
|
|
||||||
Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не
|
Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не
|
||||||
тянуть тяжёлые модули из app.main (lifespan task + DB session + scheduler).
|
тянуть тяжёлые модули из 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
|
from __future__ import annotations
|
||||||
|
|
||||||
import re
|
import os
|
||||||
from collections.abc import Awaitable, Callable
|
|
||||||
|
# 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
|
import pytest
|
||||||
from fastapi import FastAPI, Request
|
from fastapi import FastAPI
|
||||||
from fastapi.responses import JSONResponse, Response
|
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
from app.api.v1 import me as me_router
|
from app.api.v1 import me as me_router
|
||||||
from app.core import auth as auth_mod
|
from app.core import auth as auth_mod
|
||||||
|
from app.core.rbac import rbac_guard
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture(autouse=True)
|
@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:
|
def _build_test_app() -> FastAPI:
|
||||||
app = FastAPI()
|
app = FastAPI()
|
||||||
|
app.middleware("http")(rbac_guard)
|
||||||
@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.include_router(me_router.router, prefix="/api/v1", tags=["me"])
|
app.include_router(me_router.router, prefix="/api/v1", tags=["me"])
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue