mirror of
https://github.com/suitenumerique/meet.git
synced 2026-10-12 07:57:50 +00:00
🔒️(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:
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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"
|
||||
|
||||
@@ -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 RequestContext, request_context
|
||||
from .targets import describe_target
|
||||
from .targets import describe_target, user_target_of
|
||||
from .utils import prune_empty, render_value
|
||||
|
||||
AUDIT_LOGGER_NAME = "audit"
|
||||
@@ -76,6 +76,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,
|
||||
@@ -96,12 +97,14 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
|
||||
default from the context of 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.
|
||||
"""
|
||||
context = (
|
||||
request_context() if request is None else RequestContext.from_request(request)
|
||||
@@ -111,6 +114,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)
|
||||
actor_fields = describe_actor(
|
||||
context.request,
|
||||
actor=actor,
|
||||
@@ -144,7 +149,10 @@ def build_document( # noqa: PLR0913 # pylint: disable=too-many-arguments,too-m
|
||||
},
|
||||
"url": {"path": context.path},
|
||||
"user_agent": {"original": context.user_agent},
|
||||
"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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -63,3 +63,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
|
||||
|
||||
@@ -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``
|
||||
|
||||
|
||||
@@ -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") == []
|
||||
@@ -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()
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user