From 3f907120f0969b5ef965ff521cf6d6554c01c6e1 Mon Sep 17 00:00:00 2001 From: Tymofii Pidlisnyi Date: Wed, 30 Sep 2026 15:18:21 -0700 Subject: [PATCH 1/3] refactor(signing): walk AgentCard descriptors when pruning canonical payload Agent Card canonicalization removed empty values with a type-blind pass over the MessageToDict output. This walks the AgentCard descriptor alongside the JSON, so the field each value belongs to is known at every level. No canonical bytes change. Nested messages, repeated fields and maps are pruned exactly as before, free-form google.protobuf values still go through _clean_empty, and the depth bound is kept. A differential run over 6000 randomly populated cards found no difference from the previous pruning. Refs #1278 --- src/a2a/utils/signing.py | 74 ++++++++++++++++++++++- tests/utils/test_signing.py | 117 ++++++++++++++++++++++++++++++++++++ 2 files changed, 188 insertions(+), 3 deletions(-) diff --git a/src/a2a/utils/signing.py b/src/a2a/utils/signing.py index c85a80072..96566167e 100644 --- a/src/a2a/utils/signing.py +++ b/src/a2a/utils/signing.py @@ -3,6 +3,7 @@ from collections.abc import Callable from typing import Any, TypedDict +from google.protobuf.descriptor import Descriptor, FieldDescriptor from google.protobuf.json_format import MessageToDict @@ -22,6 +23,7 @@ from a2a.types import AgentCard, AgentCardSignature from a2a.utils._jcs import MAX_DEPTH, CanonicalizationError, canonicalize +from a2a.utils.proto_utils import _field_is_repeated class SignatureVerificationError(Exception): @@ -195,6 +197,72 @@ def _clean_empty(d: Any, depth: int = 0) -> Any: return d +def _is_map(field: FieldDescriptor) -> bool: + """Returns True if the field is a protobuf map.""" + message_type = field.message_type + return message_type is not None and message_type.GetOptions().map_entry + + +def _is_well_known(descriptor: Descriptor | Any) -> bool: + """Returns True for `google.protobuf` types, which carry free-form JSON.""" + return descriptor.full_name.startswith('google.protobuf.') + + +def _clean_field(value: Any, field: FieldDescriptor, depth: int) -> Any: + """Removes empty values from the JSON form of one message field.""" + message_type = field.message_type + if message_type is None or _is_well_known(message_type): + return _clean_empty(value, depth) + if _is_map(field): + value_type = message_type.fields_by_name['value'].message_type + if value_type is None or _is_well_known(value_type): + return _clean_empty(value, depth) + cleaned_map = { + k: cleaned_v + for k, v in value.items() + if (cleaned_v := _clean_message(v, value_type, depth + 1)) + } + return cleaned_map or None + if _field_is_repeated(field): + cleaned_list = [ + cleaned_v + for v in value + if (cleaned_v := _clean_message(v, message_type, depth + 1)) + ] + return cleaned_list or None + return _clean_message(value, message_type, depth) or None + + +def _clean_message( + message_dict: dict[str, Any], + descriptor: Descriptor | Any, + depth: int = 0, +) -> dict[str, Any]: + """Removes empty values from the JSON form of a message, by descriptor. + + `message_dict` is the `MessageToDict` output for a message of type + `descriptor`. Walking the descriptor alongside the JSON keeps the field + each value belongs to known at every level, which `_clean_empty` alone + cannot tell. Free-form values (`google.protobuf.Struct` and friends) and + keys the descriptor does not know fall back to `_clean_empty`. + """ + if depth > MAX_DEPTH: + raise CanonicalizationError( + f'nesting exceeds the maximum depth of {MAX_DEPTH}' + ) + fields = {field.json_name: field for field in descriptor.fields} + cleaned: dict[str, Any] = {} + for key, value in message_dict.items(): + field = fields.get(key) + if field is None: + cleaned_value = _clean_empty(value, depth + 1) + else: + cleaned_value = _clean_field(value, field, depth + 1) + if cleaned_value is not None: + cleaned[key] = cleaned_value + return cleaned + + def _canonicalize_agent_card(agent_card: AgentCard) -> str: """Canonicalizes the Agent Card JSON according to RFC 8785 (JCS).""" card_dict = MessageToDict( @@ -203,6 +271,6 @@ def _canonicalize_agent_card(agent_card: AgentCard) -> str: # Remove signatures field if present card_dict.pop('signatures', None) - # Recursively remove empty values - cleaned_dict = _clean_empty(card_dict) - return canonicalize(cleaned_dict) + # Remove empty values, walking the AgentCard descriptor + cleaned_dict = _clean_message(card_dict, AgentCard.DESCRIPTOR) + return canonicalize(cleaned_dict or None) diff --git a/tests/utils/test_signing.py b/tests/utils/test_signing.py index 616aab9f3..fd8939815 100644 --- a/tests/utils/test_signing.py +++ b/tests/utils/test_signing.py @@ -3,14 +3,24 @@ import pytest from a2a.types.a2a_pb2 import ( + APIKeySecurityScheme, AgentCapabilities, AgentCard, AgentCardSignature, + AgentExtension, AgentInterface, + AgentProvider, AgentSkill, + AuthorizationCodeOAuthFlow, + OAuth2SecurityScheme, + OAuthFlows, + SecurityRequirement, + SecurityScheme, + StringList, ) from a2a.utils import signing from cryptography.hazmat.primitives.asymmetric import ec +from google.protobuf.json_format import MessageToDict from jwt.utils import base64url_encode @@ -287,3 +297,110 @@ def test_clean_empty_does_not_mutate_input(): signing._clean_empty(original) assert original == original_copy + + +@pytest.fixture +def full_agent_card() -> AgentCard: + """A card that exercises nested messages, maps and free-form values.""" + card = AgentCard( + name='Full Agent', + description='A card that exercises nested messages', + supported_interfaces=[ + AgentInterface( + url='https://example.com/a2a/v1', + protocol_binding='JSONRPC', + protocol_version='1.0', + tenant='', + ) + ], + provider=AgentProvider( + url='https://example.com', organization='Example' + ), + version='1.0.0', + capabilities=AgentCapabilities( + streaming=False, + extensions=[ + AgentExtension(uri='https://example.com/ext/1'), + AgentExtension(uri='https://example.com/ext/2', description=''), + ], + ), + security_schemes={ + 'key': SecurityScheme( + api_key_security_scheme=APIKeySecurityScheme( + location='header', name='X-API-Key', description='' + ) + ), + 'oauth': SecurityScheme( + oauth2_security_scheme=OAuth2SecurityScheme( + flows=OAuthFlows( + authorization_code=AuthorizationCodeOAuthFlow( + authorization_url='https://example.com/auth', + token_url='https://example.com/token', + scopes={'read': 'Read access'}, + ) + ) + ) + ), + }, + security_requirements=[ + SecurityRequirement(schemes={'oauth': StringList(list=['read'])}), + SecurityRequirement(), + ], + default_input_modes=['text/plain'], + default_output_modes=['text/plain'], + skills=[ + AgentSkill( + id='skill1', + name='Skill', + description='A skill', + tags=['test'], + examples=[], + ) + ], + icon_url='', + ) + card.capabilities.extensions[0].params.update( + {'empty': '', 'nested': {'list': [], 'kept': 0}} + ) + return card + + +def test_clean_message_matches_clean_empty_on_full_card( + full_agent_card: AgentCard, +): + """Descriptor-aware cleaning gives the same result as `_clean_empty`.""" + card_dict = MessageToDict(full_agent_card) + assert signing._clean_message( + card_dict, AgentCard.DESCRIPTOR + ) == signing._clean_empty(card_dict) + + +def test_canonicalize_full_card_prunes_optional_defaults( + full_agent_card: AgentCard, +): + """Optional and free-form empty values are pruned at every level.""" + result = signing._canonicalize_agent_card(full_agent_card) + assert '"tenant"' not in result + assert '"iconUrl"' not in result + assert '"examples"' not in result + assert '"empty"' not in result + assert '"list"' in result # the StringList inside the security requirement + assert '"params":{"nested":{"kept":0}}' in result + assert '"streaming":false' in result + # The empty SecurityRequirement element is dropped, the other one stays. + assert ( + '"securityRequirements":[{"schemes":{"oauth":{"list":["read"]}}}]' + in result + ) + + +def test_clean_message_bounds_depth(): + """Descriptor-aware cleaning keeps the depth bound of `_clean_empty`.""" + nested: dict[str, Any] = {} + cursor = nested + for _ in range(signing.MAX_DEPTH + 5): + cursor['a'] = {} + cursor = cursor['a'] + card_dict = {'capabilities': {'extensions': [{'params': nested}]}} + with pytest.raises(signing.CanonicalizationError): + signing._clean_message(card_dict, AgentCard.DESCRIPTOR) From cdee28e53f2166ca04fc360868cae9fc2b7a3182 Mon Sep 17 00:00:00 2001 From: Tymofii Pidlisnyi Date: Wed, 30 Sep 2026 15:19:23 -0700 Subject: [PATCH 2/3] fix(signing): keep REQUIRED fields at their default value in the canonical payload Section 8.4.1 of the A2A specification says a REQUIRED field stays in the canonical payload even when its value matches the default. The canonicalizer dropped it, so a card signed by an SDK that keeps description "" or skills [] in the signed payload failed verification here. REQUIRED is read from google.api.field_behavior on the descriptors. REQUIRED strings are kept as "", REQUIRED repeated fields as [], REQUIRED maps and messages as {}. A REQUIRED message is present whether or not it was set, so the REQUIRED set does not depend on what the serializer emitted. Fields that are not REQUIRED are pruned as before. This changes the canonical bytes of any card that has an empty REQUIRED field, including cards signed by earlier versions of this SDK. Cards with no empty REQUIRED field keep their bytes. The rule itself is under discussion in a2aproject/A2A#2122, and this commit is separate from the previous one so it can wait for that decision. Fixes #1278 --- src/a2a/utils/signing.py | 37 ++++++ tests/utils/test_signing.py | 227 +++++++++++++++++++++++++++++++++++- 2 files changed, 261 insertions(+), 3 deletions(-) diff --git a/src/a2a/utils/signing.py b/src/a2a/utils/signing.py index 96566167e..84788e35a 100644 --- a/src/a2a/utils/signing.py +++ b/src/a2a/utils/signing.py @@ -3,6 +3,7 @@ from collections.abc import Callable from typing import Any, TypedDict +from google.api import field_behavior_pb2 as fb from google.protobuf.descriptor import Descriptor, FieldDescriptor from google.protobuf.json_format import MessageToDict @@ -197,6 +198,11 @@ def _clean_empty(d: Any, depth: int = 0) -> Any: return d +def _is_required(field: FieldDescriptor) -> bool: + """Returns True if the field carries google.api.field_behavior = REQUIRED.""" + return fb.REQUIRED in field.GetOptions().Extensions[fb.field_behavior] # type: ignore[index] # ty: ignore[invalid-argument-type] + + def _is_map(field: FieldDescriptor) -> bool: """Returns True if the field is a protobuf map.""" message_type = field.message_type @@ -208,6 +214,25 @@ def _is_well_known(descriptor: Descriptor | Any) -> bool: return descriptor.full_name.startswith('google.protobuf.') +def _required_default(field: FieldDescriptor) -> Any: + """Returns the JSON value a REQUIRED field has when it is empty, or None. + + A REQUIRED message is an empty object whether or not it was set, so the + REQUIRED set does not depend on what the serializer chose to emit. REQUIRED + fields of a scalar type other than string do not exist in the Agent Card + today and are left out. + """ + if _is_map(field): + return {} + if _field_is_repeated(field): + return [] + if field.message_type is not None: + return {} + if field.type == FieldDescriptor.TYPE_STRING: + return '' + return None + + def _clean_field(value: Any, field: FieldDescriptor, depth: int) -> Any: """Removes empty values from the JSON form of one message field.""" message_type = field.message_type @@ -240,6 +265,9 @@ def _clean_message( ) -> dict[str, Any]: """Removes empty values from the JSON form of a message, by descriptor. + REQUIRED fields are kept at their default value. Every other empty value + is removed. + `message_dict` is the `MessageToDict` output for a message of type `descriptor`. Walking the descriptor alongside the JSON keeps the field each value belongs to known at every level, which `_clean_empty` alone @@ -260,6 +288,15 @@ def _clean_message( cleaned_value = _clean_field(value, field, depth + 1) if cleaned_value is not None: cleaned[key] = cleaned_value + # A2A specification section 8.4.1: a REQUIRED field stays in the canonical + # payload even when its value matches the default. + for field in descriptor.fields: + name = field.json_name + if name in cleaned or not _is_required(field): + continue + default = _required_default(field) + if default is not None: + cleaned[name] = default return cleaned diff --git a/tests/utils/test_signing.py b/tests/utils/test_signing.py index fd8939815..19a31df03 100644 --- a/tests/utils/test_signing.py +++ b/tests/utils/test_signing.py @@ -21,6 +21,7 @@ from a2a.utils import signing from cryptography.hazmat.primitives.asymmetric import ec from google.protobuf.json_format import MessageToDict +from jwt import api_jws from jwt.utils import base64url_encode @@ -195,7 +196,8 @@ def test_canonicalize_agent_card(sample_agent_card: AgentCard): """Test canonicalize_agent_card with defaults, optionals, and exceptions. - extensions is omitted as it's not set and optional. - - protocolVersion is included because it's always added by canonicalize_agent_card. + - protocolVersion is REQUIRED on AgentInterface, so it is kept at its + default value (A2A specification section 8.4.1). - signatures should be omitted. """ expected_jcs = ( @@ -203,7 +205,7 @@ def test_canonicalize_agent_card(sample_agent_card: AgentCard): '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],' '"description":"A test agent","name":"Test Agent",' '"skills":[{"description":"A test skill","id":"skill1","name":"Test Skill","tags":["test"]}],' - '"supportedInterfaces":[{"protocolBinding":"HTTP+JSON","url":"http://localhost"}],' + '"supportedInterfaces":[{"protocolBinding":"HTTP+JSON","protocolVersion":"","url":"http://localhost"}],' '"version":"1.0.0"}' ) result = signing._canonicalize_agent_card(sample_agent_card) @@ -365,10 +367,43 @@ def full_agent_card() -> AgentCard: return card +def test_required_fields_are_read_from_descriptors(): + """REQUIRED comes from `google.api.field_behavior`, not from a name list.""" + card = AgentCard.DESCRIPTOR + required = {f.json_name for f in card.fields if signing._is_required(f)} + assert required == { + 'name', + 'description', + 'supportedInterfaces', + 'version', + 'capabilities', + 'defaultInputModes', + 'defaultOutputModes', + 'skills', + } + skill = AgentSkill.DESCRIPTOR + assert {f.json_name for f in skill.fields if signing._is_required(f)} == { + 'id', + 'name', + 'description', + 'tags', + } + + +def test_implicit_presence_fields_are_not_all_required(): + """Fields that the blanket `always_print` flag emits stay non-REQUIRED.""" + tenant = AgentInterface.DESCRIPTOR.fields_by_name['tenant'] + ext_required = AgentExtension.DESCRIPTOR.fields_by_name['required'] + ext_list = AgentCapabilities.DESCRIPTOR.fields_by_name['extensions'] + assert not signing._is_required(tenant) + assert not signing._is_required(ext_required) + assert not signing._is_required(ext_list) + + def test_clean_message_matches_clean_empty_on_full_card( full_agent_card: AgentCard, ): - """Descriptor-aware cleaning gives the same result as `_clean_empty`.""" + """With no empty REQUIRED field, descriptor-aware cleaning changes nothing.""" card_dict = MessageToDict(full_agent_card) assert signing._clean_message( card_dict, AgentCard.DESCRIPTOR @@ -404,3 +439,189 @@ def test_clean_message_bounds_depth(): card_dict = {'capabilities': {'extensions': [{'params': nested}]}} with pytest.raises(signing.CanonicalizationError): signing._clean_message(card_dict, AgentCard.DESCRIPTOR) + + +def _reachable_messages(descriptor, seen=None): + seen = {} if seen is None else seen + if descriptor.full_name in seen or signing._is_well_known(descriptor): + return seen + seen[descriptor.full_name] = descriptor + for field in descriptor.fields: + message_type = field.message_type + if message_type is None: + continue + if signing._is_map(field): + message_type = message_type.fields_by_name['value'].message_type + if message_type is None: + continue + _reachable_messages(message_type, seen) + return seen + + +def _example_card(**overrides: Any) -> AgentCard: + fields: dict[str, Any] = { + 'name': 'Example Agent', + 'description': 'An example', + 'version': '1.0.0', + 'supported_interfaces': [ + AgentInterface( + url='https://example.com/a2a/v1', + protocol_binding='JSONRPC', + protocol_version='1.0', + ) + ], + 'capabilities': AgentCapabilities( + streaming=False, push_notifications=False + ), + 'default_input_modes': ['text/plain'], + 'default_output_modes': ['text/plain'], + 'skills': [ + AgentSkill( + id='skill1', name='Skill', description='A skill', tags=['t'] + ) + ], + } + fields.update(overrides) + return AgentCard(**fields) + + +def test_canonicalize_keeps_required_fields_at_default(): + """Section 8.4.1: `description: ""` and `skills: []` stay in the payload. + + The card is the worked example of section 8.4.1 with the remaining + REQUIRED fields of AgentCard filled in. + """ + card = _example_card(description='', skills=[]) + card.capabilities.extensions.extend([]) + expected_jcs = ( + '{"capabilities":{"pushNotifications":false,"streaming":false},' + '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],' + '"description":"","name":"Example Agent","skills":[],' + '"supportedInterfaces":[{"protocolBinding":"JSONRPC",' + '"protocolVersion":"1.0","url":"https://example.com/a2a/v1"}],' + '"version":"1.0.0"}' + ) + assert signing._canonicalize_agent_card(card) == expected_jcs + + +@pytest.mark.parametrize( + ('overrides', 'kept'), + [ + pytest.param({'description': ''}, '"description":""', id='description'), + pytest.param({'skills': []}, '"skills":[]', id='skills'), + pytest.param({'version': ''}, '"version":""', id='version'), + pytest.param( + {'default_input_modes': []}, + '"defaultInputModes":[]', + id='default-input-modes', + ), + pytest.param( + { + 'skills': [ + AgentSkill(id='s', name='Skill', description='A skill') + ] + }, + '"tags":[]', + id='skill-tags', + ), + ], +) +def test_canonicalize_keeps_each_empty_required_field( + overrides: dict[str, Any], kept: str +): + assert kept in signing._canonicalize_agent_card(_example_card(**overrides)) + + +def test_canonicalize_still_prunes_non_required_defaults(): + """Implicit-presence fields that are not REQUIRED stay out of the payload.""" + card = _example_card(description='', skills=[]) + card.supported_interfaces[0].tenant = '' + card.capabilities.extensions.add( + uri='https://example.com/ext', required=False + ) + result = signing._canonicalize_agent_card(card) + assert '"tenant"' not in result + assert '"required"' not in result + assert '"extensions":[{"uri":"https://example.com/ext"}]' in result + assert '"securitySchemes"' not in result + assert '"securityRequirements"' not in result + + +def test_canonicalize_keeps_an_unset_required_message(): + """A REQUIRED message is present even when it was never set.""" + card = _example_card() + card.ClearField('capabilities') + assert not card.HasField('capabilities') + assert '"capabilities":{}' in signing._canonicalize_agent_card(card) + + +def test_canonicalize_keeps_a_set_but_empty_required_message(): + card = _example_card(capabilities=AgentCapabilities()) + assert '"capabilities":{}' in signing._canonicalize_agent_card(card) + + +def test_every_required_field_has_a_default_this_module_can_emit(): + """Fails when a REQUIRED field of an unhandled scalar type is added.""" + unhandled = [] + for descriptor in _reachable_messages(AgentCard.DESCRIPTOR).values(): + for field in descriptor.fields: + if not signing._is_required(field): + continue + if signing._required_default(field) is None: + unhandled.append(f'{descriptor.name}.{field.name}') + assert unhandled == [] + + +def test_required_messages_have_no_required_fields_of_their_own(): + """`{}` is the complete default of every REQUIRED singular message. + + Fails when a REQUIRED message gains a REQUIRED field, because an injected + empty object would then be missing it. + """ + incomplete = [] + for descriptor in _reachable_messages(AgentCard.DESCRIPTOR).values(): + for field in descriptor.fields: + message_type = field.message_type + if ( + not signing._is_required(field) + or message_type is None + or signing._is_map(field) + or signing._field_is_repeated(field) + ): + continue + if any(signing._is_required(f) for f in message_type.fields): + incomplete.append(f'{descriptor.name}.{field.name}') + assert incomplete == [] + + +def test_signature_over_spec_canonical_bytes_verifies(): + """A card signed over the section 8.4.1 form of its payload verifies. + + The signing input is built by hand, not by this module, which is how a + signer in another SDK that keeps REQUIRED fields at their default value + produces it. + """ + card = _example_card(description='', skills=[]) + payload = ( + '{"capabilities":{"pushNotifications":false,"streaming":false},' + '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],' + '"description":"","name":"Example Agent","skills":[],' + '"supportedInterfaces":[{"protocolBinding":"JSONRPC",' + '"protocolVersion":"1.0","url":"https://example.com/a2a/v1"}],' + '"version":"1.0.0"}' + ) + key = 'key12345' + token = api_jws.encode( + payload.encode('utf-8'), + key, + algorithm='HS384', + headers={'kid': 'key1'}, + ) + protected, _, signature = token.split('.') + card.signatures.append( + AgentCardSignature(protected=protected, signature=signature) + ) + verifier = signing.create_signature_verifier( + create_key_provider(key), ['HS384'] + ) + verifier(card) From a091c839ef636cb727f138e09d28b922060e87d1 Mon Sep 17 00:00:00 2001 From: Tymofii Pidlisnyi Date: Thu, 1 Oct 2026 12:49:01 -0700 Subject: [PATCH 3/3] revert: hold the REQUIRED-default change until a2aproject/A2A#2122 is decided This reverts commit cdee28e53f2166ca04fc360868cae9fc2b7a3182. The canonicalization scope is still open on a2aproject/A2A#2122. The branch keeps 3f90712, which changes no canonical bytes. --- src/a2a/utils/signing.py | 37 ------ tests/utils/test_signing.py | 227 +----------------------------------- 2 files changed, 3 insertions(+), 261 deletions(-) diff --git a/src/a2a/utils/signing.py b/src/a2a/utils/signing.py index 84788e35a..96566167e 100644 --- a/src/a2a/utils/signing.py +++ b/src/a2a/utils/signing.py @@ -3,7 +3,6 @@ from collections.abc import Callable from typing import Any, TypedDict -from google.api import field_behavior_pb2 as fb from google.protobuf.descriptor import Descriptor, FieldDescriptor from google.protobuf.json_format import MessageToDict @@ -198,11 +197,6 @@ def _clean_empty(d: Any, depth: int = 0) -> Any: return d -def _is_required(field: FieldDescriptor) -> bool: - """Returns True if the field carries google.api.field_behavior = REQUIRED.""" - return fb.REQUIRED in field.GetOptions().Extensions[fb.field_behavior] # type: ignore[index] # ty: ignore[invalid-argument-type] - - def _is_map(field: FieldDescriptor) -> bool: """Returns True if the field is a protobuf map.""" message_type = field.message_type @@ -214,25 +208,6 @@ def _is_well_known(descriptor: Descriptor | Any) -> bool: return descriptor.full_name.startswith('google.protobuf.') -def _required_default(field: FieldDescriptor) -> Any: - """Returns the JSON value a REQUIRED field has when it is empty, or None. - - A REQUIRED message is an empty object whether or not it was set, so the - REQUIRED set does not depend on what the serializer chose to emit. REQUIRED - fields of a scalar type other than string do not exist in the Agent Card - today and are left out. - """ - if _is_map(field): - return {} - if _field_is_repeated(field): - return [] - if field.message_type is not None: - return {} - if field.type == FieldDescriptor.TYPE_STRING: - return '' - return None - - def _clean_field(value: Any, field: FieldDescriptor, depth: int) -> Any: """Removes empty values from the JSON form of one message field.""" message_type = field.message_type @@ -265,9 +240,6 @@ def _clean_message( ) -> dict[str, Any]: """Removes empty values from the JSON form of a message, by descriptor. - REQUIRED fields are kept at their default value. Every other empty value - is removed. - `message_dict` is the `MessageToDict` output for a message of type `descriptor`. Walking the descriptor alongside the JSON keeps the field each value belongs to known at every level, which `_clean_empty` alone @@ -288,15 +260,6 @@ def _clean_message( cleaned_value = _clean_field(value, field, depth + 1) if cleaned_value is not None: cleaned[key] = cleaned_value - # A2A specification section 8.4.1: a REQUIRED field stays in the canonical - # payload even when its value matches the default. - for field in descriptor.fields: - name = field.json_name - if name in cleaned or not _is_required(field): - continue - default = _required_default(field) - if default is not None: - cleaned[name] = default return cleaned diff --git a/tests/utils/test_signing.py b/tests/utils/test_signing.py index 19a31df03..fd8939815 100644 --- a/tests/utils/test_signing.py +++ b/tests/utils/test_signing.py @@ -21,7 +21,6 @@ from a2a.utils import signing from cryptography.hazmat.primitives.asymmetric import ec from google.protobuf.json_format import MessageToDict -from jwt import api_jws from jwt.utils import base64url_encode @@ -196,8 +195,7 @@ def test_canonicalize_agent_card(sample_agent_card: AgentCard): """Test canonicalize_agent_card with defaults, optionals, and exceptions. - extensions is omitted as it's not set and optional. - - protocolVersion is REQUIRED on AgentInterface, so it is kept at its - default value (A2A specification section 8.4.1). + - protocolVersion is included because it's always added by canonicalize_agent_card. - signatures should be omitted. """ expected_jcs = ( @@ -205,7 +203,7 @@ def test_canonicalize_agent_card(sample_agent_card: AgentCard): '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],' '"description":"A test agent","name":"Test Agent",' '"skills":[{"description":"A test skill","id":"skill1","name":"Test Skill","tags":["test"]}],' - '"supportedInterfaces":[{"protocolBinding":"HTTP+JSON","protocolVersion":"","url":"http://localhost"}],' + '"supportedInterfaces":[{"protocolBinding":"HTTP+JSON","url":"http://localhost"}],' '"version":"1.0.0"}' ) result = signing._canonicalize_agent_card(sample_agent_card) @@ -367,43 +365,10 @@ def full_agent_card() -> AgentCard: return card -def test_required_fields_are_read_from_descriptors(): - """REQUIRED comes from `google.api.field_behavior`, not from a name list.""" - card = AgentCard.DESCRIPTOR - required = {f.json_name for f in card.fields if signing._is_required(f)} - assert required == { - 'name', - 'description', - 'supportedInterfaces', - 'version', - 'capabilities', - 'defaultInputModes', - 'defaultOutputModes', - 'skills', - } - skill = AgentSkill.DESCRIPTOR - assert {f.json_name for f in skill.fields if signing._is_required(f)} == { - 'id', - 'name', - 'description', - 'tags', - } - - -def test_implicit_presence_fields_are_not_all_required(): - """Fields that the blanket `always_print` flag emits stay non-REQUIRED.""" - tenant = AgentInterface.DESCRIPTOR.fields_by_name['tenant'] - ext_required = AgentExtension.DESCRIPTOR.fields_by_name['required'] - ext_list = AgentCapabilities.DESCRIPTOR.fields_by_name['extensions'] - assert not signing._is_required(tenant) - assert not signing._is_required(ext_required) - assert not signing._is_required(ext_list) - - def test_clean_message_matches_clean_empty_on_full_card( full_agent_card: AgentCard, ): - """With no empty REQUIRED field, descriptor-aware cleaning changes nothing.""" + """Descriptor-aware cleaning gives the same result as `_clean_empty`.""" card_dict = MessageToDict(full_agent_card) assert signing._clean_message( card_dict, AgentCard.DESCRIPTOR @@ -439,189 +404,3 @@ def test_clean_message_bounds_depth(): card_dict = {'capabilities': {'extensions': [{'params': nested}]}} with pytest.raises(signing.CanonicalizationError): signing._clean_message(card_dict, AgentCard.DESCRIPTOR) - - -def _reachable_messages(descriptor, seen=None): - seen = {} if seen is None else seen - if descriptor.full_name in seen or signing._is_well_known(descriptor): - return seen - seen[descriptor.full_name] = descriptor - for field in descriptor.fields: - message_type = field.message_type - if message_type is None: - continue - if signing._is_map(field): - message_type = message_type.fields_by_name['value'].message_type - if message_type is None: - continue - _reachable_messages(message_type, seen) - return seen - - -def _example_card(**overrides: Any) -> AgentCard: - fields: dict[str, Any] = { - 'name': 'Example Agent', - 'description': 'An example', - 'version': '1.0.0', - 'supported_interfaces': [ - AgentInterface( - url='https://example.com/a2a/v1', - protocol_binding='JSONRPC', - protocol_version='1.0', - ) - ], - 'capabilities': AgentCapabilities( - streaming=False, push_notifications=False - ), - 'default_input_modes': ['text/plain'], - 'default_output_modes': ['text/plain'], - 'skills': [ - AgentSkill( - id='skill1', name='Skill', description='A skill', tags=['t'] - ) - ], - } - fields.update(overrides) - return AgentCard(**fields) - - -def test_canonicalize_keeps_required_fields_at_default(): - """Section 8.4.1: `description: ""` and `skills: []` stay in the payload. - - The card is the worked example of section 8.4.1 with the remaining - REQUIRED fields of AgentCard filled in. - """ - card = _example_card(description='', skills=[]) - card.capabilities.extensions.extend([]) - expected_jcs = ( - '{"capabilities":{"pushNotifications":false,"streaming":false},' - '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],' - '"description":"","name":"Example Agent","skills":[],' - '"supportedInterfaces":[{"protocolBinding":"JSONRPC",' - '"protocolVersion":"1.0","url":"https://example.com/a2a/v1"}],' - '"version":"1.0.0"}' - ) - assert signing._canonicalize_agent_card(card) == expected_jcs - - -@pytest.mark.parametrize( - ('overrides', 'kept'), - [ - pytest.param({'description': ''}, '"description":""', id='description'), - pytest.param({'skills': []}, '"skills":[]', id='skills'), - pytest.param({'version': ''}, '"version":""', id='version'), - pytest.param( - {'default_input_modes': []}, - '"defaultInputModes":[]', - id='default-input-modes', - ), - pytest.param( - { - 'skills': [ - AgentSkill(id='s', name='Skill', description='A skill') - ] - }, - '"tags":[]', - id='skill-tags', - ), - ], -) -def test_canonicalize_keeps_each_empty_required_field( - overrides: dict[str, Any], kept: str -): - assert kept in signing._canonicalize_agent_card(_example_card(**overrides)) - - -def test_canonicalize_still_prunes_non_required_defaults(): - """Implicit-presence fields that are not REQUIRED stay out of the payload.""" - card = _example_card(description='', skills=[]) - card.supported_interfaces[0].tenant = '' - card.capabilities.extensions.add( - uri='https://example.com/ext', required=False - ) - result = signing._canonicalize_agent_card(card) - assert '"tenant"' not in result - assert '"required"' not in result - assert '"extensions":[{"uri":"https://example.com/ext"}]' in result - assert '"securitySchemes"' not in result - assert '"securityRequirements"' not in result - - -def test_canonicalize_keeps_an_unset_required_message(): - """A REQUIRED message is present even when it was never set.""" - card = _example_card() - card.ClearField('capabilities') - assert not card.HasField('capabilities') - assert '"capabilities":{}' in signing._canonicalize_agent_card(card) - - -def test_canonicalize_keeps_a_set_but_empty_required_message(): - card = _example_card(capabilities=AgentCapabilities()) - assert '"capabilities":{}' in signing._canonicalize_agent_card(card) - - -def test_every_required_field_has_a_default_this_module_can_emit(): - """Fails when a REQUIRED field of an unhandled scalar type is added.""" - unhandled = [] - for descriptor in _reachable_messages(AgentCard.DESCRIPTOR).values(): - for field in descriptor.fields: - if not signing._is_required(field): - continue - if signing._required_default(field) is None: - unhandled.append(f'{descriptor.name}.{field.name}') - assert unhandled == [] - - -def test_required_messages_have_no_required_fields_of_their_own(): - """`{}` is the complete default of every REQUIRED singular message. - - Fails when a REQUIRED message gains a REQUIRED field, because an injected - empty object would then be missing it. - """ - incomplete = [] - for descriptor in _reachable_messages(AgentCard.DESCRIPTOR).values(): - for field in descriptor.fields: - message_type = field.message_type - if ( - not signing._is_required(field) - or message_type is None - or signing._is_map(field) - or signing._field_is_repeated(field) - ): - continue - if any(signing._is_required(f) for f in message_type.fields): - incomplete.append(f'{descriptor.name}.{field.name}') - assert incomplete == [] - - -def test_signature_over_spec_canonical_bytes_verifies(): - """A card signed over the section 8.4.1 form of its payload verifies. - - The signing input is built by hand, not by this module, which is how a - signer in another SDK that keeps REQUIRED fields at their default value - produces it. - """ - card = _example_card(description='', skills=[]) - payload = ( - '{"capabilities":{"pushNotifications":false,"streaming":false},' - '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],' - '"description":"","name":"Example Agent","skills":[],' - '"supportedInterfaces":[{"protocolBinding":"JSONRPC",' - '"protocolVersion":"1.0","url":"https://example.com/a2a/v1"}],' - '"version":"1.0.0"}' - ) - key = 'key12345' - token = api_jws.encode( - payload.encode('utf-8'), - key, - algorithm='HS384', - headers={'kid': 'key1'}, - ) - protected, _, signature = token.split('.') - card.signatures.append( - AgentCardSignature(protected=protected, signature=signature) - ) - verifier = signing.create_signature_verifier( - create_key_provider(key), ['HS384'] - ) - verifier(card)