From f6b650c793f16d33cd0c04ad454259b607bba5a4 Mon Sep 17 00:00:00 2001 From: arunmm8335 Date: Mon, 20 Jul 2026 14:48:32 +0530 Subject: [PATCH] fix(client): apply all schemes in multi-scheme security requirement (AND semantics) AuthInterceptor.before() previously returned after applying the first credential, even when a SecurityRequirement listed multiple schemes that must all be satisfied together (AND semantics per OpenAPI spec). This change introduces a two-pass approach per requirement: - Pass 1: collect credentials for every scheme; if any is unavailable or unsupported, mark the requirement as unsatisfiable and skip it. - Pass 2: apply all collected credentials at once, then return. The outer loop over security_requirements retains OR semantics: the first fully satisfiable requirement is used. A _resolve_header() static helper was extracted to map each scheme type to its (header_name, value) pair, improving readability and keeping the collection loop clean. Added 5 new test cases covering: - Multi-scheme AND (the exact issue reproduction) - Multi-requirement OR fallback - Partial requirement prevention - Mixed Bearer + API key in a single requirement - Unsupported scheme type (query API key) fallback Fixes #1138 --- src/a2a/client/auth/interceptor.py | 159 ++++++++++++------- tests/client/test_auth_interceptor.py | 218 ++++++++++++++++++++++++++ 2 files changed, 321 insertions(+), 56 deletions(-) diff --git a/src/a2a/client/auth/interceptor.py b/src/a2a/client/auth/interceptor.py index 973c91cd7..88deeda12 100644 --- a/src/a2a/client/auth/interceptor.py +++ b/src/a2a/client/auth/interceptor.py @@ -1,6 +1,9 @@ -import logging # noqa: I001 +from __future__ import annotations + +import logging + +from typing import TYPE_CHECKING -from a2a.client.auth.credentials import CredentialService from a2a.client.client import ClientCallContext from a2a.client.interceptors import ( AfterArgs, @@ -8,6 +11,11 @@ ClientCallInterceptor, ) + +if TYPE_CHECKING: + from a2a.client.auth.credentials import CredentialService + from a2a.types.a2a_pb2 import SecurityScheme + logger = logging.getLogger(__name__) @@ -21,7 +29,17 @@ def __init__(self, credential_service: CredentialService): self._credential_service = credential_service async def before(self, args: BeforeArgs) -> None: - """Applies authentication headers to the request if credentials are available.""" + """Applies authentication headers to the request if credentials are available. + + Follows OpenAPI Security Requirement semantics: + - The outer ``security_requirements`` list uses **OR** semantics: + satisfying any single requirement is sufficient. + - Multiple schemes **within** a single requirement use **AND** + semantics: all of them must be satisfied together. + + A two-pass approach is used per requirement to avoid partial + application when one scheme's credential is unavailable. + """ agent_card = args.agent_card # Proto3 repeated fields (security) and maps (security_schemes) do not track presence. @@ -34,63 +52,92 @@ async def before(self, args: BeforeArgs) -> None: return for requirement in agent_card.security_requirements: + # Pass 1: collect credentials for every scheme in this + # requirement. If any scheme cannot be satisfied the whole + # requirement is skipped (AND semantics). + collected: dict[str, tuple[str, str]] = {} + satisfiable = True + for scheme_name in requirement.schemes: credential = await self._credential_service.get_credentials( scheme_name, args.context ) - if credential and scheme_name in agent_card.security_schemes: - scheme = agent_card.security_schemes[scheme_name] - - if args.context is None: - args.context = ClientCallContext() - - if args.context.service_parameters is None: - args.context.service_parameters = {} - - # HTTP Bearer authentication - if ( - scheme.HasField('http_auth_security_scheme') - and scheme.http_auth_security_scheme.scheme.lower() - == 'bearer' - ): - args.context.service_parameters['Authorization'] = ( - f'Bearer {credential}' - ) - logger.debug( - "Added Bearer token for scheme '%s'.", - scheme_name, - ) - return - - # OAuth2 and OIDC schemes are implicitly Bearer - if scheme.HasField( - 'oauth2_security_scheme' - ) or scheme.HasField('open_id_connect_security_scheme'): - args.context.service_parameters['Authorization'] = ( - f'Bearer {credential}' - ) - logger.debug( - "Added Bearer token for scheme '%s'.", - scheme_name, - ) - return - - # API Key in Header - if ( - scheme.HasField('api_key_security_scheme') - and scheme.api_key_security_scheme.location.lower() - == 'header' - ): - args.context.service_parameters[ - scheme.api_key_security_scheme.name - ] = credential - logger.debug( - "Added API Key Header for scheme '%s'.", - scheme_name, - ) - return - - # Note: Other cases like API keys in query/cookie are not handled and will be skipped. + if ( + not credential + or scheme_name not in agent_card.security_schemes + ): + satisfiable = False + break + + scheme = agent_card.security_schemes[scheme_name] + header = self._resolve_header(scheme_name, scheme, credential) + + if header is None: + # Unsupported scheme type (e.g. API key in query/cookie). + # Treat as unsatisfiable so the requirement is skipped. + satisfiable = False + break + + collected[scheme_name] = header + + if not satisfiable: + continue # OR: try the next requirement + + # Pass 2: apply all collected credentials at once. + if args.context is None: + args.context = ClientCallContext() + + if args.context.service_parameters is None: + args.context.service_parameters = {} + + for scheme_name, (header_name, header_value) in collected.items(): + args.context.service_parameters[header_name] = header_value + logger.debug( + "Applied credential for scheme '%s' (header '%s').", + scheme_name, + header_name, + ) + + return # One requirement fully satisfied — done. + + @staticmethod + def _resolve_header( + scheme_name: str, + scheme: SecurityScheme, + credential: str, + ) -> tuple[str, str] | None: + """Map a security scheme + credential to a ``(header_name, value)`` pair. + + Returns ``None`` when the scheme type is not supported (e.g. API + key in query or cookie), signalling that the enclosing requirement + cannot be satisfied via HTTP headers alone. + """ + # HTTP Bearer authentication + if ( + scheme.HasField('http_auth_security_scheme') + and scheme.http_auth_security_scheme.scheme.lower() == 'bearer' + ): + return ('Authorization', f'Bearer {credential}') + + # OAuth2 and OIDC schemes are implicitly Bearer + if scheme.HasField('oauth2_security_scheme') or scheme.HasField( + 'open_id_connect_security_scheme' + ): + return ('Authorization', f'Bearer {credential}') + + # API Key in Header + if ( + scheme.HasField('api_key_security_scheme') + and scheme.api_key_security_scheme.location.lower() == 'header' + ): + return (scheme.api_key_security_scheme.name, credential) + + # Unsupported scheme type + logger.debug( + "Scheme '%s' has an unsupported type or location; skipping.", + scheme_name, + ) + return None async def after(self, args: AfterArgs) -> None: """Invoked after the method is executed.""" diff --git a/tests/client/test_auth_interceptor.py b/tests/client/test_auth_interceptor.py index f7979bff7..926c5c1ff 100644 --- a/tests/client/test_auth_interceptor.py +++ b/tests/client/test_auth_interceptor.py @@ -332,3 +332,221 @@ async def test_auth_interceptor_skips_when_scheme_not_in_security_schemes( await auth_interceptor.before(args) assert context.service_parameters is None + + +# --------------------------------------------------------------------------- +# Tests for multi-scheme AND / multi-requirement OR semantics (#1138) +# --------------------------------------------------------------------------- + + +class DictCredentialService: + """A simple credential service backed by a plain dict.""" + + def __init__(self, creds: dict[str, str]) -> None: + self._creds = creds + + async def get_credentials( + self, + security_scheme_name: str, + context: ClientCallContext | None, + ) -> str | None: + return self._creds.get(security_scheme_name) + + +def _make_agent_card( + *, + security_schemes: dict[str, SecurityScheme], + security_requirements: list[SecurityRequirement], +) -> AgentCard: + """Helper to build a minimal AgentCard with security configuration.""" + return AgentCard( + name='testbot', + description='test', + version='1.0', + default_input_modes=[], + default_output_modes=[], + skills=[], + capabilities=AgentCapabilities(), + security_schemes=security_schemes, + security_requirements=security_requirements, + ) + + +@pytest.mark.asyncio +async def test_multi_scheme_and_semantics() -> None: + """A single requirement with two API-key-in-header schemes must apply both headers. + + This is the exact reproduction from issue #1138. + """ + creds = {'AUTH_KEY': 'gw-secret-key', 'USER_ID': 'user-123'} + card = _make_agent_card( + security_schemes={ + 'AUTH_KEY': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='AUTH_KEY' + ) + ), + 'USER_ID': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='USER_ID' + ) + ), + }, + security_requirements=[ + SecurityRequirement( + schemes={'AUTH_KEY': StringList(), 'USER_ID': StringList()} + ) + ], + ) + + args = BeforeArgs(input=None, method='message/send', agent_card=card) + await AuthInterceptor(DictCredentialService(creds)).before(args) + + assert args.context is not None + assert args.context.service_parameters is not None + assert args.context.service_parameters['AUTH_KEY'] == 'gw-secret-key' + assert args.context.service_parameters['USER_ID'] == 'user-123' + + +@pytest.mark.asyncio +async def test_multi_requirement_or_fallback() -> None: + """When the first requirement is unsatisfiable, the second (OR) should be used.""" + creds = {'BACKUP_KEY': 'backup-secret'} + card = _make_agent_card( + security_schemes={ + 'PRIMARY_KEY': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='X-Primary' + ) + ), + 'BACKUP_KEY': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='X-Backup' + ) + ), + }, + security_requirements=[ + # First requirement: credential not available + SecurityRequirement(schemes={'PRIMARY_KEY': StringList()}), + # Second requirement: credential available + SecurityRequirement(schemes={'BACKUP_KEY': StringList()}), + ], + ) + + args = BeforeArgs(input=None, method='message/send', agent_card=card) + await AuthInterceptor(DictCredentialService(creds)).before(args) + + assert args.context is not None + assert args.context.service_parameters is not None + assert args.context.service_parameters['X-Backup'] == 'backup-secret' + # The primary key should NOT be present + assert 'X-Primary' not in args.context.service_parameters + + +@pytest.mark.asyncio +async def test_partial_requirement_not_applied() -> None: + """A requirement with two schemes where one credential is missing must not be partially applied.""" + creds = {'AUTH_KEY': 'gw-secret-key'} # USER_ID is missing + card = _make_agent_card( + security_schemes={ + 'AUTH_KEY': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='AUTH_KEY' + ) + ), + 'USER_ID': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='USER_ID' + ) + ), + }, + security_requirements=[ + SecurityRequirement( + schemes={'AUTH_KEY': StringList(), 'USER_ID': StringList()} + ) + ], + ) + + args = BeforeArgs(input=None, method='message/send', agent_card=card) + await AuthInterceptor(DictCredentialService(creds)).before(args) + + # Nothing should be applied — partial application is not allowed + assert args.context is None or args.context.service_parameters is None + + +@pytest.mark.asyncio +async def test_mixed_bearer_and_apikey_in_single_requirement() -> None: + """A single requirement with both Bearer and API-key-in-header schemes must apply both.""" + creds = {'bearer_scheme': 'my-token', 'apikey_scheme': 'my-api-key'} + card = _make_agent_card( + security_schemes={ + 'bearer_scheme': SecurityScheme( + http_auth_security_scheme=HTTPAuthSecurityScheme( + scheme='bearer' + ) + ), + 'apikey_scheme': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='X-API-Key' + ) + ), + }, + security_requirements=[ + SecurityRequirement( + schemes={ + 'bearer_scheme': StringList(), + 'apikey_scheme': StringList(), + } + ) + ], + ) + + args = BeforeArgs(input=None, method='message/send', agent_card=card) + await AuthInterceptor(DictCredentialService(creds)).before(args) + + assert args.context is not None + assert args.context.service_parameters is not None + assert args.context.service_parameters['Authorization'] == 'Bearer my-token' + assert args.context.service_parameters['X-API-Key'] == 'my-api-key' + + +@pytest.mark.asyncio +async def test_unsupported_scheme_type_makes_requirement_unsatisfiable() -> ( + None +): + """An unsupported scheme type (API key in query) makes its requirement unsatisfiable. + + A fallback requirement should be used instead. + """ + creds = { + 'query_key': 'query-secret', + 'header_key': 'header-secret', + } + card = _make_agent_card( + security_schemes={ + 'query_key': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='query', name='api_key' + ) + ), + 'header_key': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='X-Header-Key' + ) + ), + }, + security_requirements=[ + # First requirement: uses unsupported query location + SecurityRequirement(schemes={'query_key': StringList()}), + # Second requirement: uses supported header location + SecurityRequirement(schemes={'header_key': StringList()}), + ], + ) + + args = BeforeArgs(input=None, method='message/send', agent_card=card) + await AuthInterceptor(DictCredentialService(creds)).before(args) + + assert args.context is not None + assert args.context.service_parameters is not None + assert args.context.service_parameters['X-Header-Key'] == 'header-secret' + assert 'api_key' not in args.context.service_parameters