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

Merged
bot-backend merged 1 commit from fix/tradein-rbac-scope into main 2026-07-12 19:28:39 +00:00
2 changed files with 113 additions and 1 deletions

View file

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

View file

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