fix(auth): gate cache deny on non-null user_id to prevent cross-user pollution

api/oss/src/middlewares/auth.py

@@ -767,13 +767,14 @@ def _deny(stage: str, reason: str, exc: Optional[Exception] = None):
 
     except UnauthorizedException as exc:
         _deny("catch_unauthorized", "UnauthorizedException propagated", exc)
-        await set_cache(
-            project_id=query_project_id,
-            user_id=user_id,
-            namespace="verify_bearer_token",
-            key=cache_key,
-            value={"deny": True},
-        )
+        if user_id is not None:
+            await set_cache(
+                project_id=query_project_id,
+                user_id=user_id,
+                namespace="verify_bearer_token",
+                key=cache_key,
+                value={"deny": True},
+            )
 
         raise exc
 
@@ -784,13 +785,14 @@ def _deny(stage: str, reason: str, exc: Optional[Exception] = None):
 
     except Exception as exc:  # pylint: disable=bare-except
         _deny("catch_all", "unexpected exception", exc)
-        await set_cache(
-            project_id=query_project_id,
-            user_id=user_id,
-            namespace="verify_bearer_token",
-            key=cache_key,
-            value={"deny": True},
-        )
+        if user_id is not None:
+            await set_cache(
+                project_id=query_project_id,
+                user_id=user_id,
+                namespace="verify_bearer_token",
+                key=cache_key,
+                value={"deny": True},
+            )
 
         raise UnauthorizedException() from exc
 

api/oss/tests/pytest/unit/middlewares/test_cache_deny.py

@@ -0,0 +1,88 @@
+import pytest
+from unittest.mock import AsyncMock, patch, MagicMock
+
+
+@pytest.mark.asyncio
+async def test_cache_deny_not_written_when_user_id_is_none():
+    """Verify that set_cache deny entries are not written when user_id is None.
+
+    When user_id is None (anonymous requests), writing a deny entry to the cache
+    produces a generic key shared by all anonymous users. One user's auth failure
+    must not deny all other anonymous users for the cache TTL.
+    """
+    from oss.src.middlewares.auth import auth_user
+
+    mock_request = MagicMock()
+    mock_request.url.path = "/test"
+    mock_request.method = "GET"
+    mock_request.headers = {}
+
+    mock_session = AsyncMock()
+    mock_session.get_user_id.return_value = None
+
+    set_cache_calls = []
+
+    async def capture_set_cache(**kwargs):
+        set_cache_calls.append(kwargs)
+        return True
+
+    with patch("oss.src.middlewares.auth.get_session", return_value=mock_session), \
+         patch("oss.src.middlewares.auth.set_cache", side_effect=capture_set_cache) as mock_set_cache, \
+         patch("oss.src.middlewares.auth.get_cache", return_value=None), \
+         pytest.raises(Exception):
+        await auth_user(
+            request=mock_request,
+            bearer_token="",
+            query_project_id="test-project-id",
+            query_workspace_id="test-workspace-id",
+        )
+
+    # set_cache should NOT have been called with a deny value when user_id is None
+    deny_calls = [c for c in set_cache_calls if c.get("value") == {"deny": True}]
+    assert len(deny_calls) == 0, (
+        f"set_cache was called with deny value {len(deny_calls)} time(s) "
+        f"when user_id is None. This would pollute the shared anonymous cache."
+    )
+
+
+@pytest.mark.asyncio
+async def test_cache_deny_written_when_user_id_is_set():
+    """Verify that set_cache deny entries ARE written when user_id is set.
+
+    When user_id is known, caching the deny is correct — it prevents repeated
+    failing lookups for the same specific user.
+    """
+    from oss.src.middlewares.auth import auth_user
+
+    mock_request = MagicMock()
+    mock_request.url.path = "/test"
+    mock_request.method = "GET"
+    mock_request.headers = {}
+
+    mock_session = AsyncMock()
+    mock_session.get_user_id.return_value = "user-123"
+
+    set_cache_calls = []
+
+    async def capture_set_cache(**kwargs):
+        set_cache_calls.append(kwargs)
+        return True
+
+    with patch("oss.src.middlewares.auth.get_session", return_value=mock_session), \
+         patch("oss.src.middlewares.auth.set_cache", side_effect=capture_set_cache), \
+         patch("oss.src.middlewares.auth.get_cache", return_value=None), \
+         patch("oss.src.middlewares.auth.get_supertokens_user_by_id", side_effect=Exception("timeout")), \
+         pytest.raises(Exception):
+        await auth_user(
+            request=mock_request,
+            bearer_token="",
+            query_project_id="test-project-id",
+            query_workspace_id="test-workspace-id",
+        )
+
+    # set_cache SHOULD have been called with a deny value when user_id is set
+    deny_calls = [c for c in set_cache_calls if c.get("value") == {"deny": True}]
+    assert len(deny_calls) >= 1, (
+        "set_cache was not called with deny value when user_id was set. "
+        "Auth failures for known users should be cached."
+    )