Skip to content

fix(client): apply all schemes in multi-scheme security requirement (AND semantics) #1140

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
arunmm8335 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(client): apply all schemes in multi-scheme security requirement (AND semantics) #1140

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
159 changes: 103 additions & 56 deletions src/a2a/client/auth/interceptor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,13 +1,21 @@
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,
BeforeArgs,
ClientCallInterceptor,
)


if TYPE_CHECKING:
from a2a.client.auth.credentials import CredentialService
from a2a.types.a2a_pb2 import SecurityScheme

logger = logging.getLogger(__name__)


Expand All @@ -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.
Comment on lines +34 to +41

@sokoliva sokoliva Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's remove implementation details and just keep high level description to not clutter the code with comments.

@arunmm8335 arunmm8335 Jul 30, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes sure will do that could you please review it and assign this issue to me

@sokoliva sokoliva Jul 31, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thank you! I submitted the review.

"""
agent_card = args.agent_card

# Proto3 repeated fields (security) and maps (security_schemes) do not track presence.
Expand All @@ -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).
Comment on lines +55 to +57

@sokoliva sokoliva Jul 30, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This detailed description is not needed, it adds unnecessary clutter, the behaviour from the description in visible in the next few lines of code.

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.

@sokoliva sokoliva Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remove the comment.

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."""
Loading
Loading