feat(api): add SSE authentication with access_token fallback

Browser EventSource cannot set the Authorization header, so add an
SSEAuthentication class that extends the standard JWT/API-key stack with
an ?access_token=<jwt> query-parameter fallback (RFC 6750 section 2.3),
consulted only when no Authorization header is present. The query path
accepts a JWT only; API keys remain header-only.
This commit is contained in:
Rubén De la Torre Vico committed 2026-06-11 16:17:26 +02:00
1 parent a55f276965
commit 92c270ffc4
2 files changed
+73 -2

No files matched your search

+28
View File
@@ -93,3 +93,31 @@ class CombinedJWTOrAPIKeyAuthentication(BaseAuthentication):
# Default fallback
return self.jwt_auth.authenticate(request)
class SSEAuthentication(CombinedJWTOrAPIKeyAuthentication):
"""JWT/API-Key auth that also accepts `?access_token=<jwt>`.
Browser `EventSource` is the only widely available SSE client API
and it cannot set the `Authorization` header (its constructor takes
only a URL and `withCredentials`). To keep browser SSE clients on
the same auth stack as the rest of the API, SSE endpoints additionally
accept a JWT via the `?access_token=<jwt>` query parameter — the
standard parameter name defined in RFC 6750 Section 2.3 for bearer tokens.
"""
def authenticate(self, request: Request):
auth_header = request.headers.get("Authorization", "")
if auth_header:
return super().authenticate(request)
raw_token = request.query_params.get("access_token")
if not raw_token:
# No header and no query token — let the default path raise
# the canonical AuthenticationFailed via the parent class.
return super().authenticate(request)
jwt_auth = JWTAuthentication()
validated_token = jwt_auth.get_validated_token(raw_token)
user = jwt_auth.get_user(validated_token)
return user, validated_token
@@ -1,13 +1,13 @@
import time
from datetime import datetime, timedelta, timezone
from unittest.mock import patch
from unittest.mock import MagicMock, patch
from uuid import uuid4
import pytest
from django.test import RequestFactory
from rest_framework.exceptions import AuthenticationFailed
from api.authentication import TenantAPIKeyAuthentication
from api.authentication import SSEAuthentication, TenantAPIKeyAuthentication
from api.db_router import MainRouter
from api.models import TenantAPIKey
@@ -382,3 +382,46 @@ class TestTenantAPIKeyAuthentication:
auth_backend.authenticate(request)
assert str(exc_info.value.detail) == "API Key has already expired."
class TestSSEAuthentication:
"""`SSEAuthentication` adds an `?access_token=<jwt>` fallback for
browser `EventSource` clients while keeping the standard
`Authorization` header as the authoritative source."""
def test_header_present_delegates_to_super(self):
request = MagicMock()
request.headers = {"Authorization": "Bearer header-token"}
with patch.object(
SSEAuthentication.__bases__[0], "authenticate", return_value=("user", "tok")
) as super_auth:
result = SSEAuthentication().authenticate(request)
super_auth.assert_called_once_with(request)
assert result == ("user", "tok")
def test_no_header_no_query_token_delegates_to_super(self):
request = MagicMock()
request.headers = {}
request.query_params = {}
with patch.object(
SSEAuthentication.__bases__[0], "authenticate", return_value=None
) as super_auth:
result = SSEAuthentication().authenticate(request)
super_auth.assert_called_once_with(request)
assert result is None
def test_query_token_used_only_as_fallback(self):
request = MagicMock()
request.headers = {}
request.query_params = {"access_token": "query-jwt"}
jwt_instance = MagicMock()
jwt_instance.get_validated_token.return_value = "validated"
jwt_instance.get_user.return_value = "query-user"
with patch("api.authentication.JWTAuthentication", return_value=jwt_instance):
user, token = SSEAuthentication().authenticate(request)
jwt_instance.get_validated_token.assert_called_once_with("query-jwt")
assert user == "query-user"
assert token == "validated"