mirror of
https://gitverse.ru/kpa39l/md2vk.git
synced 2026-09-29 09:55:04 +00:00
estorozhenko
121 lines
4.9 KiB
Python
121 lines
4.9 KiB
Python
"""Аудит-лог авторизации и запросов API (JSONL, ротация по дням).
|
|
|
|
Пишет строку JSON на каждый запрос /api/v1/*:
|
|
ts, ip, method, path, api_key_prefix, user_id, status, success, latency_ms, error.
|
|
|
|
Параметры из env:
|
|
- AUDIT_LOG_DIR — каталог для логов (по умолчанию "logs").
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import time
|
|
from datetime import date, datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.requests import Request
|
|
|
|
logger = logging.getLogger("md2vk.audit")
|
|
|
|
|
|
class AuditMiddleware(BaseHTTPMiddleware):
|
|
"""Логирует каждый запрос /api/v1/* в JSONL с ротацией по дням."""
|
|
|
|
def __init__(self, app, log_dir: str | None = None):
|
|
super().__init__(app)
|
|
self.log_dir = Path(log_dir or os.getenv("AUDIT_LOG_DIR", "logs"))
|
|
self.log_dir.mkdir(parents=True, exist_ok=True)
|
|
self._fh = None
|
|
self._fh_date: date | None = None
|
|
|
|
def _ensure_file(self) -> None:
|
|
today = date.today()
|
|
if self._fh is None or self._fh_date != today:
|
|
if self._fh is not None:
|
|
try:
|
|
self._fh.close()
|
|
except Exception:
|
|
pass
|
|
path = self.log_dir / f"access.{today.isoformat()}.log"
|
|
self._fh = open(path, "a", encoding="utf-8")
|
|
self._fh_date = today
|
|
|
|
def _write(self, record: dict) -> None:
|
|
try:
|
|
self._ensure_file()
|
|
self._fh.write(json.dumps(record, ensure_ascii=False, default=str) + "\n")
|
|
self._fh.flush()
|
|
except Exception:
|
|
# Логгер не должен ронять API
|
|
logger.exception("audit write failed")
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
start = time.monotonic()
|
|
response = None
|
|
error = None
|
|
try:
|
|
response = await call_next(request)
|
|
return response
|
|
except Exception as exc: # noqa: BLE001
|
|
error = str(exc)
|
|
raise
|
|
finally:
|
|
path = request.url.path
|
|
if path.startswith("/api/v1"):
|
|
try:
|
|
latency_ms = round((time.monotonic() - start) * 1000, 1)
|
|
status = response.status_code if response is not None else 500
|
|
auth = request.headers.get("authorization", "")
|
|
|
|
# api_key может быть в теле (POST) — пытаемся достать
|
|
api_key_prefix = ""
|
|
api_key_hash_short = ""
|
|
user_id = None
|
|
if auth.startswith("Bearer "):
|
|
api_key_prefix = auth[len("Bearer "):][:12]
|
|
elif request.method == "POST":
|
|
api_key_prefix = self._api_key_from_body(request)
|
|
if "md2vk_" in api_key_prefix:
|
|
# вычислим короткий хэш для привязки к user (без хранения ключа)
|
|
import hashlib
|
|
api_key_hash_short = hashlib.sha256(
|
|
api_key_prefix.encode()
|
|
).hexdigest()[:12]
|
|
|
|
record = {
|
|
"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"),
|
|
"ip": (request.client.host if request.client else ""),
|
|
"method": request.method,
|
|
"path": path,
|
|
"query": str(request.url.query) or "",
|
|
"api_key_prefix": api_key_prefix,
|
|
"api_key_hash_short": api_key_hash_short,
|
|
"user_id": user_id,
|
|
"status": status,
|
|
"success": status < 400,
|
|
"latency_ms": latency_ms,
|
|
"error": error,
|
|
}
|
|
self._write(record)
|
|
except Exception: # noqa: BLE001
|
|
logger.exception("audit dispatch failed")
|
|
|
|
@staticmethod
|
|
def _api_key_from_body(request: Request) -> str:
|
|
"""Достаёт api_key из JSON-тела, не ломая повторное чтение."""
|
|
try:
|
|
# starlette кэширует _body — повторное чтение в роутере безопасно
|
|
body = getattr(request, "_body", None)
|
|
if body is None:
|
|
body = request.body() if hasattr(request, "body") else b""
|
|
if isinstance(body, bytes) and body:
|
|
data = json.loads(body)
|
|
key = data.get("api_key", "")
|
|
return str(key)[:12]
|
|
except Exception:
|
|
return ""
|
|
return "" |