|
- import json
- import time
- import logging
- from datetime import datetime, timezone
- from typing import Any, Dict, Optional
-
- from fastapi import Request
- from starlette.middleware.base import BaseHTTPMiddleware
- from jose import jwt
-
- from security import _get_key, KEYCLOAK_ISSUER, ALGORITHMS
-
- audit_logger = logging.getLogger("audit")
- audit_logger.setLevel(logging.INFO)
-
-
- def _ts_minute_utc() -> str:
- # formato: 2025-12-12 15:47 (UTC)
- return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M")
-
-
- def _compact_json(obj: Any, max_len: int = 2000) -> str:
- try:
- s = json.dumps(obj, ensure_ascii=False, separators=(",", ":"))
- except Exception:
- s = str(obj)
- if len(s) > max_len:
- return s[:max_len] + "...(truncated)"
- return s
-
-
- async def _user_from_bearer(request: Request) -> Optional[str]:
- auth = request.headers.get("authorization", "")
- if not auth.lower().startswith("bearer "):
- return None
- token = auth.split(" ", 1)[1].strip()
-
- try:
- key = await _get_key(token)
- claims = jwt.decode(
- token,
- key,
- algorithms=ALGORITHMS,
- issuer=KEYCLOAK_ISSUER,
- options={"verify_aud": False, "verify_iss": True},
- )
- return (
- claims.get("preferred_username")
- or claims.get("username")
- or claims.get("email")
- )
- except Exception:
- return None
-
-
- async def _read_request_body_if_any(request: Request) -> Optional[str]:
- # logga solo metodi che tipicamente hanno body
- if request.method not in ("POST", "PUT", "PATCH", "DELETE"):
- return None
-
- ctype = (request.headers.get("content-type") or "").lower()
- if not ctype:
- return None
-
- # Evita file upload / form-data grossi
- if "multipart/form-data" in ctype:
- return "<multipart/form-data omitted>"
-
- try:
- raw = await request.body()
- if not raw:
- return None
-
- # Prova JSON
- if "application/json" in ctype:
- try:
- return _compact_json(json.loads(raw.decode("utf-8", errors="replace")))
- except Exception:
- # se non parseabile, logga raw truncato
- txt = raw.decode("utf-8", errors="replace")
- return (txt[:2000] + "...(truncated)") if len(txt) > 2000 else txt
-
- # Testo semplice
- if "text/" in ctype or "application/x-www-form-urlencoded" in ctype:
- txt = raw.decode("utf-8", errors="replace")
- return (txt[:2000] + "...(truncated)") if len(txt) > 2000 else txt
-
- # Altro (binary)
- return f"<body omitted content-type={ctype} bytes={len(raw)}>"
- except Exception:
- return "<body read error>"
-
-
- class AuditMiddleware(BaseHTTPMiddleware):
- async def dispatch(self, request: Request, call_next):
- start = time.time()
-
- method = request.method
- path = request.url.path
-
- # 1) via header del proxy (preferito)
- user = request.headers.get("x-authenticated-user")
-
- # 2) fallback: decodifica bearer senza usare verify_token (evita spam del logger security)
- if not user:
- user = await _user_from_bearer(request) or "-"
-
- body_str = await _read_request_body_if_any(request)
-
- status_code = 500
- try:
- response = await call_next(request)
- status_code = response.status_code
- return response
- finally:
- duration_ms = int((time.time() - start) * 1000)
- ts = _ts_minute_utc()
-
- if body_str:
- audit_logger.info(
- "%s user=%s %s %s status=%s ms=%s body=%s",
- ts, user, method, path, status_code, duration_ms, body_str
- )
- else:
- audit_logger.info(
- "%s user=%s %s %s status=%s ms=%s",
- ts, user, method, path, status_code, duration_ms
- )
|
