(backend) add connection-test API

Currently users have no way to reliably test their connection before
joining a room. To address this, we plan to build a connection-test
page.

The testing requires a dedicated LiveKit token, issued without going
through the room API, which is tied to registered meetings, lobby
rules, and longer-lived access tokens.

Introduce a new viewset for all diagnostics-related features. The
first route issues a token for diagnostics, even for anonymous
users. Each request creates a new dedicated room so users never
share the same LiveKit room during tests. Tokens are short-lived
(default 10 minutes) to limit reuse, and the endpoint is throttled
to prevent abuse.

A Celery worker also schedules a callback that deletes the room
after a certain delay, in every case.
This commit is contained in:
Arnaud Robin
2026-07-01 17:47:22 +02:00
committed by aleb_the_flash
parent 06e73d7a5e
commit b01a47bfd7
16 changed files with 518 additions and 0 deletions
+3
View File
@@ -104,3 +104,6 @@ APPLICATION_JWT_AUDIENCE=http://localhost:8071/external-api/v1.0/
APPLICATION_JWT_SECRET_KEY=devKey
APPLICATION_BASE_URL=http://localhost:3000
# Diagnostics
CONNECTION_TEST_ENABLED = True
+1
View File
@@ -65,6 +65,7 @@ def get_frontend_configuration(request):
"default_access_level": settings.RESOURCE_DEFAULT_ACCESS_LEVEL,
},
"subtitle": {"enabled": settings.ROOM_SUBTITLE_ENABLED},
"diagnostics": {"connection_test_enabled": settings.CONNECTION_TEST_ENABLED},
"livekit": {
"url": settings.LIVEKIT_CONFIGURATION["url"],
"force_wss_protocol": settings.LIVEKIT_FORCE_WSS_PROTOCOL,
+1
View File
@@ -17,6 +17,7 @@ class FeatureFlag:
"addons": "ADDONS_ENABLED",
"application": "APPLICATION_ENABLED",
"roomkit": "ROOMKIT_ENABLED",
"connection_test": "CONNECTION_TEST_ENABLED",
}
@classmethod
+12
View File
@@ -85,3 +85,15 @@ class RoomKitJoinRateThrottle(MonitoredUserRateThrottle):
"""
scope = "roomkit_join"
class ConnectionTestUserRateThrottle(MonitoredUserRateThrottle):
"""Throttle authenticated users requesting connection test tokens."""
scope = "connection_test"
class ConnectionTestAnonRateThrottle(MonitoredAnonRateThrottle):
"""Throttle anonymous users requesting connection test tokens."""
scope = "connection_test"
+69
View File
@@ -2,8 +2,10 @@
# pylint: disable=too-many-lines
import uuid
from datetime import timedelta
from logging import getLogger
from urllib.parse import unquote, urlparse
from uuid import uuid4
from django.conf import settings
from django.core.exceptions import ValidationError as DjangoValidationError
@@ -27,6 +29,9 @@ from rest_framework import (
from rest_framework import (
exceptions as drf_exceptions,
)
from rest_framework import (
permissions as drf_permissions,
)
from rest_framework import (
response as drf_response,
)
@@ -36,6 +41,7 @@ from rest_framework import (
from rest_framework.settings import api_settings
from core import analytics, enums, models, utils
from core.api import throttling
from core.api.filters import ListFileFilter
from core.enums import MEDIA_STORAGE_URL_PATTERN
from core.recording.enums import FileExtension
@@ -93,7 +99,9 @@ from core.services.room_roles import (
RoomRoleService,
)
from core.services.subtitle import SubtitleException, SubtitleService
from core.tasks.connection_test import delete_connection_test_room
from core.tasks.file import process_file_deletion
from core.utils import generate_token
from ..authentication.livekit import LiveKitTokenAuthentication
from ..models import RoomAccessLevel
@@ -1563,3 +1571,64 @@ class FileViewSet(
request = utils.generate_s3_authorization_headers(f"{url_params.get('key'):s}")
return drf_response.Response("authorized", headers=request.headers, status=200)
class DiagnosticsViewSet(viewsets.ViewSet):
"""Endpoints helping users and support diagnose connectivity issues.
Diagnostics are grouped behind a single prefix so upcoming checks
(rtcstats collection, ICE candidate reports, etc.) can be added as new
actions rather than new top-level routes.
They are open to anonymous users: someone who cannot join a room is
exactly who needs to run a test, and they may well not be logged in.
Each action therefore carries its own throttle scope.
"""
permission_classes = [drf_permissions.AllowAny]
@decorators.action(
detail=False,
methods=["POST"],
url_path="connection",
url_name="connection",
throttle_classes=[
throttling.ConnectionTestUserRateThrottle,
throttling.ConnectionTestAnonRateThrottle,
],
)
@FeatureFlag.require("connection_test")
def connection(self, request):
"""Return a short-lived LiveKit token for an ephemeral test room.
Going through the room API is not an option here: it is tied to
registered meetings, lobby rules and longer-lived tokens. Each call
gets its own room so two people testing at the same time never meet.
"""
room = f"{settings.CONNECTION_TEST_ROOM_PREFIX}-{uuid4()}"
expires_in = settings.CONNECTION_TEST_TOKEN_TTL_SECONDS
# LiveKit refreshes tokens for connected clients, so JWT TTL alone does not
# eject someone who stays connected. Schedule a hard DeleteRoom when Celery
# is available.
if settings.CELERY_ENABLED:
delete_connection_test_room.apply_async(
args=[room],
countdown=settings.CONNECTION_TEST_ROOM_MAX_AGE_SECONDS,
)
return drf_response.Response(
{
"livekit": {
"url": settings.LIVEKIT_CONFIGURATION["url"],
"room": room,
"token": generate_token(
room=room,
user=request.user,
username="Connection Test",
ttl=timedelta(seconds=expires_in),
),
"expires_in": expires_in,
},
}
)
@@ -137,6 +137,13 @@ class LiveKitEventsService:
room_name = data.room.name or data.egress_info.room_name
if self._is_connection_test_room(room_name):
logger.info(
"Ignoring webhook event for connection test room '%s'.",
room_name,
)
return
if self._filter_regex and not self._filter_regex.search(room_name):
logger.info("Filtered webhook event for room '%s'", room_name)
return
@@ -228,6 +235,11 @@ class LiveKitEventsService:
# Silently ignoring EGRESS_ABORTED, EGRESS_FAILED
@staticmethod
def _is_connection_test_room(room_name: str) -> bool:
"""Return True for ephemeral rooms created by the connection test endpoint."""
return room_name.startswith(settings.CONNECTION_TEST_ROOM_PREFIX)
def _handle_room_started(self, data):
"""Handle 'room_started' event."""
@@ -8,6 +8,7 @@ from typing import Dict, Optional
from asgiref.sync import async_to_sync
from livekit.api import (
DeleteRoomRequest,
ListRoomsRequest,
TwirpError,
UpdateRoomMetadataRequest,
@@ -88,3 +89,30 @@ class RoomManagement:
finally:
await lkapi.aclose()
@async_to_sync
async def delete_room(self, room_name: str):
"""Delete a LiveKit room and disconnect all participants.
Raises:
RoomNotFoundException: the room does not exist in LiveKit.
RoomManagementException: the deletion otherwise fails.
"""
lkapi = utils.create_livekit_client()
try:
await lkapi.room.delete_room(DeleteRoomRequest(room=room_name))
logger.info("Deleted LiveKit room %s", room_name)
except TwirpError as e:
if e.code == "not_found":
logger.warning(
"Room %s not found in LiveKit, skipping deletion",
room_name,
)
raise RoomNotFoundException("Room does not exist") from e
logger.exception("Unexpected error deleting room %s", room_name)
raise RoomManagementException("Could not delete room") from e
finally:
await lkapi.aclose()
+9
View File
@@ -0,0 +1,9 @@
"""Celery tasks for the core app."""
from core.tasks.connection_test import delete_connection_test_room
from core.tasks.file import process_file_deletion
__all__ = (
"delete_connection_test_room",
"process_file_deletion",
)
+39
View File
@@ -0,0 +1,39 @@
"""Tasks related to connection test rooms."""
import logging
from django.conf import settings
from core.services.room_management import (
RoomManagement,
RoomManagementException,
RoomNotFoundException,
)
from core.tasks._task import task
logger = logging.getLogger(__name__)
@task
def delete_connection_test_room(room_name: str):
"""Force-delete an ephemeral connection-test room.
Used as a hard cap so a participant cannot keep an auto-refreshed
LiveKit session open indefinitely after requesting a test token.
"""
prefix = settings.CONNECTION_TEST_ROOM_PREFIX
if not room_name.startswith(prefix):
logger.error(
"Refusing to delete room '%s': expected prefix '%s'.",
room_name,
prefix,
)
return
try:
RoomManagement().delete_room(room_name)
except RoomNotFoundException:
# Room may already be gone after empty/departure timeout.
logger.info("Connection test room '%s' already gone.", room_name)
except RoomManagementException:
logger.exception("Failed to delete connection test room '%s'.", room_name)
@@ -720,6 +720,7 @@ def test_receive_unsupported_event(mock_receive, service):
# Mock returned data with unsupported event type
mock_data = mock.MagicMock()
mock_data.room.name = str(uuid.uuid4())
mock_data.event = "unsupported_event"
mock_receive.return_value = mock_data
@@ -823,3 +824,33 @@ def test_receive_filter_processes_matching_events(
service.receive(mock_request)
mock_handle_room_started.assert_called_once()
@mock.patch.object(api.WebhookReceiver, "receive")
@mock.patch.object(LiveKitEventsService, "_handle_room_finished")
@mock.patch.object(LiveKitEventsService, "_handle_room_started")
def test_receive_ignores_connection_test_room(
mock_handle_room_started,
mock_handle_room_finished,
mock_receive,
mock_livekit_config,
settings,
):
"""Should ignore all webhook events for connection test rooms in receive()."""
settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
mock_request = mock.MagicMock()
mock_request.headers = {"Authorization": "test_token"}
mock_request.body = b"{}"
mock_data = mock.MagicMock()
mock_data.room.name = f"{settings.CONNECTION_TEST_ROOM_PREFIX}-{uuid.uuid4()}"
mock_data.event = "room_started"
mock_receive.return_value = mock_data
service = LiveKitEventsService()
service.receive(mock_request)
mock_handle_room_started.assert_not_called()
mock_handle_room_finished.assert_not_called()
@@ -0,0 +1,60 @@
"""Tests for the RoomManagement service."""
from unittest import mock
import pytest
from livekit.api import TwirpError
from core.services.room_management import (
RoomManagement,
RoomManagementException,
RoomNotFoundException,
)
@mock.patch("core.services.room_management.utils.create_livekit_client")
def test_delete_room_calls_livekit(mock_create_livekit_client):
"""DeleteRoom is forwarded to the LiveKit API."""
mock_api = mock.MagicMock()
mock_api.room.delete_room = mock.AsyncMock()
mock_api.aclose = mock.AsyncMock()
mock_create_livekit_client.return_value = mock_api
RoomManagement().delete_room("room-abc")
mock_api.room.delete_room.assert_awaited_once()
request = mock_api.room.delete_room.await_args.args[0]
assert request.room == "room-abc"
mock_api.aclose.assert_awaited_once()
@mock.patch("core.services.room_management.utils.create_livekit_client")
def test_delete_room_raises_not_found(mock_create_livekit_client):
"""Missing rooms raise RoomNotFoundException."""
mock_api = mock.MagicMock()
mock_api.room.delete_room = mock.AsyncMock(
side_effect=TwirpError("not_found", "room not found", status=404)
)
mock_api.aclose = mock.AsyncMock()
mock_create_livekit_client.return_value = mock_api
with pytest.raises(RoomNotFoundException):
RoomManagement().delete_room("missing-room")
mock_api.aclose.assert_awaited_once()
@mock.patch("core.services.room_management.utils.create_livekit_client")
def test_delete_room_raises_management_exception(mock_create_livekit_client):
"""Unexpected Twirp errors raise RoomManagementException."""
mock_api = mock.MagicMock()
mock_api.room.delete_room = mock.AsyncMock(
side_effect=TwirpError("internal", "boom", status=500)
)
mock_api.aclose = mock.AsyncMock()
mock_create_livekit_client.return_value = mock_api
with pytest.raises(RoomManagementException):
RoomManagement().delete_room("room-abc")
mock_api.aclose.assert_awaited_once()
@@ -0,0 +1,51 @@
"""Tests for connection test Celery tasks."""
from unittest import mock
from django.test.utils import override_settings
from core.services.room_management import (
RoomManagementException,
RoomNotFoundException,
)
from core.tasks.connection_test import delete_connection_test_room
@mock.patch("core.tasks.connection_test.RoomManagement.delete_room")
def test_delete_connection_test_room_calls_room_management(mock_delete_room, settings):
"""RoomManagement.delete_room is called for connection-test rooms."""
settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
delete_connection_test_room("connection-test-abc")
mock_delete_room.assert_called_once_with("connection-test-abc")
@mock.patch("core.tasks.connection_test.RoomManagement.delete_room")
def test_delete_connection_test_room_refuses_other_rooms(mock_delete_room, settings):
"""Refuse to delete rooms outside the connection-test namespace."""
settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
delete_connection_test_room("production-room")
mock_delete_room.assert_not_called()
@mock.patch("core.tasks.connection_test.RoomManagement.delete_room")
def test_delete_connection_test_room_ignores_missing_room(mock_delete_room, settings):
"""Missing rooms are treated as already cleaned up."""
settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
mock_delete_room.side_effect = RoomNotFoundException("Room does not exist")
delete_connection_test_room("connection-test-gone")
mock_delete_room.assert_called_once_with("connection-test-gone")
@mock.patch("core.tasks.connection_test.RoomManagement.delete_room")
def test_delete_connection_test_room_logs_other_failures(mock_delete_room, settings):
"""Unexpected LiveKit failures are swallowed after logging."""
settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
mock_delete_room.side_effect = RoomManagementException("Could not delete room")
delete_connection_test_room("connection-test-fail")
mock_delete_room.assert_called_once_with("connection-test-fail")
@@ -0,0 +1,165 @@
"""Test diagnostics API endpoints."""
import uuid
from unittest import mock
from django.test.utils import override_settings
from django.urls import reverse
import jwt
import pytest
from rest_framework.test import APIClient
from core.api.throttling import (
ConnectionTestAnonRateThrottle,
ConnectionTestUserRateThrottle,
)
from core.factories import UserFactory
pytestmark = pytest.mark.django_db
def test_api_diagnostics_connection_url():
"""The connection check is exposed under the diagnostics namespace."""
assert reverse("diagnostics-connection") == "/api/v1.0/diagnostics/connection/"
def test_api_diagnostics_connection_rejects_get():
"""Only POST is exposed, the endpoint has no side effect to trigger."""
client = APIClient()
response = client.get("/api/v1.0/diagnostics/connection/")
assert response.status_code == 405
def test_api_diagnostics_connection_returns_ephemeral_livekit_config(settings, client):
"""Each request gets a dedicated room and a short-lived token."""
settings.CONNECTION_TEST_TOKEN_TTL_SECONDS = 600
settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
response_a = client.post("/api/v1.0/diagnostics/connection/")
response_b = client.post("/api/v1.0/diagnostics/connection/")
assert response_a.status_code == 200
assert response_b.status_code == 200
data_a = response_a.json()
data_b = response_b.json()
room_a = data_a["livekit"]["room"]
room_b = data_b["livekit"]["room"]
assert room_a.startswith("connection-test-")
assert room_b.startswith("connection-test-")
uuid.UUID(room_a.removeprefix("connection-test-"))
uuid.UUID(room_b.removeprefix("connection-test-"))
assert room_a != room_b
assert data_a["livekit"]["url"]
assert data_a["livekit"]["token"]
assert data_a["livekit"]["expires_in"] == 600
assert data_a["livekit"]["token"] != data_b["livekit"]["token"]
def test_api_diagnostics_connection_token_is_short_lived_for_user(settings, client):
"""Connection test tokens expire quickly for users."""
settings.CONNECTION_TEST_TOKEN_TTL_SECONDS = 300
client = APIClient()
response = client.post("/api/v1.0/diagnostics/connection/")
assert response.status_code == 200
config = response.json()["livekit"]
payload = jwt.decode(
config["token"],
settings.LIVEKIT_CONFIGURATION["api_secret"],
algorithms=["HS256"],
options={"verify_exp": False},
)
assert config["expires_in"] == 300
assert payload["video"]["room"] == config["room"]
assert payload["name"] == "Connection Test"
assert payload["video"]["roomAdmin"] is False
assert payload["exp"] - payload["nbf"] == 300
@override_settings()
def test_api_diagnostics_connection_token_for_authenticated_user(settings, client):
"""Logged-in users get a token bound to their own identity."""
settings.CONNECTION_TEST_TOKEN_TTL_SECONDS = 300
user = UserFactory()
client.force_login(user)
response = client.post("/api/v1.0/diagnostics/connection/")
assert response.status_code == 200
payload = jwt.decode(
response.json()["livekit"]["token"],
settings.LIVEKIT_CONFIGURATION["api_secret"],
algorithms=["HS256"],
options={"verify_exp": False},
)
assert payload["sub"] == str(user.sub)
assert payload["video"]["roomAdmin"] is False
assert payload["exp"] - payload["nbf"] == 300
@mock.patch("core.api.viewsets.delete_connection_test_room.apply_async")
def test_api_diagnostics_connection_schedules_room_deletion(
mock_apply_async, settings, client
):
"""When Celery is enabled, schedule a hard room delete after max age."""
settings.CELERY_ENABLED = True
settings.CONNECTION_TEST_ROOM_MAX_AGE_SECONDS = 300
settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
response = client.post("/api/v1.0/diagnostics/connection/")
assert response.status_code == 200
room = response.json()["livekit"]["room"]
mock_apply_async.assert_called_once_with(args=[room], countdown=300)
@mock.patch("core.api.viewsets.delete_connection_test_room.apply_async")
def test_api_diagnostics_connection_skips_room_deletion_without_celery(
mock_apply_async, settings, client
):
"""Without Celery, do not schedule deletion (apply_async would run immediately)."""
settings.CELERY_ENABLED = False
response = client.post("/api/v1.0/diagnostics/connection/")
assert response.status_code == 200
mock_apply_async.assert_not_called()
@pytest.mark.parametrize(
"throttle_class",
[ConnectionTestAnonRateThrottle, ConnectionTestUserRateThrottle],
)
def test_api_diagnostics_connection_is_throttled(throttle_class, client):
"""Both throttles stay wired to the action once routed through the viewset."""
with (
mock.patch.object(throttle_class, "allow_request", return_value=False),
mock.patch.object(throttle_class, "wait", return_value=42),
):
response = client.post("/api/v1.0/diagnostics/connection/")
assert response.status_code == 429
def test_api_diagnostics_connection_feature_flag(client, settings):
"""Should return a not found error when the connection diagnostics feature is disabled."""
settings.CONNECTION_TEST_ENABLED = False
response = client.post("/api/v1.0/diagnostics/connection/")
assert response.status_code == 404
+5
View File
@@ -30,6 +30,11 @@ router.register(
addons_viewsets.SessionViewSet,
basename="addons_sessions",
)
router.register(
"diagnostics",
viewsets.DiagnosticsViewSet,
basename="diagnostics",
)
# - External API
external_router = SimpleRouter()
+5
View File
@@ -12,6 +12,7 @@ import mimetypes
import random
import secrets
import string
from datetime import timedelta
from functools import lru_cache
from typing import List, Optional
from uuid import uuid4
@@ -67,6 +68,7 @@ def generate_token( # noqa: PLR0917
sources: Optional[List[str]] = None,
role: Optional[str] = None,
participant_id: Optional[str] = None,
ttl: Optional[timedelta] = None,
) -> str:
"""Generate a LiveKit access token for a user in a specific room.
@@ -82,6 +84,7 @@ def generate_token( # noqa: PLR0917
role (Optional[str]): Room's access role if any
participant_id (Optional[str]): Stable identifier for anonymous users;
used as identity when user.is_anonymous.
ttl (Optional[timedelta]): Token validity duration. Defaults to LiveKit SDK default.
Returns:
str: The LiveKit JWT access token.
@@ -135,6 +138,8 @@ def generate_token( # noqa: PLR0917
}
)
)
if ttl is not None:
token = token.with_ttl(ttl)
return token.to_jwt()
+27
View File
@@ -354,6 +354,11 @@ class Base(Configuration):
environ_name="ROOMKIT_JOIN_THROTTLE_RATES",
environ_prefix=None,
),
"connection_test": values.Value(
default="30/minute",
environ_name="CONNECTION_TEST_THROTTLE_RATES",
environ_prefix=None,
),
},
}
MONITORED_THROTTLE_FAILURE_CALLBACK = (
@@ -660,6 +665,26 @@ class Base(Configuration):
environ_prefix=None,
default=False,
)
CONNECTION_TEST_ENABLED = values.BooleanValue(
environ_name="CONNECTION_TEST_ENABLED",
environ_prefix=None,
default=False,
)
CONNECTION_TEST_TOKEN_TTL_SECONDS = values.PositiveIntegerValue(
300,
environ_name="CONNECTION_TEST_TOKEN_TTL_SECONDS",
environ_prefix=None,
)
CONNECTION_TEST_ROOM_MAX_AGE_SECONDS = values.PositiveIntegerValue(
300,
environ_name="CONNECTION_TEST_ROOM_MAX_AGE_SECONDS",
environ_prefix=None,
)
CONNECTION_TEST_ROOM_PREFIX = values.Value(
"connection-test",
environ_name="CONNECTION_TEST_ROOM_PREFIX",
environ_prefix=None,
)
LIVEKIT_VERIFY_SSL = values.BooleanValue(
True, environ_name="LIVEKIT_VERIFY_SSL", environ_prefix=None
)
@@ -1270,6 +1295,8 @@ class Test(Base):
ADDONS_CSRF_SECRET = "secret-key-padded-for-minimum-len!-addons" # noqa:S105
ADDONS_TOKEN_SECRET_KEY = "secret-key-padded-for-minimum-len!-addons" # noqa:S105
CONNECTION_TEST_ENABLED = True
def __init__(self):
# pylint: disable=invalid-name
self.INSTALLED_APPS += ["drf_spectacular_sidecar"]