🔒️(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:
the target itself when it is a user, else the account its model
registers as user_target, as the user an access grants a role to. The
emitter derives it for every event, so user.provision names the user it
creates as well.
This commit is contained in:
briquet
2026-10-04 21:11:15 +02:00
parent 82f028623f
commit 49cda567db
16 changed files with 1114 additions and 29 deletions
+1
View File
@@ -20,6 +20,7 @@ and this project adheres to
- ♿️(frontend) make participant pagination readable and keyboard reachable #1775
- ✨(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
+57 -7
View File
@@ -73,8 +73,9 @@ hold what ECS does not define: `lasuite.*`, and `entity.target.raw.*` for the fi
| `lasuite.actor.sub` | OIDC sub of the account in `user.*`, when it has one: the identity it shares with the other La Suite products |
| `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.domain` | The (delegated) human user: primary key and email domain. The email address is never recorded |
| `user.id`, `user.domain` | The account whose authority the action used, see [Actors](#actors): primary key and email domain. The email address is never recorded |
| `user.roles` | Its privileges at the time of the event: `superuser`, `staff` |
| `user.target.id`, `user.target.domain` | The account the action was about: the target when it is a user, or the user an access grants a role to. `user.*` stays the actor |
| `organization.id` | Tenant: the application client id when present, else the user's email domain |
| `entity.target.*` | The resource acted on, see [Targets](#targets) |
| `lasuite.details` | Action-specific fields (see catalogue) |
@@ -135,17 +136,63 @@ an account a user. An event emitted with neither a request nor an actor is the s
| `event.action` | Emitted when | Notable fields |
|---|---|---|
| `application.token.issue` | An application requests a delegated token (`POST /external-api/v1.0/application/token/`), whether it obtains one or is refused: bad credentials, inactive application, invalid or unauthorized email domain, unknown user, provisioning conflict | On success: `user.*` = delegated user, `entity.target` = application, `lasuite.details.scopes`, `user_provisioned`, `expires_in`. On refusal: `event.reason`, `http.response.status_code`, `lasuite.details.requested_domain`. Until the credentials are verified, the submitted client id is only `lasuite.details.claimed_client_id`: it never sets `lasuite.application` or `organization` |
| `user.provision` | An application creates a provisional user by email, or fails to: `failure` with reason `conflict` when a concurrent request created it first | `entity.target` = user, the existing one on a conflict. A provisional user has no `sub` yet: its id joins this event to the winning one and to its later `user.login` |
| `user.provision` | An application creates a provisional user by email, or fails to: `failure` with reason `conflict` when a concurrent request created it first | `user.target` and `entity.target` = user, the existing one on a conflict. A provisional user has no `sub` yet: its id joins this event to the winning one and to its later `user.login` |
| `room.create` | A room is created through the external API, or the attempt fails | `entity.target` = room |
| `room.update` | A room is updated through the external API, or the attempt fails | `entity.target` = room, refusals included, `lasuite.details.updated_fields`, `previous_access_level` |
| `room.retrieve` | A room is read through the external API, or the attempt fails | `entity.target` = room |
| `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.category` = `web`, `event.type` = `access`, `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 `http.request.id`: granting a role
on a room produces both `admin.room.update` and `admin.resourceaccess.create`, whose `user.target` is the account
granted the role.
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
@@ -207,8 +254,9 @@ the actor is the system. It is possible to override the actor with `actor=`, `ac
for a caller failing to authenticate, an application acting before it is given a delegated user, or the system.
- **Targets are described** by their model name, primary key and the `fields` their model is registered with, see
[Targets](#targets). A model that is not registered is still identified. A peer service the backend calls is named
with `target_service=`. Extra keyword arguments land under `lasuite.details`.
[Targets](#targets). A model that is not registered is still identified. The account an action is about,
`user.target`, is found from the target unless given as `user_target=`. A peer service the backend calls is named with
`target_service=`. Extra keyword arguments land under `lasuite.details`.
- **Emission never raises.** A broken configuration or value is reported on the application logger (and Sentry) and the
business operation proceeds. A registered field that cannot be read is left out of the target, and the event is still
@@ -229,9 +277,11 @@ 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=..., entity_type=...)`: the `fields` describing a model as a target, and its ECS
`entity_type` when one of the allowed values fits (`application`, `user`...). A proxy model falls back to its
concrete model. Registering a model twice raises `AlreadyRegistered`, an entity type ECS does not allow
- `audit.register(Model, fields=..., admin_values=..., category=..., entity_type=..., user_target=...)`: the `fields`
describing a model as a target, the `admin_values` whose before and after values may be recorded in the admin, the
`category` of its admin writes, its ECS `entity_type` when one of the allowed values fits (`application`, `user`...),
and the attribute holding the account an event on it is about, as `user` for an access. A proxy model falls back to
its concrete model. Registering a model twice raises `AlreadyRegistered`, an entity type ECS does not allow
`ValueError`.
- `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
+361
View File
@@ -0,0 +1,361 @@
"""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, for instance ``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.WEB, types=(EventType.ACCESS,)
)
DIFF_ATTRIBUTE = "audit_admin_diff"
PENDING_DELETIONS_ATTRIBUTE = "audit_admin_pending_deletions"
DELETED_COPIES_ATTRIBUTE = "audit_admin_deleted_copies"
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. Deleted objects are reported through the copies ``save_formset``
took, as deleting an instance clears its primary key.
"""
touched = []
for formset in formsets or ():
forms = {id(form.instance): form for form in formset.forms}
deleted_copies = getattr(formset, DELETED_COPIES_ATTRIBUTE, {})
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((deleted_copies.get(id(obj), 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 save_formset(self, request, form, formset, change):
"""Save the inline objects, keeping copies of those about to be deleted."""
setattr(
formset,
DELETED_COPIES_ATTRIBUTE,
{id(f.instance): copy.copy(f.instance) for f in formset.deleted_forms},
)
super().save_formset(request, form, formset, change)
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,
**(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"
+17 -9
View File
@@ -12,7 +12,7 @@ from typing import Any
from django.conf import settings
from .actions import Action
from .actor import FROM_REQUEST, describe_actor
from .actor import FROM_REQUEST, describe_actor, describe_user
from .ecs import (
DEFAULT_CATEGORY,
ECS_VERSION,
@@ -22,7 +22,7 @@ from .ecs import (
)
from .enums import ActorType, EventCategory, EventType, Outcome, Reason
from .request import current_request, current_request_id, resolve_client_ip
from .targets import describe_target
from .targets import describe_target, user_target_of
from .utils import prune_empty, render_value
AUDIT_LOGGER_NAME = "audit"
@@ -78,6 +78,7 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
category: EventCategory | str | Iterable[EventCategory | str] | None = None,
types: list[EventType | str] | None = None,
target: Any = None,
user_target: Any = None,
actor: Any = FROM_REQUEST,
actor_type: ActorType | str | None = None,
auth_method: str | None = None,
@@ -98,12 +99,14 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
default the request being served.
``actor``, ``actor_type``, ``auth_method`` and ``client_id`` override them;
``actor=None`` records no account even when the request is signed in.
``target`` is the resource acted on, reported as ``entity.target``.
``target_service`` names the peer service the backend called, reported as
``service.target.name``. Any other keyword argument lands under
``lasuite.details``, unless ``None`` or an empty mapping. The value of a
detail is data and is kept whole: a ``None`` or an empty mapping inside
it, as in the ``from`` and ``to`` of a change, stays.
``target`` is the resource acted on, reported as ``entity.target``, and
``user_target`` the account the action was about, reported as
``user.target``: by default the target if it is a user, or the account
registered for its model. ``target_service`` names the peer service the
backend called, reported as ``service.target.name``. Any other keyword
argument lands under ``lasuite.details``, unless ``None`` or an empty
mapping. The value of a detail is data and is kept whole: a ``None`` or an
empty mapping inside it, as in the ``from`` and ``to`` of a change, stays.
"""
if request is None:
request = current_request()
@@ -112,6 +115,8 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
if isinstance(action, Action):
category = category or action.category
types = types or list(action.types)
if user_target is None and target is not None:
user_target = user_target_of(target)
client_ip = resolve_client_ip(request) if request is not None else None
actor_fields = describe_actor(
request,
@@ -149,7 +154,10 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
},
"url": {"path": getattr(request, "path", None) or None},
"user_agent": {"original": _user_agent(request)},
"user": actor_fields["user"],
"user": {
**(actor_fields["user"] or {}),
"target": describe_user(user_target) if user_target else None,
},
"organization": actor_fields["organization"],
"entity": {
"target": describe_target(target) if target is not None else None
+34 -5
View File
@@ -3,8 +3,14 @@
The project registers them from an ``auditing`` module in one of its apps,
imported once the audit app is ready::
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(Application, entity_type="application") # ECS ``entity.type``
audit.register(ResourceAccess, user_target="user") # the account it is about
audit.register_auth_method(ApplicationJWTAuthentication, "application_jwt")
A target is always identified by its model name and primary key, so a model
@@ -16,6 +22,7 @@ from dataclasses import dataclass
from django.db.models import Model
from .ecs import ENTITY_TYPES
from .enums import EventCategory
class AlreadyRegistered(Exception):
@@ -27,19 +34,35 @@ class ModelOptions:
"""What audit events may say about a model.
``fields`` describe the model when it is the target of an event, under
``entity.target``. ``entity_type`` is the ECS ``entity.type`` of the
model, when one of its allowed values fits.
``entity.target``. ``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. ``entity_type`` is the ECS
``entity.type`` of the model, when one of its allowed values fits.
``user_target`` names the attribute holding the account an event on the
model is about, reported as ``user.target``.
"""
fields: tuple[str, ...] = ()
admin_values: tuple[str, ...] = ()
category: EventCategory | None = None
entity_type: str | None = None
user_target: str | None = None
_models: dict[type[Model], ModelOptions] = {}
_auth_methods: dict[str, str] = {}
def register(model: type[Model], *, fields=(), entity_type: str | None = None) -> None:
def register( # noqa: PLR0913 # pylint: disable=too-many-arguments
model: type[Model],
*,
fields=(),
admin_values=(),
category: EventCategory | str | None = None,
entity_type: str | None = None,
user_target: 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
@@ -48,7 +71,13 @@ def register(model: type[Model], *, fields=(), entity_type: str | None = None) -
f"{entity_type!r} is not an ECS entity type: "
f"use one of {', '.join(sorted(ENTITY_TYPES))}"
)
_models[model] = ModelOptions(fields=tuple(fields), entity_type=entity_type)
_models[model] = ModelOptions(
fields=tuple(fields),
admin_values=tuple(admin_values),
category=EventCategory(category) if category is not None else None,
entity_type=entity_type,
user_target=user_target,
)
def unregister(model: type[Model]) -> ModelOptions | None:
+24
View File
@@ -64,3 +64,27 @@ def describe_target(obj: Any) -> dict[str, Any]:
"raw": raw,
}
)
def user_target_of(obj: Any) -> Any:
"""Return the account an event on ``obj`` is about, if any.
It is ``obj`` itself for a user, else the attribute registered as the
``user_target`` of its model, as the user an access is granted to.
"""
if _is_user(obj):
return obj
if not isinstance(obj, Model):
return None
meta = obj._meta # noqa: SLF001
attribute = model_options(meta.model).user_target
if attribute is None:
return None
try:
user = getattr(obj, attribute)
except Exception: # pylint: disable=broad-exception-caught
_logger.exception(
"Audit user target %r of %s could not be read", attribute, meta.label
)
return None
return user if _is_user(user) else None
+44 -3
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
@@ -35,14 +37,53 @@ ROOM_UPDATE = audit.Action("room.update")
# Models
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,
entity_type="application",
fields=("client_id", "name", "is_active", "scopes"),
admin_values=("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.ApplicationDomain, category=EventCategory.IAM, admin_values=("domain",)
)
audit.register(
models.ResourceAccess,
category=EventCategory.IAM,
fields=("resource_id", "user_id", "role"),
admin_values=("role",),
user_target="user",
)
audit.register(
models.RecordingAccess,
category=EventCategory.IAM,
admin_values=("role",),
user_target="user",
)
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``
+447
View File
@@ -0,0 +1,447 @@
"""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["entity"]["target"]["sub_type"] == "room"
assert event["entity"]["target"]["raw"]["slug"] == "weekly-sync"
assert event["lasuite"]["details"]["changes"]["name"] == {
"from": None,
"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_configuration_change_keeps_nulls_and_empty_objects(
audit_events, staff_client
):
"""A null or an empty object in the configuration is a value, recorded as is."""
room = RoomFactory(configuration={"a": None})
response = staff_client.post(
f"/admin/core/room/{room.pk}/change/",
room_payload(
room,
configuration='{"a": {}}',
**{"initial-configuration": '{"a": null}'},
),
)
assert response.status_code == 302
[event] = find_events(audit_events, "admin.room.update")
assert event["lasuite"]["details"]["changes"] == {
"configuration": {"from": {"a": None}, "to": {"a": {}}}
}
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["entity"]["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["entity"]["target"]["raw"]["role"] == "owner"
assert access_event["lasuite"]["details"]["changes"]["role"] == {
"from": None,
"to": "owner",
}
# The account granted the role is the target user
assert access_event["user"]["target"]["id"] == str(user.pk)
# The grant and the room change belong to the same request.
assert access_event["http"]["request"]["id"] == room_event["http"]["request"]["id"]
def test_inline_access_removal_identifies_the_removed_access(
audit_events, staff_client
):
"""An access removed through the inline is identified by its primary key."""
room = RoomFactory()
owner = UserResourceAccessFactory(resource=room, role=models.RoleChoices.OWNER)
member = UserResourceAccessFactory(resource=room, role=models.RoleChoices.MEMBER)
inline = {"accesses-INITIAL_FORMS": "2"}
for index, access in enumerate((owner, member)):
inline.update(
{
f"accesses-{index}-id": str(access.pk),
f"accesses-{index}-resource": str(room.pk),
f"accesses-{index}-user": str(access.user.pk),
f"accesses-{index}-role": access.role,
}
)
inline["accesses-1-DELETE"] = "on"
response = staff_client.post(
f"/admin/core/room/{room.pk}/change/",
room_payload(room, accesses=2, **inline),
)
assert response.status_code == 302
assert not models.ResourceAccess.objects.filter(pk=member.pk).exists()
[event] = find_events(audit_events, "admin.resourceaccess.delete")
assert event["event"]["type"] == ["deletion"]
assert event["entity"]["target"]["id"] == str(member.pk)
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), "domain": "example.com"}
assert event["entity"]["target"]["raw"] == {"sub": user.sub}
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["entity"]["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["entity"]["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["entity"]["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["entity"]["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"] == ["web"]
assert event["event"]["type"] == ["access"]
assert event["event"]["outcome"] == "failure"
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") == []
+11 -1
View File
@@ -1,9 +1,12 @@
"""Tests holding audit events to the Elastic Common Schema 9.5.0."""
from django.apps import apps
import pytest
from core import audit, auditing
from core.audit import ecs
from core.audit.admin import ADMIN_ACCESS_ACTION, AdminVerb, category_for, types_for
def declared_actions() -> list[audit.Action]:
@@ -11,7 +14,7 @@ def declared_actions() -> list[audit.Action]:
actions = [
value for value in vars(auditing).values() if isinstance(value, audit.Action)
]
return [*actions, audit.LOGIN_ACTION, audit.LOGOUT_ACTION]
return [*actions, audit.LOGIN_ACTION, audit.LOGOUT_ACTION, ADMIN_ACCESS_ACTION]
def test_check_classification_accepts_expected_types():
@@ -45,3 +48,10 @@ def test_declared_actions_are_classified_as_ecs_expects(action):
``Action`` refuses anything else when it is declared: this lists them.
"""
ecs.check_classification([action.category or ecs.DEFAULT_CATEGORY], action.types)
@pytest.mark.parametrize("verb", list(AdminVerb))
def test_admin_writes_are_classified_as_ecs_expects(verb):
"""Writes made through the admin are classified as ECS expects, for any model."""
for model in apps.get_models():
ecs.check_classification([category_for(model)], types_for(model, verb))
@@ -335,6 +335,48 @@ def test_audit_log_describes_registered_targets(audit_events):
assert "user" not in audit_events[0]
def test_audit_log_reports_a_user_target_as_user_target(audit_events):
"""A user acted on is the ``user.target``, its sub only in ``entity.target``."""
user = UserFactory(email="jane@example.org")
audit.log("user.provision", target=user)
event = audit_events[0]
assert event["user"] == {"target": {"id": str(user.pk), "domain": "example.org"}}
assert event["entity"]["target"] == {
"id": str(user.pk),
"type": ["user"],
"sub_type": "user",
"raw": {"sub": user.sub},
}
def test_audit_log_reports_the_registered_user_target(audit_events):
"""An access names the user it grants a role to as the ``user.target``."""
room = RoomFactory(users=[(UserFactory(email="jane@example.org"), "member")])
access = room.accesses.get()
audit.log("thing.grant", target=access)
event = audit_events[0]
assert event["user"] == {
"target": {"id": str(access.user_id), "domain": "example.org"}
}
assert event["entity"]["target"]["sub_type"] == "resourceaccess"
def test_audit_log_explicit_user_target_wins(audit_events):
"""A user target given to ``log`` wins over the one of the target."""
room = RoomFactory(users=[(UserFactory(), "member")])
other = UserFactory(email="other@example.net")
audit.log("thing.grant", target=room.accesses.get(), user_target=other)
assert audit_events[0]["user"]["target"]["id"] == str(other.pk)
def test_audit_log_registered_entity_type(audit_events):
"""A model registered with an ECS entity type reports it."""
application = ApplicationFactory()
+13 -1
View File
@@ -2,6 +2,8 @@
from types import SimpleNamespace
from django.contrib.auth.models import Group
import pytest
from core import audit
@@ -15,7 +17,7 @@ from core.audit.registry import (
)
from core.audit.testing import override_registration
from core.external_api.authentication import ApplicationJWTAuthentication
from core.models import Application, Resource, Room
from core.models import Application, Resource, ResourceAccess, Room
def test_register_twice_is_refused():
@@ -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_register_refuses_unknown_entity_types():
"""An entity type ECS does not allow fails where it is registered."""
with pytest.raises(ValueError, match="not an ECS entity type"):
@@ -65,7 +75,9 @@ 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 model_options(Application).entity_type == "application"
assert model_options(ResourceAccess).user_target == "user"
assert auth_methods()[dotted_path(ApplicationJWTAuthentication)] == (
"application_jwt"
)
+49 -2
View File
@@ -5,15 +5,16 @@ from django.utils.functional import SimpleLazyObject
import pytest
from core.audit.registry import ModelOptions, model_options
from core.audit.targets import describe_target
from core.audit.targets import describe_target, user_target_of
from core.audit.testing import override_registration
from core.factories import (
ApplicationFactory,
RecordingFactory,
RoomFactory,
UserFactory,
UserRecordingAccessFactory,
)
from core.models import Application, Recording, Resource, Room
from core.models import Application, Recording, Resource, ResourceAccess, Room
pytestmark = pytest.mark.django_db
@@ -148,3 +149,49 @@ def test_model_options_by_model():
with override_registration(Room, fields=("slug",)):
assert model_options(Room) == ModelOptions(fields=("slug",))
assert model_options(Resource) == ModelOptions()
def test_user_target_of_a_user_is_itself():
"""An event on an account is about that account."""
user = UserFactory()
assert user_target_of(user) == user
def test_user_target_of_reads_the_registered_attribute():
"""An event on an access is about the user it grants a role to."""
user = UserFactory()
access = RoomFactory(users=[(user, "member")]).accesses.get()
assert user_target_of(access) == user
def test_user_target_of_a_recording_access():
"""``core.auditing`` registers the user of a recording access."""
access = UserRecordingAccessFactory()
assert user_target_of(access) == access.user
def test_user_target_of_an_unregistered_model_is_none():
"""A model registered without a user target is about no account."""
assert user_target_of(RoomFactory()) is None
assert user_target_of("something") is None
def test_user_target_of_a_broken_registration_is_none(caplog):
"""An attribute that cannot be read costs the user target, not the event."""
access = RoomFactory(users=[(UserFactory(), "member")]).accesses.get()
with override_registration(ResourceAccess, user_target="no_such_attribute"):
assert user_target_of(access) is None
assert "no_such_attribute" in caplog.text
def test_user_target_of_an_attribute_that_is_not_a_user_is_none():
"""A registered attribute holding anything but an account is ignored."""
access = RoomFactory(users=[(UserFactory(), "member")]).accesses.get()
with override_registration(ResourceAccess, user_target="role"):
assert user_target_of(access) is None
@@ -61,6 +61,7 @@ def test_get_or_create_creation_is_audited(audit_events):
"type": ["user"],
"sub_type": "user",
}
assert event["user"] == {"target": {"id": str(user.pk), "domain": "example.com"}}
assert "john.doe" not in json.dumps(event)
@@ -89,6 +90,9 @@ def test_get_or_create_lost_race_is_audited_with_the_existing_user(
"type": ["user"],
"sub_type": "user",
}
assert event["user"] == {
"target": {"id": str(existing_user.pk), "domain": "example.com"}
}
assert "john.doe" not in json.dumps(event)
@@ -892,6 +892,7 @@ def test_api_applications_generate_token_provisioning_is_audited(
"type": ["user"],
"sub_type": "user",
}
assert provision["user"]["target"] == {"id": str(user.pk), "domain": "example.com"}
assert provision["http"]["request"]["id"] == response["X-Request-ID"]
assert provision["url"]["path"] == "/external-api/v1.0/application/token/"
assert "new.user@example.com" not in json.dumps(provision)
+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",