diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f6d2c767..922ad7b4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ ## Unreleased +### Security fixes + +* **auth:** MCP and A2A reject conflicting accepted credentials (401), + including repeated headers; identical decoded duplicates are accepted. + Empty or malformed accepted headers cannot bypass auth or fall back to an + alias. Auth diagnostics include reason codes, never credential values or + validator exception text. During migration, send the same token in every + accepted carrier, or send only one. Refs #1305. + ### Bug Fixes * **router:** Optional legacy creative and compact lifecycle tools now follow diff --git a/src/adcp/server/auth.py b/src/adcp/server/auth.py index c29d98c3c..f659224aa 100644 --- a/src/adcp/server/auth.py +++ b/src/adcp/server/auth.py @@ -116,6 +116,50 @@ def _parse_bearer_header(header: str) -> str | None: return token.strip() or None +@dataclass(frozen=True) +class _RejectedCredential: + reason: Literal["ambiguous_header", "wrong_scheme"] + + +def _resolve_bearer_headers( + headers: Sequence[tuple[bytes, bytes]], + aliases: Sequence[str], + alias_prefix_required: bool, +) -> str | _RejectedCredential | None: + """Inspect every accepted carrier, including repeated HTTP headers. + + Absence alone permits anonymous/fallback paths. Empty or malformed + accepted headers fail closed, even alongside a valid credential. + Identical decoded tokens are accepted; different tokens are ambiguous. + """ + tokens: list[str] = [] + alias_names = set(aliases) + for raw_name, raw_value in headers: + name = raw_name.decode("latin-1").lower() + if name != "authorization" and name not in alias_names: + continue + value = raw_value.decode("latin-1") + token = ( + _parse_bearer_header(value) + if name == "authorization" or alias_prefix_required + else value.strip() or None + ) + if token is None: + return _RejectedCredential("wrong_scheme") + tokens.append(token) + if not tokens: + return None + digest = hashlib.sha256(tokens[0].encode("utf-8")).digest() + if any( + not hmac.compare_digest(digest, hashlib.sha256(token.encode("utf-8")).digest()) + for token in tokens[1:] + ): + return _RejectedCredential("ambiguous_header") + if len(tokens) > 1: + logger.debug("identical authentication credentials accepted") + return tokens[0] + + if TYPE_CHECKING: from starlette.requests import Request @@ -341,9 +385,9 @@ class BearerTokenAuthMiddleware(BaseHTTPMiddleware): response body. Default is ``{"error": "unauthenticated"}``. :param legacy_header_aliases: Optional list of legacy header names to accept in addition to the spec-canonical ``Authorization: - Bearer``. Resolution order on every request: ``Authorization: - Bearer `` first; if absent, each alias in order; first - non-empty wins. The aliases path is for adopters mid-migration + Bearer``. Every accepted carrier is checked. Different tokens + are rejected; identical decoded duplicates are accepted. + The aliases path is for adopters mid-migration from a custom header (e.g. ``x-adcp-auth``) — both work simultaneously so no flag-day cutover is needed. **The spec-canonical header is always accepted; aliases are @@ -475,7 +519,15 @@ async def dispatch(self, request: Request, call_next: Any) -> Any: tenant_token = None metadata_token = None try: - bearer = self._extract_bearer(request) + credential = _resolve_bearer_headers( + request.scope.get("headers", ()), + self._alias_header_names, + self._alias_prefix_required, + ) + if isinstance(credential, _RejectedCredential): + logger.info("mcp auth rejected", extra={"reason": credential.reason}) + return self._unauthenticated(signature_error) + bearer = credential if is_discovery and not bearer: principal_token = current_principal.set(None) tenant_token = current_tenant.set(None) @@ -519,7 +571,7 @@ async def dispatch(self, request: Request, call_next: Any) -> Any: # Validator failure must not leak stack info to the caller. # Fail closed — a buggy validator is an auth failure, not a # 500. Logged for operators. - logger.exception("token validator raised") + logger.error("mcp auth rejected", extra={"reason": "validator_error"}) return self._unauthenticated(signature_error) if principal is None: @@ -554,38 +606,6 @@ async def dispatch(self, request: Request, call_next: Any) -> Any: if metadata_token is not None: current_principal_metadata.reset(metadata_token) - def _extract_bearer(self, request: Request) -> str | None: - """Resolve the token from incoming headers. - - Per RFC 6750 §2.1 the canonical carrier is ``Authorization: - Bearer ``; check that first. If absent, walk the - configured ``legacy_header_aliases`` in order — first non-empty - wins. Legacy aliases carry raw tokens (no scheme prefix) unless - ``legacy_aliases_bearer_prefix_required=True``. Both paths - coexist so adopters mid-migration can move clients from a - custom header to ``Authorization: Bearer`` without a flag day - (#720). - """ - # 1. Spec-canonical first. - canonical = request.headers.get("authorization", "") - bearer = _parse_bearer_header(canonical) - if bearer: - return bearer - - # 2. Legacy aliases — additive opt-in. - for alias in self._alias_header_names: - raw = request.headers.get(alias, "") - if not raw: - continue - if self._alias_prefix_required: - token = _parse_bearer_header(raw) - else: - token = raw.strip() or None - if token: - return token - - return None - def is_discovery_request(self, method: str | None, tool: str | None) -> bool: """True when the request should bypass auth. @@ -981,7 +1001,8 @@ class BearerTokenAuth: header carrying a raw token (no scheme prefix) before the spec settled. Sellers with deployed clients that can't be updated opt in additively — ``Authorization: Bearer`` is still accepted, the - alias is consulted only when the canonical header is absent:: + alias is accepted alongside the canonical header. Different tokens + are rejected; identical decoded duplicates are accepted:: # Recommended new-shape (#720). Accepts both wire carriers. BearerTokenAuth( @@ -1062,8 +1083,8 @@ class BearerTokenAuth: # NEW (#720) — additive legacy aliases. ``Authorization: Bearer`` # is ALWAYS accepted regardless of these fields; this is purely # additive opt-in for adopters mid-migration from custom headers. - # Resolution order on each request: ``Authorization: Bearer`` - # first; if absent, each alias in order, first non-empty wins. + # All accepted carriers must decode to the same token. Empty/malformed + # carriers and conflicting tokens fail closed on both transports. # # Pick cross-leg ``legacy_header_aliases`` when both MCP and A2A # adopters send the same custom header (most common case during @@ -1471,13 +1492,6 @@ def resolved_a2a_discovery_skills(self) -> frozenset[str] | None: ) -class _AmbiguousA2ACredential: - """Sentinel type for duplicate accepted authentication carriers.""" - - -_AMBIGUOUS_A2A_CREDENTIAL = _AmbiguousA2ACredential() - - class A2ABearerAuthMiddleware: """Pure-ASGI middleware that gates A2A JSON-RPC on a bearer token. @@ -1552,25 +1566,10 @@ def _has_bearer(self, scope: Any) -> bool: """ return self._resolve_credential(scope) is not None - def _resolve_credential( - self, scope: Any - ) -> tuple[bytes, bool] | _AmbiguousA2ACredential | None: - """Resolve one accepted carrier and flag duplicate credentials.""" - - accepted: list[tuple[bytes, bool]] = [] - aliases = set(self._alias_header_names) - for raw_name, raw_value in scope.get("headers", ()): - if not raw_value.strip(): - continue - name = raw_name.decode("latin-1").lower() - if name == "authorization": - accepted.append((raw_value, True)) - elif name in aliases: - accepted.append((raw_value, self._alias_prefix_required)) - - if len(accepted) > 1: - return _AMBIGUOUS_A2A_CREDENTIAL - return accepted[0] if accepted else None + def _resolve_credential(self, scope: Any) -> str | _RejectedCredential | None: + return _resolve_bearer_headers( + scope.get("headers", ()), self._alias_header_names, self._alias_prefix_required + ) async def __call__(self, scope: Any, receive: Any, send: Any) -> None: # Lifespan + websocket pass through unchanged. Auth applies to @@ -1725,33 +1724,18 @@ def _authenticate_scope(self, scope: Any) -> Principal | None: so SOC dashboards can detect scanning without bloating logs. """ credential = self._resolve_credential(scope) - if isinstance(credential, _AmbiguousA2ACredential): - logger.info("a2a auth rejected", extra={"reason": "ambiguous_header"}) + if isinstance(credential, _RejectedCredential): + logger.info("a2a auth rejected", extra={"reason": credential.reason}) return None if credential is None: logger.info("a2a auth rejected", extra={"reason": "missing_header"}) return None - raw_value, prefix_required = credential - - try: - raw_header = raw_value.decode("latin-1") - except UnicodeDecodeError: - logger.info("a2a auth rejected", extra={"reason": "header_decode"}) - return None - - if prefix_required: - bearer = _parse_bearer_header(raw_header) - else: - stripped = raw_header.strip() - bearer = stripped or None - if not bearer: - logger.info("a2a auth rejected", extra={"reason": "wrong_scheme"}) - return None + bearer = credential try: raw = self._config.validate_token(bearer) except Exception: - logger.exception("token validator raised on A2A request") + logger.error("a2a auth rejected", extra={"reason": "validator_error"}) return None if inspect.isawaitable(raw): diff --git a/tests/test_auth_credential_parity.py b/tests/test_auth_credential_parity.py new file mode 100644 index 000000000..b812de475 --- /dev/null +++ b/tests/test_auth_credential_parity.py @@ -0,0 +1,234 @@ +"""Authentication carrier policy is identical on the MCP and A2A boundaries.""" + +from __future__ import annotations + +import asyncio +import logging +from contextlib import nullcontext +from typing import Any + +import httpx +import pytest +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import JSONResponse +from starlette.routing import Route + +from adcp.server.auth import BearerTokenAuth, Principal, current_principal +from adcp.server.serve import _wrap_a2a_with_auth, _wrap_mcp_with_auth + + +def middleware_app(leg: str, config: BearerTokenAuth) -> Any: + async def endpoint(request: Request) -> JSONResponse: + await asyncio.sleep(0) + return JSONResponse({"principal": current_principal.get(), "body": await request.json()}) + + inner = Starlette(routes=[Route("/", endpoint, methods=["POST"])]) + return (_wrap_mcp_with_auth if leg == "mcp" else _wrap_a2a_with_auth)(inner, config) + + +CALL = {"method": "tools/call", "params": {"name": "get_products"}} + + +@pytest.mark.asyncio +@pytest.mark.parametrize("leg", ["mcp", "a2a"]) +@pytest.mark.parametrize("mode", ["required", "anonymous", "discovery"]) +@pytest.mark.parametrize("custom", [False, True]) +@pytest.mark.parametrize( + "headers,status", + [ + ([("Authorization", "Bearer secret-a")], 200), + ([("x-legacy", "secret-a")], 200), + ([("Authorization", "Bearer secret-a"), ("x-legacy", "secret-a")], 200), + ([("Authorization", "Bearer secret-a"), ("Authorization", "bearer\tsecret-a")], 200), + ([("x-legacy", "secret-a"), ("x-legacy", " secret-a ")], 200), + ([("Authorization", "Bearer secret-a"), ("x-legacy", "secret-b")], 401), + ([("Authorization", "Bearer secret-a"), ("Authorization", "Bearer secret-b")], 401), + ([("x-legacy", "secret-a"), ("x-legacy", "secret-b")], 401), + ([("Authorization", "")], 401), + ([("x-legacy", "\t ")], 401), + ([("Authorization", "Basic secret-a"), ("x-legacy", "secret-a")], 401), + ([("Authorization", ""), ("x-legacy", "secret-a")], 401), + ([("Authorization", "Bearer secret-a"), ("x-legacy", "")], 401), + ([("Authorization", "Bearer secret-a"), ("Authorization", "Bearer")], 401), + ([("Authorization", "Bearer unknown-secret")], 401), + ], +) +async def test_accepted_carriers( + leg: str, + mode: str, + custom: bool, + headers: list[tuple[str, str]], + status: int, + caplog: pytest.LogCaptureFixture, +) -> None: + validated: list[str] = [] + + def validate(token: str) -> Principal | None: + validated.append(token) + return Principal(token) if token in {"secret-a", "secret-b"} else None + + options = {f"{leg}_header_name": "x-legacy", f"{leg}_bearer_prefix_required": False} + with pytest.warns(DeprecationWarning) if custom and leg == "mcp" else nullcontext(): + config = BearerTokenAuth( + validate_token=validate, + allow_unauthenticated=mode == "anonymous", + **(options if custom else {"legacy_header_aliases": ["x-legacy"]}), + ) + app = middleware_app(leg, config) + body = CALL if mode != "discovery" else {"method": "tools/list"} + with caplog.at_level(logging.DEBUG, logger="adcp.server.auth"): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + response = await client.post("/", headers=headers, json=body) + assert response.status_code == status + if status == 200: + assert response.json() == {"principal": "secret-a", "body": body} + assert validated == ["secret-a"] + else: + assert "www-authenticate" in response.headers + if headers != [("Authorization", "Bearer unknown-secret")]: + assert validated == [] + assert current_principal.get() is None + for record in caplog.records: + if record.name == "adcp.server.auth": + assert "secret" not in record.getMessage() + assert record.exc_info is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("leg", ["mcp", "a2a"]) +async def test_alias_prefix_policy(leg: str) -> None: + config = BearerTokenAuth( + validate_token=Principal, + legacy_header_aliases=["x-legacy"], + legacy_aliases_bearer_prefix_required=True, + ) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=middleware_app(leg, config)), base_url="http://test" + ) as client: + for alias, status in [ + ("Bearer secret-a", 200), + ("bearer\tsecret-a", 200), + ("secret-a", 401), + ("Bearer secret-b", 401), + ]: + response = await client.post( + "/", json=CALL, headers={"authorization": "Bearer secret-a", "x-legacy": alias} + ) + assert response.status_code == status + + +@pytest.mark.asyncio +@pytest.mark.parametrize("leg", ["mcp", "a2a"]) +async def test_concurrent_credentials_and_anonymous_requests(leg: str) -> None: + config = BearerTokenAuth(validate_token=Principal, allow_unauthenticated=True) + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=middleware_app(leg, config)), base_url="http://test" + ) as client: + + async def call(identity: str | None) -> None: + headers = {"authorization": f"Bearer {identity}"} if identity else {} + response = await client.post("/", headers=headers, json=CALL) + assert response.json()["principal"] == identity + + await asyncio.gather(*(call(identity) for identity in ["a", None, "b", None] * 8)) + assert current_principal.get() is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("leg", ["mcp", "a2a"]) +async def test_validator_exception_does_not_log_credentials( + leg: str, caplog: pytest.LogCaptureFixture +) -> None: + def validate(token: str) -> Principal | None: + raise RuntimeError(f"credential={token}") + + async with httpx.AsyncClient( + transport=httpx.ASGITransport( + app=middleware_app(leg, BearerTokenAuth(validate_token=validate)) + ), + base_url="http://test", + ) as client: + with caplog.at_level(logging.INFO, logger="adcp.server.auth"): + response = await client.post( + "/", headers={"authorization": "Bearer private-token"}, json=CALL + ) + assert response.status_code == 401 + assert "private-token" not in caplog.text + response.text + errors = [record for record in caplog.records if record.levelno >= logging.ERROR] + assert len(errors) == 1 + assert errors[0].reason == "validator_error" + assert errors[0].exc_info is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("alias,status", [("token-a", 200), ("token-b", 401)]) +async def test_combined_server_credential_parity(alias: str, status: int) -> None: + """The #1305 reproduction, through the actual combined transport server.""" + from asgi_lifespan import LifespanManager + + from adcp.server import ADCPHandler + from adcp.server.auth import auth_context_factory + from adcp.server.serve import _build_mcp_and_a2a_app + + seen: list[str | None] = [] + + class Seller(ADCPHandler): + advertised_tools = {"get_products"} + + async def get_products(self, params: Any, context: Any = None) -> dict[str, Any]: + seen.append(context.caller_identity) + return {"products": []} + + app = _build_mcp_and_a2a_app( + Seller(), + name="seller", + port=3001, + host="127.0.0.1", + instructions=None, + test_controller=None, + validation=None, + stateless_http=True, + enable_dns_rebinding_protection=False, + context_factory=auth_context_factory, + auth=BearerTokenAuth(validate_token=Principal, legacy_header_aliases=["x-legacy"]), + ) + params = {"buying_mode": "brief", "brief": "video"} + mcp = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": {"name": "get_products", "arguments": params}, + } + a2a = { + "jsonrpc": "2.0", + "id": 1, + "method": "SendMessage", + "params": { + "message": { + "messageId": "m1", + "role": "ROLE_USER", + "parts": [{"data": {"skill": "get_products", "parameters": params}}], + } + }, + } + async with LifespanManager(app): + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test", follow_redirects=True + ) as client: + for path, body in [("/mcp", mcp), ("/", a2a)]: + seen.clear() + response = await client.post( + path, + json=body, + headers={ + "accept": "application/json, text/event-stream", + "A2A-Version": "1.0", + "authorization": "Bearer token-a", + "x-legacy": alias, + }, + ) + assert response.status_code == status + assert seen == (["token-a"] if status == 200 else []) diff --git a/tests/test_auth_middleware.py b/tests/test_auth_middleware.py index 08e729a79..f091f74c3 100644 --- a/tests/test_auth_middleware.py +++ b/tests/test_auth_middleware.py @@ -907,18 +907,8 @@ def validator(token: str) -> Principal | None: @pytest.mark.asyncio -async def test_authorization_wins_when_both_headers_present() -> None: - """Per #720, ``Authorization: Bearer`` is the spec-canonical - carrier and is always checked first. When both ``Authorization: - Bearer X`` and the legacy alias ``x-adcp-auth: Y`` are present, - ``X`` wins — the alias is the fallback path, not a competing - primary. - - (This test replaces the pre-#720 exclusive-mode assertion that - pinned the silent-401 bug — adopters with ``header_name`` set - used to reject every spec-compliant client; now those clients - are accepted on the canonical header.) - """ +async def test_conflicting_authorization_and_alias_are_rejected() -> None: + """All accepted carriers must identify the same credential (#1305).""" received: list[str] = [] def validator(token: str) -> Principal | None: @@ -940,8 +930,8 @@ def validator(token: str) -> Principal | None: "x-adcp-auth": "tok_y", }, ) - assert resp.status_code == 200 - assert received == ["tok_x"] # Authorization: Bearer wins per #720 + assert resp.status_code == 401 + assert received == [] @pytest.mark.asyncio @@ -1023,9 +1013,7 @@ def validator(token: str) -> Principal | None: @pytest.mark.asyncio async def test_alias_falls_through_when_authorization_missing() -> None: - """The alias path is the fallback — only consulted when - ``Authorization`` is absent or empty. Confirms the resolution - order is canonical-first.""" + """An alias can supply the sole accepted credential.""" received: list[str] = [] def validator(token: str) -> Principal | None: @@ -1048,11 +1036,8 @@ def validator(token: str) -> Principal | None: @pytest.mark.asyncio -async def test_empty_authorization_falls_through_to_alias() -> None: - """``Authorization: `` (empty value) shouldn't short-circuit the - chain — adopters mid-migration sometimes have a client that sets - the header to an empty string in error. The alias path must still - be consulted.""" +async def test_empty_authorization_rejects_even_with_valid_alias() -> None: + """A supplied empty credential must not become an auth fallback.""" received: list[str] = [] def validator(token: str) -> Principal | None: @@ -1068,21 +1053,18 @@ def validator(token: str) -> Principal | None: "/", json={"method": "tools/call", "params": {"name": "get_products"}}, headers={ - "Authorization": "", # empty — shouldn't 401 by itself + "Authorization": "", # supplied but malformed "x-adcp-auth": "fallback-token", }, ) - assert resp.status_code == 200 - assert received == ["fallback-token"] + assert resp.status_code == 401 + assert received == [] @pytest.mark.asyncio -async def test_multiple_aliases_walked_in_order() -> None: - """When two aliases are configured and both are present on the - request, the first one in the list wins. Adopters with multiple - legacy carriers (different generations of clients) get - deterministic resolution.""" +async def test_conflicting_aliases_rejected() -> None: + """Multiple legacy aliases follow the same conflict policy.""" received: list[str] = [] def validator(token: str) -> Principal | None: @@ -1103,8 +1085,8 @@ def validator(token: str) -> Principal | None: }, ) - assert resp.status_code == 200 - assert received == ["first-alias"] # first alias in the list wins + assert resp.status_code == 401 + assert received == [] @pytest.mark.asyncio diff --git a/tests/test_auth_per_leg_headers.py b/tests/test_auth_per_leg_headers.py index 5ed891843..6dc94ce13 100644 --- a/tests/test_auth_per_leg_headers.py +++ b/tests/test_auth_per_leg_headers.py @@ -662,7 +662,7 @@ async def send(message: dict) -> None: ], [ (b"authorization", b"Bearer good-token"), - (b"x-legacy-auth", b"good-token"), + (b"x-legacy-auth", b"bad-token"), ], ], )