From 92c270ffc455211e4307839dae574ab404744769 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Rub=C3=A9n=20De=20la=20Torre=20Vico?= <ruben@prowler.com>
Date: Thu, 11 Jun 2026 16:17:26 +0200
Subject: [PATCH] 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.
---
 api/src/backend/api/authentication.py         | 28 +++++++++++
 .../backend/api/tests/test_authentication.py  | 47 ++++++++++++++++++-
 2 files changed, 73 insertions(+), 2 deletions(-)

diff --git a/api/src/backend/api/authentication.py b/api/src/backend/api/authentication.py
index 499e290bb7..afca6ecadd 100644
--- a/api/src/backend/api/authentication.py
+++ b/api/src/backend/api/authentication.py
@@ -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
diff --git a/api/src/backend/api/tests/test_authentication.py b/api/src/backend/api/tests/test_authentication.py
index 6745c36e91..3af362d210 100644
--- a/api/src/backend/api/tests/test_authentication.py
+++ b/api/src/backend/api/tests/test_authentication.py
@@ -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"