Add di support w/ eddsa-jcs-2022

aries_cloudagent/config/default_context.py

@@ -131,6 +131,7 @@ async def load_plugins(self, context: InjectionContext):
             plugin_registry.register_package("aries_cloudagent.protocols")
 
         # Currently providing admin routes only
+        plugin_registry.register_plugin("aries_cloudagent.did")
         plugin_registry.register_plugin("aries_cloudagent.holder")
 
         plugin_registry.register_plugin("aries_cloudagent.ledger")

aries_cloudagent/did/__init__.py

@@ -0,0 +1,11 @@
+from .key.manager import DidKeyManager
+
+
+class DidOperationError(Exception):
+    """Generic DID operation Error."""
+
+
+__all__ = [
+    "DidKeyManager",
+    "DidOperationError",
+]

aries_cloudagent/did/key/manager.py

@@ -0,0 +1,73 @@
+"""DID Key manager class."""
+
+from ...wallet.key_type import KeyType
+from ...wallet.did_method import KEY
+from ...wallet.base import BaseWallet
+from ...core.profile import Profile
+
+
+class DidKeyManager:
+    """Class for managing key dids."""
+
+    def __init__(self, profile: Profile):
+        """Initialize a new `DidKeyManager` instance."""
+        self.profile = profile
+
+    async def register(
+        self,
+        key_type: KeyType,
+    ):
+        """Register a new key DID.
+
+        Args:
+            key_type: The key type to use for the DID
+
+        Returns:
+            A `DIDDocument` instance representing the created DID
+
+        Raises:
+            DidOperationError: If the an error occures during did registration
+
+        """
+        async with self.profile.session() as session:
+            wallet = session.inject(BaseWallet)
+        info = await wallet.create_local_did(method=KEY, key_type=key_type)
+        return await self.create_did_doc(info.did)
+
+    async def create_did_doc(
+        self,
+        did: str,
+    ):
+        """Creates a DID doc based on a did:key value.
+
+        Args:
+            did: The did:key value
+
+        Returns:
+            A `DIDDocument` instance representing the created DID
+
+        Raises:
+            DidOperationError: If the an error occures during did document creation
+
+        """
+        verification_method = f"{did}#" + did.split(":")[-1]
+        return {
+            "@context": [
+                "https://www.w3.org/ns/did/v1",
+                "https://w3id.org/security/multikey/v1",
+            ],
+            "id": did,
+            "verificationMethod": [
+                {
+                    "id": verification_method,
+                    "type": "MultiKey",
+                    "controller": did,
+                    "publicKeyMultibase": did.split(":")[-1],
+                },
+            ],
+            "authentication": [verification_method],
+            "assertionMethod": [verification_method],
+            "capabilityDelegation": [verification_method],
+            "capabilityInvocation": [verification_method],
+            "keyAgreement": [],
+        }

aries_cloudagent/did/routes.py

@@ -0,0 +1,64 @@
+"""DID Management Routes."""
+
+from aiohttp import web
+from aiohttp_apispec import docs, request_schema, response_schema
+from marshmallow.exceptions import ValidationError
+
+from ..wallet.key_type import ED25519
+from ..admin.decorators.auth import tenant_authentication
+from ..admin.request_context import AdminRequestContext
+from .web_requests import (
+    DIDKeyRegistrationRequest,
+    DIDKeyRegistrationResponse,
+)
+from . import DidOperationError, DidKeyManager
+
+KEY_MAPPINGS = {"ed25519": ED25519}
+
+
+@docs(tags=["did"], summary="Register Key DID")
+@request_schema(DIDKeyRegistrationRequest())
+@response_schema(DIDKeyRegistrationResponse(), 201, description="Register new DID key")
+@tenant_authentication
+async def register_did_key(request: web.BaseRequest):
+    """Request handler for registering a Key DID.
+
+    Args:
+        request: aiohttp request object
+
+    """
+    body = await request.json()
+    context: AdminRequestContext = request["context"]
+    try:
+        key_type = body["key_type"]
+        did_doc = await DidKeyManager(context.profile).register(KEY_MAPPINGS[key_type])
+        return web.json_response({"didDocument": did_doc}, status=201)
+    except (KeyError, ValidationError, DidOperationError) as err:
+        return web.json_response({"message": str(err)}, status=400)
+
+
+async def register(app: web.Application):
+    """Register routes."""
+
+    app.add_routes(
+        [
+            web.post("/did/key", register_did_key),
+        ]
+    )
+
+
+def post_process_routes(app: web.Application):
+    """Amend swagger API."""
+    # Add top-level tags description
+    if "tags" not in app._state["swagger_dict"]:
+        app._state["swagger_dict"]["tags"] = []
+    app._state["swagger_dict"]["tags"].append(
+        {
+            "name": "did",
+            "description": "Endpoints for managing dids",
+            "externalDocs": {
+                "description": "Specification",
+                "url": "https://www.w3.org/TR/did-core/",
+            },
+        }
+    )

aries_cloudagent/did/web_requests.py

@@ -0,0 +1,23 @@
+"""DID routes web requests schemas."""
+
+from marshmallow import fields
+from ..messaging.models.openapi import OpenAPISchema
+
+
+class DIDKeyRegistrationRequest(OpenAPISchema):
+    """Request schema for registering key dids."""
+
+    key_type = fields.Str(
+        default="ed25519",
+        required=False,
+        metadata={
+            "description": "Key Type",
+            "example": "ed25519",
+        },
+    )
+
+
+class DIDKeyRegistrationResponse(OpenAPISchema):
+    """Response schema for registering web dids."""
+
+    did_document = fields.Dict()

aries_cloudagent/vc/contexts.py

@@ -0,0 +1,14 @@
+"""Context URLS."""
+
+DID_V1 = "https://www.w3.org/ns/did/v1"
+CREDENTIALS_V1 = "https://www.w3.org/2018/credentials/v1"
+CREDENTIALS_V2 = "https://www.w3.org/ns/credentials/v2"
+
+SECURITY_V1 = "https://w3id.org/security/v1"
+SECURITY_V2 = "https://w3id.org/security/v2"
+SECURITY_V3_UNSTABLE = "https://w3id.org/security/v3-unstable"
+SECURITY_DATA_INTEGRITY_V2 = "https://w3id.org/security/data-integrity/v2"
+SECURITY_BBS_V1 = "https://w3id.org/security/bbs/v1"
+SECURITY_SUITES_ED25519_2020_V1 = "https://w3id.org/security/suites/ed25519-2020/v1"
+SECURITY_MULTIKEY_V1 = "https://w3id.org/security/multikey/v1"
+SECURITY_JWK_V1 = "https://w3id.org/security/jwk/v1"

aries_cloudagent/vc/vc_di/__init__.py

@@ -5,7 +5,13 @@
 )
 from .verify import verify_signed_anoncredspresentation
 
+
+class DataIntegrityProofException(Exception):
+    """Base exception for data integrity proofs."""
+
+
 __all__ = [
+    "DataIntegrityProofException",
     "verify_signed_anoncredspresentation",
     "create_signed_anoncreds_presentation",
     "prepare_data_for_presentation",

aries_cloudagent/vc/vc_di/cryptosuites/__init__.py

@@ -0,0 +1,9 @@
+from .eddsa_jcs_2022 import EddsaJcs2022
+
+CRYPTOSUITES = {
+    "eddsa-jcs-2022": EddsaJcs2022,
+}
+
+__all__ = [
+    "EddsaJcs2022",
+]

aries_cloudagent/vc/vc_di/cryptosuites/eddsa_jcs_2022.py

@@ -0,0 +1,128 @@
+"""EddsaJcs2022 cryptosuite."""
+
+from hashlib import sha256
+import canonicaljson
+import nacl
+
+from ....wallet.base import BaseWallet
+from ....utils.multiformats import multibase
+from ....core.profile import Profile
+from .. import DataIntegrityProofException
+
+
+class EddsaJcs2022:
+    """EddsaJcs2022 suite."""
+
+    def __init__(self, *, profile: Profile):
+        """Create new EddsaJcs2022 Cryptosuite instance.
+
+        https://www.w3.org/TR/vc-di-eddsa/#eddsa-rdfc-2022
+
+        Args:
+            profile: Key profile to use.
+
+        """
+        super().__init__()
+        self.profile = profile
+
+    async def _serialization(self, hash_data, options):
+        """Data Integrity Proof Serialization Algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#proof-serialization-eddsa-jcs-2022
+
+        """
+        async with self.profile.session() as session:
+            did_info = await session.inject(BaseWallet).get_local_did(
+                options["verificationMethod"].split("#")[0]
+            )
+        async with self.profile.session() as session:
+            wallet = session.inject(BaseWallet)
+        proof_bytes = await wallet.sign_message(
+            message=hash_data,
+            from_verkey=did_info.verkey,
+        )
+        return proof_bytes
+
+    async def add_proof(self, document, proof_options):
+        """Data Integrity Add Proof Algorithm.
+
+        https://www.w3.org/TR/vc-data-integrity/#add-proof
+
+        Args:
+            document: The data to sign.
+            proof_options: The proof options.
+
+        Returns:
+            secured_document: The document with a new proof attached
+
+        """
+
+        existing_proof = document.pop("proof", [])
+        assert isinstance(existing_proof, list) or isinstance(existing_proof, dict)
+        existing_proof = (
+            [existing_proof] if isinstance(existing_proof, dict) else existing_proof
+        )
+
+        assert proof_options["type"] == "DataIntegrityProof"
+        assert proof_options["cryptosuite"] == "eddsa-jcs-2022"
+        assert proof_options["proofPurpose"]
+        assert proof_options["verificationMethod"]
+
+        try:
+            hash_data = (
+                sha256(canonicaljson.encode_canonical_json(document)).digest()
+                + sha256(canonicaljson.encode_canonical_json(proof_options)).digest()
+            )
+            proof_bytes = await self._serialization(hash_data, proof_options)
+
+            proof = proof_options.copy()
+            proof["proofValue"] = multibase.encode(proof_bytes, "base58btc")
+
+            secured_document = document.copy()
+            secured_document["proof"] = existing_proof
+            secured_document["proof"].append(proof)
+
+            return secured_document
+        except Exception:
+            raise DataIntegrityProofException()
+
+    async def verify_proof(self, unsecured_document, proof):
+        """Data Integrity Verify Proof Algorithm.
+
+        https://www.w3.org/TR/vc-data-integrity/#verify-proof
+
+        Args:
+            unsecured_document: The data to check.
+            proof: The proof.
+
+        Returns:
+            verification_response: Whether the signature is valid for the data
+
+        """
+        try:
+            assert proof["type"] == "DataIntegrityProof"
+            assert proof["cryptosuite"] == "eddsa-jcs-2022"
+            assert proof["proofPurpose"]
+            assert proof["proofValue"]
+            assert proof["verificationMethod"]
+
+            proof_options = proof.copy()
+            proof_value = proof_options.pop("proofValue")
+            proof_bytes = multibase.decode(proof_value)
+
+            hash_data = (
+                sha256(canonicaljson.encode_canonical_json(unsecured_document)).digest()
+                + sha256(canonicaljson.encode_canonical_json(proof_options)).digest()
+            )
+            verification_method = proof["verificationMethod"]
+            did = verification_method.split("#")[0]
+            if did.split(":")[1] == "key":
+                pub_key = multibase.decode(did.split(":")[-1])
+                public_key_bytes = bytes(bytearray(pub_key)[2:])
+            try:
+                nacl.bindings.crypto_sign_open(proof_bytes + hash_data, public_key_bytes)
+                return True
+            except nacl.exceptions.BadSignatureError:
+                return False
+        except Exception:
+            raise DataIntegrityProofException()

aries_cloudagent/vc/vc_di/models/__init__.py

@@ -0,0 +1,4 @@
+from .proof import DIProof
+from .proof_options import DIProofOptions, DIProofOptionsSchema
+
+__all__ = ["DIProof", "DIProofSchema", "DIProofOptions", "DIProofOptionsSchema"]