🔒️(backend) audit writes made through the Django admin

Every ModelAdmin now emits an ECS audit event when an object is created,
changed or deleted, and when a bulk action runs. Actions are templated
as admin.<target>.<verb>, after the model name. A signed-in account
without staff access reaching the admin is recorded as a denied
admin.access.

The wiring is a custom AdminSite installed through AdminConfig.default_site:
it mixes the auditing into every admin class at registration, including the
ones Django declares, so a new ModelAdmin is covered without doing anything.
It hangs off log_addition, log_change, log_deletions and get_actions, which
Django calls in every path, rather than off save_model. Deletions noted by
log_deletions are emitted from delete_model and delete_queryset, once their
outcome is known.

Changed field names are always reported; their values only for the
admin_values each model is registered with, and never for anything that
looks like a secret. An account acted upon is reported as user.target.
This commit is contained in:
briquet
2026-10-04 21:11:15 +02:00
parent 4ce08479ee
commit b769faeeff
10 changed files with 885 additions and 16 deletions
+1
View File
@@ -17,6 +17,7 @@ and this project adheres to
- ✨(backend) expose `allow_unregistered_rooms` in the frontend configuration
- ✨(backend) add structured audit logging facility
- ✨(backend) audit external API token and room operations
- 🔒️(backend) audit writes and bulk actions made in the Django admin
### Changed
+50 -3
View File
@@ -60,7 +60,8 @@ Standard fields follow the [Elastic Common Schema](https://www.elastic.co/guide/
| `lasuite.actor.name` | Name of a `service` actor: `roomkit`, `summary` |
| `lasuite.auth.method` | `session`, `application_jwt`, `addons_jwt`, `resource_server`, `livekit_token`, `shared_secret`, `client_credentials`, `oidc`, `password`, `none`, or `unknown` for a class that is not registered. Requests served outside DRF, as the admin and logout are, report `session` when signed in |
| `lasuite.application.client_id` | The external application acting, when there is one. Only set once its credentials are verified |
| `user.id`, `user.sub`, `user.domain` | The (delegated) human user: primary key, OIDC sub when the account has one, and email domain. The email address is never recorded |
| `user.id`, `user.sub`, `user.domain` | The account whose authority the action used, see [Actors](#actors): primary key, OIDC sub when the account has one, and email domain. The email address is never recorded |
| `user.target.id`, `user.target.sub`, `user.target.domain` | The account an IAM action was performed on, when the target is a user. `user.*` stays the actor |
| `organization.id` | Tenant: the application client id when present, else the user's email domain |
| `lasuite.target` | The resource acted on: `type`, `id` and a few stable fields per type |
| `lasuite.details` | Action-specific fields (see catalogue) |
@@ -102,10 +103,55 @@ an account a user. An event emitted with neither a request nor an actor is the s
| `room.list` | Rooms are listed through the external API, or the attempt fails | `lasuite.details.total` |
| `user.login` | A user logs in or a login attempt fails, `denied` with reason `authentication_failed` | `lasuite.auth.method` = `oidc` or `password`, or `unknown`: named after the backend on success, `lasuite.details.auth_backend`, and after the credentials submitted on failure (a password, or the nonce of the OIDC callback) |
| `user.logout` | A user logs out | |
| `admin.access` | A signed-in account without staff access reaches an admin page (always denied), once per refused page | `event.reason`, `http.response.status_code`: the redirect to the login page |
| `admin.<target>.<verb>` | A write is made through the Django admin, see below | |
Actions are always dotted, lower-case, with the format `<target>.<verb>`, and name what was attempted: whether it
succeeded is told by `event.outcome`, `lasuite.outcome` and `event.reason`, never by the action.
## Django admin
The admin is the most sensitive surface of the product, so every write made through it emits an audit event next to
the `LogEntry` Django writes itself. Nothing is replaced and there is no extra table: the admin history keeps working.
The action is templated rather than listed: `admin.<target>.<verb>`, where `<target>` is the model name and `<verb>`
one of:
| `<verb>` | Emitted when | Notable fields |
|---|---|---|
| `create` | An object is added | `lasuite.details.changed_fields`, `changes` |
| `update` | An object is changed | `lasuite.details.changed_fields`, `changes` |
| `delete` | An object is deleted, one event per object, once the deletion has run. A deletion that raises is a `failure` with reason `internal_error`; in a bulk deletion every selected object is then reported as failed | `error.message` on failure |
| `action` | A bulk action runs | `lasuite.details.admin_action`, `count` |
So `admin.room.update`, `admin.user.delete`, `admin.recording.action`. `event.category` is `iam` for anything granting
access to the product and `configuration` otherwise. Writes on a user or a group lead `event.type` with `user` or
`group`, as in `["user", "change"]`.
`lasuite.details.changed_fields` always carries the **names** of the fields a form changed, exactly the ones Django
reports in its own history. `lasuite.details.changes` carries their **values**, as `{"from": ..., "to": ...}`, and only
for the fields a model explicitly allows in the `admin_values` it is registered with. Anything that
looks like a secret is refused there whatever the allow-list says, so a password change is reported as a change to `password` and never with its value.
A `JSONField` on the allow-list, such as a room's `configuration`, is recorded as JSON rather than stringified, and both versions are kept
whole.
Objects edited through an **inline** emit their own event, joined to the parent's by `trace.id`: granting a role on a
room produces both `admin.room.update` and `admin.resourceaccess.create`.
What is deliberately **not** covered:
- **Reads.** Opening a change list, a change form or the history page emits nothing. Django's own `LogEntry` remains
the record of who touched what.
- **A custom action bypassing the ORM hooks.** An action calling `queryset.update()` or `queryset.delete()` directly
is reported as `admin.<target>.action` with its name and the number of objects, not one event per object.
`delete_selected` is the exception: Django reports its objects through `log_deletions`, so it emits one
`admin.<target>.delete` each and no `action` event.
The wiring lives in `core/audit/admin.py`: `AuditedAdminSite` mixes the auditing into every admin class at
registration, including those declared by Django itself, and is installed through
`core.audit.apps.AuditedAdminConfig` in `INSTALLED_APPS`. A new `ModelAdmin` is therefore covered without doing
anything; registering its model (see below) only adds its category and its allowed values.
## Emitting events
Actions are declared once, in `core/auditing.py`, as `audit.Action` constants. An action may carry its ECS category
@@ -187,8 +233,9 @@ emitted.
The project describes itself to the facility in code, from `core/auditing.py`. The audit app imports the `auditing`
module of every installed app once it is ready:
- `audit.register(Model, fields=...)`: the `fields` describing a model as a target. A proxy model falls back to its
concrete model. Registering a model twice raises `AlreadyRegistered`.
- `audit.register(Model, fields=..., admin_values=..., category=...)`: the `fields` describing a model as a target,
the `admin_values` whose before and after values may be recorded in the admin, and the `category` of its admin
writes. A proxy model falls back to its concrete model. Registering a model twice raises `AlreadyRegistered`.
- `audit.register_auth_method(klass, name)`: the `lasuite.auth.method` of a DRF authentication class or of a login
backend. A DRF class inherits the name of its closest registered base, and DRF's own classes are built in. A login
backend must be registered itself, as custom backends often subclass `ModelBackend` for its permission checks
+348
View File
@@ -0,0 +1,348 @@
"""Audit the writes performed through the Django admin.
Every ``ModelAdmin`` registered on :class:`AuditedAdminSite` emits an audit
event when an object is created, changed or deleted, and when a bulk action
runs. Django's own ``LogEntry`` keeps being written exactly as before: this
stream is additive.
Actions are named ``admin.<target>.<verb>`` where ``<target>`` is the model
name, so ``admin.room.update`` or ``admin.user.delete``.
Unlike the rest of the catalogue this family is templated rather than
enumerated: it follows whatever models are registered.
Only writes are audited. Browsing a change list or a change form emits
nothing.
Which field values may be recorded, and the event category, are registered
per model; see ``core.audit.registry``.
"""
import copy
import logging
from contextlib import contextmanager
from enum import StrEnum
from functools import wraps
from typing import Any
from django.contrib.admin import ModelAdmin
from django.contrib.admin.sites import AdminSite
from django.contrib.auth import get_user_model
from django.contrib.auth.models import Group, Permission
from .actions import Action
from .emitter import log
from .enums import EventCategory, EventType, Outcome, Reason
from .registry import model_options
from .utils import render_value
ADMIN_ACCESS_ACTION = Action("admin.access", category=EventCategory.IAM)
DIFF_ATTRIBUTE = "audit_admin_diff"
PENDING_DELETIONS_ATTRIBUTE = "audit_admin_pending_deletions"
UNAUDITED_ACTIONS = frozenset({"delete_selected"})
SENSITIVE_FIELD_NAMES = frozenset(
{"api_key", "client_secret", "pin_code", "secret", "sub", "token"}
)
SENSITIVE_FIELD_MARKERS = ("password", "secret", "token")
_logger = logging.getLogger(__name__)
class AdminVerb(StrEnum):
"""What was done to an object through the admin."""
CREATE = "create"
UPDATE = "update"
DELETE = "delete"
ACTION = "action"
_VERB_TYPES: dict[AdminVerb, list[EventType]] = {
AdminVerb.CREATE: [EventType.CREATION],
AdminVerb.UPDATE: [EventType.CHANGE],
AdminVerb.DELETE: [EventType.DELETION],
AdminVerb.ACTION: [EventType.CHANGE],
}
def is_sensitive(field_name: str) -> bool:
"""Tell whether the value of a field must never be recorded."""
return field_name in SENSITIVE_FIELD_NAMES or any(
marker in field_name for marker in SENSITIVE_FIELD_MARKERS
)
def value_fields_for(model: type) -> frozenset[str]:
"""Return the fields of ``model`` whose before and after values may be recorded.
Anything that looks like a secret is dropped from the ``admin_values`` of
the model here, so a mistake in the registration cannot leak one.
"""
names = model_options(model).admin_values
return frozenset(name for name in names if not is_sensitive(name))
def category_for(model: type) -> EventCategory:
"""Return the registered category, else IAM for Django's auth models.
Anything granting access to the product is IAM, the rest configuration.
"""
if category := model_options(model).category:
return category
if model is get_user_model() or issubclass(model, (Group, Permission)):
return EventCategory.IAM
return EventCategory.CONFIGURATION
def types_for(model: type, verb: AdminVerb) -> list[EventType]:
"""Return the event types of ``verb`` on ``model``.
ECS expects ``user`` or ``group`` before the verb when one was the target.
"""
if issubclass(model, get_user_model()):
return [EventType.USER, *_VERB_TYPES[verb]]
if issubclass(model, Group):
return [EventType.GROUP, *_VERB_TYPES[verb]]
return _VERB_TYPES[verb]
def action_name(model: type, verb: AdminVerb) -> str:
"""Return the audit action for ``verb`` on ``model``."""
return f"admin.{model._meta.model_name}.{verb}" # noqa: SLF001
def form_diff(form, value_fields: frozenset[str]) -> dict[str, Any]:
"""Return the names of the fields a form changed, and the allowed values.
Field names are always reported. Values are reported for allow-listed
fields only, as ``{"from": ..., "to": ...}``.
"""
changed = sorted(form.changed_data)
changes = {
name: {
"from": render_value(form.initial.get(name)),
"to": render_value(form.cleaned_data.get(name)),
}
for name in changed
if name in value_fields
}
return {"changed_fields": changed, "changes": changes}
def related_diffs(formsets) -> list[tuple[Any, AdminVerb, dict[str, Any] | None]]:
"""Return one ``(object, verb, diff)`` triple per inline object touched.
Called after ``save_related``, so the formsets already carry what they
saved. The objects they list are the very instances their forms bound, so
the matching form, and with it the before and after values, is found by
identity.
"""
touched = []
for formset in formsets or ():
forms = {id(form.instance): form for form in formset.forms}
value_fields = value_fields_for(formset.model)
def diff_of(obj, forms=forms, value_fields=value_fields):
form = forms.get(id(obj))
return form_diff(form, value_fields) if form is not None else None
for obj in getattr(formset, "new_objects", ()):
touched.append((obj, AdminVerb.CREATE, diff_of(obj)))
for obj, _fields in getattr(formset, "changed_objects", ()):
touched.append((obj, AdminVerb.UPDATE, diff_of(obj)))
for obj in getattr(formset, "deleted_objects", ()):
# A deleted inline has no meaningful diff
touched.append((obj, AdminVerb.DELETE, None))
return touched
class AuditedModelAdminMixin:
"""Emit an audit event for every write made through this ModelAdmin."""
def construct_change_message(self, request, form, formsets, add=False):
"""Stash the structured diff for the ``log_*`` hook that follows."""
message = super().construct_change_message(request, form, formsets, add)
try:
diff = {
"own": form_diff(form, value_fields_for(self.model)),
"related": related_diffs(formsets),
}
except Exception: # pylint: disable=broad-exception-caught
_logger.exception("Admin audit diff could not be built")
diff = None
setattr(request, DIFF_ATTRIBUTE, diff)
return message
def log_addition(self, request, obj, message):
"""Record the creation, and that of any inline object saved with it."""
entry = super().log_addition(request, obj, message)
self.audit_form_write(request, AdminVerb.CREATE, obj)
return entry
def log_change(self, request, obj, message):
"""Record the change, and that of any inline object saved with it."""
entry = super().log_change(request, obj, message)
self.audit_form_write(request, AdminVerb.UPDATE, obj)
return entry
def log_deletions(self, request, queryset):
"""Note the objects about to be deleted.
Django calls this before ``delete_model`` and ``delete_queryset``, in
both the single and the bulk path. Those emit the events, once the
deletion has succeeded or failed. Copies are kept because deleting an
instance clears its primary key.
"""
targets = list(queryset)
entries = super().log_deletions(request, targets)
setattr(
request, PENDING_DELETIONS_ATTRIBUTE, [copy.copy(obj) for obj in targets]
)
return entries
def delete_model(self, request, obj):
"""Delete the object, then record one deletion."""
with self.auditing_deletions(request, lambda: [copy.copy(obj)]):
super().delete_model(request, obj)
def delete_queryset(self, request, queryset):
"""Delete the objects, then record one deletion per object."""
with self.auditing_deletions(request, lambda: list(queryset)):
super().delete_queryset(request, queryset)
@contextmanager
def auditing_deletions(self, request, default_targets):
"""Record the deletions noted by ``log_deletions`` with their outcome.
``default_targets`` lists the objects when ``log_deletions`` did not
run, as when a custom action deletes through these methods directly.
"""
targets = getattr(request, PENDING_DELETIONS_ATTRIBUTE, None)
setattr(request, PENDING_DELETIONS_ATTRIBUTE, None)
if targets is None:
targets = default_targets()
try:
yield
except Exception as error:
for obj in targets:
self.audit_write(request, AdminVerb.DELETE, obj, error=error)
raise
for obj in targets:
self.audit_write(request, AdminVerb.DELETE, obj)
def get_actions(self, request):
"""Return the available actions, each wrapped so that running it is audited."""
return {
name: (self.audited_action(func, name), name, description)
for name, (func, _name, description) in super().get_actions(request).items()
}
def audited_action(self, func, name):
"""Wrap an admin action so every run emits an event, success or not."""
if name in UNAUDITED_ACTIONS:
return func
@wraps(func)
def run(modeladmin, request, queryset):
count = queryset.count()
try:
response = func(modeladmin, request, queryset)
except Exception as error:
modeladmin.audit_action(request, name, count, error=error)
raise
modeladmin.audit_action(request, name, count)
return response
return run
def audit_form_write(self, request, verb, obj):
"""Emit the event for a form write and for the inlines saved with it."""
diff = getattr(request, DIFF_ATTRIBUTE, None) or {}
setattr(request, DIFF_ATTRIBUTE, None)
self.audit_write(request, verb, obj, diff.get("own"))
for related_obj, related_verb, related_diff in diff.get("related", ()):
self.audit_write(request, related_verb, related_obj, related_diff)
def audit_write(self, request, verb, obj, diff=None, *, error=None): # pylint: disable=too-many-arguments
"""Emit one event for a write on ``obj``, a failed one if ``error`` is set."""
model = obj.__class__
log(
action_name(model, verb),
request=request,
outcome=Outcome.SUCCESS if error is None else Outcome.FAILURE,
reason=None if error is None else Reason.INTERNAL_ERROR,
error=error,
category=category_for(model),
types=types_for(model, verb),
target=obj,
user_target=obj if isinstance(obj, get_user_model()) else None,
**(diff or {}),
)
def audit_action(self, request, name, count, error=None):
"""Emit one event for a bulk action run on ``count`` objects."""
log(
action_name(self.model, AdminVerb.ACTION),
request=request,
outcome=Outcome.SUCCESS if error is None else Outcome.FAILURE,
reason=None if error is None else Reason.INTERNAL_ERROR,
category=category_for(self.model),
types=types_for(self.model, AdminVerb.ACTION),
error=error,
admin_action=name,
count=count,
)
def audited(admin_class: type) -> type:
"""Return ``admin_class`` with the audit mixin."""
if issubclass(admin_class, AuditedModelAdminMixin):
return admin_class
return type(
f"Audited{admin_class.__name__}",
(AuditedModelAdminMixin, admin_class),
{"__module__": admin_class.__module__, "__doc__": admin_class.__doc__},
)
class AuditedAdminSite(AdminSite):
"""Admin site whose model admins all emit audit events.
Installed through ``AdminConfig.default_site`` so that admin classes
declared by Django itself, or by a third-party app, are covered as well as
the project's own.
"""
def register(self, model_or_iterable, admin_class=None, **options):
"""Register the audited flavour of the given admin class."""
super().register(
model_or_iterable, audited(admin_class or ModelAdmin), **options
)
def admin_view(self, view, cacheable=False):
"""Record when a signed-in account without staff access tries an admin view.
Django asks ``has_permission`` several times per request, the login
page included, so the refusal is recorded here instead: once per
refused view. The answer is taken before the view runs, which may log
the user out.
"""
guarded = super().admin_view(view, cacheable)
@wraps(guarded)
def inner(request, *args, **kwargs):
refused = getattr(
request.user, "is_authenticated", False
) and not self.has_permission(request)
response = guarded(request, *args, **kwargs)
if refused:
log(
ADMIN_ACCESS_ACTION,
outcome=Outcome.DENIED,
reason=Reason.PERMISSION_DENIED,
request=request,
status_code=response.status_code,
)
return response
return inner
+7
View File
@@ -1,6 +1,7 @@
"""Application configurations of the audit facility."""
from django.apps import AppConfig
from django.contrib.admin.apps import AdminConfig
from django.utils.module_loading import autodiscover_modules
from .signals import connect_auth_signals
@@ -20,3 +21,9 @@ class AuditConfig(AppConfig):
"""
connect_auth_signals()
autodiscover_modules("auditing")
class AuditedAdminConfig(AdminConfig):
"""Serve the admin from the site that audits every write."""
default_site = "core.audit.admin.AuditedAdminSite"
+9 -4
View File
@@ -8,7 +8,7 @@ from typing import Any
from django.conf import settings
from .actions import Action
from .actor import describe_actor
from .actor import describe_actor, describe_user
from .enums import ActorType, EventCategory, EventType, Outcome, Reason
from .request import current_request_id, resolve_client_ip
from .targets import describe_target
@@ -55,6 +55,7 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
category: EventCategory | str | None = None,
types: list[EventType | str] | None = None,
target: Any = None,
user_target: Any = None,
actor: Any = None,
actor_type: ActorType | str | None = None,
auth_method: str | None = None,
@@ -71,8 +72,9 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
apply unless given here, or a bare dotted name (``room.create``).
The actor, auth method and network fields are read from ``request``;
``actor``, ``actor_type``, ``auth_method`` and ``client_id`` override them.
``target`` is the resource acted on. Any other keyword argument lands under
``lasuite.details``.
``target`` is the resource acted on and ``user_target`` the account an IAM
action was performed on, reported as ``user.target``. Any other keyword
argument lands under ``lasuite.details``.
"""
outcome = Outcome(outcome)
reason = Reason(reason) if reason is not None else None
@@ -107,7 +109,10 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
"response": {"status_code": status_code},
},
"url": {"path": getattr(request, "path", None) or None},
"user": actor_fields["user"],
"user": {
**(actor_fields["user"] or {}),
"target": describe_user(user_target) if user_target else None,
},
"organization": actor_fields["organization"],
"lasuite": {
**actor_fields["lasuite"],
+26 -3
View File
@@ -3,7 +3,12 @@
The project registers them from an ``auditing`` module in one of its apps,
imported once the audit app is ready, so models stay free of audit concerns::
audit.register(Room, fields=("slug", "access_level")) # describe the target
audit.register(
Room,
fields=("slug", "access_level"), # describe the target
admin_values=("name", "access_level"), # values diffed in the admin
category=audit.EventCategory.CONFIGURATION, # ECS category of admin writes
)
audit.register_auth_method(ApplicationJWTAuthentication, "application_jwt")
A target is always identified by its model name and primary key, so a model
@@ -14,6 +19,8 @@ from dataclasses import dataclass
from django.db.models import Model
from .enums import EventCategory
class AlreadyRegistered(Exception):
"""A model or an authentication class that was registered twice."""
@@ -24,20 +31,36 @@ class ModelOptions:
"""What audit events may say about a model.
``fields`` describe the model when it is the target of an event.
``admin_values`` are the fields whose before and after values may be
recorded when they change in the Django admin. ``category`` is the ECS
category of admin writes: ``iam`` for anything granting access to the
product, ``configuration`` by default.
"""
fields: tuple[str, ...] = ()
admin_values: tuple[str, ...] = ()
category: EventCategory | None = None
_models: dict[type[Model], ModelOptions] = {}
_auth_methods: dict[str, str] = {}
def register(model: type[Model], *, fields=()) -> None:
def register(
model: type[Model],
*,
fields=(),
admin_values=(),
category: EventCategory | str | None = None,
) -> None:
"""Declare what audit events may say about ``model``."""
if model in _models:
raise AlreadyRegistered(f"{model._meta.label} is already registered") # noqa: SLF001
_models[model] = ModelOptions(fields=tuple(fields))
_models[model] = ModelOptions(
fields=tuple(fields),
admin_values=tuple(admin_values),
category=EventCategory(category) if category is not None else None,
)
def unregister(model: type[Model]) -> ModelOptions | None:
+48 -5
View File
@@ -3,6 +3,8 @@
Imported by the audit app once it is ready, see ``core.audit.apps``.
"""
from django.contrib.auth.models import Group
from lasuite.oidc_resource_server.authentication import ResourceServerAuthentication
from core import audit, models
@@ -33,12 +35,53 @@ ROOM_LIST = audit.Action("room.list")
ROOM_RETRIEVE = audit.Action("room.retrieve")
ROOM_UPDATE = audit.Action("room.update")
# Models: the ``fields`` describing them as a target.
# Models: the ``fields`` describing them as a target, the ``admin_values``
# whose before and after values may be recorded in the admin, and the
# ``category`` of their admin writes: ``iam`` for anything granting access to
# the product, ``configuration`` by default.
audit.register(models.Application, fields=("client_id", "name", "is_active", "scopes"))
audit.register(models.ResourceAccess, fields=("resource_id", "user_id", "role"))
audit.register(models.Room, fields=("slug", "name", "access_level"))
audit.register(models.Recording, fields=("room_id", "status", "mode"))
audit.register(
models.User,
category=EventCategory.IAM,
admin_values=(
"is_active",
"is_staff",
"is_superuser",
"is_device",
"groups",
"user_permissions",
),
)
audit.register(Group, category=EventCategory.IAM, admin_values=("name", "permissions"))
audit.register(
models.Application,
category=EventCategory.IAM,
fields=("client_id", "name", "is_active", "scopes"),
admin_values=("name", "is_active", "scopes"),
)
audit.register(
models.ApplicationDomain, category=EventCategory.IAM, admin_values=("domain",)
)
audit.register(
models.ResourceAccess,
category=EventCategory.IAM,
fields=("resource_id", "user_id", "role"),
admin_values=("role",),
)
audit.register(
models.RecordingAccess, category=EventCategory.IAM, admin_values=("role",)
)
audit.register(
models.Room,
fields=("slug", "name", "access_level"),
admin_values=("name", "slug", "access_level", "configuration"),
)
audit.register(
models.Recording,
fields=("room_id", "status", "mode"),
admin_values=("status", "mode"),
)
audit.register(models.File, admin_values=("title", "upload_state"))
# Authentication classes and login backends -> ``lasuite.auth.method``
+383
View File
@@ -0,0 +1,383 @@
"""Tests for the audit of writes made through the Django admin."""
import json
from unittest import mock
from django.test import override_settings
import pytest
from core import models
from core.audit.testing import find_events
from core.factories import (
FileFactory,
RecordingFactory,
RoomFactory,
UserFactory,
UserResourceAccessFactory,
)
pytestmark = pytest.mark.django_db
# Admin pages render static files: serve them without a manifest.
plain_storages = override_settings(
STORAGES={
"default": {"BACKEND": "django.core.files.storage.FileSystemStorage"},
"staticfiles": {
"BACKEND": "django.contrib.staticfiles.storage.StaticFilesStorage"
},
}
)
@pytest.fixture(name="staff_client")
def staff_client_fixture(client):
"""A client signed in as a superuser, able to reach every admin page."""
client.force_login(UserFactory(is_staff=True, is_superuser=True))
return client
def room_payload(room=None, accesses=0, **overrides):
"""Return what the room change form expects, inline management included."""
payload = {
"name": room.name if room else "Weekly sync",
"slug": room.slug if room else "weekly-sync",
"access_level": room.access_level if room else models.RoomAccessLevel.PUBLIC,
"configuration": "{}",
# ``configuration`` has a callable default, so the form renders a hidden
# ``initial-`` input. Without it the field always looks changed.
"initial-configuration": "{}",
"pin_code": (room.pin_code if room else None) or "",
"accesses-TOTAL_FORMS": str(accesses),
"accesses-INITIAL_FORMS": "0",
"accesses-MIN_NUM_FORMS": "0",
"accesses-MAX_NUM_FORMS": "1000",
}
payload.update(overrides)
return payload
def test_room_creation_is_audited(audit_events, staff_client):
"""Adding a room through the admin records a creation on the room."""
response = staff_client.post("/admin/core/room/add/", room_payload())
assert response.status_code == 302
[event] = find_events(audit_events, "admin.room.create")
assert event["event"]["category"] == ["configuration"]
assert event["event"]["type"] == ["creation"]
assert event["event"]["outcome"] == "success"
assert event["lasuite"]["target"]["type"] == "room"
assert event["lasuite"]["target"]["slug"] == "weekly-sync"
assert event["lasuite"]["details"]["changes"]["name"] == {"to": "Weekly sync"}
def test_room_change_records_field_names_and_allowed_values(audit_events, staff_client):
"""A change reports the raw field names, and the values of allowed fields."""
room = RoomFactory(access_level=models.RoomAccessLevel.PUBLIC)
response = staff_client.post(
f"/admin/core/room/{room.pk}/change/",
room_payload(room, access_level=models.RoomAccessLevel.RESTRICTED),
)
assert response.status_code == 302
[event] = find_events(audit_events, "admin.room.update")
assert event["event"]["type"] == ["change"]
assert event["lasuite"]["details"]["changed_fields"] == ["access_level"]
assert event["lasuite"]["details"]["changes"] == {
"access_level": {"from": "public", "to": "restricted"}
}
def test_room_configuration_change_records_both_versions(audit_events, staff_client):
"""The configuration is allow-listed, and kept as JSON rather than stringified."""
room = RoomFactory(configuration={"a": 1})
response = staff_client.post(
f"/admin/core/room/{room.pk}/change/",
room_payload(
room,
configuration='{"a": 2, "b": "new"}',
**{"initial-configuration": '{"a": 1}'},
),
)
assert response.status_code == 302
[event] = find_events(audit_events, "admin.room.update")
assert event["lasuite"]["details"]["changed_fields"] == ["configuration"]
assert event["lasuite"]["details"]["changes"]["configuration"] == {
"from": {"a": 1},
"to": {"a": 2, "b": "new"},
}
def test_room_deletion_is_audited(audit_events, staff_client):
"""Deleting a room from its own page records a deletion."""
room = RoomFactory()
response = staff_client.post(f"/admin/core/room/{room.pk}/delete/", {"post": "yes"})
assert response.status_code == 302
[event] = find_events(audit_events, "admin.room.delete")
assert event["event"]["type"] == ["deletion"]
assert event["lasuite"]["target"]["id"] == str(room.pk)
def test_inline_access_grant_emits_its_own_iam_event(audit_events, staff_client):
"""A role granted through the inline is an IAM event of its own."""
room = RoomFactory()
user = UserFactory()
response = staff_client.post(
f"/admin/core/room/{room.pk}/change/",
room_payload(
room,
accesses=1,
**{
"accesses-0-user": str(user.pk),
"accesses-0-role": models.RoleChoices.OWNER,
"accesses-0-id": "",
"accesses-0-resource": str(room.pk),
},
),
)
assert response.status_code == 302
[room_event] = find_events(audit_events, "admin.room.update")
[access_event] = find_events(audit_events, "admin.resourceaccess.create")
assert access_event["event"]["category"] == ["iam"]
assert access_event["event"]["type"] == ["creation"]
assert access_event["lasuite"]["target"]["role"] == "owner"
assert access_event["lasuite"]["details"]["changes"]["role"] == {"to": "owner"}
# The grant and the room change belong to the same request.
assert access_event["trace"]["id"] == room_event["trace"]["id"]
def test_user_change_targets_the_user_and_never_leaks_the_password(
audit_events, staff_client
):
"""Promoting a user is an IAM event naming the account, never its secret."""
user = UserFactory(email="promoted@example.com", is_staff=False)
response = staff_client.post(
f"/admin/core/user/{user.pk}/password/",
{
"usable_password": "true",
"password1": "sup3r-s3cret-value",
"password2": "sup3r-s3cret-value",
},
)
assert response.status_code == 302
[event] = find_events(audit_events, "admin.user.update")
assert event["event"]["category"] == ["iam"]
assert event["event"]["type"] == ["user", "change"]
assert event["user"]["target"] == {
"id": str(user.pk),
"sub": user.sub,
"domain": "example.com",
}
assert event["lasuite"]["details"]["changed_fields"] == ["password"]
assert "changes" not in event["lasuite"]["details"]
assert "sup3r-s3cret-value" not in json.dumps(event)
def test_user_permission_change_records_the_flag_values(audit_events, staff_client):
"""Staff and superuser flags are allow-listed, so their values are kept."""
user = UserFactory(is_staff=False, is_superuser=False)
response = staff_client.post(
f"/admin/core/user/{user.pk}/change/",
{
"admin_email": "",
"language": user.language,
"timezone": str(user.timezone),
"is_active": "on",
"is_staff": "on",
"files_created-TOTAL_FORMS": "0",
"files_created-INITIAL_FORMS": "0",
"files_created-MIN_NUM_FORMS": "0",
"files_created-MAX_NUM_FORMS": "1000",
},
)
assert response.status_code == 302
[event] = find_events(audit_events, "admin.user.update")
assert event["lasuite"]["details"]["changes"]["is_staff"] == {
"from": False,
"to": True,
}
assert event["user"]["target"]["id"] == str(user.pk)
def test_bulk_delete_audits_each_object_but_not_the_action(audit_events, staff_client):
"""``delete_selected`` reports its objects, and nothing about itself."""
recordings = RecordingFactory.create_batch(2)
response = staff_client.post(
"/admin/core/recording/",
{
"action": "delete_selected",
"_selected_action": [str(recording.pk) for recording in recordings],
"post": "yes",
},
)
assert response.status_code == 302
events = find_events(audit_events, "admin.recording.delete")
assert {event["lasuite"]["target"]["id"] for event in events} == {
str(recording.pk) for recording in recordings
}
assert find_events(audit_events, "admin.recording.action") == []
def test_custom_action_is_audited(audit_events, staff_client):
"""A custom admin action reports its name and how many objects it ran on."""
recordings = RecordingFactory.create_batch(2)
response = staff_client.post(
"/admin/core/recording/",
{
"action": "mark_as_failed_to_stop",
"_selected_action": [str(recording.pk) for recording in recordings],
},
)
assert response.status_code == 302
[event] = find_events(audit_events, "admin.recording.action")
assert event["event"]["outcome"] == "success"
assert event["lasuite"]["details"] == {
"admin_action": "mark_as_failed_to_stop",
"count": 2,
}
def test_hard_deleted_file_is_audited_once(audit_events, staff_client):
"""``FileAdmin`` hard deletes without going through ``Model.delete``."""
file = FileFactory()
response = staff_client.post(f"/admin/core/file/{file.pk}/delete/", {"post": "yes"})
assert response.status_code == 302
[event] = find_events(audit_events, "admin.file.delete")
assert event["lasuite"]["target"]["id"] == str(file.pk)
def test_failed_deletion_is_audited_as_a_failure(audit_events, staff_client):
"""A deletion that raises is recorded as failed, never as done."""
file = FileFactory()
with (
mock.patch("core.admin.hard_delete_file", side_effect=RuntimeError("S3 down")),
pytest.raises(RuntimeError),
):
staff_client.post(f"/admin/core/file/{file.pk}/delete/", {"post": "yes"})
[event] = find_events(audit_events, "admin.file.delete")
assert event["event"]["type"] == ["deletion"]
assert event["event"]["reason"] == "internal_error"
assert event["lasuite"]["outcome"] == "failure"
assert event["lasuite"]["target"]["id"] == str(file.pk)
assert event["error"] == {"message": "S3 down"}
assert event["log"]["level"] == "error"
def test_failed_bulk_deletion_is_audited_as_a_failure(audit_events, staff_client):
"""A bulk deletion that raises reports every selected object as failed."""
files = FileFactory.create_batch(2)
with (
mock.patch("core.admin.hard_delete_file", side_effect=RuntimeError("S3 down")),
pytest.raises(RuntimeError),
):
staff_client.post(
"/admin/core/file/",
{
"action": "delete_selected",
"_selected_action": [str(file.pk) for file in files],
"post": "yes",
},
)
events = find_events(audit_events, "admin.file.delete")
assert {event["lasuite"]["target"]["id"] for event in events} == {
str(file.pk) for file in files
}
assert {event["lasuite"]["outcome"] for event in events} == {"failure"}
@plain_storages
def test_reading_the_admin_emits_nothing(audit_events, staff_client):
"""Browsing is not audited: only writes are."""
room = RoomFactory()
UserResourceAccessFactory(resource=room, user=UserFactory())
assert staff_client.get("/admin/").status_code == 200
assert staff_client.get("/admin/core/room/").status_code == 200
assert staff_client.get(f"/admin/core/room/{room.pk}/change/").status_code == 200
assert staff_client.get(f"/admin/core/room/{room.pk}/history/").status_code == 200
assert [
event["event"]["action"]
for event in audit_events
if event["event"]["action"].startswith("admin.")
] == []
def test_non_staff_user_reaching_the_admin_is_recorded(audit_events, client):
"""A signed-in account without staff access trying the admin is a denial."""
client.force_login(UserFactory(is_staff=False))
response = client.get("/admin/core/room/")
assert response.status_code == 302
[event] = find_events(audit_events, "admin.access")
assert event["event"]["category"] == ["iam"]
assert event["event"]["reason"] == "permission_denied"
assert event["lasuite"]["outcome"] == "denied"
assert event["lasuite"]["auth"] == {"method": "session"}
assert event["http"]["response"] == {"status_code": 302}
assert event["log"]["level"] == "warning"
@plain_storages
def test_non_staff_user_is_recorded_once_per_refused_view(audit_events, client):
"""Landing on the login page after the refusal records nothing more."""
client.force_login(UserFactory(is_staff=False))
response = client.get("/admin/", follow=True)
assert response.redirect_chain[-1][0].startswith("/admin/login/")
assert response.status_code == 200
assert len(find_events(audit_events, "admin.access")) == 1
def test_anonymous_visitor_is_not_recorded(audit_events, client):
"""An anonymous hit is a redirect to the login page, not a denial worth keeping."""
assert client.get("/admin/").status_code == 302
assert find_events(audit_events, "admin.access") == []
@@ -2,6 +2,8 @@
from types import SimpleNamespace
from django.contrib.auth.models import Group
import pytest
from core import audit
@@ -36,6 +38,14 @@ def test_register_refuses_unknown_options():
assert model_options(Resource) == ModelOptions()
def test_register_refuses_unknown_categories():
"""A category outside the ECS subset fails where it is registered."""
with pytest.raises(ValueError):
audit.register(Resource, category="nonsense")
assert model_options(Resource) == ModelOptions()
def test_model_options_falls_back_to_the_concrete_model():
"""A proxy model is described as the model it proxies."""
proxy = type("ProxyRoom", (), {"_meta": SimpleNamespace(concrete_model=Room)})
@@ -57,6 +67,7 @@ def test_override_registration_restores_the_previous_one():
def test_project_declarations_are_discovered():
"""``core.auditing`` is imported when the audit app is ready."""
assert model_options(Room).fields == ("slug", "name", "access_level")
assert model_options(Group).category == audit.EventCategory.IAM
assert auth_methods()[dotted_path(ApplicationJWTAuthentication)] == (
"application_jwt"
)
+2 -1
View File
@@ -342,7 +342,8 @@ class Base(Configuration):
"parler",
"easy_thumbnails",
# Django
"django.contrib.admin",
# The admin is served by a site that audits every write it performs.
"core.audit.apps.AuditedAdminConfig",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.postgres",