fix: prefer JWS header kid over jwk.kid in attach decorator verify (fixes openwallet-foundation#4077)

When verifying JWS in attach decorator data, use kid from the JWS
unprotected header first (canonical per spec). Fall back to jwk.kid
only when header has no kid. Fixes DIDComm connection failure with
agents (e.g. Credo) that put kid only in header and not in jwk.

Backport of openwallet-foundation#4085 for 1.3.lts.

- Add test_verify_uses_kid_from_header_when_jwk_has_no_kid
- Add test_verify_uses_kid_from_jwk_when_header_has_no_kid
- Add test_verify_returns_false_when_signer_verkey_does_not_match

Signed-off-by: Patrick St-Louis <patrick.st-louis@opsecid.ca>
Made-with: Cursor

Author: PatStLouis

acapy_agent/messaging/decorators/attach_decorator.py

@@ -454,8 +454,15 @@ async def verify(
             if not await wallet.verify_message(sign_input, b_sig, verkey, ED25519):
                 return False
 
-            if "kid" in jwk:
-                encoded_pk = DIDKey.from_did(protected["jwk"]["kid"]).public_key_b58
+            # Prefer kid from JWS header (canonical per spec); fall back to jwk.kid
+            kid = None
+            if getattr(sig, "header", None) and getattr(sig.header, "kid", None):
+                kid = sig.header.kid
+            elif "kid" in jwk:
+                kid = protected["jwk"]["kid"]
+
+            if kid:
+                encoded_pk = DIDKey.from_did(kid).public_key_b58
                 verkey_to_check.append(encoded_pk)
                 if not await wallet.verify_message(
                     sign_input, b_sig, encoded_pk, ED25519

acapy_agent/messaging/decorators/tests/test_attach_decorator.py

@@ -1,6 +1,7 @@
 import json
 from copy import deepcopy
 from datetime import datetime, timezone
+from types import SimpleNamespace
 from unittest import TestCase
 
 import pytest
@@ -12,7 +13,7 @@
 from ....wallet.base import BaseWallet
 from ....wallet.did_method import SOV, DIDMethods
 from ....wallet.key_type import ED25519
-from ....wallet.util import b64_to_bytes, bytes_to_b64
+from ....wallet.util import b58_to_bytes, b64_to_bytes, bytes_to_b64, str_to_b64
 from ..attach_decorator import (
     AttachDecorator,
     AttachDecoratorData,
@@ -522,3 +523,84 @@ async def test_indy_sign(self, wallet, seed):
         deco_dict["data"]["links"] = "https://en.wikipedia.org/wiki/Potato"
         with pytest.raises(BaseModelError):
             AttachDecorator.deserialize(deco_dict)  # now has base64 and links
+
+    @pytest.mark.asyncio
+    async def test_verify_uses_kid_from_header_when_jwk_has_no_kid(self, wallet, seed):
+        """Verify uses kid from JWS header when jwk has no kid (Credo/interop case)."""
+        # Build a JWS where kid is only in the unprotected header, not in jwk.
+        # This is the format some agents (e.g. Credo) send; verification must use header.kid.
+        did_info = await wallet.create_local_did(SOV, ED25519, seed[0])
+        verkey_b58 = did_info.verkey
+        kid_full = did_key(verkey_b58)
+
+        payload = b"payload for header-kid test"
+        b64_payload = bytes_to_b64(payload)
+        # Protected header with jwk but NO kid in jwk
+        protected = {
+            "alg": "EdDSA",
+            "jwk": {
+                "kty": "OKP",
+                "crv": "Ed25519",
+                "x": bytes_to_b64(b58_to_bytes(verkey_b58), urlsafe=True, pad=False),
+            },
+        }
+        b64_protected = str_to_b64(
+            json.dumps(protected, separators=(",", ":")),
+            urlsafe=True,
+            pad=False,
+        )
+        sign_input = (b64_protected + "." + b64_payload).encode("ascii")
+        sig_bytes = await wallet.sign_message(
+            message=sign_input,
+            from_verkey=verkey_b58,
+        )
+        b64_sig = bytes_to_b64(sig_bytes, urlsafe=True, pad=False)
+
+        data = AttachDecoratorData.deserialize(
+            {
+                "base64": b64_payload,
+                "jws": {
+                    "header": {"kid": kid_full},
+                    "protected": b64_protected,
+                    "signature": b64_sig,
+                },
+            }
+        )
+        assert await data.verify(wallet)
+        assert await data.verify(wallet, signer_verkey=verkey_b58)
+
+    @pytest.mark.asyncio
+    async def test_verify_uses_kid_from_jwk_when_header_has_no_kid(self, wallet, seed):
+        """Verify uses jwk.kid when header has no kid (fallback path)."""
+        deco = AttachDecorator.data_base64(
+            mapping=INDY_CRED,
+            ident=IDENT,
+            description=DESCRIPTION,
+        )
+        did_info = await wallet.create_local_did(SOV, ED25519, seed[0])
+        await deco.data.sign(did_info.verkey, wallet)
+        # Force fallback: header has no kid, so verify must use jwk.kid from protected
+        deco.data.jws_.header = SimpleNamespace(kid=None)
+        assert await deco.data.verify(wallet)
+        assert await deco.data.verify(wallet, signer_verkey=did_info.verkey)
+
+    @pytest.mark.asyncio
+    async def test_verify_returns_false_when_signer_verkey_does_not_match(
+        self, wallet, seed
+    ):
+        """Verify returns False when signer_verkey is not the signing key."""
+        deco = AttachDecorator.data_base64(
+            mapping=INDY_CRED,
+            ident=IDENT,
+            description=DESCRIPTION,
+        )
+        did_infos = [
+            await wallet.create_local_did(SOV, ED25519, seed[i]) for i in [0, 1]
+        ]
+        await deco.data.sign(did_infos[0].verkey, wallet)
+        # Sign with key 0; require key 1 -> should fail
+        assert not await deco.data.verify(wallet, signer_verkey=did_infos[1].verkey)
+        # No signer_verkey constraint -> should pass
+        assert await deco.data.verify(wallet)
+        # Correct signer_verkey -> should pass
+        assert await deco.data.verify(wallet, signer_verkey=did_infos[0].verkey)