From d314d45c6ec0e69f2338b53e692a2133d8c21005 Mon Sep 17 00:00:00 2001 From: leo <260626284+cameledev@users.noreply.github.com> Date: Wed, 10 Jun 2026 14:44:32 +0200 Subject: [PATCH] wip --- src/backend/core/audit.py | 194 +++++++++++++++++++++ src/backend/core/audit_meet.py | 19 +++ src/backend/core/tests/test_audit_log.py | 206 +++++++++++++++++++++++ 3 files changed, 419 insertions(+) create mode 100644 src/backend/core/audit.py create mode 100644 src/backend/core/audit_meet.py create mode 100644 src/backend/core/tests/test_audit_log.py diff --git a/src/backend/core/audit.py b/src/backend/core/audit.py new file mode 100644 index 00000000..616e53af --- /dev/null +++ b/src/backend/core/audit.py @@ -0,0 +1,194 @@ +"""Structured audit logging.""" + +# Audit helpers intentionally expose many optional keyword fields. +# pylint: disable=R0913,R0917 +# ruff: noqa: PLR0913 + +import json +import logging +from datetime import datetime, timezone +from functools import partialmethod +from typing import TYPE_CHECKING, Any, Protocol + +from django.http import HttpRequest + +AUDIT_LOGGER_NAME = "audit" + + +def resolve_source_ip(request: HttpRequest): + """Return the best-effort client IP for ``request``. + + Reads the original client from ``X-Forwarded-For`` when present, + falling back to ``REMOTE_ADDR``. + + NB: behind a proxy/load-balancer chain, correctness depends on the ingress + being configured to set and trust ``X-Forwarded-For``. Confirm the + trusted-proxy chain before relying on this value for security decisions. + """ + forwarded = request.META.get("HTTP_X_FORWARDED_FOR") + if forwarded: + return forwarded.split(",")[0].strip() + return request.META.get("REMOTE_ADDR") + + +def extract_request_fields(request: HttpRequest) -> dict[str, Any]: + """Return the audit fields derivable from ``request``.""" + meta = request.META + return { + "source_ip": resolve_source_ip(request), + "source_port": meta.get("REMOTE_PORT"), + "request_path": request.path, + "request_url": request.build_absolute_uri(), + "request_method": request.method, + "request_body_bytes": meta.get("CONTENT_LENGTH"), + "http_version": meta.get("SERVER_PROTOCOL"), + "user_agent": meta.get("HTTP_USER_AGENT"), + "referer": meta.get("HTTP_REFERER"), + "server_user": meta.get("REMOTE_USER"), + } + + +class AuditJsonFormatter(logging.Formatter): + """Render audit records as single-line JSON. + + Read the structured payload attached to the record under ``audit`` and + wrap it in a small envelope. + """ + + def format(self, record): + payload = {"log_type": "audit"} + + audit = getattr(record, "audit", None) + if isinstance(audit, dict): + payload.update(audit) + else: + payload["event_type"] = record.getMessage() + + payload.setdefault( + "timestamp", + datetime.fromtimestamp(record.created, tz=timezone.utc).isoformat(), + ) + payload["level"] = record.levelname + payload["logger"] = record.name + + # ``default=str`` serialises UUIDs, datetimes, etc.; ``ensure_ascii`` + # off keeps emails and non-ASCII identifiers readable. + return json.dumps(payload, default=str, ensure_ascii=False) + + +class _AuditEmit(Protocol): + """Public signature shared by the per-level audit methods. + + Declared so editors and type checkers see the real keyword fields despite using `partialwrapper`. + """ + + def __call__( + self, + event_type: str, + *, + auth_type: str | None = ..., + actor: dict[str, Any] | None = ..., + source_ip: str | None = ..., + target: dict[str, Any] | None = ..., + request: HttpRequest | None = ..., + **extra: Any, + ) -> None: + pass + + +class AuditLogger: + """Wrapper around the named ``audit`` logger.""" + + def __init__( + self, + logger_name=AUDIT_LOGGER_NAME, + custom_serializers: list[tuple] | None = None, + ): + self._logger = logging.getLogger(logger_name) + self._serializers = [] + if custom_serializers is not None: + self._serializers = custom_serializers + + def _emit( + self, + level, + event_type, + request: HttpRequest | None, # Intentionnaly mandatory + *, + exc_info: bool = False, + **extra, + ): + """Assemble the structured payload and emit it on the audit logger. + + ``exc_info`` is forwarded to the stdlib logger (set by ``exception``) so + the active traceback is captured; it is a logging concern and never + enters the audit payload. + """ + + request_fields = extract_request_fields(request) if request is not None else {} + + # Create fields from custom serializers + new_extra = {} + for key, value in extra.items(): + for cls, serializer in self._serializers: + if isinstance(value, cls): + new_extra = new_extra | { + f"{key}.{ser_key}": ser_field + for ser_key, ser_field in serializer(value).items() + } + break + else: + new_extra[key] = value + + audit = { + "timestamp": datetime.now(timezone.utc).isoformat(), + "event_type": event_type, + **request_fields, + **new_extra, + } + audit = {key: value for key, value in audit.items() if value is not None} + + self._logger.log(level, event_type, extra={"audit": audit}, exc_info=exc_info) + + # One public method per standard logging level, all sharing ``_emit``. + # The ``TYPE_CHECKING`` declarations expose the real signature to editors; + # the ``else`` branch is what runs, binding the level via ``partialmethod``. + if TYPE_CHECKING: + debug: _AuditEmit + info: _AuditEmit + warning: _AuditEmit + error: _AuditEmit + critical: _AuditEmit + # ``exception`` mirrors stdlib: ERROR level with the active traceback. + exception: _AuditEmit + else: + debug = partialmethod(_emit, logging.DEBUG) + info = partialmethod(_emit, logging.INFO) + warning = partialmethod(_emit, logging.WARNING) + error = partialmethod(_emit, logging.ERROR) + critical = partialmethod(_emit, logging.CRITICAL) + exception = partialmethod(_emit, logging.ERROR, exc_info=True) + + +def getLogger( + name: str | None = None, custom_serializers: list[tuple | None] = None +) -> AuditLogger: + """Return an :class:`AuditLogger`, mirroring :func:`logging.getLogger`. + + Pass ``__name__`` to tag audit records with the calling module while still + emitting on the dedicated ``audit`` handler:: + + from core import audit + + logger = audit.getLogger(__name__) + logger.info("external_api.token.issued", ...) + + Names are nested under ``AUDIT_LOGGER_NAME`` (e.g. ``audit.core.foo``) so + they inherit its handlers through the standard logging hierarchy, keeping + the module visible in the ``logger`` field of the emitted JSON. + """ + if not name or name == AUDIT_LOGGER_NAME: + return AuditLogger(AUDIT_LOGGER_NAME, custom_serializers=custom_serializers) + return AuditLogger( + f"{AUDIT_LOGGER_NAME}.{name}", custom_serializers=custom_serializers + ) diff --git a/src/backend/core/audit_meet.py b/src/backend/core/audit_meet.py new file mode 100644 index 00000000..8d2d583e --- /dev/null +++ b/src/backend/core/audit_meet.py @@ -0,0 +1,19 @@ +"""Audit logging with custom serializers.""" + +from core.audit import getLogger as get_logger_base +from core.models import Room + + +def serialize_room(room: Room): + return { + "name": room.name, + "slug": room.slug, + "access_level": room.access_level, + } + + +custom_serializers = [(Room, serialize_room)] + + +def getLogger(name: str | None = None): + return get_logger_base(name=name, custom_serializers=custom_serializers) diff --git a/src/backend/core/tests/test_audit_log.py b/src/backend/core/tests/test_audit_log.py new file mode 100644 index 00000000..cd68169a --- /dev/null +++ b/src/backend/core/tests/test_audit_log.py @@ -0,0 +1,206 @@ +""" +Test audit.py +""" + +# pylint: disable=W0621,redefined-outer-name + +import json +import logging + +from django.test import RequestFactory + +import pytest + +from core.audit import AuditJsonFormatter, extract_request_fields, getLogger + + +class ExampleRoomModel: + """Example for an audited object.""" + + def __init__(self, name, slug, access_level): + self.name = name + self.slug = slug + self.access_level = access_level + + +def serialize_example_room(room): + """Flatten a ``ExampleRoomModel`` into the audit fields a service would expose.""" + return { + "name": room.name, + "slug": room.slug, + "access_level": room.access_level, + } + + +# Audit logger wired with the local serializer, mirroring how a service +# registers its own object serializers. +test_audit_logger = getLogger( + "core.tests.test_audit_2", + custom_serializers=[(ExampleRoomModel, serialize_example_room)], +) + + +class _CaptureHandler(logging.Handler): + """Collect every record routed to the ``audit`` logger tree.""" + + def __init__(self): + super().__init__() + self.records = [] + + def emit(self, record): + self.records.append(record) + + +@pytest.fixture +def audit_records(): + """Yield audit records, capturing children via propagation to ``audit``. + + The viewsets log on ``audit.core.external_api.viewsets``; attaching to the + root ``audit`` logger captures those propagated records as well as ones + emitted directly on it. + """ + handler = _CaptureHandler() + logger = logging.getLogger("audit") + logger.addHandler(handler) + previous_level = logger.level + logger.setLevel(logging.DEBUG) + try: + yield handler.records + finally: + logger.removeHandler(handler) + logger.setLevel(previous_level) + + +def _payloads(records, event_type): + """Return the audit payloads matching ``event_type``.""" + return [r.audit for r in records if r.audit.get("event_type") == event_type] + + +# --------------------------------------------------------------------------- +# core.audit - custom serializer flattening +# --------------------------------------------------------------------------- + + +def test_room_extra_is_flattened_via_custom_serializer(audit_records): + """A ``Room`` keyword is expanded into dotted ``room.`` keys.""" + room = ExampleRoomModel( + name="Talking About Vacation", + slug="talking-about-vacations", + access_level="restricted", + ) + + test_audit_logger.info("event", request=None, room=room) + + audit = audit_records[0].audit + assert audit["room.name"] == "Talking About Vacation" + assert audit["room.slug"] == "talking-about-vacations" + assert audit["room.access_level"] == "restricted" + # The raw object is replaced by its serialized fields, not kept. + assert "room" not in audit + + +def test_non_registered_extras_pass_through_unchanged(audit_records): + """Values without a matching serializer are kept verbatim.""" + room = ExampleRoomModel( + name="Back To Work", slug="back-to-work", access_level="public" + ) + + test_audit_logger.info( + "event", + request=None, + room=room, + client_id="test-id", + target={"user_id": "u1"}, + ) + + audit = audit_records[0].audit + assert audit["client_id"] == "test-id" + # A dict has no registered serializer, so it stays nested as-is. + assert audit["target"] == {"user_id": "u1"} + # The Room alongside it is still flattened. + assert audit["room.name"] == "Back To Work" + + +# --------------------------------------------------------------------------- +# core.audit - getLogger naming & request field extraction +# --------------------------------------------------------------------------- + + +def test_getlogger_nests_module_name_under_audit(): + """A named logger sits under ``audit.`` so it inherits its handlers.""" + assert getLogger("core.foo")._logger.name == "audit.core.foo" + + +def test_getlogger_without_name_uses_base_audit_logger(): + """Empty ``audit`` names resolve to the bare ``audit`` logger.""" + assert getLogger()._logger.name == "audit" + assert getLogger("audit")._logger.name == "audit" + + +def test_extract_request_fields_collects_request_metadata(): + """All request-derived fields are populated from request META.""" + request = RequestFactory().post( + "/external-api/v1.0/rooms/", + REMOTE_ADDR="10.0.0.5", + REMOTE_PORT="54321", + HTTP_USER_AGENT="pytest-agent", + HTTP_REFERER="https://example.test/from", + HTTP_X_FORWARDED_FOR="203.0.113.7, 10.0.0.5", + ) + + fields = extract_request_fields(request) + + # X-Forwarded-For wins over REMOTE_ADDR for the client IP. + assert fields["source_ip"] == "203.0.113.7" + assert fields["source_port"] == "54321" + assert fields["request_path"] == "/external-api/v1.0/rooms/" + assert fields["request_method"] == "POST" + assert fields["user_agent"] == "pytest-agent" + assert fields["referer"] == "https://example.test/from" + assert fields["request_url"].endswith("/external-api/v1.0/rooms/") + + +# --------------------------------------------------------------------------- +# core.audit - exception level & JSON rendering of flattened fields +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "method_name,expected_level", + [ + ("debug", logging.DEBUG), + ("info", logging.INFO), + ("warning", logging.WARNING), + ("error", logging.ERROR), + ("critical", logging.CRITICAL), + ], +) +def test_level_is_passed_to_record(audit_records, method_name, expected_level): + """Each per-level method emits a record carrying that level.""" + getattr(test_audit_logger, method_name)("event", request=None) + + record = audit_records[0] + assert record.levelno == expected_level + assert record.levelname == method_name.upper() + + +def test_level_is_rendered_in_json_payload(audit_records): + """The JSON formatter surfaces the record level under ``level``.""" + test_audit_logger.warning("event", request=None) + + rendered = json.loads(AuditJsonFormatter().format(audit_records[0])) + assert rendered["level"] == "WARNING" + + +def test_exception_logs_error_with_active_traceback(audit_records): + """``exception`` emits at ERROR and captures the active traceback.""" + try: + raise ValueError("boom") + except ValueError: + test_audit_logger.exception("operation.failed", request=None) + + record = audit_records[0] + assert record.levelno == logging.ERROR + # exc_info is the live traceback tuple, never part of the audit payload. + assert record.exc_info is not None + assert "exc_info" not in record.audit