feat(rbac): add analyst role + §19 audit-log middleware (#962)
Some checks failed
CI / changes (push) Successful in 6s
CI / frontend-tests (push) Has been skipped
CI / changes (pull_request) Successful in 6s
Deploy Trade-In / changes (push) Successful in 7s
Deploy / changes (push) Successful in 6s
CI / frontend-tests (pull_request) Has been skipped
Deploy Trade-In / build-frontend (push) Has been skipped
Deploy Trade-In / build-browser (push) Has been skipped
Deploy Trade-In / test (push) Successful in 36s
Deploy / build-backend (push) Successful in 1m39s
Deploy / build-frontend (push) Has been skipped
Deploy Trade-In / build-backend (push) Successful in 48s
Deploy / build-worker (push) Successful in 2m41s
Deploy / deploy (push) Failing after 4s
Deploy Trade-In / deploy (push) Successful in 47s
CI / backend-tests (push) Successful in 6m38s
CI / backend-tests (pull_request) Successful in 6m30s

EPIC18/§19. analyst sees everything (deals, insights, exports, site-finder,
analytics, concept) EXCEPT admin/data-management. Enforcement is backend-hard
(the existing rbac_guard already 403s any non-admin role on /api/v1/admin/*, so
adding the role auto-blocks it) + frontend (deny_paths via /me) + audit.

§19 audit: new best-effort HTTP middleware logs the sensitive actions
(analyze / forecast / forecast-export) to a new audit_log table after the
response. Audit failures never break or delay-fail the request (2-layer
try/except + finally close). Registered INNER to rbac_guard so only authorized
requests are audited (a 403 short-circuits before audit). classify_path matches
export before forecast (anchored).

- auth/roles.yaml: analyst role (paths /**, deny admin-mgmt) + analysttest QA user
- core/auth.py (+ tradein mirror): Role Literal += analyst
- core/audit_middleware.py (new) + main.py registration
- data/sql/144_audit_log.sql (idempotent; auto-applies)
- tests: analyst rbac (403 admin / 200 parcels) + 11 audit cases

No data-level ACL (analyst sees data per policy) -> no #948 dependency. No new
deps. parcels.py untouched. Real analyst logins still need adding to
caddy/users.caddy.snippet (devops).

Refs #962.
This commit is contained in:
Light1YT 2026-06-08 12:16:19 +05:00
parent 25e21c2bff
commit 86828d0388
8 changed files with 652 additions and 4 deletions

View file

@ -39,7 +39,26 @@ roles:
- "/admin/**" - "/admin/**"
- "/api/v1/admin/**" - "/api/v1/admin/**"
- "/trade-in/api/v1/admin/**" - "/trade-in/api/v1/admin/**"
analyst:
# #962 (EPIC18, ТЗ §19): analyst видит ВСЁ (deals, insights, exports,
# site-finder, analytics, concept) КРОМЕ admin/data-management.
# paths "/**" = доступ ко всему; deny = админ-управление (scraper-триггеры,
# sync-таски, data-management, user/role management).
# Backend hard-gate: app/main.py rbac_guard уже блокирует /api/v1/admin/*
# для любого role != "admin" → analyst авто-403 на admin-API без доп. кода.
# deny ниже драйвит фронтовый RouteGuard (deny_paths из /me) для UI-gating
# /admin/** страниц.
paths:
- "/**"
deny:
- "/admin/**"
- "/api/v1/admin/**"
- "/trade-in/api/v1/admin/**"
# NB: реальные analyst-логины ДОЛЖНЫ быть добавлены и в
# caddy/users.caddy.snippet (Caddy basic_auth) силами devops — иначе Caddy не
# пропустит юзера и не выставит X-Authenticated-User. Здесь только role-mapping;
# Caddy НЕ редактируем из этого таска.
users: users:
admin: admin admin: admin
kopylov: pilot kopylov: pilot
@ -56,3 +75,4 @@ users:
praktika: pilot # пилот-аккаунт агентства «Практика» 2026-05-29 praktika: pilot # пилот-аккаунт агентства «Практика» 2026-05-29
admintest: admin # temp QA 2026-05-26 admintest: admin # temp QA 2026-05-26
pilottest: pilot # temp QA 2026-05-26 pilottest: pilot # temp QA 2026-05-26
analysttest: analyst # temp QA 2026-06-07 (#962)

View file

@ -0,0 +1,172 @@
"""§19 audit log middleware (#962, EPIC18).
Best-effort аудит ключевых/чувствительных действий через FastAPI HTTP-middleware.
Регистрируется в ``app/main.py`` ПОСЛЕ ``rbac_guard`` (Starlette applies middleware
в обратном порядке регистрации, поэтому rbac_guard должен оставаться внешним
аудитим только запросы, которые RBAC уже пропустил).
Аудируем РОВНО три пути (sensitive actions из ТЗ §19):
* ``POST /api/v1/parcels/{cad}/analyze`` action='analyze'
* ``GET /api/v1/parcels/{cad}/forecast`` action='forecast'
* ``GET /api/v1/parcels/{cad}/forecast/export`` action='export'
``/health``, статика, любые non-matched пути НЕ аудируются.
Гарантии:
* Сбой записи в ``audit_log`` НИКОГДА не ломает и не замедляет-фейлит реальный
запрос write обёрнут в try/except, ошибка логируется, отдаётся настоящий
response.
* Запись делается ПОСЛЕ ``call_next`` (когда response уже готов и status_code
известен) off the critical latency path.
* В ``settings.testing`` middleware пропускает аудит (как rbac_guard) юнит-тесты
бьют по app мимо Caddy; аудит-логика покрыта напрямую через
``classify_path`` / ``write_audit_row`` в tests/test_audit_middleware.py.
DB-write использует свежий ``SessionLocal()`` (сессия запроса в middleware
недоступна), commit + close всё в try/except.
"""
from __future__ import annotations
import logging
import re
from collections.abc import Awaitable, Callable
from fastapi import Request
from fastapi.responses import Response
from sqlalchemy import text
from app.core.auth import get_role
from app.core.config import settings
from app.core.db import SessionLocal
logger = logging.getLogger(__name__)
# Кадастровый номер в пути: 66:41:0204016:10 (цифры/двоеточия). Берём «жадный»
# сегмент между /parcels/ и следующим действием — не парсим формат строго,
# чтобы не ломаться на нестандартных кад. номерах.
_CAD = r"(?P<cad>[^/]+)"
# Порядок ВАЖЕН: forecast/export проверяем раньше forecast, иначе export
# заматчится как forecast. Каждый паттерн якорный ($) → точное совпадение пути.
_AUDIT_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
("export", re.compile(rf"^/api/v1/parcels/{_CAD}/forecast/export$")),
("analyze", re.compile(rf"^/api/v1/parcels/{_CAD}/analyze$")),
("forecast", re.compile(rf"^/api/v1/parcels/{_CAD}/forecast$")),
)
def classify_path(path: str) -> tuple[str, str | None] | None:
"""Сопоставить *path* с аудируемым действием.
Returns ``(action, cad_num)`` если путь один из sensitive endpoints,
иначе ``None`` (путь не аудируется). ``cad_num`` извлекается из пути.
"""
for action, pattern in _AUDIT_PATTERNS:
m = pattern.match(path)
if m:
return action, m.group("cad")
return None
def write_audit_row(
*,
username: str,
action: str,
method: str,
path: str,
status_code: int,
cad_num: str | None,
) -> None:
"""Best-effort вставка одной строки в ``audit_log``.
Открывает свежий ``SessionLocal()``, commit, close всё в try/except.
Любая ошибка логируется и ГЛОТАЕТСЯ (re-raise здесь сломал бы основной
запрос) это намеренное исключение из «no silent except»: аудит вторичен
по отношению к запросу пользователя.
"""
role: str | None
try:
role = get_role(username)
except KeyError:
# Юзер прошёл сюда через rbac_guard, так что обычно известен; но
# на гонке reload roles.yaml / прямом вызове — пишем role=None, не падаем.
role = None
db = SessionLocal()
try:
db.execute(
text(
"INSERT INTO audit_log "
"(username, role, action, method, path, cad_num, status_code) "
"VALUES (:username, :role, :action, :method, :path, :cad_num, "
":status_code)"
),
{
"username": username,
"role": role,
"action": action,
"method": method,
"path": path,
"cad_num": cad_num,
"status_code": status_code,
},
)
db.commit()
except Exception:
# Best-effort: НЕ re-raise — аудит не должен ломать запрос.
logger.exception(
"audit_log write failed (user=%s action=%s path=%s) — request unaffected",
username,
action,
path,
)
try:
db.rollback()
except Exception:
logger.exception("audit_log rollback failed")
finally:
db.close()
async def audit_log_middleware(
request: Request,
call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
"""Аудируем sensitive actions ПОСЛЕ выполнения запроса (best-effort)."""
# Test-mode bypass (как rbac_guard). Аудит-логика покрыта напрямую в
# tests/test_audit_middleware.py через classify_path / write_audit_row.
if settings.testing:
return await call_next(request)
matched = classify_path(request.url.path)
# Выполняем сам запрос ВСЕГДА (даже non-matched) — сначала получаем response.
response = await call_next(request)
if matched is None:
return response
action, cad_num = matched
username = request.headers.get("X-Authenticated-User")
if not username:
# Без auth-header (локальный curl мимо Caddy) rbac_guard уже отдал бы 401,
# но если мы тут — нечего атрибутировать, аудит пропускаем.
return response
# Write после готового response — off the critical path. Любой сбой внутри
# write_audit_row проглатывается там же; здесь дополнительная страховка.
try:
write_audit_row(
username=username,
action=action,
method=request.method,
path=request.url.path,
status_code=response.status_code,
cad_num=cad_num,
)
except Exception:
logger.exception("audit middleware unexpected error — request unaffected")
return response

View file

@ -30,7 +30,7 @@ import yaml
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
Role = Literal["admin", "pilot"] Role = Literal["admin", "pilot", "analyst"]
class UserScope(TypedDict): class UserScope(TypedDict):

View file

@ -36,6 +36,7 @@ from app.api.v1 import (
trade_in, trade_in,
users, users,
) )
from app.core.audit_middleware import audit_log_middleware
from app.core.auth import get_role from app.core.auth import get_role
from app.core.config import settings from app.core.config import settings
from app.observability.sentry_scrub import scrub_sensitive_query from app.observability.sentry_scrub import scrub_sensitive_query
@ -77,6 +78,15 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
app = FastAPI(title="GenDesign API", version="0.1.0", lifespan=lifespan) app = FastAPI(title="GenDesign API", version="0.1.0", lifespan=lifespan)
# §19 audit log (#962, EPIC18): аудит sensitive actions (analyze/forecast/export).
# Регистрируем ПЕРВЫМ → Starlette применяет HTTP-middleware в LIFO-порядке
# (add_middleware insert(0); build_middleware_stack wraps reversed), поэтому
# rbac_guard (зарегистрирован НИЖЕ) оказывается ВНЕШНИМ слоем, а audit — внутри:
# request flow = rbac_guard → audit → router. Аудитим ТОЛЬКО запросы, которые
# rbac_guard уже пропустил (его 401/403 short-circuit'ы не доходят до audit).
# Best-effort: сбой записи не ломает запрос (см. app/core/audit_middleware.py).
app.middleware("http")(audit_log_middleware)
# RBAC: defense-in-depth поверх Caddy basic_auth + X-Authenticated-User # RBAC: defense-in-depth поверх Caddy basic_auth + X-Authenticated-User
# (см. app/core/auth.py + auth/roles.yaml). Правила: # (см. app/core/auth.py + auth/roles.yaml). Правила:
# 1) Любой non-public path требует X-Authenticated-User — иначе 401 # 1) Любой non-public path требует X-Authenticated-User — иначе 401

View file

@ -0,0 +1,319 @@
"""§19 audit-log middleware tests (#962, EPIC18).
Coverage:
- classify_path: analyze / forecast / forecast-export матчатся, cad_num
извлекается; forecast/export НЕ путается с forecast; non-matched None.
- write_audit_row: best-effort INSERT ровно одна строка с верными полями;
сбой записи (execute кидает) проглатывается, не re-raise; rollback вызван.
- middleware (копия, минующая settings.testing-bypass, как в test_rbac.py):
* matched request ровно один audit write с верными полями + реальный
status_code из ответа.
* non-matched path НИ одной записи.
* сбой audit write запрос всё равно отдаёт настоящий response.
DB не нужен: SessionLocal патчится фейковой сессией, ловящей execute()/commit().
auth.get_role патчится, чтобы не зависеть от roles.yaml-кэша.
"""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from typing import Any
import pytest
from fastapi import FastAPI, Request
from fastapi.responses import Response
from fastapi.testclient import TestClient
from app.core import audit_middleware as audit_mod
# ---------------------------------------------------------------------------
# Fake DB session — записывает execute()/commit()/rollback()/close().
# ---------------------------------------------------------------------------
class _FakeSession:
def __init__(self, *, fail_on_execute: bool = False) -> None:
self.executed: list[tuple[str, dict[str, Any]]] = []
self.committed = 0
self.rolled_back = 0
self.closed = 0
self._fail_on_execute = fail_on_execute
def execute(self, stmt: Any, params: dict[str, Any]) -> None:
if self._fail_on_execute:
raise RuntimeError("simulated DB failure")
# stmt — SQLAlchemy TextClause; .text несёт сырой SQL.
self.executed.append((str(getattr(stmt, "text", stmt)), params))
def commit(self) -> None:
self.committed += 1
def rollback(self) -> None:
self.rolled_back += 1
def close(self) -> None:
self.closed += 1
@pytest.fixture
def fake_session(monkeypatch: pytest.MonkeyPatch) -> _FakeSession:
"""Патчит SessionLocal в audit-модуле, возвращает единственную сессию."""
sess = _FakeSession()
monkeypatch.setattr(audit_mod, "SessionLocal", lambda: sess)
# get_role детерминированно → не зависим от roles.yaml lru_cache.
monkeypatch.setattr(audit_mod, "get_role", lambda u: "analyst")
return sess
# ---------------------------------------------------------------------------
# classify_path
# ---------------------------------------------------------------------------
def test_classify_path_analyze() -> None:
assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10/analyze") == (
"analyze",
"66:41:0204016:10",
)
def test_classify_path_forecast() -> None:
assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10/forecast") == (
"forecast",
"66:41:0204016:10",
)
def test_classify_path_forecast_export_not_confused_with_forecast() -> None:
"""forecast/export должен дать action='export', НЕ 'forecast'."""
assert audit_mod.classify_path(
"/api/v1/parcels/66:41:0204016:10/forecast/export"
) == ("export", "66:41:0204016:10")
def test_classify_path_unmatched_returns_none() -> None:
assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10") is None
assert audit_mod.classify_path("/api/v1/parcels/dummy/runs") is None
assert audit_mod.classify_path("/api/v1/analytics/dashboard") is None
assert audit_mod.classify_path("/health") is None
assert audit_mod.classify_path("/api/v1/admin/jobs/settings") is None
# Похожие, но не точные пути (якорь $) не матчатся.
assert audit_mod.classify_path("/api/v1/parcels/X/analyze/extra") is None
assert audit_mod.classify_path("/api/v1/parcels/X/forecast/export/extra") is None
# ---------------------------------------------------------------------------
# write_audit_row
# ---------------------------------------------------------------------------
def test_write_audit_row_inserts_one_row(fake_session: _FakeSession) -> None:
audit_mod.write_audit_row(
username="analysttest",
action="analyze",
method="POST",
path="/api/v1/parcels/66:41:0204016:10/analyze",
status_code=200,
cad_num="66:41:0204016:10",
)
assert len(fake_session.executed) == 1
sql, params = fake_session.executed[0]
assert "INSERT INTO audit_log" in sql
# psycopg v3 CAST-конвенция: никаких ':param::type' в SQL.
assert "::" not in sql
assert params == {
"username": "analysttest",
"role": "analyst",
"action": "analyze",
"method": "POST",
"path": "/api/v1/parcels/66:41:0204016:10/analyze",
"cad_num": "66:41:0204016:10",
"status_code": 200,
}
assert fake_session.committed == 1
assert fake_session.closed == 1
assert fake_session.rolled_back == 0
def test_write_audit_row_failure_is_swallowed(monkeypatch: pytest.MonkeyPatch) -> None:
"""Сбой execute НЕ должен пробрасываться — best-effort. rollback + close."""
sess = _FakeSession(fail_on_execute=True)
monkeypatch.setattr(audit_mod, "SessionLocal", lambda: sess)
monkeypatch.setattr(audit_mod, "get_role", lambda u: "analyst")
# Не должно бросить.
audit_mod.write_audit_row(
username="analysttest",
action="forecast",
method="GET",
path="/api/v1/parcels/66:41:0204016:10/forecast",
status_code=500,
cad_num="66:41:0204016:10",
)
assert sess.committed == 0
assert sess.rolled_back == 1
assert sess.closed == 1
def test_write_audit_row_unknown_user_role_none(monkeypatch: pytest.MonkeyPatch) -> None:
"""get_role KeyError → role=None, запись всё равно делается."""
sess = _FakeSession()
monkeypatch.setattr(audit_mod, "SessionLocal", lambda: sess)
def _raise(_: str) -> str:
raise KeyError("unknown")
monkeypatch.setattr(audit_mod, "get_role", _raise)
audit_mod.write_audit_row(
username="ghost",
action="export",
method="GET",
path="/api/v1/parcels/66:41:0204016:10/forecast/export",
status_code=403,
cad_num="66:41:0204016:10",
)
assert len(sess.executed) == 1
_, params = sess.executed[0]
assert params["role"] is None
assert params["username"] == "ghost"
# ---------------------------------------------------------------------------
# Middleware — копия audit_log_middleware БЕЗ settings.testing-bypass (как
# rbac_guard в test_rbac.py: прод-middleware на settings.testing смотрит,
# поэтому здесь гоняем «развязанную» копию, идентичную по логике).
# ---------------------------------------------------------------------------
def _build_audited_app(captured: list[dict[str, Any]]) -> FastAPI:
app = FastAPI()
@app.middleware("http")
async def audit_mw(
request: Request,
call_next: Callable[[Request], Awaitable[Response]],
) -> Response:
matched = audit_mod.classify_path(request.url.path)
response = await call_next(request)
if matched is None:
return response
action, cad_num = matched
username = request.headers.get("X-Authenticated-User")
if not username:
return response
captured.append(
{
"username": username,
"action": action,
"method": request.method,
"path": request.url.path,
"status_code": response.status_code,
"cad_num": cad_num,
}
)
return response
@app.post("/api/v1/parcels/{cad}/analyze")
async def analyze(cad: str) -> dict:
return {"ok": True, "cad": cad}
@app.get("/api/v1/parcels/{cad}/forecast")
async def forecast(cad: str) -> dict:
return {"ok": True, "cad": cad}
@app.get("/api/v1/parcels/{cad}/forecast/export")
async def forecast_export(cad: str) -> dict:
return {"ok": True, "cad": cad}
@app.get("/api/v1/analytics/dashboard")
async def analytics() -> dict:
return {"ok": True}
@app.get("/health")
async def health() -> dict:
return {"status": "ok"}
return app
def test_middleware_audits_matched_request() -> None:
captured: list[dict[str, Any]] = []
client = TestClient(_build_audited_app(captured))
resp = client.post(
"/api/v1/parcels/66:41:0204016:10/analyze",
headers={"X-Authenticated-User": "analysttest"},
)
assert resp.status_code == 200
assert len(captured) == 1
row = captured[0]
assert row["username"] == "analysttest"
assert row["action"] == "analyze"
assert row["method"] == "POST"
assert row["cad_num"] == "66:41:0204016:10"
assert row["status_code"] == 200
def test_middleware_does_not_audit_unmatched_paths() -> None:
captured: list[dict[str, Any]] = []
client = TestClient(_build_audited_app(captured))
r1 = client.get(
"/api/v1/analytics/dashboard",
headers={"X-Authenticated-User": "analysttest"},
)
r2 = client.get("/health")
assert r1.status_code == 200
assert r2.status_code == 200
assert captured == []
def test_middleware_export_action_distinct_from_forecast() -> None:
captured: list[dict[str, Any]] = []
client = TestClient(_build_audited_app(captured))
client.get(
"/api/v1/parcels/66:41:0204016:10/forecast",
headers={"X-Authenticated-User": "analysttest"},
)
client.get(
"/api/v1/parcels/66:41:0204016:10/forecast/export",
headers={"X-Authenticated-User": "analysttest"},
)
assert [r["action"] for r in captured] == ["forecast", "export"]
def test_middleware_audit_failure_does_not_break_request(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""Сбой записи аудита НЕ ломает основной запрос — real response отдан.
Здесь гоняем реальный audit_log_middleware (settings.testing выключаем
для этого app), но write_audit_row патчим на бросающий проверяем что
клиент всё равно получает 200 от handler'а."""
from app.core.config import settings
monkeypatch.setattr(settings, "testing", False)
def _boom(**_: Any) -> None:
raise RuntimeError("audit exploded")
monkeypatch.setattr(audit_mod, "write_audit_row", _boom)
app = FastAPI()
app.middleware("http")(audit_mod.audit_log_middleware)
@app.post("/api/v1/parcels/{cad}/analyze")
async def analyze(cad: str) -> dict:
return {"ok": True, "cad": cad}
client = TestClient(app)
resp = client.post(
"/api/v1/parcels/66:41:0204016:10/analyze",
headers={"X-Authenticated-User": "analysttest"},
)
# write_audit_row взорвался, но middleware проглотил → handler-ответ дошёл.
assert resp.status_code == 200
assert resp.json()["ok"] is True

View file

@ -2,13 +2,14 @@
Coverage: Coverage:
- get_role / get_user_scope happy + error paths - get_role / get_user_scope happy + error paths
- is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**) - is_path_allowed glob semantics (admin everywhere, pilot blocked from /admin/**,
analyst sees /** except admin/data-management #962)
- GET /me header missing / unknown user / pilot / admin - GET /me header missing / unknown user / pilot / admin
- rbac_guard middleware: - rbac_guard middleware:
* любой non-public path требует X-Authenticated-User (no-header 401) * любой non-public path требует X-Authenticated-User (no-header 401)
* неизвестный юзер 403 на ВСЁ (включая non-admin paths) * неизвестный юзер 403 на ВСЁ (включая non-admin paths)
«человек без ролей вообще ничего не видит» (decided 2026-05-25) «человек без ролей вообще ничего не видит» (decided 2026-05-25)
* /api/v1/admin/* только role=admin (pilot 403) * /api/v1/admin/* только role=admin (pilot/analyst 403; #962)
* public /health пропускаем без проверки * public /health пропускаем без проверки
Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не Тесты используют изолированный FastAPI app (см. _build_test_app), чтобы не
@ -87,6 +88,11 @@ def _build_test_app() -> FastAPI:
async def admin_dummy() -> dict: async def admin_dummy() -> dict:
return {"ok": True} return {"ok": True}
# Non-admin /api/v1/* endpoint — analyst/pilot/admin должны доходить до него.
@app.get("/api/v1/parcels/dummy")
async def parcels_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"}
@ -158,6 +164,28 @@ def test_is_path_allowed_pilot_only_tradein() -> None:
assert not auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/admin/scrape") assert not auth_mod.is_path_allowed("pilot", "/trade-in/api/v1/admin/scrape")
def test_is_path_allowed_analyst_everything_except_admin() -> None:
"""Analyst (#962) видит ВСЁ кроме admin/data-management.
paths=/** ; deny=/admin/**, /api/v1/admin/**, /trade-in/api/v1/admin/**."""
# Analyst ALLOWED — данные, аналитика, site-finder, concept, trade-in
assert auth_mod.is_path_allowed("analyst", "/")
assert auth_mod.is_path_allowed("analyst", "/analytics/dashboard")
assert auth_mod.is_path_allowed("analyst", "/site-finder/123")
assert auth_mod.is_path_allowed("analyst", "/concept/abc")
assert auth_mod.is_path_allowed("analyst", "/trade-in/123")
assert auth_mod.is_path_allowed("analyst", "/api/v1/parcels/66:41:0204016:10/analyze")
assert auth_mod.is_path_allowed("analyst", "/api/v1/parcels/66:41:0204016:10/forecast")
assert auth_mod.is_path_allowed("analyst", "/api/v1/analytics/dashboard")
# Analyst DENIED — admin/data-management (scraper, sync, user/role mgmt)
assert not auth_mod.is_path_allowed("analyst", "/admin")
assert not auth_mod.is_path_allowed("analyst", "/admin/jobs")
assert not auth_mod.is_path_allowed("analyst", "/admin/scrape/runs/42")
assert not auth_mod.is_path_allowed("analyst", "/api/v1/admin/jobs/settings")
assert not auth_mod.is_path_allowed("analyst", "/api/v1/admin/scrape/status")
assert not auth_mod.is_path_allowed("analyst", "/trade-in/api/v1/admin/scrape")
def test_is_path_allowed_unknown_role_denied() -> None: 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", "/")
assert not auth_mod.is_path_allowed("ghost", "/admin") assert not auth_mod.is_path_allowed("ghost", "/admin")
@ -189,6 +217,18 @@ def test_get_user_scope_pilot() -> None:
assert "/api/v1/admin/**" in scope["deny_paths"] assert "/api/v1/admin/**" in scope["deny_paths"]
def test_get_user_scope_analyst() -> None:
"""analysttest (#962) → analyst scope: allowed /** + admin deny_paths."""
scope = auth_mod.get_user_scope("analysttest")
assert scope["username"] == "analysttest"
assert scope["role"] == "analyst"
assert "/**" in scope["allowed_paths"]
# deny_paths драйвит фронтовый RouteGuard для /admin/** страниц.
assert "/admin/**" in scope["deny_paths"]
assert "/api/v1/admin/**" in scope["deny_paths"]
assert "/trade-in/api/v1/admin/**" in scope["deny_paths"]
def test_get_user_scope_unknown_raises() -> None: def test_get_user_scope_unknown_raises() -> None:
with pytest.raises(KeyError): with pytest.raises(KeyError):
auth_mod.get_user_scope("nobody") auth_mod.get_user_scope("nobody")
@ -255,6 +295,28 @@ def test_rbac_guard_admin_allowed(client: TestClient) -> None:
assert resp.json() == {"ok": True} assert resp.json() == {"ok": True}
def test_rbac_guard_analyst_blocked_on_admin_api(client: TestClient) -> None:
"""#962 backend-hard: existing rbac_guard АВТО-блокирует analyst на
/api/v1/admin/* (role != admin) без доп. кода, просто потому что роль
добавлена. Подтверждаем что 403 'admin only'."""
resp = client.get(
"/api/v1/admin/dummy",
headers={"X-Authenticated-User": "analysttest"},
)
assert resp.status_code == 403
assert resp.json()["detail"] == "admin only"
def test_rbac_guard_analyst_allowed_on_non_admin_api(client: TestClient) -> None:
"""Analyst свободно ходит в обычные /api/v1/* (не admin) — 200."""
resp = client.get(
"/api/v1/parcels/dummy",
headers={"X-Authenticated-User": "analysttest"},
)
assert resp.status_code == 200
assert resp.json() == {"ok": True}
def test_rbac_guard_no_header_on_admin_path_returns_401(client: TestClient) -> None: def test_rbac_guard_no_header_on_admin_path_returns_401(client: TestClient) -> None:
"""Локальный curl без Caddy → /api/v1/admin/* отдаёт 401.""" """Локальный curl без Caddy → /api/v1/admin/* отдаёт 401."""
resp = client.get("/api/v1/admin/dummy") resp = client.get("/api/v1/admin/dummy")

View file

@ -0,0 +1,65 @@
-- 144_audit_log.sql
-- #962 (EPIC18, ТЗ §19): аудит-лог чувствительных действий аналитика/пользователей.
--
-- Пишется best-effort FastAPI-middleware (app/core/audit_middleware.py) ПОСЛЕ
-- ответа на запрос для ключевых действий:
-- POST /api/v1/parcels/{cad}/analyze → action='analyze'
-- GET /api/v1/parcels/{cad}/forecast → action='forecast'
-- GET /api/v1/parcels/{cad}/forecast/export → action='export'
-- Прочие matched-нестандартные пути → action='other'. /health, static,
-- non-matched пути НЕ аудируются.
--
-- Best-effort: сбой записи в audit_log НИКОГДА не ломает основной запрос
-- (middleware оборачивает write в try/except).
--
-- Deploy: auto-applied deploy.yml через _schema_migrations (ровно один раз NN=144).
-- Idempotent: CREATE TABLE/INDEX IF NOT EXISTS + guarded DO-блок для CHECK.
BEGIN;
-- ── 1. Таблица audit_log ──────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS audit_log (
id BIGSERIAL PRIMARY KEY,
username TEXT NOT NULL, -- из X-Authenticated-User
role TEXT, -- get_role(username) на момент запроса
action TEXT NOT NULL, -- analyze | forecast | export | other
method TEXT NOT NULL, -- HTTP-метод (GET/POST/...)
path TEXT NOT NULL, -- request.url.path
cad_num TEXT, -- кадастровый номер из пути (NULL если нет)
status_code INTEGER, -- HTTP-статус ответа
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
detail JSONB -- произвольный контекст (резерв)
);
COMMENT ON TABLE audit_log IS
'Аудит чувствительных действий (#962, ТЗ §19). Пишется best-effort '
'middleware app/core/audit_middleware.py ПОСЛЕ ответа на запрос. '
'Сбой записи не влияет на основной запрос.';
COMMENT ON COLUMN audit_log.action IS
'analyze | forecast | export | other — выводится из request path в middleware.';
COMMENT ON COLUMN audit_log.cad_num IS
'Кадастровый номер из /parcels/{cad}/... (напр. 66:41:0204016:10), NULL если нет.';
COMMENT ON COLUMN audit_log.detail IS
'Резервное JSONB-поле под доп. контекст (пока не заполняется).';
-- ── 2. CHECK на допустимые action (guarded — idempotent при ре-apply) ────────
DO $$
BEGIN
IF NOT EXISTS (
SELECT 1 FROM pg_constraint WHERE conname = 'ck_audit_log_action'
) THEN
ALTER TABLE audit_log
ADD CONSTRAINT ck_audit_log_action
CHECK (action IN ('analyze', 'forecast', 'export', 'other'));
END IF;
END
$$;
-- ── 3. Индексы: lookup по юзеру во времени + по кадастру ──────────────────────
CREATE INDEX IF NOT EXISTS idx_audit_log_username_created
ON audit_log (username, created_at);
CREATE INDEX IF NOT EXISTS idx_audit_log_cad_num
ON audit_log (cad_num);
COMMIT;

View file

@ -31,7 +31,7 @@ import yaml
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
Role = Literal["admin", "pilot"] Role = Literal["admin", "pilot", "analyst"]
class UserScope(TypedDict): class UserScope(TypedDict):