Merge pull request #3279 from dbluhm/feature/vm-selection

More robust verification method selection by did

acapy_agent/protocols/present_proof/dif/pres_exch_handler.py

@@ -40,7 +40,10 @@
 from ....vc.vc_di.prove import create_signed_anoncreds_presentation
 from ....vc.vc_ld.prove import create_presentation, derive_credential, sign_presentation
 from ....wallet.base import BaseWallet, DIDInfo
-from ....wallet.default_verification_key_strategy import BaseVerificationKeyStrategy
+from ....wallet.default_verification_key_strategy import (
+    BaseVerificationKeyStrategy,
+    ProofPurposeStr,
+)
 from ....wallet.error import WalletError, WalletNotFoundError
 from ....wallet.key_type import BLS12381G2, ED25519
 from .pres_exch import (
@@ -115,19 +118,19 @@ async def _get_issue_suite(
         self,
         *,
         issuer_id: str,
+        proof_purpose: Optional[ProofPurposeStr] = None,
     ):
         """Get signature suite for signing presentation."""
+        proof_purpose = proof_purpose or "assertionMethod"
         did_info = await self._did_info_for_did(issuer_id)
-        verkey_id_strategy = self.profile.context.inject(BaseVerificationKeyStrategy)
-        verification_method = await verkey_id_strategy.get_verification_method_id_for_did(
-            issuer_id, self.profile, proof_purpose="assertionMethod"
+        vm_id_strategy = self.profile.context.inject(BaseVerificationKeyStrategy)
+        verification_method = await vm_id_strategy.get_verification_method_id_for_did(
+            issuer_id,
+            self.profile,
+            proof_type=self.proof_type,
+            proof_purpose=proof_purpose,
         )
 
-        if verification_method is None:
-            raise DIFPresExchError(
-                f"Unable to get retrieve verification method for did {issuer_id}"
-            )
-
         # Get signature class based on proof type
         SignatureClass = self.PROOF_TYPE_SIGNATURE_SUITE_MAPPING[self.proof_type]
 
@@ -1302,8 +1305,9 @@ async def create_vp(
                 )
             else:
                 vp = self.__add_dif_fields_to_vp(vp, submission_property)
+                assert issuer_id
                 issue_suite = await self._get_issue_suite(
-                    issuer_id=issuer_id,
+                    issuer_id=issuer_id, proof_purpose="authentication"
                 )
                 signed_vp = await sign_presentation(
                     presentation=vp,

acapy_agent/vc/vc_ld/manager.py

@@ -344,15 +344,13 @@ async def _get_suite_for_document(
         verification_method = (
             options.verification_method
             or await verkey_id_strategy.get_verification_method_id_for_did(
-                issuer_id, self.profile, proof_purpose="assertionMethod"
+                issuer_id,
+                self.profile,
+                proof_type=proof_type,
+                proof_purpose="assertionMethod",
             )
         )
 
-        if verification_method is None:
-            raise VcLdpManagerError(
-                f"Unable to get retrieve verification method for did {issuer_id}"
-            )
-
         suite = await self._get_suite(
             proof_type=proof_type,
             verification_method=verification_method,

acapy_agent/wallet/default_verification_key_strategy.py

@@ -1,11 +1,35 @@
 """Utilities for specifying which verification method is in use for a given DID."""
 
 from abc import ABC, abstractmethod
-from typing import List, Optional
+import logging
+from typing import Literal, Optional
 
-from acapy_agent.core.profile import Profile
-from acapy_agent.did.did_key import DIDKey
-from acapy_agent.wallet.key_type import KeyType
+from pydid import DIDDocument
+
+from ..core.error import BaseError
+from ..core.profile import Profile
+from ..did.did_key import DIDKey
+from ..resolver.did_resolver import DIDResolver
+
+LOGGER = logging.getLogger(__name__)
+
+
+ProofPurposeStr = Literal[
+    "assertionMethod",
+    "authentication",
+    "capabilityDelegation",
+    "capabilityInvocation",
+]
+PROOF_PURPOSES = (
+    "authentication",
+    "assertionMethod",
+    "capabilityInvocation",
+    "capabilityDelegation",
+)
+
+
+class VerificationKeyStrategyError(BaseError):
+    """Raised on issues with verfication method derivation."""
 
 
 class BaseVerificationKeyStrategy(ABC):
@@ -15,10 +39,11 @@ class BaseVerificationKeyStrategy(ABC):
     async def get_verification_method_id_for_did(
         self,
         did: str,
-        profile: Optional[Profile],
-        allowed_verification_method_types: Optional[List[KeyType]] = None,
-        proof_purpose: Optional[str] = None,
-    ) -> Optional[str]:
+        profile: Profile,
+        *,
+        proof_type: Optional[str] = None,
+        proof_purpose: Optional[ProofPurposeStr] = None,
+    ) -> str:
         """Given a DID, returns the verification key ID in use.
 
         Returns None if no strategy is specified for this DID.
@@ -29,7 +54,7 @@ async def get_verification_method_id_for_did(
         :params proof_purpose: the verkey relationship (assertionMethod, keyAgreement, ..)
         :returns Optional[str]: the current verkey ID
         """
-        pass
+        ...
 
 
 class DefaultVerificationKeyStrategy(BaseVerificationKeyStrategy):
@@ -38,13 +63,21 @@ class DefaultVerificationKeyStrategy(BaseVerificationKeyStrategy):
     Supports did:key: and did:sov only.
     """
 
+    def __init__(self):
+        """Initialize the key types mapping."""
+        self.key_types_mapping = {
+            "Ed25519Signature2018": ["Ed25519VerificationKey2018"],
+            "Ed25519Signature2020": ["Ed25519VerificationKey2020", "Multikey"],
+        }
+
     async def get_verification_method_id_for_did(
         self,
         did: str,
-        profile: Optional[Profile],
-        allowed_verification_method_types: Optional[List[KeyType]] = None,
-        proof_purpose: Optional[str] = None,
-    ) -> Optional[str]:
+        profile: Profile,
+        *,
+        proof_type: Optional[str] = None,
+        proof_purpose: Optional[ProofPurposeStr] = None,
+    ) -> str:
         """Given a did:key or did:sov, returns the verification key ID in use.
 
         Returns None if no strategy is specified for this DID.
@@ -55,10 +88,55 @@ async def get_verification_method_id_for_did(
         :params proof_purpose: the verkey relationship (assertionMethod, keyAgreement, ..)
         :returns Optional[str]: the current verkey ID
         """
+        proof_type = proof_type or "Ed25519Signature2018"
+        proof_purpose = proof_purpose or "assertionMethod"
+
+        if proof_purpose not in PROOF_PURPOSES:
+            raise ValueError("Invalid proof purpose")
+
         if did.startswith("did:key:"):
             return DIDKey.from_did(did).key_id
         elif did.startswith("did:sov:"):
             # key-1 is what uniresolver uses for key id
             return did + "#key-1"
 
-        return None
+        resolver = profile.inject(DIDResolver)
+        doc_raw = await resolver.resolve(profile=profile, did=did)
+        doc = DIDDocument.deserialize(doc_raw)
+
+        methods_or_refs = getattr(doc, proof_purpose, [])
+        # Dereference any refs in the verification relationship
+        methods = [
+            await resolver.dereference_verification_method(profile, method, document=doc)
+            if isinstance(method, str)
+            else method
+            for method in methods_or_refs
+        ]
+
+        method_types = self.key_types_mapping.get(proof_type)
+        if not method_types:
+            raise VerificationKeyStrategyError(
+                f"proof type {proof_type} is not supported"
+            )
+
+        # Filter methods by type expected for proof_type
+        methods = [vm for vm in methods if vm.type in method_types]
+        if not methods:
+            raise VerificationKeyStrategyError(
+                f"No matching verification method found for did {did} with proof "
+                f"type {proof_type} and purpose {proof_purpose}"
+            )
+
+        if len(methods) > 1:
+            LOGGER.info(
+                (
+                    "More than 1 verification method matched for did %s with proof "
+                    "type %s and purpose %s; returning the first: %s"
+                ),
+                did,
+                proof_type,
+                proof_purpose,
+                methods[0].id,
+            )
+
+        return methods[0].id

acapy_agent/wallet/jwt.py

@@ -61,8 +61,6 @@ async def jwt_sign(
         verification_method = await verkey_strat.get_verification_method_id_for_did(
             did, profile
         )
-        if not verification_method:
-            raise ValueError("Could not determine verification method from DID")
     else:
         # We look up keys by did for now
         did = DIDUrl.parse(verification_method).did

acapy_agent/wallet/tests/test_default_verification_key_strategy.py

@@ -1,4 +1,7 @@
 from unittest import IsolatedAsyncioTestCase
+import pytest
+
+from acapy_agent.resolver.did_resolver import DIDResolver
 
 from ...did.did_key import DIDKey
 from ...utils.testing import create_test_profile
@@ -13,6 +16,8 @@
 class TestDefaultVerificationKeyStrategy(IsolatedAsyncioTestCase):
     async def asyncSetUp(self) -> None:
         self.profile = await create_test_profile()
+        resolver = DIDResolver()
+        self.profile.context.injector.bind_instance(DIDResolver, resolver)
 
     async def test_with_did_sov(self):
         strategy = DefaultVerificationKeyStrategy()
@@ -30,9 +35,7 @@ async def test_with_did_key(self):
 
     async def test_unsupported_did_method(self):
         strategy = DefaultVerificationKeyStrategy()
-        assert (
+        with pytest.raises(Exception):
             await strategy.get_verification_method_id_for_did(
                 "did:test:test", self.profile
             )
-            is None
-        )