feat(api): add Server-Sent Events infrastructure

Add the platform SSE layer that wires django-eventstream into the API:

- BaseSSEViewSet: a base viewset features subclass to expose an SSE
  endpoint, reusing the regular DRF stack (auth, RBAC permissions, tenant
  transaction) and delegating the stream to django-eventstream.
- SSEChannelManager: resolves the channel set off the request and enforces
  a tenant gate by parsing the tenant id embedded in the channel name.
- make_channel_name/tenant_id_from_channel: the single source of truth for
  the <prefix>:<tenant_id>:<resource_id> channel format.
- eventstream settings: Valkey Pub/Sub backend on a dedicated DB, the
  channel manager, and allowed headers; registered in Django settings.

No endpoint streams over SSE yet; this is the reusable base.
This commit is contained in:
Rubén De la Torre Vico committed 2026-06-11 16:19:41 +02:00
1 parent 92c270ffc4
commit 6c46dbce8a
7 files changed
+277

No files matched your search

+13
View File
@@ -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"]
+46
View File
@@ -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)
+33
View File
@@ -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
+40
View File
@@ -0,0 +1,40 @@
"""Channel-name convention shared by SSE publishers, consumers, and the
channel manager. The format is `<prefix>:<tenant_id>:<resource_id>`.
"""
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
+101
View File
@@ -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
)
+3
View File
@@ -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",
@@ -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"