test(mcp): cover per-request auth and workspace isolation

surfsense_mcp/tests/test_auth_headers.py

@@ -0,0 +1,40 @@
+"""API key extraction from request headers: Bearer, fallback, and rejection."""
+
+from __future__ import annotations
+
+from starlette.datastructures import Headers
+
+from surfsense_mcp.core.auth.headers import extract_api_key
+
+
+def _headers(**pairs: str) -> Headers:
+    return Headers(pairs)
+
+
+def test_reads_bearer_token():
+    assert extract_api_key(_headers(authorization="Bearer ss_pat_abc")) == "ss_pat_abc"
+
+
+def test_bearer_scheme_is_case_insensitive():
+    assert extract_api_key(_headers(authorization="bearer ss_pat_abc")) == "ss_pat_abc"
+
+
+def test_falls_back_to_x_api_key():
+    assert extract_api_key(Headers({"x-api-key": "ss_pat_xyz"})) == "ss_pat_xyz"
+
+
+def test_bearer_wins_over_fallback():
+    headers = Headers({"authorization": "Bearer primary", "x-api-key": "secondary"})
+    assert extract_api_key(headers) == "primary"
+
+
+def test_missing_headers_return_none():
+    assert extract_api_key(_headers()) is None
+
+
+def test_empty_bearer_is_rejected():
+    assert extract_api_key(_headers(authorization="Bearer   ")) is None
+
+
+def test_non_bearer_authorization_is_ignored():
+    assert extract_api_key(_headers(authorization="Basic abc123")) is None

surfsense_mcp/tests/test_client_errors.py

@@ -15,7 +15,7 @@ def _response(status: int, **kwargs) -> httpx.Response:
 
 def test_explains_401_with_token_hint():
     message = SurfSenseClient._explain_failure(_response(401, json={"detail": "bad"}))
-    assert "SURFSENSE_API_KEY" in message
+    assert "API key" in message
     assert "bad" in message
 
 

surfsense_mcp/tests/test_client_params.py

@@ -25,7 +25,9 @@ def _capture(client: SurfSenseClient) -> dict:
 
 
 def test_none_params_are_dropped():
-    client = SurfSenseClient(api_base="http://test/api/v1", api_key="ss_pat_x", timeout=5)
+    client = SurfSenseClient(
+        api_base="http://test/api/v1", timeout=5, fallback_api_key="ss_pat_x"
+    )
     seen = _capture(client)
     asyncio.run(
         client.request(

surfsense_mcp/tests/test_request_auth.py

@@ -0,0 +1,100 @@
+"""Per-request key resolution and the Authorization header the backend receives.
+
+Covers the security-critical behaviors: the per-request key wins over the env
+fallback, the fallback covers stdio, a missing key is refused, and concurrent
+callers never see each other's key.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import httpx
+import pytest
+
+from surfsense_mcp.core.auth import identity
+from surfsense_mcp.core.client import SurfSenseClient
+from surfsense_mcp.core.errors import ToolError
+
+
+def _client_recording_auth(seen: dict, *, fallback: str | None) -> SurfSenseClient:
+    async def handler(request: httpx.Request) -> httpx.Response:
+        seen["authorization"] = request.headers.get("authorization")
+        return httpx.Response(200, json={"ok": True})
+
+    client = SurfSenseClient(
+        api_base="http://test/api/v1", timeout=5, fallback_api_key=fallback
+    )
+    client._http = httpx.AsyncClient(
+        base_url="http://test/api/v1", transport=httpx.MockTransport(handler)
+    )
+    return client
+
+
+async def _get(client: SurfSenseClient) -> None:
+    await client.request("GET", "/workspaces")
+
+
+def test_request_key_is_sent_as_bearer():
+    seen: dict = {}
+    client = _client_recording_auth(seen, fallback=None)
+
+    async def run() -> None:
+        token = identity.bind_api_key("ss_pat_request")
+        try:
+            await _get(client)
+        finally:
+            identity.unbind_api_key(token)
+
+    asyncio.run(run())
+    assert seen["authorization"] == "Bearer ss_pat_request"
+
+
+def test_request_key_overrides_env_fallback():
+    seen: dict = {}
+    client = _client_recording_auth(seen, fallback="ss_pat_env")
+
+    async def run() -> None:
+        token = identity.bind_api_key("ss_pat_request")
+        try:
+            await _get(client)
+        finally:
+            identity.unbind_api_key(token)
+
+    asyncio.run(run())
+    assert seen["authorization"] == "Bearer ss_pat_request"
+
+
+def test_env_fallback_used_without_request_key():
+    seen: dict = {}
+    client = _client_recording_auth(seen, fallback="ss_pat_env")
+    asyncio.run(_get(client))
+    assert seen["authorization"] == "Bearer ss_pat_env"
+
+
+def test_missing_key_is_refused():
+    client = _client_recording_auth({}, fallback=None)
+    with pytest.raises(ToolError):
+        asyncio.run(_get(client))
+
+
+def test_concurrent_callers_do_not_leak_keys():
+    seen_by_caller: dict[str, str | None] = {}
+
+    async def call_as(key: str) -> None:
+        # Each caller runs in its own task, so the contextvar is isolated.
+        recorded: dict = {}
+        client = _client_recording_auth(recorded, fallback=None)
+        token = identity.bind_api_key(key)
+        try:
+            await _get(client)
+        finally:
+            identity.unbind_api_key(token)
+        seen_by_caller[key] = recorded["authorization"]
+
+    async def run() -> None:
+        await asyncio.gather(call_as("ss_pat_A"), call_as("ss_pat_B"))
+
+    asyncio.run(run())
+    assert seen_by_caller["ss_pat_A"] == "Bearer ss_pat_A"
+    assert seen_by_caller["ss_pat_B"] == "Bearer ss_pat_B"

surfsense_mcp/tests/test_workspace_context.py

@@ -6,6 +6,7 @@ import asyncio
 
 import pytest
 
+from surfsense_mcp.core.auth import identity
 from surfsense_mcp.core.errors import ToolError
 from surfsense_mcp.core.workspace_context import WorkspaceContext
 
@@ -96,3 +97,30 @@ def test_resolution_is_remembered_as_active():
     assert ctx.active is not None and ctx.active.id == 2
     # a later default call reuses the active selection without re-choosing
     assert asyncio.run(ctx.resolve(None)).id == 2
+
+
+def test_active_workspace_is_isolated_per_identity():
+    ctx = _context(_rows(("A", 1), ("B", 2)))
+
+    async def select_as(key: str, reference: str) -> None:
+        token = identity.bind_api_key(key)
+        try:
+            await ctx.resolve(reference)
+        finally:
+            identity.unbind_api_key(token)
+
+    async def active_for(key: str) -> int | None:
+        token = identity.bind_api_key(key)
+        try:
+            return ctx.active.id if ctx.active else None
+        finally:
+            identity.unbind_api_key(token)
+
+    asyncio.run(select_as("ss_pat_A", "A"))
+    asyncio.run(select_as("ss_pat_B", "B"))
+
+    # Each caller keeps its own selection; no bleed across identities.
+    assert asyncio.run(active_for("ss_pat_A")) == 1
+    assert asyncio.run(active_for("ss_pat_B")) == 2
+    # An unknown caller has no active selection.
+    assert asyncio.run(active_for("ss_pat_C")) is None