diff --git a/tradein-mvp/backend/app/main.py b/tradein-mvp/backend/app/main.py index 55a13911..cdab1d88 100644 --- a/tradein-mvp/backend/app/main.py +++ b/tradein-mvp/backend/app/main.py @@ -24,7 +24,7 @@ from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration from sentry_sdk.integrations.starlette import StarletteIntegration from app.api.v1 import admin, brand, buildings, geocode, lead, me, search, trade_in -from app.core.auth import get_role +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 @@ -108,6 +108,14 @@ app = FastAPI( # 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") @@ -160,6 +168,39 @@ async def rbac_guard( 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) diff --git a/tradein-mvp/backend/tests/test_rbac.py b/tradein-mvp/backend/tests/test_rbac.py index e4b261b3..6ff637b9 100644 --- a/tradein-mvp/backend/tests/test_rbac.py +++ b/tradein-mvp/backend/tests/test_rbac.py @@ -44,6 +44,9 @@ def _reset_auth_cache() -> None: _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: @@ -76,6 +79,18 @@ def _build_test_app() -> FastAPI: 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"]) @@ -84,6 +99,18 @@ def _build_test_app() -> FastAPI: async def admin_dummy() -> dict: return {"ok": True} + @app.get("/api/v1/search") + async def search_dummy() -> dict: + return {"ok": True} + + @app.get("/api/v1/trade-in/dummy") + async def tradein_dummy() -> dict: + return {"ok": True} + + @app.get("/api/v1/brand/dummy") + async def brand_dummy() -> dict: + return {"ok": True} + @app.get("/health") async def health() -> dict: return {"status": "ok"} @@ -288,6 +315,50 @@ def test_rbac_guard_unknown_user_blocked_on_non_admin_path(client: TestClient) - assert "not in roles config" in resp.json()["detail"].lower() +# --- #R2-H3: roles.yaml scope enforced on ALL non-admin paths (not just /admin/*) --- + + +def test_rbac_expired_denied_on_non_admin_api(client: TestClient) -> None: + """#R2-H3: revoked (role=expired, paths:[] deny:/**) НЕ достаёт non-admin API. + Раньше rbac_guard гейтил только /admin/* → expired имел полный non-admin доступ + (POST /search экспорт листингов и т.д.). Теперь scope-blocked → 403.""" + for user in ("user2", "praktika"): # оба role=expired (trial отозван) + for path in ("/api/v1/search", "/api/v1/trade-in/dummy"): + resp = client.get(path, headers={"X-Authenticated-User": user}) + assert resp.status_code == 403, f"{user} {path}: {resp.status_code}" + assert "forbidden for role" in resp.json()["detail"].lower() + + +def test_rbac_expired_allowed_on_bootstrap(client: TestClient) -> None: + """expired ДОЛЖЕН достучаться до /me (получить role=expired → trial-экран) и + /brand/* (брендинг trial-экрана) — иначе UX сломан. Bootstrap-исключение.""" + resp_me = client.get("/api/v1/me", headers={"X-Authenticated-User": "user2"}) + assert resp_me.status_code == 200, resp_me.text + assert resp_me.json()["role"] == "expired" + resp_brand = client.get("/api/v1/brand/dummy", headers={"X-Authenticated-User": "user2"}) + assert resp_brand.status_code == 200, resp_brand.text + + +def test_rbac_pilot_allowed_on_tradein_api(client: TestClient) -> None: + """pilot по roles.yaml имеет /trade-in/api/v1/** → доступ к trade-in/search сохраняется.""" + for path in ("/api/v1/search", "/api/v1/trade-in/dummy"): + resp = client.get(path, headers={"X-Authenticated-User": "kopylov"}) + assert resp.status_code == 200, f"pilot {path}: {resp.text}" + + +def test_rbac_admin_allowed_everywhere(client: TestClient) -> None: + for path in ("/api/v1/search", "/api/v1/trade-in/dummy", "/api/v1/admin/dummy"): + resp = client.get(path, headers={"X-Authenticated-User": "admin"}) + assert resp.status_code == 200, f"admin {path}: {resp.text}" + + +def test_rbac_analyst_allowed_non_admin_denied_admin(client: TestClient) -> None: + ok = client.get("/api/v1/search", headers={"X-Authenticated-User": "analysttest"}) + assert ok.status_code == 200, ok.text + denied = client.get("/api/v1/admin/dummy", headers={"X-Authenticated-User": "analysttest"}) + assert denied.status_code == 403 + + def test_rbac_guard_no_header_on_non_admin_path_returns_401(client: TestClient) -> None: """Любой non-public path без X-Authenticated-User → 401.""" resp = client.get("/api/v1/me")