75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Callable, Mapping
|
|
from datetime import datetime, timedelta
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_audit.backend.db.models import AuditLog
|
|
from govoplan_audit.backend.recording import _record_ref
|
|
from govoplan_core.core.access import AuditRecordRef, AuditRetentionProvider
|
|
|
|
|
|
class SqlAuditRetentionProvider(AuditRetentionProvider):
|
|
def apply_detail_retention(
|
|
self,
|
|
session: object,
|
|
*,
|
|
dry_run: bool,
|
|
now: datetime,
|
|
policy_for_record: Callable[[AuditRecordRef], object],
|
|
) -> Mapping[str, int]:
|
|
db = _session(session)
|
|
result = {"eligible": 0, "redacted": 0, "already_redacted": 0}
|
|
items = db.query(AuditLog).order_by(AuditLog.created_at.asc()).all()
|
|
for item in items:
|
|
policy = policy_for_record(_record_ref(item))
|
|
cutoff = _cutoff(getattr(policy, "audit_detail_retention_days", None), now=now)
|
|
if not _is_before_cutoff(item.created_at, cutoff):
|
|
continue
|
|
details = item.details if isinstance(item.details, dict) else {}
|
|
if details.get("_retention", {}).get("audit_detail_redacted"):
|
|
result["already_redacted"] += 1
|
|
continue
|
|
result["eligible"] += 1
|
|
if dry_run:
|
|
continue
|
|
item.details = {
|
|
"_retention": {
|
|
"audit_detail_redacted": True,
|
|
"redacted_at": now.isoformat(),
|
|
"original_detail_sha256": _json_sha256(details),
|
|
}
|
|
}
|
|
db.add(item)
|
|
result["redacted"] += 1
|
|
return result
|
|
|
|
|
|
def _cutoff(days: int | None, *, now: datetime) -> datetime | None:
|
|
if days is None:
|
|
return None
|
|
return now - timedelta(days=days)
|
|
|
|
|
|
def _is_before_cutoff(value: datetime | None, cutoff: datetime | None) -> bool:
|
|
if value is None or cutoff is None:
|
|
return False
|
|
candidate = value
|
|
if candidate.tzinfo is None:
|
|
candidate = candidate.replace(tzinfo=cutoff.tzinfo)
|
|
return candidate < cutoff
|
|
|
|
|
|
def _json_sha256(value: object) -> str:
|
|
import hashlib
|
|
import json
|
|
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _session(session: object) -> Session:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Audit retention requires a SQLAlchemy Session")
|
|
return session
|