✨(backend) add two new endpoints to start and stop a recording

The LiveKit egress worker interactions are proxied through the backend for
security reasons. Allowing clients to directly use tokens with sufficient
grants to start recordings could lead to misuse, enabling users to spam the
egress worker API and potentially initiate a DDOS attack on the egress
service. To prevent this, only users with room-specific privileges can
initiate recordings.

We make sure only one recording at the time can be made on a room.

The requested recording mode is stored so it can be referenced later when
the recording is saved, triggering a callback action as needed.

A feature flag was also introduced for this capability; while this is a simple
approach, a more robust system for managing feature flags could be valuable
long-term. For now, KISS (Keep It Simple, Stupid) applies.

The viewset endpoints were designed to be as straightforward as possible—
let me know if anything can be improved.

src/backend/core/api/permissions.py

@@ -1,5 +1,7 @@
 """Permission handlers for the Meet core app."""
 
+from django.conf import settings
+
 from rest_framework import permissions
 
 from ..models import RoleChoices
@@ -87,3 +89,23 @@ class HasAbilityPermission(IsAuthenticated):
     def has_object_permission(self, request, view, obj):
         """Check permission for a given object."""
         return obj.get_abilities(request.user).get(view.action, False)
+
+
+class HasPrivilegesOnRoom(IsAuthenticated):
+    """Check if user has privileges on a given room."""
+
+    message = "You must have privileges to start a recording."
+
+    def has_object_permission(self, request, view, obj):
+        """Determine if user has privileges on room."""
+        return obj.is_owner(request.user) or obj.is_administrator(request.user)
+
+
+class IsRecordingEnabled(permissions.BasePermission):
+    """Check if the recording feature is enabled."""
+
+    message = "Access denied, recording is disabled."
+
+    def has_permission(self, request, view):
+        """Determine if access is allowed based on settings."""
+        return settings.RECORDING_ENABLE

src/backend/core/api/serializers.py

@@ -156,3 +156,25 @@ class RecordingSerializer(serializers.ModelSerializer):
         model = models.Recording
         fields = ["id", "room", "created_at", "updated_at", "status"]
         read_only_fields = fields
+
+
+class StartRecordingSerializer(serializers.Serializer):
+    """Validate start recording requests."""
+
+    mode = serializers.ChoiceField(
+        choices=models.RecordingModeChoices.choices,
+        required=True,
+        error_messages={
+            "required": "Recording mode is required.",
+            "invalid_choice": "Invalid recording mode. Choose between "
+            "screen_recording or transcript.",
+        },
+    )
+
+    def create(self, validated_data):
+        """Not implemented as this is a validation-only serializer."""
+        raise NotImplementedError("StartRecordingSerializer is validation-only")
+
+    def update(self, instance, validated_data):
+        """Not implemented as this is a validation-only serializer."""
+        raise NotImplementedError("StartRecordingSerializer is validation-only")

src/backend/core/api/viewsets.py

@@ -1,6 +1,7 @@
 """API endpoints"""
 
 import uuid
+from logging import getLogger
 
 from django.conf import settings
 from django.db.models import Q
@@ -14,16 +15,34 @@ from rest_framework import (
     pagination,
     viewsets,
 )
+from rest_framework import (
+    exceptions as drf_exceptions,
+)
 from rest_framework import (
     response as drf_response,
 )
+from rest_framework import (
+    status as drf_status,
+)
 
 from core import models, utils
+from core.recording.worker.exceptions import (
+    RecordingStartError,
+    RecordingStopError,
+)
+from core.recording.worker.factories import (
+    get_worker_service,
+)
+from core.recording.worker.mediator import (
+    WorkerServiceMediator,
+)
 
 from . import permissions, serializers
 
 # pylint: disable=too-many-ancestors
 
+logger = getLogger(__name__)
+
 
 class NestedGenericViewSet(viewsets.GenericViewSet):
     """
@@ -233,6 +252,89 @@ class RoomViewSet(
             role=models.RoleChoices.OWNER,
         )
 
+    @decorators.action(
+        detail=True,
+        methods=["post"],
+        url_path="start-recording",
+        permission_classes=[
+            permissions.HasPrivilegesOnRoom,
+            permissions.IsRecordingEnabled,
+        ],
+    )
+    def start_room_recording(self, request, pk=None):  # pylint: disable=unused-argument
+        """Start recording a room."""
+
+        serializer = serializers.StartRecordingSerializer(data=request.data)
+
+        if not serializer.is_valid():
+            return drf_response.Response(
+                {"detail": "Invalid request."}, status=drf_status.HTTP_400_BAD_REQUEST
+            )
+
+        mode = serializer.validated_data["mode"]
+        room = self.get_object()
+
+        # May raise exception if an active or initiated recording already exist for the room
+        recording = models.Recording.objects.create(room=room, mode=mode)
+
+        models.RecordingAccess.objects.create(
+            user=self.request.user, role=models.RoleChoices.OWNER, recording=recording
+        )
+
+        worker_service = get_worker_service(mode=recording.mode)
+        worker_manager = WorkerServiceMediator(worker_service=worker_service)
+
+        try:
+            worker_manager.start(recording)
+        except RecordingStartError:
+            return drf_response.Response(
+                {"error": f"Recording failed to start for room {room.slug}"},
+                status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )
+
+        return drf_response.Response(
+            {"message": f"Recording successfully started for room {room.slug}"},
+            status=drf_status.HTTP_201_CREATED,
+        )
+
+    @decorators.action(
+        detail=True,
+        methods=["post"],
+        url_path="stop-recording",
+        permission_classes=[
+            permissions.HasPrivilegesOnRoom,
+            permissions.IsRecordingEnabled,
+        ],
+    )
+    def stop_room_recording(self, request, pk=None):  # pylint: disable=unused-argument
+        """Stop room recording."""
+
+        room = self.get_object()
+
+        try:
+            recording = models.Recording.objects.get(
+                room=room, status=models.RecordingStatusChoices.ACTIVE
+            )
+        except models.Recording.DoesNotExist as e:
+            raise drf_exceptions.NotFound(
+                "No active recording found for this room."
+            ) from e
+
+        worker_service = get_worker_service(mode=recording.mode)
+        worker_manager = WorkerServiceMediator(worker_service=worker_service)
+
+        try:
+            worker_manager.stop(recording)
+        except RecordingStopError:
+            return drf_response.Response(
+                {"error": f"Recording failed to stop for room {room.slug}"},
+                status=drf_status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )
+
+        return drf_response.Response(
+            {"message": f"Recording stopped for room {room.slug}."}
+        )
+
 
 class ResourceAccessListModelMixin:
     """List mixin for resource access API."""

src/backend/core/factories.py

@@ -75,6 +75,7 @@ class RecordingFactory(factory.django.DjangoModelFactory):
 
     room = factory.SubFactory(RoomFactory)
     status = models.RecordingStatusChoices.INITIATED
+    mode = models.RecordingModeChoices.SCREEN_RECORDING
     worker_id = None
 
     @factory.post_generation

src/backend/core/migrations/0007_recording_mode.py

@@ -0,0 +1,18 @@
+# Generated by Django 5.1.1 on 2024-11-12 10:59
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0006_merge_duplicate_users'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='recording',
+            name='mode',
+            field=models.CharField(choices=[('screen_recording', 'SCREEN_RECORDING'), ('transcript', 'TRANSCRIPT')], default='screen_recording', help_text='Defines the mode of recording being called.', max_length=20, verbose_name='Recording mode'),
+        ),
+    ]

src/backend/core/models.py

@@ -72,6 +72,13 @@ class RecordingStatusChoices(models.TextChoices):
         return status in {cls.ABORTED, cls.FAILED_TO_START, cls.FAILED_TO_STOP}
 
 
+class RecordingModeChoices(models.TextChoices):
+    """Recording mode choices."""
+
+    SCREEN_RECORDING = "screen_recording", _("SCREEN_RECORDING")
+    TRANSCRIPT = "transcript", _("TRANSCRIPT")
+
+
 class BaseModel(models.Model):
     """
     Serves as an abstract base model for other models, ensuring that records are validated
@@ -482,6 +489,13 @@ class Recording(BaseModel):
             "This ID is retained even when the worker stops, allowing for easy tracking."
         ),
     )
+    mode = models.CharField(
+        max_length=20,
+        choices=RecordingModeChoices.choices,
+        default=RecordingModeChoices.SCREEN_RECORDING,
+        verbose_name=_("Recording mode"),
+        help_text=_("Defines the mode of recording being called."),
+    )
 
     class Meta:
         db_table = "meet_recording"

src/backend/core/tests/rooms/test_api_rooms_start_recording.py

@@ -0,0 +1,200 @@
+"""
+Test rooms API endpoints in the Meet core app: start recording.
+"""
+
+# pylint: disable=W0621,W0613
+
+from unittest import mock
+
+import pytest
+from rest_framework.test import APIClient
+
+from ...factories import RoomFactory, UserFactory
+from ...models import Recording
+from ...recording.worker.exceptions import RecordingStartError
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def mock_worker_service():
+    """Mock worker service."""
+    return mock.Mock()
+
+
+@pytest.fixture
+def mock_worker_service_factory(mock_worker_service):
+    """Mock worker service factory."""
+    with mock.patch(
+        "core.api.viewsets.get_worker_service",
+        return_value=mock_worker_service,
+    ) as mock_worker_service_factory:
+        yield mock_worker_service_factory
+
+
+@pytest.fixture
+def mock_worker_manager(mock_worker_service):
+    """Mock worker service mediator."""
+    with mock.patch("core.api.viewsets.WorkerServiceMediator") as mock_mediator_class:
+        mock_mediator = mock.Mock()
+        mock_mediator_class.return_value = mock_mediator
+        yield mock_mediator
+
+
+def test_start_recording_anonymous():
+    """Anonymous users should not be allowed to start room recordings."""
+    room = RoomFactory()
+    client = APIClient()
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "screen_recording"},
+    )
+
+    assert response.status_code == 401
+    assert Recording.objects.count() == 0
+
+
+def test_start_recording_non_owner_and_non_administrator():
+    """Non-owner and Non-Administrator users should not be allowed to start room recordings."""
+    room = RoomFactory()
+    user = UserFactory()
+    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 == 403
+    assert Recording.objects.count() == 0
+
+
+def test_start_recording_recording_disabled(settings):
+    """Should fail if recording is disabled for the room."""
+    settings.RECORDING_ENABLE = False
+
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    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 == 403
+    assert response.json() == {"detail": "Access denied, recording is disabled."}
+    assert Recording.objects.count() == 0
+
+
+def test_start_recording_missing_mode(settings):
+    """Should fail if recording mode is not provided."""
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    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/", {})
+
+    assert response.status_code == 400
+    assert response.json() == {"detail": "Invalid request."}
+    assert Recording.objects.count() == 0
+
+
+def test_start_recording_worker_error(
+    mock_worker_service_factory, mock_worker_manager, settings
+):
+    """Should handle worker service errors appropriately."""
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    # Configure mock mediator to raise error
+    mock_start = mock.Mock()
+    mock_start.side_effect = RecordingStartError("Failed to connect to worker")
+
+    mock_worker_manager.start = mock_start
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "screen_recording"},
+    )
+
+    mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
+
+    assert response.status_code == 500
+    assert response.json() == {
+        "error": f"Recording failed to start for room {room.slug}"
+    }
+
+    # Recording object should be created even if worker fails
+    assert Recording.objects.count() == 1
+    recording = Recording.objects.first()
+    assert recording.room == room
+    assert recording.mode == "screen_recording"
+
+    # Verify recording access details
+    assert recording.accesses.count() == 1
+    access = recording.accesses.first()
+    assert access.user == user
+    assert access.role == "owner"
+
+
+def test_start_recording_success(
+    mock_worker_service_factory, mock_worker_manager, settings
+):
+    """Should successfully start recording when everything is configured correctly."""
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    mock_start = mock.Mock()
+    mock_worker_manager.start = mock_start
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(
+        f"/api/v1.0/rooms/{room.id}/start-recording/",
+        {"mode": "screen_recording"},
+    )
+
+    mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
+
+    assert response.status_code == 201
+    assert response.json() == {
+        "message": f"Recording successfully started for room {room.slug}"
+    }
+
+    # Verify the mediator was called with the recording
+    recording = Recording.objects.first()
+    mock_start.assert_called_once_with(recording)
+
+    assert recording.room == room
+    assert recording.mode == "screen_recording"
+
+    # Verify recording access details
+    assert recording.accesses.count() == 1
+    access = recording.accesses.first()
+    assert access.user == user
+    assert access.role == "owner"

src/backend/core/tests/rooms/test_api_rooms_stop_recording.py

@@ -0,0 +1,182 @@
+"""
+Test rooms API endpoints in the Meet core app: stop recording.
+"""
+
+# pylint: disable=W0621,W0613
+
+from unittest import mock
+
+import pytest
+from rest_framework.test import APIClient
+
+from ...factories import RecordingFactory, RoomFactory, UserFactory
+from ...models import Recording, RecordingStatusChoices
+from ...recording.worker.exceptions import RecordingStopError
+
+pytestmark = pytest.mark.django_db
+
+
+@pytest.fixture
+def mock_worker_service():
+    """Mock worker service."""
+    return mock.Mock()
+
+
+@pytest.fixture
+def mock_worker_service_factory(mock_worker_service):
+    """Mock worker service factory."""
+    with mock.patch(
+        "core.api.viewsets.get_worker_service",
+        return_value=mock_worker_service,
+    ) as mock_worker_service_factory:
+        yield mock_worker_service_factory
+
+
+@pytest.fixture
+def mock_worker_manager(mock_worker_service):
+    """Mock worker service mediator."""
+    with mock.patch("core.api.viewsets.WorkerServiceMediator") as mock_mediator_class:
+        mock_mediator = mock.Mock()
+        mock_mediator_class.return_value = mock_mediator
+        yield mock_mediator
+
+
+def test_stop_recording_anonymous():
+    """Anonymous users should not be allowed to stop room recordings."""
+    room = RoomFactory()
+    RecordingFactory(room=room, status=RecordingStatusChoices.ACTIVE)
+    client = APIClient()
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    assert response.status_code == 401
+    # Verify recording status hasn't changed
+    assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
+
+
+def test_stop_recording_non_owner_and_non_administrator():
+    """Non-owner and Non-Administrator users should not be allowed to stop room recordings."""
+    room = RoomFactory()
+    user = UserFactory()
+    RecordingFactory(room=room, 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 == 403
+    # Verify recording status hasn't changed
+    assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
+
+
+def test_stop_recording_recording_disabled(settings):
+    """Should fail if recording is disabled for the room."""
+
+    settings.RECORDING_ENABLE = False
+
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    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 == 403
+    assert response.json() == {"detail": "Access denied, recording is disabled."}
+    # Verify no recording exists
+    assert Recording.objects.count() == 0
+
+
+def test_stop_recording_no_active_recording(settings):
+    """Should fail when there is no active recording for the room."""
+
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    # Make user the room owner
+    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
+    assert response.json() == {"detail": "No active recording found for this room."}
+
+
+def test_stop_recording_worker_error(
+    mock_worker_service_factory, mock_worker_manager, settings
+):
+    """Should handle worker service errors appropriately."""
+
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    recording = RecordingFactory(
+        room=room,
+        status=RecordingStatusChoices.ACTIVE,
+        mode="screen_recording",
+    )
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    # Configure mock mediator to raise error
+    mock_stop = mock.Mock()
+    mock_stop.side_effect = RecordingStopError("Failed to connect to worker")
+    mock_worker_manager.stop = mock_stop
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
+    mock_stop.assert_called_once_with(recording)
+
+    assert response.status_code == 500
+    assert response.json() == {
+        "error": f"Recording failed to stop for room {room.slug}"
+    }
+    # Verify recording status hasn't changed
+    assert Recording.objects.filter(status=RecordingStatusChoices.ACTIVE).count() == 1
+
+
+def test_stop_recording_success(
+    mock_worker_service_factory, mock_worker_manager, settings
+):
+    """Should successfully stop recording when everything is configured correctly."""
+
+    settings.RECORDING_ENABLE = True
+
+    room = RoomFactory()
+    user = UserFactory()
+    recording = RecordingFactory(
+        room=room,
+        status=RecordingStatusChoices.ACTIVE,
+        mode="screen_recording",
+    )
+    # Make user the room owner
+    room.accesses.create(user=user, role="owner")
+
+    mock_stop = mock.Mock()
+    mock_worker_manager.stop = mock_stop
+
+    client = APIClient()
+    client.force_login(user)
+
+    response = client.post(f"/api/v1.0/rooms/{room.id}/stop-recording/")
+
+    mock_worker_service_factory.assert_called_once_with(mode="screen_recording")
+    mock_stop.assert_called_once_with(recording)
+
+    assert response.status_code == 200
+    assert response.json() == {"message": f"Recording stopped for room {room.slug}."}
+
+    # Verify the recording still exists
+    assert Recording.objects.count() == 1

src/backend/meet/settings.py

@@ -407,12 +407,23 @@ class Base(Configuration):
     )
 
     # Recording settings
+    RECORDING_ENABLE = values.BooleanValue(
+        False, environ_name="RECORDING_ENABLE", environ_prefix=None
+    )
     RECORDING_OUTPUT_FOLDER = values.Value(
         "recordings", environ_name="RECORDING_OUTPUT_FOLDER", environ_prefix=None
     )
     RECORDING_VERIFY_SSL = values.BooleanValue(
         True, environ_name="RECORDING_VERIFY_SSL", environ_prefix=None
     )
+    RECORDING_WORKER_CLASSES = values.DictValue(
+        {
+            "screen_recording": "core.recording.worker.services.VideoCompositeEgressService",
+            "transcript": "core.recording.worker.services.AudioCompositeEgressService",
+        },
+        environ_name="RECORDING_WORKER_CLASSES",
+        environ_prefix=None,
+    )
 
     # pylint: disable=invalid-name
     @property