Merge pull request #2559 from dbluhm/refactor/replace-multiformats

refactor: replace multiformats library

aries_cloudagent/connections/base_manager.py

@@ -6,7 +6,6 @@
 import logging
 from typing import List, Optional, Sequence, Text, Tuple, Union
 
-from multiformats import multibase, multicodec
 from pydid import (
     BaseDIDDocument as ResolvedDocument,
     DIDCommService,
@@ -18,6 +17,7 @@
     Ed25519VerificationKey2020,
     JsonWebKey2020,
 )
+
 from ..cache.base import BaseCache
 from ..config.base import InjectionError
 from ..config.logging import get_logger_inst
@@ -41,6 +41,7 @@
 from ..storage.error import StorageDuplicateError, StorageError, StorageNotFoundError
 from ..storage.record import StorageRecord
 from ..transport.inbound.receipt import MessageReceipt
+from ..utils.multiformats import multibase, multicodec
 from ..wallet.base import BaseWallet
 from ..wallet.crypto import create_keypair, seed_to_did
 from ..wallet.did_info import DIDInfo

aries_cloudagent/connections/tests/test_base_manager.py

@@ -3,7 +3,6 @@
 from unittest.mock import call
 
 from asynctest import TestCase as AsyncTestCase, mock as async_mock
-from multiformats import multibase, multicodec
 from pydid import DID, DIDDocument, DIDDocumentBuilder
 from pydid.doc.builder import ServiceBuilder
 from pydid.verification_method import (
@@ -41,6 +40,7 @@
 from ...storage.error import StorageNotFoundError
 from ...storage.record import StorageRecord
 from ...transport.inbound.receipt import MessageReceipt
+from ...utils.multiformats import multibase, multicodec
 from ...wallet.base import DIDInfo
 from ...wallet.did_method import DIDMethods, SOV
 from ...wallet.error import WalletNotFoundError

aries_cloudagent/resolver/default/peer3.py

@@ -6,27 +6,22 @@
 the did:peer:2 has been replaced with the did:peer:3.
 """
 
-import re
 from copy import deepcopy
 from hashlib import sha256
+import re
 from typing import Optional, Pattern, Sequence, Text
-from multiformats import multibase, multicodec
-
-from peerdid.dids import (
-    DID,
-    MalformedPeerDIDError,
-    DIDDocument,
-)
-from peerdid.keys import to_multibase, MultibaseFormat
-from ...wallet.util import bytes_to_b58
 
-from ...connections.base_manager import BaseConnectionManager
+from peerdid.dids import DID, DIDDocument, MalformedPeerDIDError
+from peerdid.keys import MultibaseFormat, to_multibase
+
 from ...config.injection_context import InjectionContext
+from ...connections.base_manager import BaseConnectionManager
 from ...core.profile import Profile
 from ...storage.base import BaseStorage
 from ...storage.error import StorageNotFoundError
 from ...storage.record import StorageRecord
-
+from ...utils.multiformats import multibase, multicodec
+from ...wallet.util import bytes_to_b58
 from ..base import BaseDIDResolver, DIDNotFound, ResolverType
 
 RECORD_TYPE_DID_DOCUMENT = "did_document"  # pydid DIDDocument

aries_cloudagent/utils/multiformats/__init__.py

@@ -0,0 +1 @@
+"""Multiformats utility functions."""

aries_cloudagent/utils/multiformats/multibase.py

@@ -0,0 +1,104 @@
+"""MultiBase encoding and decoding utilities."""
+
+from abc import ABC, abstractmethod
+from enum import Enum
+from typing import ClassVar, Literal, Union
+
+
+class MultibaseEncoder(ABC):
+    """Encoding details."""
+
+    name: ClassVar[str]
+    character: ClassVar[str]
+
+    @abstractmethod
+    def encode(self, value: bytes) -> str:
+        """Encode a byte string using this encoding."""
+
+    @abstractmethod
+    def decode(self, value: str) -> bytes:
+        """Decode a string using this encoding."""
+
+
+class Base58BtcEncoder(MultibaseEncoder):
+    """Base58BTC encoding."""
+
+    name = "base58btc"
+    character = "z"
+
+    def encode(self, value: bytes) -> str:
+        """Encode a byte string using the base58btc encoding."""
+        import base58
+
+        return base58.b58encode(value).decode()
+
+    def decode(self, value: str) -> bytes:
+        """Decode a multibase encoded string."""
+        import base58
+
+        return base58.b58decode(value)
+
+
+class Encoding(Enum):
+    """Enum for supported encodings."""
+
+    base58btc = Base58BtcEncoder()
+    # Insert additional encodings here
+
+    @classmethod
+    def from_name(cls, name: str) -> MultibaseEncoder:
+        """Get encoding from name."""
+        for encoding in cls:
+            if encoding.value.name == name:
+                return encoding.value
+        raise ValueError(f"Unsupported encoding: {name}")
+
+    @classmethod
+    def from_character(cls, character: str) -> MultibaseEncoder:
+        """Get encoding from character."""
+        for encoding in cls:
+            if encoding.value.character == character:
+                return encoding.value
+        raise ValueError(f"Unsupported encoding: {character}")
+
+
+EncodingStr = Literal[
+    "base58btc",
+    # Insert additional encoding names here
+]
+
+
+def encode(value: bytes, encoding: Union[Encoding, EncodingStr]) -> str:
+    """Encode a byte string using the given encoding.
+
+    Args:
+        value: The byte string to encode
+        encoding: The encoding to use
+
+    Returns:
+        The encoded string
+    """
+    if isinstance(encoding, str):
+        encoder = Encoding.from_name(encoding)
+    elif isinstance(encoding, Encoding):
+        encoder = encoding.value
+    else:
+        raise TypeError("encoding must be an Encoding or EncodingStr")
+
+    return encoder.character + encoder.encode(value)
+
+
+def decode(value: str) -> bytes:
+    """Decode a multibase encoded string.
+
+    Args:
+        value: The string to decode
+
+    Returns:
+        The decoded byte string
+    """
+    encoding = value[0]
+    encoded = value[1:]
+    encoder = Encoding.from_character(encoding)
+
+    return encoder.decode(encoded)

aries_cloudagent/utils/multiformats/multicodec.py

@@ -0,0 +1,72 @@
+"""Multicodec wrap and unwrap functions."""
+
+from enum import Enum
+from typing import Literal, NamedTuple, Optional, Union
+
+
+class Multicodec(NamedTuple):
+    """Multicodec base class."""
+
+    name: str
+    code: bytes
+
+
+class SupportedCodecs(Enum):
+    """Enumeration of supported multicodecs."""
+
+    ed25519_pub = Multicodec("ed25519-pub", b"\xed\x01")
+    x25519_pub = Multicodec("x25519-pub", b"\xec\x01")
+    bls12381g1 = Multicodec("bls12_381-g1-pub", b"\xea\x01")
+    bls12381g2 = Multicodec("bls12_381-g2-pub", b"\xeb\x01")
+    bls12381g1g2 = Multicodec("bls12_381-g1g2-pub", b"\xee\x01")
+    secp256k1_pub = Multicodec("secp256k1-pub", b"\xe7\x01")
+
+    @classmethod
+    def by_name(cls, name: str) -> Multicodec:
+        """Get multicodec by name."""
+        for codec in cls:
+            if codec.value.name == name:
+                return codec.value
+        raise ValueError(f"Unsupported multicodec: {name}")
+
+    @classmethod
+    def for_data(cls, data: bytes) -> Multicodec:
+        """Get multicodec by data."""
+        for codec in cls:
+            if data.startswith(codec.value.code):
+                return codec.value
+        raise ValueError("Unsupported multicodec")
+
+
+MulticodecStr = Literal[
+    "ed25519-pub",
+    "x25519-pub",
+    "bls12_381-g1-pub",
+    "bls12_381-g2-pub",
+    "bls12_381-g1g2-pub",
+    "secp256k1-pub",
+]
+
+
+def multicodec(name: str) -> Multicodec:
+    """Get multicodec by name."""
+    return SupportedCodecs.by_name(name)
+
+
+def wrap(multicodec: Union[Multicodec, MulticodecStr], data: bytes) -> bytes:
+    """Wrap data with multicodec prefix."""
+    if isinstance(multicodec, str):
+        multicodec = SupportedCodecs.by_name(multicodec)
+    elif isinstance(multicodec, Multicodec):
+        pass
+    else:
+        raise TypeError("multicodec must be Multicodec or MulticodecStr")
+
+    return multicodec.code + data
+
+
+def unwrap(data: bytes, codec: Optional[Multicodec] = None) -> tuple[Multicodec, bytes]:
+    """Unwrap data with multicodec prefix."""
+    if not codec:
+        codec = SupportedCodecs.for_data(data)
+    return codec, data[len(codec.code) :]

aries_cloudagent/utils/tests/test_multiformats.py

@@ -0,0 +1,73 @@
+import pytest
+from ..multiformats import multibase, multicodec
+
+
+def test_encode_decode():
+    value = b"Hello World!"
+    encoded = multibase.encode(value, "base58btc")
+    assert encoded == "z2NEpo7TZRRrLZSi2U"
+    decoded = multibase.decode(encoded)
+    assert decoded == value
+
+
+def test_encode_decode_by_encoding():
+    value = b"Hello World!"
+    encoded = multibase.encode(value, multibase.Encoding.base58btc)
+    assert encoded == "z2NEpo7TZRRrLZSi2U"
+    decoded = multibase.decode(encoded)
+    assert decoded == value
+
+
+def test_x_unknown_encoding():
+    with pytest.raises(ValueError):
+        multibase.encode(b"Hello World!", "fancy-encoding")
+
+
+def test_x_unknown_character():
+    with pytest.raises(ValueError):
+        multibase.decode("fHello World!")
+
+
+def test_x_invalid_encoding():
+    with pytest.raises(TypeError):
+        multibase.encode(b"Hello World!", 123)
+
+
+def test_wrap_unwrap():
+    value = b"Hello World!"
+    wrapped = multicodec.wrap("ed25519-pub", value)
+    codec, unwrapped = multicodec.unwrap(wrapped)
+    assert codec == multicodec.multicodec("ed25519-pub")
+    assert unwrapped == value
+
+
+def test_wrap_unwrap_custom():
+    value = b"Hello World!"
+    my_codec = multicodec.Multicodec("my-codec", b"\x00\x01")
+    wrapped = multicodec.wrap(my_codec, value)
+    codec, unwrapped = multicodec.unwrap(wrapped, my_codec)
+    assert codec == my_codec
+    assert unwrapped == value
+
+
+def test_wrap_unwrap_by_codec():
+    value = b"Hello World!"
+    wrapped = multicodec.wrap(multicodec.multicodec("ed25519-pub"), value)
+    codec, unwrapped = multicodec.unwrap(wrapped, multicodec.multicodec("ed25519-pub"))
+    assert codec == multicodec.multicodec("ed25519-pub")
+    assert unwrapped == value
+
+
+def test_x_unknown_multicodec():
+    with pytest.raises(ValueError):
+        multicodec.wrap("fancy-multicodec", b"Hello World!")
+
+
+def test_x_invalid_multicodec():
+    with pytest.raises(TypeError):
+        multicodec.wrap(123, b"Hello World!")
+
+
+def test_x_invalid_multicodec_unwrap():
+    with pytest.raises(ValueError):
+        multicodec.unwrap(b"Hello World!")

aries_cloudagent/vc/ld_proofs/suites/ed25519_signature_2020.py

@@ -1,14 +1,13 @@
 """Ed25519Signature2018 suite."""
 
 from datetime import datetime
-from typing import Union, List
+from typing import List, Union
 
-from multiformats import multibase
-
-from .linked_data_signature import LinkedDataSignature
+from ....utils.multiformats import multibase
 from ..crypto import _KeyPair as KeyPair
 from ..document_loader import DocumentLoaderMethod
 from ..error import LinkedDataProofException
+from .linked_data_signature import LinkedDataSignature
 
 
 class Ed25519Signature2020(LinkedDataSignature):