diff --git a/src/a2a/utils/signing.py b/src/a2a/utils/signing.py index c6540d162..da9070c86 100644 --- a/src/a2a/utils/signing.py +++ b/src/a2a/utils/signing.py @@ -1,3 +1,4 @@ +import binascii import json from collections.abc import Callable @@ -144,6 +145,13 @@ def signature_verifier( agent_card_signature.protected.encode('utf-8') ).decode('utf-8') protected_header = json.loads(protected_header_json) + except (binascii.Error, UnicodeDecodeError, json.JSONDecodeError): + continue + + if not isinstance(protected_header, dict): + continue + + try: kid = protected_header.get('kid') jku = protected_header.get('jku') verification_key = key_provider(kid, jku) diff --git a/tests/utils/test_signing.py b/tests/utils/test_signing.py index f22c60573..035b391d0 100644 --- a/tests/utils/test_signing.py +++ b/tests/utils/test_signing.py @@ -143,6 +143,32 @@ def test_signer_and_verifier_symmetric_multiple_signatures( verifier_wrong_key(signed_card) +def test_verifier_skips_malformed_protected_header( + sample_agent_card: AgentCard, +): + key = 'key12345' + signer = signing.create_agent_card_signer( + signing_key=key, + protected_header={'alg': 'HS384', 'kid': 'key1', 'jku': None}, + ) + verifier = signing.create_signature_verifier( + create_key_provider(key), ['HS256', 'HS384', 'ES256', 'RS256'] + ) + malformed = AgentCardSignature( + protected='!!!not-base64url!!!', signature='invalid' + ) + + invalid_card = AgentCard() + invalid_card.CopyFrom(sample_agent_card) + invalid_card.signatures.append(malformed) + with pytest.raises(signing.InvalidSignaturesError): + verifier(invalid_card) + + valid_card = signer(sample_agent_card) + valid_card.signatures.insert(0, malformed) + verifier(valid_card) + + def test_signer_and_verifier_asymmetric(sample_agent_card: AgentCard): """Test the agent card signing and verification process with an asymmetric key encryption.""" private_key = ec.generate_private_key(ec.SECP256R1())