feat: add legacy interfaces

Signed-off-by: Daniel Bluhm <[email protected]>

Author: dbluhm

didcomm_messaging/legacy/askar.py

@@ -0,0 +1,133 @@
+"""Legacy DIDComm v1 Crypto implementation."""
+
+from collections import OrderedDict
+from typing import Optional, Sequence, Tuple, cast
+
+from base58 import b58decode, b58encode
+
+from didcomm_messaging.crypto.jwe import JweBuilder, JweEnvelope, JweRecipient
+from didcomm_messaging.legacy.base import (
+    LegacyUnpackResult,
+    LegacyCryptoService,
+    RecipData,
+)
+
+try:
+    from aries_askar import Key, KeyAlg, crypto_box
+    from aries_askar.bindings import key_get_secret_bytes
+    from didcomm_messaging.crypto.backend.askar import AskarKey, AskarSecretKey
+except ImportError:
+    raise ImportError("Legacy Askar backend requires the 'askar' extra to be installed")
+
+
+class AskarLegacyCryptoService(LegacyCryptoService[AskarKey, AskarSecretKey]):
+    """Legacy crypto service implementation for askar."""
+
+    def kid_to_public_key(self, kid: str) -> AskarKey:
+        """Get a public key from a kid.
+
+        In DIDComm v1, kids are the base58 encoded keys.
+        """
+        return AskarKey(Key.from_public_bytes(KeyAlg.ED25519, b58decode(kid)), kid)
+
+    async def pack_message(
+        self,
+        to_verkeys: Sequence[AskarKey],
+        from_key: Optional[AskarSecretKey],
+        message: bytes,
+    ) -> JweEnvelope:
+        """Encode a message using the DIDComm v1 'pack' algorithm."""
+        builder = JweBuilder(
+            with_protected_recipients=True, with_flatten_recipients=False
+        )
+        cek = Key.generate(KeyAlg.C20P)
+        # avoid converting to bytes object: this way the only copy is zeroed afterward
+        # tell type checking it's bytes to make it happy
+        cek_b = cast(bytes, key_get_secret_bytes(cek._handle))
+        sender_vk = (
+            b58encode(from_key.key.get_public_bytes()).decode("utf-8")
+            if from_key
+            else None
+        )
+        sender_xk = from_key.key.convert_key(KeyAlg.X25519) if from_key else None
+
+        for target_vk in to_verkeys:
+            target_xk = target_vk.key.convert_key(KeyAlg.X25519)
+            if sender_vk and sender_xk:
+                enc_sender = crypto_box.crypto_box_seal(target_xk, sender_vk)
+                nonce = crypto_box.random_nonce()
+                enc_cek = crypto_box.crypto_box(target_xk, sender_xk, cek_b, nonce)
+                builder.add_recipient(
+                    JweRecipient(
+                        encrypted_key=enc_cek,
+                        header=OrderedDict(
+                            [
+                                ("kid", target_vk.kid),
+                                ("sender", self.b64url.encode(enc_sender)),
+                                ("iv", self.b64url.encode(nonce)),
+                            ]
+                        ),
+                    )
+                )
+            else:
+                enc_sender = None
+                nonce = None
+                enc_cek = crypto_box.crypto_box_seal(target_xk, cek_b)
+                builder.add_recipient(
+                    JweRecipient(encrypted_key=enc_cek, header={"kid": target_vk.kid})
+                )
+        builder.set_protected(
+            OrderedDict(
+                [
+                    ("enc", "xchacha20poly1305_ietf"),
+                    ("typ", "JWM/1.0"),
+                    ("alg", "Authcrypt" if from_key else "Anoncrypt"),
+                ]
+            ),
+        )
+        enc = cek.aead_encrypt(message, aad=builder.protected_bytes)
+        ciphertext, tag, nonce = enc.parts
+        builder.set_payload(ciphertext, nonce, tag)
+        return builder.build()
+
+    async def unpack_message(
+        self,
+        wrapper: JweEnvelope,
+        recip_key: AskarSecretKey,
+        recip_data: RecipData,
+    ) -> LegacyUnpackResult:
+        """Decode a message using the DIDComm v1 'unpack' algorithm."""
+        payload_key, sender_vk = self._extract_payload_key(recip_key.key, recip_data)
+
+        cek = Key.from_secret_bytes(KeyAlg.C20P, payload_key)
+        message = cek.aead_decrypt(
+            wrapper.ciphertext,
+            nonce=wrapper.iv,
+            tag=wrapper.tag,
+            aad=wrapper.protected_b64,
+        )
+        return LegacyUnpackResult(message, recip_key.kid, sender_vk)
+
+    def _extract_payload_key(
+        self, recip_key: Key, recip_data: RecipData
+    ) -> Tuple[bytes, Optional[str]]:
+        """Extract the payload key from pack recipient details.
+
+        Returns: A tuple of the CEK and sender verkey
+        """
+        recip_x = recip_key.convert_key(KeyAlg.X25519)
+
+        if recip_data.nonce and recip_data.enc_sender:
+            sender_vk = crypto_box.crypto_box_seal_open(
+                recip_x, recip_data.enc_sender
+            ).decode("utf-8")
+            sender_x = Key.from_public_bytes(
+                KeyAlg.ED25519, b58decode(sender_vk)
+            ).convert_key(KeyAlg.X25519)
+            cek = crypto_box.crypto_box_open(
+                recip_x, sender_x, recip_data.enc_cek, recip_data.nonce
+            )
+        else:
+            sender_vk = None
+            cek = crypto_box.crypto_box_seal_open(recip_x, recip_data.enc_cek)
+        return cek, sender_vk

didcomm_messaging/legacy/base.py

@@ -0,0 +1,49 @@
+"""DIDComm v1 Base Services."""
+
+from abc import ABC, abstractmethod
+from typing import Generic, NamedTuple, Optional, Sequence
+from didcomm_messaging.crypto.base import P, S
+from didcomm_messaging.crypto.jwe import JweEnvelope
+from didcomm_messaging.multiformats.multibase import Base64UrlEncoder
+
+
+class RecipData(NamedTuple):
+    """Recipient metadata."""
+
+    kid: str
+    enc_sender: Optional[bytes]
+    nonce: Optional[bytes]
+    enc_cek: bytes
+
+
+class LegacyUnpackResult(NamedTuple):
+    """Result of unpacking."""
+
+    message: bytes
+    recip: str
+    sender: Optional[str]
+
+
+class LegacyCryptoService(ABC, Generic[P, S]):
+    """CryptoService interface for DIDComm v1."""
+
+    b64url = Base64UrlEncoder()
+
+    @abstractmethod
+    def kid_to_public_key(self, kid: str) -> P:
+        """Get a public key from a kid.
+
+        In DIDComm v1, kids are the base58 encoded keys.
+        """
+
+    @abstractmethod
+    async def pack_message(
+        self, to_verkeys: Sequence[P], from_key: Optional[S], message: bytes
+    ) -> JweEnvelope:
+        """Encode a message using the DIDComm v1 'pack' algorithm."""
+
+    @abstractmethod
+    async def unpack_message(
+        self, wrapper: JweEnvelope, recip_key: S, recip_data: RecipData
+    ) -> LegacyUnpackResult:
+        """Decode a message using DIDCvomm v1 'unpack' algorithm."""

didcomm_messaging/legacy/crypto.py

@@ -1,4 +1,8 @@
-"""DIDComm v1 packing and unpacking."""
+"""DIDComm v1 packing and unpacking.
+
+This implementation is kept around for backwards compatibility. It is
+likely that you should use the LegacyCryptoService interfaces instead.
+"""
 
 from collections import OrderedDict
 from typing import Iterable, Optional, Sequence, Dict, Union, Tuple, cast

didcomm_messaging/legacy/nacl.py

@@ -0,0 +1,151 @@
+"""LegacyCryptoService implementation for pynacl."""
+
+from dataclasses import dataclass
+from typing import Dict, Optional, Sequence
+
+import base58
+
+try:
+    import nacl.bindings
+    import nacl.exceptions
+    import nacl.utils
+except ImportError as err:
+    raise ImportError(
+        "Legacy implementation requires 'legacy' extra to be installed"
+    ) from err
+from pydid import VerificationMethod
+
+from didcomm_messaging.crypto.base import (
+    PublicKey,
+    SecretKey,
+    SecretsManager,
+)
+from didcomm_messaging.crypto.jwe import JweEnvelope
+from didcomm_messaging.multiformats import multibase, multicodec
+
+from . import crypto
+from .base import LegacyCryptoService, LegacyUnpackResult, RecipData
+
+
+@dataclass
+class KeyPair(SecretKey):
+    """Keys."""
+
+    verkey: bytes
+    sigkey: bytes
+
+    @property
+    def verkey_b58(self) -> str:
+        """Return base58 encoding of verkey."""
+        return base58.b58encode(self.verkey).decode()
+
+    @property
+    def kid(self) -> str:
+        """Get the key ID."""
+        return self.verkey_b58
+
+
+@dataclass
+class EdPublicKey(PublicKey):
+    """Simple public key representation as base58 encoded str."""
+
+    value: bytes
+
+    @classmethod
+    def from_verification_method(cls, vm: VerificationMethod) -> "EdPublicKey":
+        """Create a Key instance from a DID Document Verification Method."""
+        key_bytes = cls.key_bytes_from_verification_method(vm)
+        return EdPublicKey(key_bytes)
+
+    @property
+    def key(self) -> str:
+        """Return base58 encoded key."""
+        return base58.b58encode(self.value).decode()
+
+    @property
+    def kid(self) -> str:
+        """Get the key ID."""
+        return self.key
+
+    @property
+    def multikey(self) -> str:
+        """Get the key in multikey format."""
+        return multibase.encode(
+            multicodec.wrap("ed25519-pub", base58.b58decode(self.key)), "base58btc"
+        )
+
+
+class NaclLegacyCryptoService(LegacyCryptoService[EdPublicKey, KeyPair]):
+    """Legacy crypto service using pynacl."""
+
+    def kid_to_public_key(self, kid: str):
+        """Get a public key from a kid.
+
+        In DIDComm v1, kids are the base58 encoded keys.
+        """
+        return EdPublicKey(base58.b58decode(kid))
+
+    async def pack_message(
+        self,
+        to_verkeys: Sequence[EdPublicKey],
+        from_key: Optional[KeyPair],
+        message: bytes,
+    ) -> JweEnvelope:
+        """Encode a message using the DIDComm v1 'pack' algorithm."""
+        packed = crypto.pack_message(
+            message=message.decode(),
+            to_verkeys=[vk.value for vk in to_verkeys],
+            from_verkey=from_key.verkey if from_key else None,
+            from_sigkey=from_key.sigkey if from_key else None,
+        )
+        return JweEnvelope.deserialize(packed)
+
+    def _extract_payload_key(self, recip_key: KeyPair, recip_data: RecipData):
+        """Extract the payload key."""
+        pk = nacl.bindings.crypto_sign_ed25519_pk_to_curve25519(recip_key.verkey)
+        sk = nacl.bindings.crypto_sign_ed25519_sk_to_curve25519(recip_key.sigkey)
+
+        if recip_data.nonce and recip_data.enc_sender:
+            sender_vk = nacl.bindings.crypto_box_seal_open(
+                recip_data.enc_sender, pk, sk
+            ).decode()
+            sender_pk = nacl.bindings.crypto_sign_ed25519_pk_to_curve25519(
+                crypto.b58_to_bytes(sender_vk)
+            )
+            cek = nacl.bindings.crypto_box_open(
+                recip_data.enc_cek, recip_data.nonce, sender_pk, sk
+            )
+        else:
+            sender_vk = None
+            cek = nacl.bindings.crypto_box_seal_open(recip_data.enc_cek, pk, sk)
+        return cek, sender_vk
+
+    async def unpack_message(
+        self, wrapper: JweEnvelope, recip_key: KeyPair, recip_data: RecipData
+    ) -> LegacyUnpackResult:
+        """Decode a message using DIDCvomm v1 'unpack' algorithm."""
+        cek, sender_vk = self._extract_payload_key(recip_key, recip_data)
+
+        payload_bin = wrapper.ciphertext + wrapper.tag
+        message = crypto.decrypt_plaintext(
+            payload_bin, wrapper.protected_b64, wrapper.iv, cek
+        )
+        return LegacyUnpackResult(message.encode(), recip_key.kid, sender_vk)
+
+
+class InMemSecretsManager(SecretsManager[KeyPair]):
+    """In-memory secrets manager for ed25519 key pairs."""
+
+    def __init__(self):
+        """Initialize the manager."""
+        self.secrets: Dict[str, KeyPair] = {}
+
+    async def get_secret_by_kid(self, kid: str) -> Optional[KeyPair]:
+        """Retrieve secret by kid."""
+        return self.secrets.get(kid)
+
+    def create(self, seed: Optional[bytes] = None) -> KeyPair:
+        """Create and store a new keypair."""
+        keys = KeyPair(*crypto.create_keypair(seed))
+        self.secrets[keys.kid] = keys
+        return keys