Merge pull request 'fix(tradein/rbac): энфорсить roles.yaml scope на всех non-admin путях (R2 H3 security)' (#2498) from fix/tradein-rbac-scope into main

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)
 
 

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")