From bc93637da23781eebbb9453fd9f13acc14bbb101 Mon Sep 17 00:00:00 2001 From: briquet Date: Wed, 7 Oct 2026 00:00:43 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8(backend)=20audit=20recordings,=20tran?= =?UTF-8?q?scripts=20and=20summaries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Emit audit events across the recording lifecycle: recording.start, recording.stop and recording.delete from the API, recording.end when LiveKit reports the egress ended, recording.transcript.request when the backend sends a recording to the summary service, and recording.transcript.report and recording.summary.report when that service calls back. recording.end has an unknown outcome when LiveKit reports an egress status the backend does not map, and recording.transcript.request names the summary service as service.target. Small bugfix: AuditViewMixin keeps a copy of a deleted target, since deleting an instance clears its primary key. --- CHANGELOG.md | 1 + docs/features/audit-logging.md | 21 ++- src/backend/core/api/viewsets.py | 35 ++++- src/backend/core/audit/drf.py | 6 + src/backend/core/auditing.py | 15 +- src/backend/core/models.py | 16 ++ .../core/recording/event/notification.py | 55 +++++-- .../recording/services/recording_events.py | 2 +- src/backend/core/recording/worker/mediator.py | 2 +- src/backend/core/services/livekit_events.py | 28 +++- src/backend/core/tests/audit/test_ecs.py | 114 ++++++++++++++ .../recording/event/test_notification.py | 62 ++++++++ .../recording/test_api_recordings_delete.py | 22 +++ ...st_api_recordings_external_process_hook.py | 81 ++++++++++ .../rooms/test_api_rooms_start_recording.py | 144 +++++++++++++++++- .../rooms/test_api_rooms_stop_recording.py | 46 ++++++ .../tests/services/test_livekit_events.py | 76 ++++++++- 17 files changed, 700 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 428ee8353..f31f777c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -21,6 +21,7 @@ and this project adheres to - ✨(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 +- ✨(backend) audit recordings, transcripts and summaries ### Changed diff --git a/docs/features/audit-logging.md b/docs/features/audit-logging.md index de9daaf8f..f809d0d40 100644 --- a/docs/features/audit-logging.md +++ b/docs/features/audit-logging.md @@ -61,8 +61,8 @@ hold what ECS does not define: `lasuite.*`, and `entity.target.raw.*` for the fi | `@timestamp` | ISO 8601 with millisecond precision in UTC timezone | | `data_stream.*`, `event.dataset` | `logs`, `.audit` and `AUDIT_LOG_DATA_STREAM_NAMESPACE`: what tells the audit stream apart from the application logs | | `service.name`, `service.environment`, `service.version`, `service.node.name` | `AUDIT_LOG_SERVICE_NAME`, current environment, release and host name (the pod on Kubernetes): the emitter, never the caller | -| `service.origin.name` | The internal peer that called the backend, when the actor is a `service`: `roomkit`, `summary` | -| `service.target.name` | The peer service the backend called, when one is named | +| `service.origin.name` | The internal peer that called the backend, when the actor is a `service`: `roomkit`, `summary`, `livekit` | +| `service.target.name` | The peer service the backend called, as `summary` for `recording.transcript.request` | | `event.id` | Unique id of the event, so that a shipper retrying it cannot duplicate it | | `event.action` | What was attempted, from the catalogue below | | `event.category`, `event.type` | ECS classification (`api`, `authentication`, `iam`... / `creation`, `change`, `access`, `denied`, `user`...). Always a combination ECS expects, see [Classification](#classification) | @@ -124,7 +124,7 @@ An audited API action that raises an exception DRF does not handle is still reco |---|---|---| | `user` | A person's account acting for itself: session, OIDC or password login, add-on token, LiveKit token of a known account | That account | | `application` | A client application acting on behalf of a user: a Meet application through its client credentials or its delegated token, or another La Suite application through the resource server. `lasuite.application.client_id` names it | The delegating user | -| `service` | An internal peer of the deployment acting on its own behalf, named by `service.origin.name`: the LiveKit SIP bridge (`roomkit`), the summary service (`summary`). | Absent | +| `service` | An internal peer of the deployment, named by `service.origin.name`: the LiveKit SIP bridge (`roomkit`), the summary service (`summary`), the LiveKit server reporting on a recording (`livekit`) | Absent | | `system` | The backend itself, with no inbound request | Absent | | `anonymous` | A caller that did not authenticate, or failed to | Absent | @@ -141,11 +141,26 @@ an account a user. An event emitted with neither a request nor an actor is the s | `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` | +| `recording.start` | A room owner or administrator starts a recording, or the attempt fails: conflict with a recording in progress, worker error | `entity.target` = recording, or the room when none was created, `lasuite.details.collect_metadata` | +| `recording.stop` | A room owner or administrator stops the recording in progress, or the attempt fails | `entity.target` = recording, or the room when none is active | +| `recording.end` | LiveKit reports a recording ended (`egress_ended` webhook): `success` when its media file is available, `failure` when it was aborted or failed, `unknown` for a status the backend does not map | `service.origin.name` = `livekit`, `entity.target` = recording, as it was before the report is processed, `lasuite.details.worker_event` (`completed`, `limit reached`, `aborted`, `failed`), `error_code` | +| `recording.delete` | A recording is deleted, or the attempt fails | `entity.target` = recording | +| `recording.transcript.request` | The backend sends a recording to the summary service to be transcribed, or fails to | `lasuite.actor.type` = `system`, `service.target.name` = `summary`, `entity.target` = recording, `lasuite.details.summary_requested`: whether a summary is to be made of the transcript, `job_id` | +| `recording.transcript.report` | The summary service reports on a transcript (`external-process-hook`), or a call to the hook is refused | `service.origin.name` = `summary`, `entity.target` = recording, absent for an unknown job, `lasuite.details.job_id`, `status`. A reported `failure` is a `failure` | +| `recording.summary.report` | The summary service reports on a summary | As `recording.transcript.report` | | `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..` | A write is made through the Django admin, see below | | +### Recordings, transcripts and summaries + +A recording's target carries what was recorded: `mode` is how the media was captured (`screen_recording`, a video; +`transcript`, an audio track), `requested_mode` what the user asked for, and `is_transcribed` whether it is sent to +the summary service. They differ for a transcript started with a screen capture: `mode` is `screen_recording`, +`requested_mode` `transcript`. A transcript and a summary are not recordings but what the summary service derives +from one, audited under `recording.transcript.*` and `recording.summary.*` with the recording as their target. + Actions are always dotted, lower-case, with the format `.`, and name what was attempted: whether it succeeded is told by `event.outcome`, `lasuite.outcome` and `event.reason`, never by the action. diff --git a/src/backend/core/api/viewsets.py b/src/backend/core/api/viewsets.py index 22e6c4c07..113ee40ab 100644 --- a/src/backend/core/api/viewsets.py +++ b/src/backend/core/api/viewsets.py @@ -40,7 +40,7 @@ from rest_framework import ( ) from rest_framework.settings import api_settings -from core import analytics, enums, models, utils +from core import analytics, audit, auditing, enums, models, utils from core.api import throttling from core.api.filters import ListFileFilter from core.enums import MEDIA_STORAGE_URL_PATTERN @@ -168,6 +168,7 @@ class UserViewSet( class RoomViewSet( + audit.AuditViewMixin, mixins.CreateModelMixin, mixins.DestroyModelMixin, mixins.UpdateModelMixin, @@ -312,6 +313,7 @@ class RoomViewSet( permission_classes=[ permissions.HasPrivilegesOnRoom, ], + audit_action=auditing.RECORDING_START, ) @FeatureFlag.require("recording") def start_room_recording(self, request, pk=None): # pylint: disable=unused-argument @@ -349,6 +351,11 @@ class RoomViewSet( recording=recording, ) + self.audit_target = recording + self.audit_details = { + "collect_metadata": bool(recording.options.get("collect_metadata")), + } + except (DjangoValidationError, IntegrityError): # DjangoValidationError covers the Python-level check (full_clean); # IntegrityError covers the race where two concurrent requests both @@ -393,6 +400,7 @@ class RoomViewSet( permission_classes=[ permissions.HasPrivilegesOnRoom, ], + audit_action=auditing.RECORDING_STOP, ) @FeatureFlag.require("recording") def stop_room_recording(self, request, pk=None): # pylint: disable=unused-argument @@ -409,6 +417,8 @@ class RoomViewSet( "No active recording found for this room." ) from e + self.audit_target = recording + worker_service = get_worker_service(mode=recording.mode) worker_manager = WorkerServiceMediator(worker_service=worker_service) @@ -942,6 +952,7 @@ class ResourceAccessViewSet( class RecordingViewSet( + audit.AuditViewMixin, mixins.DestroyModelMixin, mixins.ListModelMixin, mixins.RetrieveModelMixin, @@ -956,6 +967,23 @@ class RecordingViewSet( queryset = models.Recording.objects.all() serializer_class = serializers.RecordingSerializer + audit_actions = {"destroy": auditing.RECORDING_DELETE} + # What the summary service reports on, once its event is validated + audit_process_type = None + + def get_audit_action(self): + """Audit a summary apart from the transcript it is made from.""" + if self.audit_process_type == "summary": + return auditing.RECORDING_SUMMARY_REPORT + return super().get_audit_action() + + def get_audit_fields(self, status_code, error=None): + """Report a failure of the summary service as one, though it is acknowledged.""" + fields = super().get_audit_fields(status_code, error) + if status_code < 400 and (self.audit_details or {}).get("status") == "failure": + fields["outcome"] = audit.Outcome.FAILURE + return fields + def get_queryset(self): """Restrict recordings to the user's ones.""" user = self.request.user @@ -971,6 +999,7 @@ class RecordingViewSet( url_path="external-process-hook", authentication_classes=[RecordingProcessWebhookAuthentication], serializer_class=serializers.ExternalProcessEventSerializer, + audit_action=auditing.RECORDING_TRANSCRIPT_REPORT, ) def on_external_process_event_received(self, request, pk=None): # pylint: disable=unused-argument """Handle incoming external process events for recordings.""" @@ -984,12 +1013,16 @@ class RecordingViewSet( validated_data = serializer.validated_data job_id = validated_data["job_id"] + self.audit_process_type = validated_data.get("type") + self.audit_details = {"job_id": job_id, "status": validated_data.get("status")} try: recording = models.Recording.objects.get(external_process_id=job_id) except models.Recording.DoesNotExist as e: logger.warning("No recording found for job_id %s: %s", job_id, e) return ok_response + self.audit_target = recording + if validated_data.get("type") == "transcript": if validated_data.get("status") == "success": logger.info( diff --git a/src/backend/core/audit/drf.py b/src/backend/core/audit/drf.py index 7bb887966..a69a1641e 100644 --- a/src/backend/core/audit/drf.py +++ b/src/backend/core/audit/drf.py @@ -18,6 +18,7 @@ A refusal is recorded under the action that was attempted, with its outcome and reason derived from the response status. """ +import copy import logging from collections.abc import Mapping from typing import Any @@ -94,6 +95,11 @@ class AuditViewMixin: self.audit_target = obj super().check_object_permissions(request, obj) + def perform_destroy(self, instance): + """Keep a copy of the target, since deleting an instance clears its pk.""" + self.audit_target = copy.copy(instance) + super().perform_destroy(instance) + def finalize_response(self, request, response, *args, **kwargs): """Audit the response once DRF has built it.""" response = super().finalize_response(request, response, *args, **kwargs) diff --git a/src/backend/core/auditing.py b/src/backend/core/auditing.py index 78fe6d228..ea4c5d7f3 100644 --- a/src/backend/core/auditing.py +++ b/src/backend/core/auditing.py @@ -34,6 +34,19 @@ ROOM_CREATE = audit.Action("room.create") ROOM_LIST = audit.Action("room.list") ROOM_RETRIEVE = audit.Action("room.retrieve") ROOM_UPDATE = audit.Action("room.update") +RECORDING_START = audit.Action("recording.start", types=(EventType.START,)) +RECORDING_STOP = audit.Action("recording.stop", types=(EventType.END,)) +RECORDING_END = audit.Action("recording.end", types=(EventType.END,)) +RECORDING_DELETE = audit.Action("recording.delete", types=(EventType.DELETION,)) +RECORDING_TRANSCRIPT_REQUEST = audit.Action( + "recording.transcript.request", types=(EventType.START,) +) +RECORDING_TRANSCRIPT_REPORT = audit.Action( + "recording.transcript.report", types=(EventType.END,) +) +RECORDING_SUMMARY_REPORT = audit.Action( + "recording.summary.report", types=(EventType.END,) +) # Models @@ -80,7 +93,7 @@ audit.register( ) audit.register( models.Recording, - fields=("room_id", "status", "mode"), + fields=("room_id", "status", "mode", "requested_mode", "is_transcribed"), admin_values=("status", "mode"), ) audit.register(models.File, admin_values=("title", "upload_state")) diff --git a/src/backend/core/models.py b/src/backend/core/models.py index 8fe519400..c2d9fef8e 100644 --- a/src/backend/core/models.py +++ b/src/backend/core/models.py @@ -702,6 +702,22 @@ class Recording(BaseModel): """Check if the recording is in a saved state.""" return self.status in RecordingStatusChoices.saved_statuses() + @property + def requested_mode(self) -> str: + """Return the mode the user asked for. + + A transcript started with a screen capture is stored as a screen + recording, which keeps the transcript as its original mode. + """ + return (self.options or {}).get("original_mode") or self.mode + + @property + def is_transcribed(self) -> bool: + """Tell whether the recording is sent to the summary service.""" + return self.mode == RecordingModeChoices.TRANSCRIPT or bool( + (self.options or {}).get("transcribe", False) + ) + @property def extension(self): """Get recording extension based on its mode.""" diff --git a/src/backend/core/recording/event/notification.py b/src/backend/core/recording/event/notification.py index 81624b1a3..4e7a8a04c 100644 --- a/src/backend/core/recording/event/notification.py +++ b/src/backend/core/recording/event/notification.py @@ -17,7 +17,7 @@ import requests from asgiref.sync import async_to_sync from livekit import api as livekit_api -from core import models, utils +from core import audit, auditing, models, utils from core.analytics import UserFeatureFlag, is_user_feature_flag_enabled from core.utils import generate_download_s3_url @@ -52,7 +52,7 @@ class NotificationService: if recording.mode == models.RecordingModeChoices.SCREEN_RECORDING: summary_success = True - if recording.options.get("transcribe", False): + if recording.is_transcribed: summary_success = self._notify_summary_service(recording) email_success = self._notify_user_by_email(recording) @@ -213,14 +213,32 @@ class NotificationService: @staticmethod def _notify_summary_service(recording: models.Recording): - if settings.SUMMARY_SERVICE_VERSION == 1: - return NotificationService._notify_summary_service_v1(recording) - if settings.SUMMARY_SERVICE_VERSION == 2: - return NotificationService._notify_summary_service_v2(recording) - - raise NotImplementedError( - f"Unknown summary service version: {settings.SUMMARY_SERVICE_VERSION}" - ) + succeeded = False + audit_details = {} + try: + if settings.SUMMARY_SERVICE_VERSION == 1: + succeeded = NotificationService._notify_summary_service_v1(recording) + elif settings.SUMMARY_SERVICE_VERSION == 2: + succeeded = NotificationService._notify_summary_service_v2( + recording, audit_details + ) + else: + raise NotImplementedError( + "Unknown summary service version: " + f"{settings.SUMMARY_SERVICE_VERSION}" + ) + return succeeded + finally: + audit.log( + auditing.RECORDING_TRANSCRIPT_REQUEST, + actor=None, + actor_type=audit.ActorType.SYSTEM, + target=recording, + target_service="summary", + outcome=audit.Outcome.SUCCESS if succeeded else audit.Outcome.FAILURE, + job_id=recording.external_process_id, + **audit_details, + ) @staticmethod def _notify_summary_service_v1(recording: models.Recording): @@ -298,8 +316,11 @@ class NotificationService: return True @staticmethod - def _notify_summary_service_v2(recording: models.Recording): - """Notify summary service about a new recording.""" + def _notify_summary_service_v2(recording: models.Recording, audit_details=None): + """Notify summary service about a new recording. + + Whether a summary is asked for is added to ``audit_details``. + """ if ( not settings.SUMMARY_SERVICE_ENDPOINT @@ -349,6 +370,12 @@ class NotificationService: "ended_at": ended_at.isoformat(), } + summary_requested = is_user_feature_flag_enabled( + owner_access.user, UserFeatureFlag.TRANSCRIPT_SUMMARY_ENABLED + ) + if audit_details is not None: + audit_details["summary_requested"] = summary_requested + payload = { "user_sub": owner_access.user.sub, "user_email": owner_access.user.email, @@ -372,9 +399,7 @@ class NotificationService: ), "download_link": f"{get_recording_download_base_url()}/{recording.id}", "form_link": form_link, - "auto_create_summary": is_user_feature_flag_enabled( - owner_access.user, UserFeatureFlag.TRANSCRIPT_SUMMARY_ENABLED - ), + "auto_create_summary": summary_requested, }, "metadata": metadata_payload, } diff --git a/src/backend/core/recording/services/recording_events.py b/src/backend/core/recording/services/recording_events.py index 8093a02a6..cf95084e6 100644 --- a/src/backend/core/recording/services/recording_events.py +++ b/src/backend/core/recording/services/recording_events.py @@ -84,7 +84,7 @@ class RecordingEventsService: @staticmethod def _notify_participants(recording: Recording, event: RecordingWorkerEvent): """Notify the room's participants that a recording ended on the given event.""" - recording_mode = recording.options.get("original_mode", None) or recording.mode + recording_mode = recording.requested_mode notification_type = get_notification_type(recording_mode, event) if not notification_type: diff --git a/src/backend/core/recording/worker/mediator.py b/src/backend/core/recording/worker/mediator.py index 312fcb3b0..8493561e3 100644 --- a/src/backend/core/recording/worker/mediator.py +++ b/src/backend/core/recording/worker/mediator.py @@ -68,7 +68,7 @@ class WorkerServiceMediator: finally: recording.save() - mode = recording.options.get("original_mode", None) or recording.mode + mode = recording.requested_mode try: RoomManagement.update_metadata( diff --git a/src/backend/core/services/livekit_events.py b/src/backend/core/services/livekit_events.py index 167b7b4b1..ccd47d558 100644 --- a/src/backend/core/services/livekit_events.py +++ b/src/backend/core/services/livekit_events.py @@ -12,8 +12,9 @@ from django.utils import timezone from livekit import api -from core import models -from core.recording.enums import RecordingWorkerEvent +from core import audit, auditing, models +from core.recording.enums import UNSUCCESSFUL_EVENTS, RecordingWorkerEvent +from core.recording.event.authentication import MachineUser from core.recording.services.metadata_collector import ( MetadataCollectorException, MetadataCollectorService, @@ -219,6 +220,7 @@ class LiveKitEventsService: ) from err event = to_recording_event(data.egress_info.status) + self._audit_recording_end(recording, event, data.egress_info.error_code) # Log if/why the recording failed self.recording_events.log_worker_error( @@ -254,6 +256,28 @@ class LiveKitEventsService: self.recording_events.handle_terminal_event(recording, event) + @staticmethod + def _audit_recording_end(recording, event, error_code=None): + """Audit the end of a recording, as LiveKit reports it. + + An egress status the backend does not map has an unknown outcome. + """ + if event is None: + outcome = audit.Outcome.UNKNOWN + elif event in UNSUCCESSFUL_EVENTS: + outcome = audit.Outcome.FAILURE + else: + outcome = audit.Outcome.SUCCESS + audit.log( + auditing.RECORDING_END, + actor=MachineUser("livekit"), + auth_method="shared_secret", + target=recording, + outcome=outcome, + worker_event=event.value if event is not None else None, + error_code=error_code or None, + ) + @staticmethod def _is_connection_test_room(room_name: str) -> bool: """Return True for ephemeral rooms created by the connection test endpoint.""" diff --git a/src/backend/core/tests/audit/test_ecs.py b/src/backend/core/tests/audit/test_ecs.py index 813e5d6a2..b749307c4 100644 --- a/src/backend/core/tests/audit/test_ecs.py +++ b/src/backend/core/tests/audit/test_ecs.py @@ -1,12 +1,81 @@ """Tests holding audit events to the Elastic Common Schema 9.5.0.""" +from collections.abc import Iterator, Mapping +from typing import Any + from django.apps import apps +from django.test import RequestFactory 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 +from core.factories import ApplicationFactory, RoomFactory, UserFactory +from core.recording.event.authentication import MachineUser + +pytestmark = pytest.mark.django_db + +# The ECS 9.5.0 fields audit events may carry, from +# https://github.com/elastic/ecs/blob/v9.5.0/generated/ecs/ecs_flat.yml +ECS_FIELDS = frozenset( + { + "@timestamp", + "client.ip", + "data_stream.dataset", + "data_stream.namespace", + "data_stream.type", + "ecs.version", + "entity.target.id", + "entity.target.name", + "entity.target.sub_type", + "entity.target.type", + "error.message", + "error.stack_trace", + "error.type", + "event.action", + "event.category", + "event.dataset", + "event.id", + "event.kind", + "event.outcome", + "event.reason", + "event.type", + "http.request.id", + "http.request.method", + "http.response.status_code", + "log.level", + "log.logger", + "message", + "organization.id", + "service.environment", + "service.name", + "service.node.name", + "service.origin.name", + "service.target.name", + "service.version", + "source.ip", + "url.path", + "user.domain", + "user.id", + "user.roles", + "user.target.domain", + "user.target.id", + "user_agent.original", + } +) +# What is not ECS lives in these namespaces: ``raw`` holds a target's own fields +CUSTOM_NAMESPACES = ("lasuite.", "entity.target.raw.") + + +def leaf_paths(document: Mapping[str, Any], prefix: str = "") -> Iterator[str]: + """Yield the dotted path of every value of a document.""" + for key, value in document.items(): + path = f"{prefix}{key}" + if isinstance(value, Mapping): + yield from leaf_paths(value, f"{path}.") + else: + yield path def declared_actions() -> list[audit.Action]: @@ -55,3 +124,48 @@ 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)) + + +def _documents(audit_events) -> list[dict[str, Any]]: + """Emit the events covering every field the facility fills.""" + user = UserFactory(is_staff=True) + request = RequestFactory().post( + "/external-api/v1.0/rooms/", + REMOTE_ADDR="1.2.3.4", + HTTP_USER_AGENT="Mozilla/5.0", + ) + request.user = user + request.auth = {"client_id": "app-1"} + + audit.log( + auditing.ROOM_UPDATE, + request=request, + target=RoomFactory(), + status_code=500, + outcome=audit.Outcome.FAILURE, + reason=audit.Reason.INTERNAL_ERROR, + error="boom", + error_type="builtins.RuntimeError", + message="anything", + updated_fields=["name"], + ) + audit.log(auditing.USER_PROVISION, target=user) + audit.log(auditing.APPLICATION_TOKEN_ISSUE, target=ApplicationFactory()) + audit.log( + auditing.RECORDING_TRANSCRIPT_REQUEST, + actor=MachineUser("livekit"), + target_service="summary", + ) + return audit_events + + +def test_documents_only_carry_ecs_fields_or_custom_namespaces(audit_events): + """No field outside ECS 9.5.0 lands anywhere but in a custom namespace.""" + for document in _documents(audit_events): + stray = { + path + for path in leaf_paths(document) + if path not in ECS_FIELDS and not path.startswith(CUSTOM_NAMESPACES) + } + + assert not stray, f"{document['event']['action']}: {sorted(stray)}" diff --git a/src/backend/core/tests/recording/event/test_notification.py b/src/backend/core/tests/recording/event/test_notification.py index f40bb8404..08e5ba004 100644 --- a/src/backend/core/tests/recording/event/test_notification.py +++ b/src/backend/core/tests/recording/event/test_notification.py @@ -15,6 +15,7 @@ import pytest from core import factories, models from core.analytics import UserFeatureFlag +from core.audit.testing import find_events from core.recording.event.notification import NotificationService, notification_service pytestmark = pytest.mark.django_db @@ -479,3 +480,64 @@ def test_notify_summary_service_v2_payload_json_serializable_without_timestamps( assert isinstance(title, str) # ...so the payload serializes exactly the way ``requests`` serializes it. json.dumps(payload) + + +@pytest.mark.parametrize("summary_requested", [True, False]) +@mock.patch("core.recording.event.notification.requests.post") +@mock.patch("core.recording.event.notification.generate_download_s3_url") +@mock.patch.object( + NotificationService, "_get_recording_timestamps", new_callable=mock.AsyncMock +) +def test_notify_summary_service_is_audited( # noqa: PLR0913, PLR0917 + mock_get_recording_timestamps, + mock_generate_download_s3_url, + mock_post, + summary_requested, + settings, + audit_events, +): # pylint: disable=too-many-arguments,too-many-positional-arguments + """Sending a recording to be transcribed says whether a summary is asked for.""" + settings.SUMMARY_SERVICE_VERSION = 2 + settings.SUMMARY_SERVICE_ENDPOINT = "https://summary.test/api/v2/tasks" + settings.SUMMARY_SERVICE_API_TOKEN = "summary-token" + settings.METADATA_COLLECTOR_ENABLED = False + + recording = factories.RecordingFactory(mode="transcript") + factories.UserRecordingAccessFactory( + recording=recording, role=models.RoleChoices.OWNER + ) + mock_get_recording_timestamps.return_value = (None, None) + mock_generate_download_s3_url.return_value = "https://storage.test/recording.ogg" + mock_post.return_value.json.return_value = {"job_id": "job-7"} + + with mock.patch( + "core.recording.event.notification.is_user_feature_flag_enabled", + return_value=summary_requested, + ): + assert NotificationService._notify_summary_service(recording) is True + + [event] = find_events(audit_events, "recording.transcript.request") + + assert event["lasuite"]["outcome"] == "success" + assert event["lasuite"]["actor"] == {"type": "system"} + assert event["entity"]["target"]["id"] == str(recording.id) + assert event["service"]["target"] == {"name": "summary"} + assert event["lasuite"]["details"] == { + "job_id": "job-7", + "summary_requested": summary_requested, + } + + +def test_notify_summary_service_failure_is_audited(settings, audit_events): + """A recording the summary service never received is a failure.""" + settings.SUMMARY_SERVICE_VERSION = 2 + settings.SUMMARY_SERVICE_ENDPOINT = None + + recording = factories.RecordingFactory(mode="transcript") + + assert NotificationService._notify_summary_service(recording) is False + + [event] = find_events(audit_events, "recording.transcript.request") + + assert event["lasuite"]["outcome"] == "failure" + assert event["entity"]["target"]["id"] == str(recording.id) diff --git a/src/backend/core/tests/recording/test_api_recordings_delete.py b/src/backend/core/tests/recording/test_api_recordings_delete.py index 5cf9e1fb3..d75285567 100644 --- a/src/backend/core/tests/recording/test_api_recordings_delete.py +++ b/src/backend/core/tests/recording/test_api_recordings_delete.py @@ -5,6 +5,7 @@ Test recordings API endpoints in the Meet core app: delete. import pytest from rest_framework.test import APIClient +from ...audit.testing import capture_audit, find_events from ...factories import RecordingFactory, UserFactory, UserRecordingAccessFactory from ...models import Recording @@ -112,3 +113,24 @@ def test_api_recordings_delete_final(role): assert response.status_code == 204 assert Recording.objects.count() == 0 + + +def test_api_recordings_delete_is_audited(): + """A deleted recording is still identified by the event.""" + user = UserFactory() + recording = RecordingFactory(status="saved", mode="transcript") + UserRecordingAccessFactory(role="owner", user=user, recording=recording) + client = APIClient() + client.force_login(user) + + with capture_audit() as events: + response = client.delete(f"/api/v1.0/recordings/{recording.id}/") + + assert response.status_code == 204 + + [event] = find_events(events, "recording.delete") + + assert event["event"]["type"] == ["deletion"] + assert event["lasuite"]["outcome"] == "success" + assert event["entity"]["target"]["id"] == str(recording.id) + assert event["entity"]["target"]["raw"]["mode"] == "transcript" diff --git a/src/backend/core/tests/recording/test_api_recordings_external_process_hook.py b/src/backend/core/tests/recording/test_api_recordings_external_process_hook.py index 71aae54de..7342d2a60 100644 --- a/src/backend/core/tests/recording/test_api_recordings_external_process_hook.py +++ b/src/backend/core/tests/recording/test_api_recordings_external_process_hook.py @@ -6,6 +6,7 @@ Test recordings API endpoints: external process hook. import pytest +from ...audit.testing import find_events from ...factories import RecordingFactory from ...models import RecordingStatusChoices @@ -132,3 +133,83 @@ def test_external_process_event_non_transcript_event_does_not_change_status( recording.refresh_from_db() assert recording.status == RecordingStatusChoices.SAVED + + +@pytest.mark.parametrize( + "process_type,action", + [ + ("transcript", "recording.transcript.report"), + ("summary", "recording.summary.report"), + ], +) +@pytest.mark.parametrize( + "process_status,outcome", [("success", "success"), ("failure", "failure")] +) +def test_external_process_event_is_audited( # noqa: PLR0913, PLR0917 + external_process_settings, + client, + audit_events, + process_type, + action, + process_status, + outcome, +): # pylint: disable=too-many-arguments,too-many-positional-arguments + """A transcript and a summary are audited apart, as reported by the service.""" + recording = RecordingFactory( + status=RecordingStatusChoices.SAVED, external_process_id="job-123" + ) + + response = client.post( + "/api/v1.0/recordings/external-process-hook/", + {"job_id": "job-123", "type": process_type, "status": process_status}, + HTTP_AUTHORIZATION="Bearer testWebhookToken", + ) + + assert response.status_code == 200 + + [event] = find_events(audit_events, action) + + assert event["lasuite"]["outcome"] == outcome + assert event["lasuite"]["actor"] == {"type": "service"} + assert event["service"]["origin"] == {"name": "summary"} + assert event["entity"]["target"]["id"] == str(recording.id) + assert event["lasuite"]["details"] == { + "job_id": "job-123", + "status": process_status, + } + + +def test_external_process_event_unknown_recording_is_audited( + external_process_settings, client, audit_events +): + """An event about an unknown job keeps its job id, with no target.""" + response = client.post( + "/api/v1.0/recordings/external-process-hook/", + {"job_id": "job-unknown", "type": "summary", "status": "success"}, + HTTP_AUTHORIZATION="Bearer testWebhookToken", + ) + + assert response.status_code == 200 + + [event] = find_events(audit_events, "recording.summary.report") + + assert "target" not in event["lasuite"] + assert event["lasuite"]["details"]["job_id"] == "job-unknown" + + +def test_external_process_event_refused_is_audited( + external_process_settings, client, audit_events +): + """A caller without the token is denied under the transcript action.""" + response = client.post( + "/api/v1.0/recordings/external-process-hook/", + {"job_id": "job-1", "type": "summary", "status": "success"}, + HTTP_AUTHORIZATION="Bearer wrongToken", + ) + + assert response.status_code == 401 + + [event] = find_events(audit_events, "recording.transcript.report") + + assert event["lasuite"]["outcome"] == "denied" + assert not find_events(audit_events, "recording.summary.report") diff --git a/src/backend/core/tests/rooms/test_api_rooms_start_recording.py b/src/backend/core/tests/rooms/test_api_rooms_start_recording.py index d50354a75..7923a5476 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_start_recording.py +++ b/src/backend/core/tests/rooms/test_api_rooms_start_recording.py @@ -2,7 +2,7 @@ Test rooms API endpoints in the Meet core app: start recording. """ -# pylint: disable=redefined-outer-name,unused-argument,no-member +# pylint: disable=redefined-outer-name,unused-argument,no-member,too-many-lines from unittest import mock @@ -10,6 +10,7 @@ import pytest from livekit import api as livekit_api from rest_framework.test import APIClient +from ...audit.testing import capture_audit, find_events from ...factories import RoomFactory, UserFactory from ...models import Recording from ...recording.worker.exceptions import RecordingStartError @@ -879,3 +880,144 @@ def test_start_recording_options_original_mode_invalid(settings, value): ) assert response.status_code == 400 + + +def test_start_recording_is_audited( + settings, mock_worker_service_factory, mock_worker_manager, audit_events +): + """A started screen recording names the recording and its mode.""" + settings.RECORDING_ENABLE = True + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + client = APIClient() + client.force_login(user) + + response = client.post( + f"/api/v1.0/rooms/{room.id}/start-recording/", + {"mode": "screen_recording"}, + ) + + assert response.status_code == 201 + + recording = Recording.objects.get(room=room) + [event] = find_events(audit_events, "recording.start") + + assert event["event"]["type"] == ["start"] + assert event["lasuite"]["outcome"] == "success" + assert event["user"]["id"] == str(user.pk) + assert event["entity"]["target"] == { + "id": str(recording.pk), + "sub_type": "recording", + "raw": { + "room_id": str(room.pk), + "status": "initiated", + "mode": "screen_recording", + "requested_mode": "screen_recording", + "is_transcribed": False, + }, + } + assert event["lasuite"]["details"] == {"collect_metadata": False} + + +def test_start_recording_transcript_with_screen_capture_is_audited( + settings, mock_worker_service_factory, mock_worker_manager, audit_events +): + """A transcript recorded with the screen is told apart from a screen recording.""" + settings.RECORDING_ENABLE = True + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + client = APIClient() + client.force_login(user) + + response = client.post( + f"/api/v1.0/rooms/{room.id}/start-recording/", + { + "mode": "screen_recording", + "options": {"transcribe": True, "original_mode": "transcript"}, + }, + format="json", + ) + + assert response.status_code == 201 + + [event] = find_events(audit_events, "recording.start") + target = event["entity"]["target"]["raw"] + + assert target["mode"] == "screen_recording" + assert target["requested_mode"] == "transcript" + assert target["is_transcribed"] is True + + +def test_start_recording_conflict_is_audited_on_the_room( + settings, mock_worker_service_factory, mock_worker_manager, audit_events +): + """A conflicting start is a failure aimed at the room, no recording existing.""" + settings.RECORDING_ENABLE = True + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + Recording.objects.create(room=room, mode="screen_recording", status="active") + client = APIClient() + client.force_login(user) + + response = client.post( + f"/api/v1.0/rooms/{room.id}/start-recording/", + {"mode": "transcript"}, + ) + + assert response.status_code == 409 + + [event] = find_events(audit_events, "recording.start") + + assert event["lasuite"]["outcome"] == "failure" + assert event["event"]["reason"] == "conflict" + assert event["entity"]["target"]["sub_type"] == "room" + assert event["entity"]["target"]["id"] == str(room.pk) + + +def test_start_recording_worker_error_is_audited( + settings, mock_worker_service_factory, mock_worker_manager, audit_events +): + """A recording the worker could not start is a failure on that recording.""" + settings.RECORDING_ENABLE = True + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + mock_worker_manager.start = mock.Mock(side_effect=RecordingStartError("boom")) + client = APIClient() + client.force_login(user) + + response = client.post( + f"/api/v1.0/rooms/{room.id}/start-recording/", + {"mode": "transcript"}, + ) + + assert response.status_code == 502 + + [event] = find_events(audit_events, "recording.start") + + assert event["lasuite"]["outcome"] == "failure" + assert event["http"]["response"]["status_code"] == 502 + assert event["entity"]["target"]["id"] == str(Recording.objects.get().pk) + assert event["entity"]["target"]["raw"]["mode"] == "transcript" + assert event["entity"]["target"]["raw"]["is_transcribed"] is True + + +def test_start_recording_anonymous_is_audited(): + """An anonymous attempt is denied.""" + room = RoomFactory() + + with capture_audit() as events: + response = APIClient().post( + f"/api/v1.0/rooms/{room.id}/start-recording/", + {"mode": "screen_recording"}, + ) + + assert response.status_code == 401 + + [event] = find_events(events, "recording.start") + + assert event["lasuite"]["outcome"] == "denied" + assert event["lasuite"]["actor"]["type"] == "anonymous" diff --git a/src/backend/core/tests/rooms/test_api_rooms_stop_recording.py b/src/backend/core/tests/rooms/test_api_rooms_stop_recording.py index b003e49e8..a2ee4aa19 100644 --- a/src/backend/core/tests/rooms/test_api_rooms_stop_recording.py +++ b/src/backend/core/tests/rooms/test_api_rooms_stop_recording.py @@ -9,6 +9,7 @@ from unittest import mock import pytest from rest_framework.test import APIClient +from ...audit.testing import find_events from ...factories import RecordingFactory, RoomFactory, UserFactory from ...models import Recording, RecordingStatusChoices from ...recording.worker.exceptions import RecordingStopError @@ -181,3 +182,48 @@ def test_stop_recording_success( # Verify the recording still exists assert Recording.objects.count() == 1 + + +def test_stop_recording_is_audited( + settings, mock_worker_service_factory, mock_worker_manager, audit_events +): + """A stopped recording is the target of the event.""" + settings.RECORDING_ENABLE = True + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + recording = RecordingFactory( + room=room, mode="transcript", status=RecordingStatusChoices.ACTIVE + ) + client = APIClient() + client.force_login(user) + + response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/") + + assert response.status_code == 200 + + [event] = find_events(audit_events, "recording.stop") + + assert event["event"]["type"] == ["end"] + assert event["lasuite"]["outcome"] == "success" + assert event["entity"]["target"]["id"] == str(recording.pk) + assert event["entity"]["target"]["raw"]["mode"] == "transcript" + + +def test_stop_recording_without_active_recording_is_audited(settings, audit_events): + """Stopping a room that records nothing fails on the room.""" + settings.RECORDING_ENABLE = True + room = RoomFactory() + user = UserFactory() + room.accesses.create(user=user, role="owner") + client = APIClient() + client.force_login(user) + + response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/") + + assert response.status_code == 404 + + [event] = find_events(audit_events, "recording.stop") + + assert event["event"]["reason"] == "not_found" + assert event["entity"]["target"]["sub_type"] == "room" diff --git a/src/backend/core/tests/services/test_livekit_events.py b/src/backend/core/tests/services/test_livekit_events.py index c69027fae..28fcd689a 100644 --- a/src/backend/core/tests/services/test_livekit_events.py +++ b/src/backend/core/tests/services/test_livekit_events.py @@ -1,7 +1,7 @@ """ Test LiveKitEvents service. """ -# pylint: disable=W0621,W0613, W0212, E0611 +# pylint: disable=W0621,W0613, W0212, E0611, too-many-lines import logging import uuid @@ -13,6 +13,7 @@ from django.utils import timezone import pytest from livekit.api import EgressStatus +from core.audit.testing import find_events from core.factories import RecordingFactory, RoomFactory from core.models import Room from core.recording.enums import RecordingWorkerEvent @@ -994,3 +995,76 @@ def test_participant_left_without_identity_is_ignored(mock_delete, service, sett service._handle_participant_left(data) # pylint: disable=protected-access mock_delete.assert_not_called() + + +@pytest.mark.parametrize( + ("egress_status", "worker_event", "outcome"), + ( + (EgressStatus.EGRESS_COMPLETE, "completed", "success"), + (EgressStatus.EGRESS_LIMIT_REACHED, "limit reached", "success"), + (EgressStatus.EGRESS_ABORTED, "aborted", "failure"), + (EgressStatus.EGRESS_FAILED, "failed", "failure"), + ), +) +@mock.patch( + "core.recording.services.recording_events.notification_service." + "notify_external_services" +) +@mock.patch("core.utils.notify_participants") +@mock.patch("core.services.room_management.RoomManagement.update_metadata") +def test_handle_egress_ended_is_audited( # noqa: PLR0913, PLR0917 + mock_update_metadata, + mock_notify, + mock_notify_external_services, + egress_status, + worker_event, + outcome, + service, + audit_events, +): # pylint: disable=too-many-arguments,too-many-positional-arguments + """The end of a recording is audited as LiveKit reports it.""" + + recording = RecordingFactory( + worker_id="worker-1", + status="active", + mode="screen_recording", + options={"transcribe": True}, + ) + mock_data = mock.MagicMock() + mock_data.egress_info.egress_id = recording.worker_id + mock_data.egress_info.status = egress_status + mock_data.egress_info.error_code = 0 + + service._handle_egress_ended(mock_data) + + [event] = find_events(audit_events, "recording.end") + + assert event["event"]["type"] == ["end"] + assert event["lasuite"]["outcome"] == outcome + assert event["lasuite"]["actor"] == {"type": "service"} + assert event["service"]["origin"] == {"name": "livekit"} + assert event["lasuite"]["auth"] == {"method": "shared_secret"} + assert event["entity"]["target"]["id"] == str(recording.id) + assert event["entity"]["target"]["raw"]["is_transcribed"] is True + assert event["lasuite"]["details"] == {"worker_event": worker_event} + + +@mock.patch("core.services.room_management.RoomManagement.update_metadata") +def test_handle_egress_ended_with_an_unmapped_status_is_audited_as_unknown( + mock_update_metadata, service, audit_events +): + """An egress status the backend does not map has an unknown outcome.""" + recording = RecordingFactory(worker_id="worker-1", status="active") + mock_data = mock.MagicMock() + mock_data.egress_info.egress_id = recording.worker_id + mock_data.egress_info.status = 999 + mock_data.egress_info.error_code = 0 + + service._handle_egress_ended(mock_data) + + [event] = find_events(audit_events, "recording.end") + + assert event["event"]["outcome"] == "unknown" + assert event["lasuite"]["outcome"] == "unknown" + assert event["log"]["level"] == "warning" + assert "details" not in event["lasuite"]