fix(tradein/rbac): энфорсить roles.yaml scope на всех non-admin путях (R2 H3)
All checks were successful
CI / changes (pull_request) Successful in 8s
CI Trade-In / changes (pull_request) Successful in 8s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 52s
All checks were successful
CI / changes (pull_request) Successful in 8s
CI Trade-In / changes (pull_request) Successful in 8s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
CI Trade-In / backend-tests (pull_request) Successful in 52s
rbac_guard гейтил только /api/v1/admin/* → revoked (role=expired, roles.yaml paths:[] deny:/**) и узко-скоупленные аккаунты сохраняли полный non-admin API-доступ (напр. POST /api/v1/search — экспорт листингов + estimate-quota). is_path_allowed (roles.yaml paths/deny) существовал, но НЕ вызывался (0 callers). Fix: после admin-гейта вызываем is_path_allowed(role, external_path) для всех non-bootstrap путей. roles.yaml globs — внешние (Caddy срезал /trade-in), поэтому восстанавливаем внешний путь (_EXTERNAL_PREFIX + path). Bootstrap-пути /me и /brand/* исключены — expired ДОЛЖЕН получить role=expired через /me (trial-экран) и брендинг; без исключения trial-UX сломался бы (expired paths:[] → 403 на /me). На сбой парса — fail-open + громкий лог (не лочим платящего pilot из-за конфиг-бага). roles.yaml НЕ меняю — только энфорсю уже задекларированную политику. pilot/admin/ analyst доступ сохранён (verified), expired теперь 403 на non-admin API. Тесты: +5 scope-кейсов (expired denied search/trade-in, allowed me/brand; pilot/ admin/analyst preserved) — 31 passed. Тест-harness middleware — mirror, обновлён.
This commit is contained in:
parent
0f4ed2f6e9
commit
85321d1d1a
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 sentry_sdk.integrations.starlette import StarletteIntegration
|
||||||
|
|
||||||
from app.api.v1 import admin, brand, buildings, geocode, lead, me, search, trade_in
|
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.config import settings
|
||||||
from app.core.db import SessionLocal
|
from app.core.db import SessionLocal
|
||||||
from app.core.fdw import ensure_fdw_user_mapping
|
from app.core.fdw import ensure_fdw_user_mapping
|
||||||
|
|
@ -108,6 +108,14 @@ app = FastAPI(
|
||||||
# X-Authenticated-User там не приходит из Caddy.
|
# X-Authenticated-User там не приходит из Caddy.
|
||||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
||||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
_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")
|
@app.middleware("http")
|
||||||
|
|
@ -160,6 +168,39 @@ async def rbac_guard(
|
||||||
status_code=403,
|
status_code=403,
|
||||||
content={"detail": "admin only"},
|
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)
|
return await call_next(request)
|
||||||
|
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -44,6 +44,9 @@ def _reset_auth_cache() -> None:
|
||||||
|
|
||||||
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
_ADMIN_API_RE = re.compile(r"^/api/v1/admin/")
|
||||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
_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:
|
def _build_test_app() -> FastAPI:
|
||||||
|
|
@ -76,6 +79,18 @@ def _build_test_app() -> FastAPI:
|
||||||
status_code=403,
|
status_code=403,
|
||||||
content={"detail": "admin only"},
|
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)
|
return await call_next(request)
|
||||||
|
|
||||||
app.include_router(me_router.router, prefix="/api/v1", tags=["me"])
|
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:
|
async def admin_dummy() -> dict:
|
||||||
return {"ok": True}
|
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")
|
@app.get("/health")
|
||||||
async def health() -> dict:
|
async def health() -> dict:
|
||||||
return {"status": "ok"}
|
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()
|
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:
|
def test_rbac_guard_no_header_on_non_admin_path_returns_401(client: TestClient) -> None:
|
||||||
"""Любой non-public path без X-Authenticated-User → 401."""
|
"""Любой non-public path без X-Authenticated-User → 401."""
|
||||||
resp = client.get("/api/v1/me")
|
resp = client.get("/api/v1/me")
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue