feat: add did:jwk resolver

aries_cloudagent/resolver/__init__.py

@@ -29,6 +29,12 @@ async def setup(context: InjectionContext):
     await key_resolver.setup(context)
     registry.register_resolver(key_resolver)
 
+    jwk_resolver = ClassProvider(
+        "aries_cloudagent.resolver.default.jwk.JwkDIDResolver"
+    ).provide(context.settings, context.injector)
+    await jwk_resolver.setup(context)
+    registry.register_resolver(jwk_resolver)
+
     if not context.settings.get("ledger.disabled"):
         indy_resolver = ClassProvider(
             "aries_cloudagent.resolver.default.indy.IndyDIDResolver"

aries_cloudagent/resolver/default/jwk.py

@@ -0,0 +1,71 @@
+"""did:jwk: resolver implementation."""
+
+import re
+from typing import Optional, Pattern, Sequence, Text
+from aries_cloudagent.config.injection_context import InjectionContext
+from aries_cloudagent.core.profile import Profile
+from aries_cloudagent.resolver.base import BaseDIDResolver, ResolverType, ResolverError
+from aries_cloudagent.wallet.jwt import b64_to_dict
+
+
+class JwkDIDResolver(BaseDIDResolver):
+    """did:jwk: resolver implementation."""
+
+    PATTERN = re.compile(r"^did:jwk:(?P<did>.*)$")
+
+    def __init__(self):
+        """Initialize the resolver."""
+        super().__init__(ResolverType.NATIVE)
+
+    async def setup(self, context: InjectionContext):
+        """Perform required setup for the resolver."""
+        pass
+
+    @property
+    def supported_did_regex(self) -> Pattern:
+        """Return supported DID regex."""
+        return self.PATTERN
+
+    async def _resolve(
+        self,
+        profile: Profile,
+        did: str,
+        service_accept: Optional[Sequence[Text]] = None,
+    ) -> dict:
+        """Resolve a DID."""
+        if match := self.PATTERN.match(did):
+            encoded = match.group("did")
+        else:
+            raise ResolverError(f"Invalid DID: {did}")
+
+        jwk = b64_to_dict(encoded)
+        doc = {
+            "@context": [
+                "https://www.w3.org/ns/did/v1",
+                "https://w3id.org/security/suites/jws-2020/v1",
+            ],
+            "id": f"did:jwk:{encoded}",
+            "verificationMethod": [
+                {
+                    "id": f"did:jwk:{encoded}#0",
+                    "type": "JsonWebKey2020",
+                    "controller": f"did:jwk:{encoded}",
+                    "publicKeyJwk": jwk,
+                }
+            ],
+        }
+
+        use = jwk.get("use")
+        if use == "sig":
+            doc.update(
+                {
+                    "assertionMethod": [f"did:jwk:{encoded}#0"],
+                    "authentication": [f"did:jwk:{encoded}#0"],
+                    "capabilityInvocation": [f"did:jwk:{encoded}#0"],
+                    "capabilityDelegation": [f"did:jwk:{encoded}#0"],
+                }
+            )
+        elif use == "enc":
+            doc.update({"keyAgreement": [f"did:jwk:{encoded}#0"]})
+
+        return doc

aries_cloudagent/resolver/default/tests/test_jwk.py

@@ -0,0 +1,54 @@
+"""Test JwkDIDResolver."""
+
+import pytest
+
+from ....core.in_memory import InMemoryProfile
+from ....core.profile import Profile
+
+from ...base import DIDMethodNotSupported
+from ..jwk import JwkDIDResolver
+
+# pylint: disable=W0621
+TEST_DIDS = [
+    "did:jwk:eyJjcnYiOiJQLTI1NiIsImt0eSI6IkVDIiwieCI6ImFjYklRaXVNczNpOF91c3pFakoydHBUdFJNNEVVM3l6OTFQSDZDZEgyVjAiLCJ5IjoiX0tjeUxqOXZXTXB0bm1LdG00NkdxRHo4d2Y3NEk1TEtncmwyR3pIM25TRSJ9",
+    "did:jwk:eyJrdHkiOiJPS1AiLCJjcnYiOiJYMjU1MTkiLCJ1c2UiOiJlbmMiLCJ4IjoiM3A3YmZYdDl3YlRUVzJIQzdPUTFOei1EUThoYmVHZE5yZngtRkctSUswOCJ9",
+]
+
+TEST_DID_INVALID = "did:key:z1MkmjY8GnV5i9YTDtPETC2uUAW6ejw3nk5mXF5yci5ab7th"
+
+
+@pytest.fixture
+def resolver():
+    """Resolver fixture."""
+    yield JwkDIDResolver()
+
+
+@pytest.fixture
+def profile():
+    """Profile fixture."""
+    profile = InMemoryProfile.test_profile()
+    yield profile
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("did", TEST_DIDS)
+async def test_supported_did_regex(profile, resolver: JwkDIDResolver, did: str):
+    """Test the supported_did_regex."""
+    assert await resolver.supports(
+        profile,
+        did,
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("did", TEST_DIDS)
+async def test_resolve(resolver: JwkDIDResolver, profile: Profile, did: str):
+    """Test resolve method."""
+    assert await resolver.resolve(profile, did)
+
+
+@pytest.mark.asyncio
+async def test_resolve_x_did_not_found(resolver: JwkDIDResolver, profile: Profile):
+    """Test resolve method when no did is found."""
+    with pytest.raises(DIDMethodNotSupported):
+        await resolver.resolve(profile, TEST_DID_INVALID)