Fix: add an algorithms parameter to create_signature_verifier function to security measures.

Author: sokoliva

src/a2a/utils/signing.py

@@ -11,7 +11,7 @@
     from jose.utils import base64url_decode, base64url_encode
 except ImportError as e:
     raise ImportError(
-        'A2AUtilsSigning requires python-jose to be installed. '
+        'A2A Signing requires python-jose to be installed. '
         'Install with: '
         "'pip install a2a-sdk[signing]'"
     ) from e
@@ -70,7 +70,7 @@ def create_agent_card_signer(
         signing_key: The private key for signing.
         kid: Key ID for the signing key.
         alg: The algorithm to use (e.g., "ES256", "RS256").
-        jku: Optional URL to the JWKS.
+        jku: Optional URL to the JSON Web Keys.
 
     Returns:
         A callable that takes an AgentCard and returns the modified AgentCard with a signature.
@@ -111,11 +111,13 @@ def create_signature_verifier(
     key_provider: Callable[
         [str | None, str | None], str | bytes | dict[str, Any] | Key
     ],
+    algorithms: list[str],
 ) -> Callable[[AgentCard], None]:
     """Creates a function that verifies AgentCard signatures.
 
     Args:
-        key_provider: A callable that takes key-id (kid) and JSON web key url (jku) and returns the verification key.
+        key_provider: A callable that takes key-id and JSON web key url and returns the verification key.
+        algorithms: List of acceptable algorithms for verification used to prevent algorithm confusion attacks.
 
     Returns:
         A callable that takes an AgentCard, and raises an error if none of the signatures are valid.
@@ -138,7 +140,6 @@ def signature_verifier(
                 protected_header = json.loads(protected_header_json)
                 kid = protected_header.get('kid')
                 jku = protected_header.get('jku')
-                alg = protected_header.get('alg')
                 verification_key = key_provider(kid, jku)
 
                 canonical_payload = canonicalize_agent_card(agent_card)
@@ -150,7 +151,7 @@ def signature_verifier(
                 jws.verify(
                     token=token,
                     key=verification_key,
-                    algorithms=[alg] if alg else None,
+                    algorithms=algorithms,
                 )
                 # Found a valid signature, exit the loop and function
                 break

tests/integration/test_client_server_integration.py

@@ -878,7 +878,9 @@ async def test_json_transport_get_signed_base_card_no_initial(
     )
 
     # Get the card, this will trigger verification in get_card
-    signature_verifier = create_signature_verifier(create_key_provider(key))
+    signature_verifier = create_signature_verifier(
+        create_key_provider(key), ['HS384']
+    )
     result = await transport.get_card(signature_verifier=signature_verifier)
     assert result.name == agent_card.name
     assert result.signatures is not None
@@ -929,7 +931,7 @@ async def test_json_transport_get_signed_extended_card(
 
     # Get the card, this will trigger verification in get_card
     signature_verifier = create_signature_verifier(
-        create_key_provider(public_key)
+        create_key_provider(public_key), ['HS384', 'ES256']
     )
     result = await transport.get_card(signature_verifier=signature_verifier)
     assert result.name == extended_agent_card.name
@@ -985,7 +987,7 @@ async def test_json_transport_get_signed_base_and_extended_cards(
 
     # Get the card, this will trigger verification in get_card
     signature_verifier = create_signature_verifier(
-        create_key_provider(public_key)
+        create_key_provider(public_key), ['HS384', 'ES256', 'RS256']
     )
     result = await transport.get_card(signature_verifier=signature_verifier)
     assert result.name == extended_agent_card.name
@@ -1039,8 +1041,9 @@ async def test_rest_transport_get_signed_card(
     )
 
     # Get the card, this will trigger verification in get_card
-    signature_verifier = create_signature_verifier(
-        create_key_provider(public_key)
+    signature_verifier = (
+        create_key_provider(public_key),
+        ['HS384', 'ES256', 'RS256'],
     )
     result = await transport.get_card(signature_verifier=signature_verifier)
     assert result.name == extended_agent_card.name
@@ -1093,7 +1096,7 @@ def channel_factory(address: str) -> Channel:
 
         # Get the card, this will trigger verification in get_card
         signature_verifier = create_signature_verifier(
-            create_key_provider(public_key)
+            create_key_provider(public_key), ['HS384', 'ES256', 'RS256']
         )
         result = await transport.get_card(signature_verifier=signature_verifier)
         assert result.signatures is not None

tests/utils/test_signing.py

@@ -74,15 +74,17 @@ def test_signer_and_verifier_symmetric(sample_agent_card: AgentCard):
     assert signature.signature is not None
 
     # Verify the signature
-    verifier = create_signature_verifier(create_key_provider(key))
+    verifier = create_signature_verifier(
+        create_key_provider(key), ['HS256', 'HS384', 'ES256', 'RS256']
+    )
     try:
         verifier(signed_card)
     except JOSEError:
         pytest.fail('Signature verification failed with correct key')
 
     # Verify with wrong key
     verifier_wrong_key = create_signature_verifier(
-        create_key_provider(wrong_key)
+        create_key_provider(wrong_key), ['HS256', 'HS384', 'ES256', 'RS256']
     )
     with pytest.raises(JOSEError):
         verifier_wrong_key(signed_card)
@@ -114,15 +116,17 @@ def test_signer_and_verifier_symmetric_multiple_signatures(
     assert signature.signature is not None
 
     # Verify the signature
-    verifier = create_signature_verifier(create_key_provider(key))
+    verifier = create_signature_verifier(
+        create_key_provider(key), ['HS256', 'HS384', 'ES256', 'RS256']
+    )
     try:
         verifier(signed_card)
     except JOSEError:
         pytest.fail('Signature verification failed with correct key')
 
     # Verify with wrong key
     verifier_wrong_key = create_signature_verifier(
-        create_key_provider(wrong_key)
+        create_key_provider(wrong_key), ['HS256', 'HS384', 'ES256', 'RS256']
     )
     with pytest.raises(JOSEError):
         verifier_wrong_key(signed_card)
@@ -150,15 +154,18 @@ def test_signer_and_verifier_asymmetric(sample_agent_card: AgentCard):
     assert signature.protected is not None
     assert signature.signature is not None
 
-    verifier = create_signature_verifier(create_key_provider(public_key))
+    verifier = create_signature_verifier(
+        create_key_provider(public_key), ['HS256', 'HS384', 'ES256', 'RS256']
+    )
     try:
         verifier(signed_card)
     except JOSEError:
         pytest.fail('Signature verification failed with correct key')
 
     # Verify with wrong key
     verifier_wrong_key = create_signature_verifier(
-        create_key_provider(public_key_error)
+        create_key_provider(public_key_error),
+        ['HS256', 'HS384', 'ES256', 'RS256'],
     )
     with pytest.raises(JOSEError):
         verifier_wrong_key(signed_card)
@@ -173,9 +180,11 @@ def test_canonicalize_agent_card(
     - protocolVersion is included because it's always added by canonicalize_agent_card.
     - signatures should be omitted.
     """
-    sample_agent_card.signatures = (
-        [{'protected': 'protected_header', 'signature': 'test_signature'}],
-    )
+    sample_agent_card.signatures = [
+        AgentCardSignature(
+            protected='protected_header', signature='test_signature'
+        )
+    ]
     expected_jcs = (
         '{"capabilities":{"pushNotifications":true},'
         '"defaultInputModes":["text/plain"],"defaultOutputModes":["text/plain"],'