mirror of
https://github.com/suitenumerique/meet.git
synced 2026-10-14 09:01:21 +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:
@@ -52,6 +52,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
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# Audit logging
|
||||
|
||||
La Suite Meet keeps an **audit log**: a record of the notable things that happen in the product, such as a room created
|
||||
through the external API, an application obtaining a token, or a sign-in. Every entry answers the
|
||||
through the external API, a sign-in, or a change made in the Django admin. Every entry answers the
|
||||
same questions: *who* did *what*, *to what*, *on whose behalf*, *from where*, and *did it work?*
|
||||
|
||||
It is written for:
|
||||
@@ -89,7 +89,7 @@ find each one in the JSON above.
|
||||
| **Audit Event** | One immutable record that a notable action was attempted, by whom, on what, and with what outcome. Written as a single JSON line on the Audit Log. | The whole document, identified by `event.id` |
|
||||
| **Audit Log** | The stream of Audit Events. It is distinct from the application logs: a logger and a dataset of its own. | `data_stream.*`, `event.dataset` |
|
||||
| **Action** | The dotted, stable name of what was attempted, such as `room.create` or `application.token.issue`. It names the attempt, not its result: a refused `room.create` is still a `room.create`. Every Action belongs to the Catalogue. | `event.action` |
|
||||
| **Catalogue** | The closed list of Actions the code may emit, with their meaning and fields: see [Catalogue](#catalogue). | — |
|
||||
| **Catalogue** | The closed list of Actions the code may emit, with their meaning and fields: see [Catalogue](#catalogue). Writes made through the Django admin are its one templated family, `admin.<target>.<verb>`. | — |
|
||||
| **Details** | The fields specific to one Action, such as the scopes granted with a token or the fields a room update changed. The Catalogue lists them. | `lasuite.details.*` |
|
||||
|
||||
### Who acted
|
||||
@@ -110,6 +110,7 @@ find each one in the JSON above.
|
||||
| Term | Meaning | In the event |
|
||||
|---|---|---|
|
||||
| **Target** | The resource the Action was attempted on: a room, a recording, an access grant, an application or a user. Described by its type, its id and a few stable fields. See [Targets](#targets). | `entity.target.*` |
|
||||
| **User Target** | The person the Action was *about*, when there is one: a provisioned user, the user an access grants a role to, or the account an admin edits. Not to be confused with `user.*`, the account whose authority was used. | `user.target.*` |
|
||||
|
||||
### How it went
|
||||
|
||||
@@ -140,18 +141,18 @@ find each one in the JSON above.
|
||||
These are the fields to filter on for the questions that come up most. They work the same whatever tool you search
|
||||
your logs with.
|
||||
|
||||
**Who created or changed this room?**
|
||||
**Who created, changed or deleted this room?**
|
||||
Filter on `entity.target.sub_type: room` with the room's id in `entity.target.id`, or its slug in
|
||||
`entity.target.raw.slug`. `event.action` says what happened, and `user.id` or `lasuite.application.client_id` who did
|
||||
it. Rooms are audited when they are handled through the external API (`room.*`).
|
||||
it. Rooms are audited when they are handled through the external API (`room.*`) or the Django admin (`admin.room.*`).
|
||||
|
||||
**What has this application done?**
|
||||
Filter on `lasuite.application.client_id`. To include the attempts made before its credentials were verified, such as a
|
||||
wrong secret or an inactive application, also search `lasuite.details.claimed_client_id`.
|
||||
|
||||
**What did this person do?**
|
||||
Filter on `user.id`, the account whose authority an action used. To follow someone across the other La Suite products,
|
||||
use their OIDC `sub`, in `lasuite.actor.sub`.
|
||||
**What did this person do, and what was done to their account?**
|
||||
`user.id` is the account whose authority an action used, and `user.target.id` the account an action was about: search
|
||||
both. To follow someone across the other La Suite products, use their OIDC `sub`, in `lasuite.actor.sub`.
|
||||
|
||||
**Which requests were refused, and why?**
|
||||
Filter on `lasuite.outcome: denied` and group by `event.reason`. A burst of `authentication_failed` on `user.login` or
|
||||
@@ -187,6 +188,7 @@ what ECS does not define: `lasuite.*`, and `entity.target.raw.*` for the fields
|
||||
| `lasuite.application.client_id` | The external application acting, if any. Only set once its credentials are verified |
|
||||
| `user.id`, `user.domain` | The account whose authority the action used (see [Actors](#actors)): its primary key and email domain. The email address itself is never recorded |
|
||||
| `user.roles` | That account's privileges when the event happened: `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` | The Tenant: the application's client id when there is one, otherwise the user's email domain |
|
||||
| `entity.target.*` | The resource acted on, see [Targets](#targets) |
|
||||
| `lasuite.details` | Fields specific to the action, see the [Catalogue](#catalogue) |
|
||||
@@ -257,7 +259,7 @@ is told by `event.outcome`, `lasuite.outcome` and `event.reason`, never by the n
|
||||
| `event.action` | Emitted when | Notable fields |
|
||||
|---|---|---|
|
||||
| `application.token.issue` | An application asks for a delegated token (`POST /external-api/v1.0/application/token/`), whether it gets one or is refused. It can be refused for bad credentials, an inactive application, an invalid or unauthorized email domain, an unknown user, or a provisioning conflict | On success: `user.*` is the delegated user, `entity.target` the application, and `lasuite.details` holds `scopes`, `user_provisioned` and `expires_in`. On refusal: `event.reason`, `http.response.status_code` and `lasuite.details.requested_domain`. Until its credentials are verified, the client id the caller submitted is only recorded as `lasuite.details.claimed_client_id`: it never sets `lasuite.application` or `organization` |
|
||||
| `user.provision` | An application creates a provisional user from an email address, or fails to. It is a `failure` with reason `conflict` when a concurrent request created the user first | `entity.target` is the user, the existing one on a conflict. A provisional user has no `sub` yet: its id links this event to the winning one and to its later `user.login` |
|
||||
| `user.provision` | An application creates a provisional user from an email address, or fails to. It is a `failure` with reason `conflict` when a concurrent request created the user first | `user.target` and `entity.target` are the user, the existing one on a conflict. A provisional user has no `sub` yet: its id links this event to the winning one and to its later `user.login` |
|
||||
|
||||
### Rooms
|
||||
|
||||
@@ -270,12 +272,66 @@ Rooms are audited on the external API only.
|
||||
| `room.retrieve` | A room is read through the external API, or the attempt fails | `entity.target` is the room |
|
||||
| `room.list` | Rooms are listed through the external API, or the attempt fails | `lasuite.details.total` |
|
||||
|
||||
### Sign-in
|
||||
### Sign-in and the admin
|
||||
|
||||
| `event.action` | Emitted when | Notable fields |
|
||||
|---|---|---|
|
||||
| `user.login` | A user signs in, or a sign-in attempt fails: `denied`, with reason `authentication_failed` | `lasuite.auth.method` is `oidc`, `password` or `unknown`. On success, it is named after the login backend, also given in `lasuite.details.auth_backend`. On failure, it is named after the credentials submitted: a password, or the nonce of the OIDC callback |
|
||||
| `user.logout` | A user signs out | |
|
||||
| `admin.access` | A signed-in account without staff access reaches an admin page. Always denied, once per refused page | `event.category` is `web` and `event.type` is `access`. `event.reason`, and `http.response.status_code`: the redirect to the login page |
|
||||
| `admin.<target>.<verb>` | A write is made through the Django admin, see [Django admin](#django-admin) | |
|
||||
|
||||
## Django admin
|
||||
|
||||
The admin is the most sensitive part of the product, so every write made through it produces an audit event, alongside
|
||||
the `LogEntry` Django already writes. Nothing is replaced and there is no extra table: the admin history keeps working.
|
||||
|
||||
The action name follows a template, `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`, and 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` |
|
||||
|
||||
This gives names such as `admin.room.update`, `admin.user.delete` or `admin.recording.action`. `event.category` is `iam`
|
||||
for anything that grants access to the product, and `configuration` otherwise. Writes on a user or a group put `user` or
|
||||
`group` first in `event.type`, as in `["user", "change"]`.
|
||||
|
||||
### What changed
|
||||
|
||||
- `lasuite.details.changed_fields` always lists the **names** of the fields a form changed, exactly the ones Django
|
||||
reports in its own history.
|
||||
- `lasuite.details.changes` gives their **values**, as `{"from": ..., "to": ...}`, but only for the fields a model
|
||||
explicitly allows in the `admin_values` it is registered with (see
|
||||
[Registering models](#registering-models-and-authentication-methods)).
|
||||
- Anything that looks like a secret is refused, whatever the allow-list says. A password change is reported as a change
|
||||
to `password`, never with its value.
|
||||
- A `JSONField` on the allow-list, such as a room's `configuration`, is recorded as JSON rather than as a string, and
|
||||
both versions are kept whole.
|
||||
|
||||
### Inlines
|
||||
|
||||
Objects edited through an **inline** produce their own event, linked 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 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.
|
||||
- **Custom actions that bypass the ORM hooks.** An action calling `queryset.update()` or `queryset.delete()` directly
|
||||
is reported as a single `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` per object and no `action` event.
|
||||
|
||||
### How it is wired
|
||||
|
||||
The wiring lives in `core/audit/admin.py`. `AuditedAdminSite` adds auditing to every admin class when it is registered,
|
||||
including those Django declares 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
|
||||
[Registering models](#registering-models-and-authentication-methods)) only adds its category and its allowed values.
|
||||
|
||||
## For developers: emitting events
|
||||
|
||||
@@ -352,7 +408,8 @@ and the actor is the system.
|
||||
as. `actor=None` records no account: for a caller failing to authenticate, an application acting before it is given
|
||||
a delegated user, or the system.
|
||||
- **The target** is described by its model name, its primary key and the `fields` its model is registered with, see
|
||||
[Targets](#targets). A model that is not registered is still identified. Name a service the backend calls with
|
||||
[Targets](#targets). A model that is not registered is still identified. The account the action is about,
|
||||
`user.target`, is found from the target, unless given as `user_target=`. Name a service the backend calls with
|
||||
`target_service=`. Any other keyword argument lands under `lasuite.details`.
|
||||
|
||||
### Emission never raises
|
||||
@@ -380,7 +437,10 @@ imports the `auditing` module of every installed app.
|
||||
**`audit.register(Model, ...)`** tells the audit log how to handle a model:
|
||||
|
||||
- `fields`: the fields describing it as a target;
|
||||
- `entity_type`: its ECS entity type, when one of the allowed values fits (`application`, `user`...).
|
||||
- `admin_values`: the fields whose before and after values may be recorded for admin writes;
|
||||
- `category`: the ECS category of its admin writes;
|
||||
- `entity_type`: its ECS entity type, when one of the allowed values fits (`application`, `user`...);
|
||||
- `user_target`: the attribute holding the account an event on it is about, such as `user` for an access.
|
||||
|
||||
A proxy model falls back to its concrete model. Registering a model twice raises `AlreadyRegistered`, and an entity type
|
||||
ECS does not allow raises `ValueError`.
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
|
||||
@@ -1133,6 +1133,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)
|
||||
|
||||
@@ -335,6 +335,7 @@ class Base(Configuration):
|
||||
"core.audit.apps.AuditConfig",
|
||||
"demo",
|
||||
"drf_spectacular",
|
||||
"core.audit.apps.AuditedAdminConfig",
|
||||
# Third party apps
|
||||
"corsheaders",
|
||||
"dockerflow.django",
|
||||
@@ -342,7 +343,6 @@ class Base(Configuration):
|
||||
"parler",
|
||||
"easy_thumbnails",
|
||||
# Django
|
||||
"django.contrib.admin",
|
||||
"django.contrib.auth",
|
||||
"django.contrib.contenttypes",
|
||||
"django.contrib.postgres",
|
||||
|
||||
Reference in New Issue
Block a user