686acf259a
Delivers the append-only audit trail system for all compliance-critical actions per the TRA-246 acceptance criteria. - tracking/models.py: AuditEvent model with ORM-level immutability guard (save raises on update, delete raises on direct call) - tracking/audit.py: single record() call point; never raises in production - tracking/admin.py: read-only Django admin for AuditEvent inspection - tracking/migrations/0001_initial.py: DB schema with composite indexes - tracking/serializers.py: PII metadata gating (oidc_sub stripped for non-admin callers) - tracking/views.py: read-only AuditEventViewSet (IsPrivileged + 60/min throttle) - tracking/urls.py: registers audit/events/ router - tracking/management/commands/prune_audit_log.py: retention enforcement command with --dry-run and --class filter; writes access.admin_action event on real prune runs - config/settings/base.py: AUDIT_RETENTION_DAYS per event class + audit throttle rate - api/exceptions.py: wires access.permission_denied audit event on every PermissionDenied exception (M1 integration point) - tests/test_audit.py: 26-event taxonomy coverage, immutability, retention, API permission, PII gating, and service helper unit tests Co-Authored-By: Paperclip <noreply@paperclip.ing>
104 lines
2.7 KiB
Python
104 lines
2.7 KiB
Python
"""
|
|
Single call point for all audit event writes.
|
|
|
|
Usage:
|
|
from tracking.audit import record
|
|
|
|
record(
|
|
"auth.login_success",
|
|
actor=request.user,
|
|
actor_ip=client_ip,
|
|
metadata={"oidc_sub": sub},
|
|
)
|
|
|
|
Never raises in production so that compliance events cannot break request processing.
|
|
In DEBUG mode the write error is re-raised for test visibility.
|
|
"""
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
|
|
from django.conf import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Valid event_class prefixes derived from the taxonomy
|
|
_VALID_CLASSES = {
|
|
"auth",
|
|
"progress",
|
|
"assessment",
|
|
"trainer",
|
|
"certificate",
|
|
"access",
|
|
}
|
|
|
|
|
|
def _derive_class(event_type: str) -> str:
|
|
prefix = event_type.split(".")[0]
|
|
return prefix if prefix in _VALID_CLASSES else "_unknown"
|
|
|
|
|
|
def _retention_class(event_class: str) -> str:
|
|
"""Return the retention bucket key for this event class."""
|
|
cfg = getattr(settings, "AUDIT_RETENTION_DAYS", {})
|
|
if event_class in cfg:
|
|
return event_class
|
|
return "_default"
|
|
|
|
|
|
def record(
|
|
event_type: str,
|
|
*,
|
|
actor=None,
|
|
actor_type: str = "user",
|
|
actor_ip: str | None = None,
|
|
object_type: str | None = None,
|
|
object_id: str | None = None,
|
|
occurred_at: datetime | None = None,
|
|
metadata: dict | None = None,
|
|
):
|
|
"""
|
|
Persist a compliance audit event.
|
|
|
|
Returns the saved AuditEvent instance, or None if the write failed and
|
|
DEBUG is False (the error is logged instead of propagated).
|
|
"""
|
|
from tracking.models import AuditEvent # local import avoids circular at module load
|
|
|
|
event_class = _derive_class(event_type)
|
|
ret_class = _retention_class(event_class)
|
|
ts = occurred_at or datetime.now(tz=timezone.utc)
|
|
|
|
actor_id = None
|
|
if actor is not None:
|
|
pk = getattr(actor, "pk", None) or getattr(actor, "id", None)
|
|
if pk is not None:
|
|
try:
|
|
import uuid as _uuid
|
|
actor_id = _uuid.UUID(str(pk))
|
|
except (ValueError, AttributeError):
|
|
actor_id = None
|
|
|
|
obj_id = str(object_id) if object_id is not None else None
|
|
|
|
try:
|
|
event = AuditEvent(
|
|
event_class=event_class,
|
|
event_type=event_type,
|
|
actor_id=actor_id,
|
|
actor_type=actor_type,
|
|
actor_ip=actor_ip,
|
|
object_type=object_type,
|
|
object_id=obj_id,
|
|
occurred_at=ts,
|
|
metadata=metadata or {},
|
|
retention_class=ret_class,
|
|
)
|
|
event.save()
|
|
return event
|
|
except Exception as exc:
|
|
logger.exception("audit.record failed for event_type=%s: %s", event_type, exc)
|
|
if getattr(settings, "DEBUG", False):
|
|
raise
|
|
return None
|