mirror of
https://github.com/suitenumerique/meet.git
synced 2026-10-12 07:57:50 +00:00
briquet
3af2f2f774
Add a core.audit package emitting one Elastic Common Schema JSON line per security-relevant action on a dedicated "audit" logger. Only lasuite.* and entity.target.raw.* extends on ECS core fields. Each event names the data stream it belongs to (data_stream.*, event.dataset) and carries a unique event.id. The audit logs reads the actor, auth method, tenant and network fields from the request it is given, else from the request being served, which AuditLogMiddleware keeps in a context variable so that a service needs no request argument. A person is identified by primary key, OIDC sub when the account has one and email domain, with its privileges as user.roles; an internal peer calling the backend by service.origin.name. The tenant is the application client id, else that domain. The client IP is DRF's throttle ident, which now honours a NUM_PROXIES setting. The middleware also replaces the inbound X-Request-ID with a fresh id, unless REQUEST_ID_TRUST_HEADER says the ingress overwrites it, and echoes it in the response and the gunicorn access log. Audit events carry it as http.request.id, next to the client's user agent. An action is defined with a dotted name with its default ECS category and types, validated at import against the types ECS expects in that category, so a call site only names what was attempted. A category or types given to audit.log win over them, and a bare dotted name is accepted too. Meet declares all of the events core/auditing.py: - audit.register lists the fields describing a model as a target, an ECS entity.target, and its ECS entity type. Every target carries its model name and primary key, so an unregistered one is still identified, and a user target its OIDC sub. - audit.register_auth_method names the lasuite.auth.method of a DRF authentication class AuditViewMixin audits every response of the CRUD actions a view maps in audit_actions, and of the extra actions naming theirs with @action(audit_action=...), from finalize_response: the outcome, reason and status come from the response. An exception DRF does not handle is recorded as an internal error before it propagates. The core.audit app also records Django's login, failed login and logout signals.
63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
"""Helpers for asserting on audit events in tests."""
|
|
|
|
import logging
|
|
from collections.abc import Iterator
|
|
from contextlib import contextmanager
|
|
from dataclasses import asdict
|
|
from typing import Any
|
|
|
|
from . import registry
|
|
from .actions import Action
|
|
from .emitter import AUDIT_LOGGER_NAME
|
|
|
|
|
|
class _CollectingHandler(logging.Handler):
|
|
"""Keep the documents attached to the records it receives."""
|
|
|
|
def __init__(self):
|
|
super().__init__(level=logging.DEBUG)
|
|
self.documents: list[dict[str, Any]] = []
|
|
|
|
def emit(self, record: logging.LogRecord) -> None:
|
|
document = getattr(record, "audit", None)
|
|
if not isinstance(document, dict):
|
|
document = {"message": record.getMessage()}
|
|
self.documents.append({**document, "log": {"level": record.levelname.lower()}})
|
|
|
|
|
|
@contextmanager
|
|
def capture_audit() -> Iterator[list[dict[str, Any]]]:
|
|
"""Collect the audit documents emitted inside the block"""
|
|
logger = logging.getLogger(AUDIT_LOGGER_NAME)
|
|
handler = _CollectingHandler()
|
|
previous_level = logger.level
|
|
logger.addHandler(handler)
|
|
logger.setLevel(logging.DEBUG)
|
|
try:
|
|
yield handler.documents
|
|
finally:
|
|
logger.removeHandler(handler)
|
|
logger.setLevel(previous_level)
|
|
|
|
|
|
def find_events(
|
|
events: list[dict[str, Any]], action: Action | str
|
|
) -> list[dict[str, Any]]:
|
|
"""Return the captured events whose ``event.action`` is ``action``."""
|
|
return [
|
|
event for event in events if event.get("event", {}).get("action") == str(action)
|
|
]
|
|
|
|
|
|
@contextmanager
|
|
def override_registration(model, **options) -> Iterator[None]:
|
|
"""Register ``model`` with ``options`` inside the block, whatever it was before."""
|
|
previous = registry.unregister(model)
|
|
registry.register(model, **options)
|
|
try:
|
|
yield
|
|
finally:
|
|
registry.unregister(model)
|
|
if previous is not None:
|
|
registry.register(model, **asdict(previous))
|