"""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/**) - GET /me — header missing / unknown user / pilot / admin - rbac_guard middleware: * любой non-public path требует X-Authenticated-User (no-header → 401) * неизвестный юзер → 403 на ВСЁ (включая non-admin paths) — «человек без ролей вообще ничего не видит» (decided 2026-05-25) * /api/v1/admin/* — только role=admin (pilot → 403) Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не тянуть тяжёлые модули из app.main (lifespan task + DB session + scheduler). RBAC middleware и /me router импортируются напрямую — это даёт нам ровно тот же поведение что в проде. """ from __future__ import annotations import re from collections.abc import Awaitable, Callable import pytest from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, Response from fastapi.testclient import TestClient from app.api.v1 import me as me_router from app.core import auth as auth_mod @pytest.fixture(autouse=True) def _reset_auth_cache() -> None: auth_mod.reset_cache_for_tests() # --------------------------------------------------------------------------- # Test app — копия rbac_guard из app/main.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.include_router(me_router.router, prefix="/api/v1", tags=["me"]) @app.get("/api/v1/admin/dummy") 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"} return app @pytest.fixture def client() -> TestClient: return TestClient(_build_test_app()) # --------------------------------------------------------------------------- # get_role # --------------------------------------------------------------------------- def test_get_role_known_users() -> None: assert auth_mod.get_role("admin") == "admin" assert auth_mod.get_role("kopylov") == "pilot" for n in range(1, 11): assert auth_mod.get_role(f"user{n}") == "pilot" def test_get_role_unknown_user_raises() -> None: with pytest.raises(KeyError): auth_mod.get_role("nobody") # --------------------------------------------------------------------------- # is_path_allowed # --------------------------------------------------------------------------- def test_is_path_allowed_admin_everywhere() -> None: assert auth_mod.is_path_allowed("admin", "/") assert auth_mod.is_path_allowed("admin", "/admin") assert auth_mod.is_path_allowed("admin", "/admin/scrape/runs") assert auth_mod.is_path_allowed("admin", "/api/v1/admin/scrape/status") assert auth_mod.is_path_allowed("admin", "/trade-in/api/v1/admin/scrape") assert auth_mod.is_path_allowed("admin", "/concept/123") assert auth_mod.is_path_allowed("admin", "/analytics/dashboard") def test_is_path_allowed_pilot_only_tradein() -> None: """Pilot имеет доступ ТОЛЬКО к /trade-in/** (decision 2026-05-26).""" # Pilot ALLOWED — только trade-in assert auth_mod.is_path_allowed("pilot", "/trade-in") assert auth_mod.is_path_allowed("pilot", "/trade-in/") assert auth_mod.is_path_allowed("pilot", "/trade-in/123") assert auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/search") # Pilot DENIED — landing + остальные разделы (admin-only) assert not auth_mod.is_path_allowed("pilot", "/") assert not auth_mod.is_path_allowed("pilot", "/analytics/dashboard") assert not auth_mod.is_path_allowed("pilot", "/site-finder/123") assert not auth_mod.is_path_allowed("pilot", "/concept/abc") assert not auth_mod.is_path_allowed("pilot", "/api/v1/parcels/123") # Pilot DENIED — admin paths assert not auth_mod.is_path_allowed("pilot", "/admin") assert not auth_mod.is_path_allowed("pilot", "/admin/jobs") assert not auth_mod.is_path_allowed("pilot", "/admin/scrape/runs/42") assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/scrape/status") assert not auth_mod.is_path_allowed("pilot", "/api/v1/admin/jobs") assert not auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/admin/scrape") def test_is_path_allowed_unknown_role_denied() -> None: assert not auth_mod.is_path_allowed("ghost", "/") assert not auth_mod.is_path_allowed("ghost", "/admin") # --------------------------------------------------------------------------- # get_user_scope # --------------------------------------------------------------------------- def test_get_user_scope_admin() -> None: scope = auth_mod.get_user_scope("admin") assert scope["username"] == "admin" assert scope["role"] == "admin" assert "/**" in scope["allowed_paths"] assert scope["deny_paths"] == [] def test_get_user_scope_pilot() -> None: scope = auth_mod.get_user_scope("kopylov") assert scope["username"] == "kopylov" assert scope["role"] == "pilot" # Pilot allowed только /trade-in/** (decision 2026-05-26). assert "/trade-in/**" in scope["allowed_paths"] assert "/trade-in/api/v1/**" in scope["allowed_paths"] assert "/" not in scope["allowed_paths"] assert "/analytics/**" not in scope["allowed_paths"] assert "/admin/**" in scope["deny_paths"] assert "/api/v1/admin/**" in scope["deny_paths"] def test_get_user_scope_unknown_raises() -> None: with pytest.raises(KeyError): auth_mod.get_user_scope("nobody") # --------------------------------------------------------------------------- # GET /me # --------------------------------------------------------------------------- def test_me_endpoint_admin(client: TestClient) -> None: resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "admin"}) assert resp.status_code == 200, resp.text body = resp.json() assert body["username"] == "admin" assert body["role"] == "admin" assert body["allowed_paths"] == ["/**"] def test_me_endpoint_pilot(client: TestClient) -> None: resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "kopylov"}) assert resp.status_code == 200, resp.text body = resp.json() assert body["username"] == "kopylov" assert body["role"] == "pilot" assert "/admin/**" in body["deny_paths"] def test_me_endpoint_no_header_401(client: TestClient) -> None: resp = client.get("/api/v1/me") assert resp.status_code == 401 assert "no authenticated user" in resp.json()["detail"].lower() def test_me_endpoint_unknown_user_403(client: TestClient) -> None: resp = client.get("/api/v1/me", headers={"X-Authenticated-User": "ghost"}) assert resp.status_code == 403 assert "not in roles config" in resp.json()["detail"].lower() # --------------------------------------------------------------------------- # rbac_guard middleware # --------------------------------------------------------------------------- def test_rbac_guard_pilot_blocked_on_admin_api(client: TestClient) -> None: resp = client.get( "/api/v1/admin/dummy", headers={"X-Authenticated-User": "kopylov"}, ) assert resp.status_code == 403 assert resp.json()["detail"] == "admin only" def test_rbac_guard_admin_allowed(client: TestClient) -> None: resp = client.get( "/api/v1/admin/dummy", headers={"X-Authenticated-User": "admin"}, ) assert resp.status_code == 200 assert resp.json() == {"ok": True} def test_rbac_guard_no_header_on_admin_path_returns_401(client: TestClient) -> None: resp = client.get("/api/v1/admin/dummy") assert resp.status_code == 401 def test_rbac_guard_unknown_user_on_admin_path_returns_403(client: TestClient) -> None: resp = client.get( "/api/v1/admin/dummy", headers={"X-Authenticated-User": "ghost"}, ) assert resp.status_code == 403 assert "not in roles config" in resp.json()["detail"].lower() def test_rbac_guard_skips_health(client: TestClient) -> None: resp = client.get("/health") assert resp.status_code == 200 assert resp.json()["status"] == "ok" def test_rbac_guard_pilot_can_hit_non_admin_api(client: TestClient) -> None: resp = client.get( "/api/v1/me", headers={"X-Authenticated-User": "user1"}, ) assert resp.status_code == 200 assert resp.json()["role"] == "pilot" def test_rbac_guard_unknown_user_blocked_on_non_admin_path(client: TestClient) -> None: """Неизвестный юзер → 403 даже на non-admin path. «Человек без ролей вообще ничего не видит» (decided 2026-05-25).""" resp = client.get( "/api/v1/me", headers={"X-Authenticated-User": "ghost"}, ) assert resp.status_code == 403 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 ("praktika",): # role=expired (trial отозван; user2 восстановлен 2026-07-13) 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": "praktika"}) 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": "praktika"}) 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") assert resp.status_code == 401 assert "no authenticated user" in resp.json()["detail"].lower()