"""Audit logging with request context and integrity hashing.""" from __future__ import annotations import hashlib from datetime import datetime, timezone from sqlalchemy.orm import Session from app.middleware.request_context import request_ip, request_user_agent from app.models.audit import AuditLog def _integrity_payload(entry: AuditLog) -> str: ts = entry.timestamp if ts is None: ts = datetime.now(timezone.utc) user_part = str(entry.user_id) if entry.user_id else "" entity_type = entry.entity_type or "" entity_id = entry.entity_id or "" return f"{user_part}:{entry.action}:{entity_type}:{entity_id}:{ts.isoformat()}" def compute_integrity_hash(entry: AuditLog) -> str: """Return the SHA-256 hex digest for an audit log entry.""" return hashlib.sha256(_integrity_payload(entry).encode()).hexdigest() def verify_audit_integrity(entry: AuditLog) -> bool: """Return whether the stored hash matches the entry's current fields.""" if not entry.integrity_hash: return False return entry.integrity_hash == compute_integrity_hash(entry) def log_action( db: Session, user_id, action: str, entity_type: str | None = None, entity_id: str | None = None, details: dict | None = None, *, ip_address: str | None = None, user_agent: str | None = None, session_id: str | None = None, ) -> AuditLog: """Record an audit event. Does not commit — the caller owns the transaction.""" ip = ip_address if ip_address is not None else request_ip.get("") ua = user_agent if user_agent is not None else request_user_agent.get("") entry = AuditLog( user_id=user_id, action=action, entity_type=entity_type, entity_id=str(entity_id) if entity_id else None, details=details, ip_address=ip or None, user_agent=ua or None, session_id=session_id, timestamp=datetime.now(timezone.utc), ) db.add(entry) db.flush() entry.integrity_hash = compute_integrity_hash(entry) return entry