fix(tradein/rbac): энфорсить roles.yaml scope на всех non-admin путях (R2 H3 security) #2498
2 changed files with 113 additions and 1 deletions
|
|
@ -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)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue