A2A: refuse to serve when no transport credential is configured

s16code/core/a2a/server.py

@@ -91,6 +91,11 @@ def close(self) -> None:
         self.db.close()
 
 
+def _matches(presented: str, accepted: set[str]) -> bool:
+    """Constant-time membership. `presented in accepted` leaks length and prefix."""
+    return bool(presented) and any(hmac.compare_digest(presented, item) for item in accepted)
+
+
 class A2ADemoServer:
     """Local reference adapter, not a replacement for the official A2A SDK.
 
@@ -101,13 +106,18 @@ class A2ADemoServer:
     def __init__(self, card: dict[str, Any], *, push_callback: PushCallback | None = None,
                  task_handler: TaskHandler | None = None, task_db: str | Path = ":memory:",
                  bearer_tokens: set[str] | None = None, api_keys: set[str] | None = None,
+                 allow_anonymous: bool = False,
                  push_signing_secret: str | None = None, push_http: httpx.AsyncClient | None = None,
                  push_attempts: int = 4, push_backoff: float = .05) -> None:
         validate_agent_card(card)
         self.card, self.push_callback, self.task_handler = card, push_callback, task_handler
         self.store = DurableTaskStore(task_db)
         self.tasks = self.store.load_all()
         self.bearer_tokens, self.api_keys = bearer_tokens or set(), api_keys or set()
+        # Serving anonymously is a decision, not a default. Without this the
+        # transport check below was skipped whenever no credential happened to be
+        # configured, which is the state a fresh checkout is in.
+        self.allow_anonymous = allow_anonymous
         self.push_signing_secret, self.push_http = push_signing_secret, push_http
         self.push_attempts, self.push_backoff = push_attempts, push_backoff
         self._jobs: dict[str, asyncio.Task[None]] = {}
@@ -128,11 +138,21 @@ async def start(self) -> None:
                 self._jobs[task.id] = asyncio.create_task(self._complete_later(task, 0))
 
     async def rpc(self, request: Request):
-        if self.bearer_tokens or self.api_keys:
+        if not (self.bearer_tokens or self.api_keys):
+            if not self.allow_anonymous:
+                # Refuse rather than serve everybody. `message/send` creates and
+                # runs a task, so an unconfigured transport is not a missing
+                # feature: it hands any caller the agent's time and budget.
+                return JSONResponse(
+                    self._error(None, -32003,
+                                "S16_A2A_BEARER_TOKENS or S16_A2A_API_KEYS is not configured; "
+                                "the A2A transport refuses to serve without it"),
+                    status_code=503)
+        else:
             bearer = request.headers.get("authorization", "")
             bearer = bearer[7:] if bearer.lower().startswith("bearer ") else ""
             api_key = request.headers.get("x-api-key", "")
-            if not ((bearer and bearer in self.bearer_tokens) or (api_key and api_key in self.api_keys)):
+            if not (_matches(bearer, self.bearer_tokens) or _matches(api_key, self.api_keys)):
                 return JSONResponse(self._error(None, -32003, "A2A transport authentication required"), status_code=401,
                                     headers={"WWW-Authenticate": "Bearer"})
         body = await request.json()

s16code/core/a2a/tests/test_adapter.py

@@ -37,7 +37,10 @@ async def setup(keys):
     async def pushed(event):
         received.append(event)
 
-    server = A2ADemoServer(sign_card(card(), private, kid="local-1"), push_callback=pushed)
+    server = A2ADemoServer(sign_card(card(), private, kid="local-1"), push_callback=pushed,
+                           # This fixture drives the transport without a credential;
+                           # serving anonymously is now something a caller must ask for.
+                           allow_anonymous=True)
     http = httpx.AsyncClient(transport=httpx.ASGITransport(app=server.app), base_url="http://a2a.test")
     trust = AgentCardTrustPolicy(key_resolver=lambda kid: public if kid == "local-1" else None)
     yield server, A2AClient(http, trust), received

s16code/core/a2a/tests/test_hardening.py

@@ -100,6 +100,28 @@ async def test_jsonrpc_transport_requires_bearer_or_api_key():
         assert (await http.post("/a2a", json=payload, headers={"X-API-Key": "key-good"})).status_code == 200
 
 
+@pytest.mark.asyncio
+async def test_jsonrpc_transport_refuses_to_serve_with_no_credential_configured():
+    """An unconfigured transport must refuse, not serve everybody.
+
+    The check above proves the gate works once tokens exist. Nothing covered the
+    state a fresh checkout is in -- no tokens at all -- where the guard was
+    skipped entirely and `message/send` ran work for any caller.
+    """
+    server = A2ADemoServer(card())
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=server.app), base_url="http://a2a.test") as http:
+        payload = {"jsonrpc": "2.0", "id": "1", "method": "message/send", "params": params()}
+        refused = await http.post("/a2a", json=payload)
+        assert refused.status_code == 503
+        assert "S16_A2A_BEARER_TOKENS" in refused.text
+
+    # Serving anonymously stays possible, but only when it is asked for.
+    opened = A2ADemoServer(card(), allow_anonymous=True)
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=opened.app), base_url="http://a2a.test") as http:
+        payload = {"jsonrpc": "2.0", "id": "1", "method": "message/send", "params": params()}
+        assert (await http.post("/a2a", json=payload)).status_code == 200
+
+
 def test_official_card_schema_and_version_negotiation():
     validate_agent_card(card(("1.0", "1.1")))
     assert negotiate_interface(card(("1.0", "1.1")), versions=("1.1", "1.0"))["protocolVersion"] == "1.1"