Feature multikey management

aries_cloudagent/config/default_context.py

@@ -140,6 +140,7 @@ async def load_plugins(self, context: InjectionContext):
         plugin_registry.register_plugin("aries_cloudagent.settings")
         plugin_registry.register_plugin("aries_cloudagent.vc")
         plugin_registry.register_plugin("aries_cloudagent.wallet")
+        plugin_registry.register_plugin("aries_cloudagent.wallet.keys")
 
         anoncreds_plugins = [
             "aries_cloudagent.anoncreds",

aries_cloudagent/vc/tests/test_bbs_mattr_interop.py

@@ -47,6 +47,7 @@ async def asyncSetUp(self):
             "verkey": public_key_base58,
             "metadata": {},
             "key_type": BLS12381G2,
+            "kid": None,
         }
 
         self.signature_issuer_suite = BbsBlsSignature2020(

aries_cloudagent/wallet/askar.py

@@ -114,8 +114,7 @@ async def create_key(
                     "Verification key already present in wallet"
                 ) from None
             raise WalletError("Error creating signing key") from err
-
-        return KeyInfo(verkey=verkey, metadata=metadata, key_type=key_type)
+        return KeyInfo(verkey=verkey, metadata=metadata, key_type=key_type, kid=kid)
 
     async def assign_kid_to_key(self, verkey: str, kid: str) -> KeyInfo:
         """Assign a KID to a key.
@@ -143,7 +142,7 @@ async def assign_kid_to_key(self, verkey: str, kid: str) -> KeyInfo:
             raise WalletError(f"Unknown key type {key.algorithm.value}")
 
         await self._session.handle.update_key(name=verkey, tags={"kid": kid})
-        return KeyInfo(verkey=verkey, metadata=metadata, key_type=key_type)
+        return KeyInfo(verkey=verkey, metadata=metadata, key_type=key_type, kid=kid)
 
     async def get_key_by_kid(self, kid: str) -> KeyInfo:
         """Fetch a key by looking up its kid.
@@ -170,7 +169,7 @@ async def get_key_by_kid(self, kid: str) -> KeyInfo:
         if not key_type:
             raise WalletError(f"Unknown key type {key.algorithm.value}")
 
-        return KeyInfo(verkey=verkey, metadata=metadata, key_type=key_type)
+        return KeyInfo(verkey=verkey, metadata=metadata, key_type=key_type, kid=kid)
 
     async def get_signing_key(self, verkey: str) -> KeyInfo:
         """Fetch info for a signing keypair.
@@ -194,7 +193,8 @@ async def get_signing_key(self, verkey: str) -> KeyInfo:
             raise WalletNotFoundError("Unknown key: {}".format(verkey))
         metadata = json.loads(key.metadata or "{}")
         # FIXME implement key types
-        return KeyInfo(verkey=verkey, metadata=metadata, key_type=ED25519)
+        kid = key.tags["kid"] if "kid" in key.tags else None
+        return KeyInfo(verkey=verkey, metadata=metadata, key_type=ED25519, kid=kid)
 
     async def replace_signing_key_metadata(self, verkey: str, metadata: dict):
         """Replace the metadata associated with a signing keypair.

aries_cloudagent/wallet/did_info.py

@@ -7,9 +7,16 @@
 
 INVITATION_REUSE_KEY = "invitation_reuse"
 
-KeyInfo = NamedTuple(
-    "KeyInfo", [("verkey", str), ("metadata", dict), ("key_type", KeyType)]
-)
+
+class KeyInfo(NamedTuple):
+    """Class returning key information."""
+
+    verkey: str
+    metadata: dict
+    key_type: KeyType
+    kid: str = None
+
+
 DIDInfo = NamedTuple(
     "DIDInfo",
     [

aries_cloudagent/wallet/in_memory.py

@@ -95,6 +95,7 @@ async def create_key(
             verkey=verkey_enc,
             metadata=self.profile.keys[verkey_enc]["metadata"].copy(),
             key_type=key_type,
+            kid=kid,
         )
 
     async def assign_kid_to_key(self, verkey: str, kid: str) -> KeyInfo:
@@ -120,6 +121,7 @@ async def assign_kid_to_key(self, verkey: str, kid: str) -> KeyInfo:
             verkey=key["verkey"],
             metadata=key["metadata"].copy(),
             key_type=key["key_type"],
+            kid=kid,
         )
 
     async def get_key_by_kid(self, kid: str) -> KeyInfo:
@@ -138,6 +140,7 @@ async def get_key_by_kid(self, kid: str) -> KeyInfo:
                     verkey=key["verkey"],
                     metadata=key["metadata"].copy(),
                     key_type=key["key_type"],
+                    kid=key["kid"],
                 )
 
         raise WalletNotFoundError(f"Key not found with kid {kid}")
@@ -162,6 +165,7 @@ async def get_signing_key(self, verkey: str) -> KeyInfo:
             verkey=key["verkey"],
             metadata=key["metadata"].copy(),
             key_type=key["key_type"],
+            kid=key["kid"],
         )
 
     async def replace_signing_key_metadata(self, verkey: str, metadata: dict):

aries_cloudagent/wallet/keys/manager.py

@@ -0,0 +1,99 @@
+"""Multikey class."""
+
+from ...core.profile import ProfileSession
+from ..base import BaseWallet
+from ..key_type import ED25519
+from ..util import b58_to_bytes, bytes_to_b58
+from ...utils.multiformats import multibase
+from ...wallet.error import WalletNotFoundError
+
+DEFAULT_ALG = "ed25519"
+ALG_MAPPINGS = {
+    "ed25519": {"key_type": ED25519, "prefix_hex": "ed01", "prefix_length": 2}
+}
+
+
+class MultikeyManagerError(Exception):
+    """Generic MultikeyManager Error."""
+
+
+class MultikeyManager:
+    """Class for managing wallet keys."""
+
+    def __init__(self, session: ProfileSession):
+        """Initialize the MultikeyManager."""
+        self.session = session
+
+    def _multikey_to_verkey(self, multikey: str, alg: str = DEFAULT_ALG):
+        """Transform multikey to verkey."""
+        prefix_length = ALG_MAPPINGS[alg]["prefix_length"]
+        public_bytes = bytes(bytearray(multibase.decode(multikey))[prefix_length:])
+        return bytes_to_b58(public_bytes)
+
+    def _verkey_to_multikey(self, verkey: str, alg: str = DEFAULT_ALG):
+        """Transform verkey to multikey."""
+        prefix_hex = ALG_MAPPINGS[alg]["prefix_hex"]
+        prefixed_key_hex = f"{prefix_hex}{b58_to_bytes(verkey).hex()}"
+        return multibase.encode(bytes.fromhex(prefixed_key_hex), "base58btc")
+
+    async def kid_exists(self, wallet: BaseWallet, kid: str):
+        """Check if kid exists."""
+        try:
+            key = await wallet.get_key_by_kid(kid=kid)
+            if key:
+                return True
+        except (WalletNotFoundError, AttributeError):
+            return False
+
+    async def from_kid(self, kid: str):
+        """Fetch a single key."""
+        wallet: BaseWallet | None = self.session.inject_or(BaseWallet)
+        key_info = await wallet.get_key_by_kid(kid=kid)
+        return {
+            "kid": key_info.kid,
+            "multikey": self._verkey_to_multikey(key_info.verkey),
+        }
+
+    async def from_multikey(self, multikey: str):
+        """Fetch a single key."""
+        wallet: BaseWallet | None = self.session.inject_or(BaseWallet)
+        key_info = await wallet.get_signing_key(verkey=self._multikey_to_verkey(multikey))
+        return {
+            "kid": key_info.kid,
+            "multikey": self._verkey_to_multikey(key_info.verkey),
+        }
+
+    async def create(self, seed: str = None, kid: str = None, alg: str = DEFAULT_ALG):
+        """Create a new key pair."""
+
+        if alg not in ALG_MAPPINGS:
+            raise MultikeyManagerError(
+                f"Unknown key algorithm, use one of {list(ALG_MAPPINGS.keys())}."
+            )
+
+        wallet: BaseWallet | None = self.session.inject_or(BaseWallet)
+
+        if kid and await self.kid_exists(wallet=wallet, kid=kid):
+            raise MultikeyManagerError(f"kid '{kid}' already exists in wallet.")
+
+        key_type = ALG_MAPPINGS[alg]["key_type"]
+        key_info = await wallet.create_key(key_type=key_type, seed=seed, kid=kid)
+        return {
+            "kid": key_info.kid,
+            "multikey": self._verkey_to_multikey(key_info.verkey),
+        }
+
+    async def update(self, multikey: str, kid: str):
+        """Assign a new kid to a key pair."""
+
+        wallet: BaseWallet | None = self.session.inject_or(BaseWallet)
+
+        if kid and await self.kid_exists(wallet=wallet, kid=kid):
+            raise MultikeyManagerError(f"kid '{kid}' already exists in wallet.")
+        key_info = await wallet.assign_kid_to_key(
+            verkey=self._multikey_to_verkey(multikey), kid=kid
+        )
+        return {
+            "kid": key_info.kid,
+            "multikey": self._verkey_to_multikey(key_info.verkey),
+        }