52 lines
1.4 KiB
Python
52 lines
1.4 KiB
Python
"""Audit logging helper (spec section 38).
|
|
|
|
Every privileged action writes an append-only AuditEvent row. Secrets and
|
|
password fields must never be placed in old_value/new_value — strip them
|
|
at the call site before serializing.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
from fastapi import Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import AuditEvent
|
|
|
|
|
|
def record(
|
|
db: Session,
|
|
action: str,
|
|
*,
|
|
request: Request | None = None,
|
|
user: str | None = None,
|
|
object_type: str | None = None,
|
|
object_id: str | None = None,
|
|
old_value: Any = None,
|
|
new_value: Any = None,
|
|
result: str = "success",
|
|
correlation_id: str | None = None,
|
|
) -> None:
|
|
source_ip = None
|
|
if request is not None:
|
|
source_ip = request.client.host if request.client else None
|
|
correlation_id = correlation_id or getattr(request.state, "correlation_id", None)
|
|
|
|
db.add(
|
|
AuditEvent(
|
|
id=uuid.uuid4(),
|
|
action=action,
|
|
user=user,
|
|
source_ip=source_ip,
|
|
object_type=object_type,
|
|
object_id=object_id,
|
|
old_value=old_value,
|
|
new_value=new_value,
|
|
result=result,
|
|
correlation_id=correlation_id,
|
|
)
|
|
)
|
|
# Intentionally no commit here: the request-scoped session commits
|
|
# atomically with the business change (get_db commits on success). |