diff --git a/api/src/backend/api/sse/__init__.py b/api/src/backend/api/sse/__init__.py new file mode 100644 index 0000000000..244bd3c9ad --- /dev/null +++ b/api/src/backend/api/sse/__init__.py @@ -0,0 +1,13 @@ +"""Platform Server-Sent Events (SSE) infrastructure. + +Wires `django-eventstream` into the API: a base viewset features +subclass to expose an SSE endpoint +(:class:`api.sse.base_views.BaseSSEViewSet`), the channel manager that +enforces the tenant gate (:class:`api.sse.channelmanager.SSEChannelManager`), +and the channel-name helpers (:func:`api.sse.utils.make_channel_name`). +""" + +from api.sse.utils import make_channel_name +from api.sse.base_views import BaseSSEViewSet + +__all__ = ["BaseSSEViewSet", "make_channel_name"] diff --git a/api/src/backend/api/sse/base_views.py b/api/src/backend/api/sse/base_views.py new file mode 100644 index 0000000000..c4a24540ee --- /dev/null +++ b/api/src/backend/api/sse/base_views.py @@ -0,0 +1,46 @@ +"""Base view class for SSE endpoints.""" + +from api.authentication import SSEAuthentication +from api.base_views import BaseRLSViewSet +from django_eventstream.renderers import SSEEventRenderer +from django_eventstream.views import events + + +class BaseSSEViewSet(BaseRLSViewSet): + """Base class for platform SSE endpoints. + + Subclasses override method `get_channels` to declare the channel + names the connection should subscribe to — the same way a regular + DRF viewset overrides method `get_queryset`. The channel manager + reads the result from `request.sse_channels`; there is no other + coupling between platform and feature. + """ + + authentication_classes = [SSEAuthentication] + # Pin the SSE renderer so content negotiation accepts the browser's + # `Accept: text/event-stream`. + renderer_classes = [SSEEventRenderer] + + def get_channels(self) -> set[str]: + """Return the channels this connection subscribes to. + + Implementations MUST raise the relevant DRF exceptions + (`NotAuthenticated`, `PermissionDenied`, `NotFound`) when + authorization fails. Returning an empty set would surface as + django-eventstream's "No channels specified" which masks the + real cause. + """ + raise NotImplementedError + + def get_queryset(self): + # Most SSE viewsets only need `get_channels` and never call + # `get_queryset` (the SSE list path bypasses serialization + # entirely). Subclasses that perform their own queryset lookup + # inside `get_channels` should override; the default raises + # the same error a missing override on a ModelViewSet would. + raise NotImplementedError + + def list(self, request, *_args, **kwargs): + """Resolve channels under the regular DRF stack and stream.""" + request.sse_channels = self.get_channels() + return events(request, **kwargs) diff --git a/api/src/backend/api/sse/channelmanager.py b/api/src/backend/api/sse/channelmanager.py new file mode 100644 index 0000000000..6be42747f2 --- /dev/null +++ b/api/src/backend/api/sse/channelmanager.py @@ -0,0 +1,33 @@ +"""Channel manager that wires `django-eventstream` to platform SSE views.""" + +from __future__ import annotations + +from django_eventstream.channelmanager import DefaultChannelManager + +from api.sse.utils import tenant_id_from_channel + + +class SSEChannelManager(DefaultChannelManager): + """Connect `django-eventstream` to the platform's SSE viewsets.""" + + def get_channels_for_request(self, request, _view_kwargs): + """Return the channels the viewset already computed on the request.""" + return getattr(request, "sse_channels", set()) + + def can_read_channel(self, user, channel): + """Re-verify tenant membership once the stream is established. + + The channel name embeds the tenant id; cross-tenant subscription + is rejected here even if the URL-level check ever has a bug. + Resource-level visibility was already enforced at connect. + """ + if user is None or not user.is_authenticated: + return False + tenant_id = tenant_id_from_channel(channel) + if tenant_id is None: + return False + return user.is_member_of_tenant(tenant_id) + + def is_channel_reliable(self, channel): + """Clients refetch canonical state from REST on reconnect""" + return False diff --git a/api/src/backend/api/sse/utils.py b/api/src/backend/api/sse/utils.py new file mode 100644 index 0000000000..e6d58a1ec1 --- /dev/null +++ b/api/src/backend/api/sse/utils.py @@ -0,0 +1,40 @@ +"""Channel-name convention shared by SSE publishers, consumers, and the +channel manager. The format is `::`. +""" + +from __future__ import annotations + +import uuid + +CHANNEL_SEPARATOR = ":" + + +def make_channel_name( + prefix: str, + tenant_id: str | uuid.UUID, + resource_id: str | uuid.UUID, +) -> str: + """Build the canonical channel name for a resource. + + Args: + prefix: Feature-owned prefix (e.g. `"lighthouse-session"`). + tenant_id: Tenant the resource belongs to. + resource_id: Resource identifier within the tenant. + """ + return CHANNEL_SEPARATOR.join((prefix, str(tenant_id), str(resource_id))) + + +def tenant_id_from_channel(channel: str) -> uuid.UUID | None: + """Return the tenant UUID embedded in *channel*, or `None` if + *channel* does not follow the platform convention. + + A `None` result MUST be treated by callers as "not authorized" or + a malformed channel cannot be safely read. + """ + segments = channel.split(CHANNEL_SEPARATOR) + if len(segments) < 3: + return None + try: + return uuid.UUID(segments[1]) + except ValueError: + return None diff --git a/api/src/backend/api/tests/test_sse.py b/api/src/backend/api/tests/test_sse.py new file mode 100644 index 0000000000..18cd688589 --- /dev/null +++ b/api/src/backend/api/tests/test_sse.py @@ -0,0 +1,101 @@ +"""Tests for the platform SSE infrastructure (``api.sse``). + +Cover the two security-critical platform pieces — the channel-name +convention (:mod:`api.sse.utils`) and the tenant gate enforced by +:class:`api.sse.channelmanager.SSEChannelManager`. The SSE authentication +class lives in :mod:`api.authentication` with the rest of the auth stack, +so its tests live in ``test_authentication.py``. Per-feature SSE endpoints +add their own tests on top of these. +""" + +import uuid +from unittest.mock import MagicMock + +import pytest + +from api.sse.channelmanager import SSEChannelManager +from api.sse.utils import make_channel_name, tenant_id_from_channel + + +class TestMakeChannel: + def test_round_trips_tenant_id(self): + tenant_id = uuid.uuid4() + channel = make_channel_name("lighthouse-session", tenant_id, uuid.uuid4()) + assert tenant_id_from_channel(channel) == tenant_id + + def test_accepts_str_arguments(self): + tenant_id = uuid.uuid4() + channel = make_channel_name("lighthouse-session", str(tenant_id), "resource-1") + assert channel == f"lighthouse-session:{tenant_id}:resource-1" + + def test_prefix_with_hyphen_is_not_split(self): + # Prefixes contain hyphens but never colons, so the tenant id is + # always the second colon-separated segment. + tenant_id = uuid.uuid4() + channel = make_channel_name("a-long-hyphenated-prefix", tenant_id, "res") + assert tenant_id_from_channel(channel) == tenant_id + + +class TestTenantIdFromChannel: + def test_returns_none_for_too_few_segments(self): + assert tenant_id_from_channel("prefix:only") is None + assert tenant_id_from_channel("garbage") is None + + def test_returns_none_for_non_uuid_tenant_segment(self): + assert tenant_id_from_channel("prefix:not-a-uuid:resource") is None + + def test_parses_valid_channel(self): + tenant_id = uuid.uuid4() + assert tenant_id_from_channel(f"prefix:{tenant_id}:resource") == tenant_id + + +@pytest.mark.django_db +class TestSSEChannelManager: + def test_member_can_read_own_tenant_channel( + self, create_test_user, tenants_fixture + ): + tenant = tenants_fixture[0] + channel = make_channel_name("lighthouse-session", tenant.id, uuid.uuid4()) + assert SSEChannelManager().can_read_channel(create_test_user, channel) + + def test_non_member_cannot_read_other_tenant_channel( + self, create_test_user, tenants_fixture + ): + # create_test_user is a member of tenant1 and tenant2 but not tenant3. + foreign_tenant = tenants_fixture[2] + channel = make_channel_name( + "lighthouse-session", foreign_tenant.id, uuid.uuid4() + ) + assert not SSEChannelManager().can_read_channel(create_test_user, channel) + + def test_anonymous_user_is_rejected(self, tenants_fixture): + channel = make_channel_name( + "lighthouse-session", tenants_fixture[0].id, uuid.uuid4() + ) + assert not SSEChannelManager().can_read_channel(None, channel) + + anon = MagicMock(is_authenticated=False) + assert not SSEChannelManager().can_read_channel(anon, channel) + + def test_malformed_channel_is_rejected(self, create_test_user, tenants_fixture): + assert not SSEChannelManager().can_read_channel(create_test_user, "garbage") + + def test_get_channels_for_request_reads_stashed_set(self): + request = MagicMock() + request.sse_channels = {"prefix:tenant:resource"} + manager = SSEChannelManager() + assert manager.get_channels_for_request(request, {}) == { + "prefix:tenant:resource" + } + + def test_get_channels_for_request_defaults_to_empty(self): + # A request that never went through BaseSSEViewSet.list has no + # sse_channels attribute; the manager must not raise. + request = object() + assert SSEChannelManager().get_channels_for_request(request, {}) == set() + + def test_channel_is_not_reliable(self): + # v1 ships without server-side replay storage. + assert ( + SSEChannelManager().is_channel_reliable("prefix:tenant:resource") is False + ) diff --git a/api/src/backend/config/django/base.py b/api/src/backend/config/django/base.py index 38cf047ac2..3308561b48 100644 --- a/api/src/backend/config/django/base.py +++ b/api/src/backend/config/django/base.py @@ -3,6 +3,7 @@ from datetime import timedelta from config.custom_logging import LOGGING # noqa from config.env import BASE_DIR, env # noqa from config.settings.celery import * # noqa +from config.settings.eventstream import * # noqa from config.settings.partitions import * # noqa from config.settings.sentry import * # noqa from config.settings.social_login import * # noqa @@ -44,6 +45,7 @@ INSTALLED_APPS = [ "dj_rest_auth.registration", "rest_framework.authtoken", "drf_simple_apikey", + "django_eventstream", ] MIDDLEWARE = [ @@ -136,6 +138,7 @@ SPECTACULAR_SETTINGS = { } WSGI_APPLICATION = "config.wsgi.application" +ASGI_APPLICATION = "config.asgi.application" DJANGO_GUID = { "GUID_HEADER_NAME": "Transaction-ID", diff --git a/api/src/backend/config/settings/eventstream.py b/api/src/backend/config/settings/eventstream.py new file mode 100644 index 0000000000..470062050b --- /dev/null +++ b/api/src/backend/config/settings/eventstream.py @@ -0,0 +1,41 @@ +"""Server-Sent Events (SSE) configuration. + +Wires django-eventstream into the platform: Valkey Pub/Sub backend on a +dedicated DB (separate from the Celery broker), the platform channel +manager, and headers that match the existing CORS allowlist. +""" + +from config.env import env +from config.settings.celery import ( + VALKEY_HOST, + VALKEY_PASSWORD, + VALKEY_PORT, + VALKEY_SCHEME, + VALKEY_USERNAME, +) + +# Dedicated Valkey DB for the SSE Pub/Sub bus. Kept distinct from the +# Celery broker DB so a noisy broker can't shoulder out streaming +# traffic on the same keyspace. +EVENTSTREAM_VALKEY_DB = env.int("EVENTSTREAM_VALKEY_DB", default=2) + +EVENTSTREAM_REDIS: dict = { + "host": VALKEY_HOST, + "port": int(VALKEY_PORT), + "db": EVENTSTREAM_VALKEY_DB, +} +if VALKEY_PASSWORD: + EVENTSTREAM_REDIS["password"] = VALKEY_PASSWORD +if VALKEY_USERNAME: + EVENTSTREAM_REDIS["username"] = VALKEY_USERNAME +if VALKEY_SCHEME == "rediss": + EVENTSTREAM_REDIS["ssl"] = True + +# Platform channel manager — performs the per-feature authorization and +# rewrites the placeholder channel from the URL into the canonical +# tenant-scoped channel name. See ``api.sse.channelmanager``. +EVENTSTREAM_CHANNELMANAGER_CLASS = "api.sse.channelmanager.SSEChannelManager" + +# Headers a browser EventSource may legitimately send. Keep tight; the +# stream itself reads no body, so no permissive defaults. +EVENTSTREAM_ALLOW_HEADERS = "Cache-Control, Last-Event-ID"