5025732d12
All checks were successful
CI Trade-In / changes (pull_request) Successful in 7s
CI / changes (pull_request) Successful in 7s
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 4m52s
Admin viewing the audit/analytics dashboards hits /api/v1/admin/* which the request-audit middleware logged as api_request — polluting top_paths and activity counts with the dashboards' own traffic. Skip api_request for admin paths; keep login (admin sign-in from an IP is still valid audit).
71 lines
3.5 KiB
Python
71 lines
3.5 KiB
Python
"""RequestAuditMiddleware — пишет `api_request` (+ дедуплицированный `login`)
|
||
события в `user_events` для каждого аутентифицированного `/api/*` запроса.
|
||
|
||
Foundation для Feature 2 (login/IP audit) и базы Feature 3 (behavior analytics).
|
||
Логирование выполняется ПОСЛЕ `call_next` (не задерживает и не ветвит реальный
|
||
ответ клиенту) и через fire-and-forget `schedule_event` — сбой аудита никогда
|
||
не влияет на HTTP-ответ.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from fastapi import Request
|
||
from starlette.middleware.base import BaseHTTPMiddleware
|
||
from starlette.responses import Response
|
||
|
||
from app.core.ratelimit import _client_ip
|
||
from app.services.user_events import schedule_event, should_log_login
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Зеркалит app.main._PUBLIC_PATHS. Не импортируем напрямую из app.main — оно
|
||
# импортирует этот модуль (регистрирует middleware), обратный импорт дал бы
|
||
# циклическую зависимость.
|
||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
||
|
||
|
||
class RequestAuditMiddleware(BaseHTTPMiddleware):
|
||
"""Логирует активность аутентифицированных пользователей в `user_events`."""
|
||
|
||
async def dispatch(self, request: Request, call_next): # type: ignore[no-untyped-def]
|
||
response: Response = await call_next(request)
|
||
|
||
try:
|
||
username = request.headers.get("x-authenticated-user")
|
||
path = request.url.path
|
||
if username and path.startswith("/api/") and path not in _PUBLIC_PATHS:
|
||
ip = _client_ip(request)
|
||
ua = request.headers.get("user-agent")
|
||
|
||
# Общий behavior/activity-поток — каждый authenticated API-запрос.
|
||
# /api/v1/admin/* исключаем: это ops-действия (просмотр самих
|
||
# дашбордов аудита/аналитики), а не поведение пилота — иначе
|
||
# запросы дашборда зашумляют top_paths и счётчики активности.
|
||
# login ниже логируем всегда (вход админа с IP — валидный аудит).
|
||
if not path.startswith("/api/v1/admin/"):
|
||
schedule_event(
|
||
event_type="api_request",
|
||
username=username,
|
||
ip=ip,
|
||
user_agent=ua,
|
||
path=path,
|
||
method=request.method,
|
||
)
|
||
|
||
# Дедуплицированный login/IP-audit сигнал — максимум раз в день
|
||
# на (юзер, IP, устройство).
|
||
if should_log_login(username, ip, ua):
|
||
schedule_event(
|
||
event_type="login",
|
||
username=username,
|
||
ip=ip,
|
||
user_agent=ua,
|
||
path=path,
|
||
method=request.method,
|
||
)
|
||
except Exception:
|
||
logger.warning("RequestAuditMiddleware: failed to record event", exc_info=True)
|
||
|
||
return response
|