Data integrity routes

aries_cloudagent/config/default_context.py

@@ -139,6 +139,7 @@ async def load_plugins(self, context: InjectionContext):
         plugin_registry.register_plugin("aries_cloudagent.resolver")
         plugin_registry.register_plugin("aries_cloudagent.settings")
         plugin_registry.register_plugin("aries_cloudagent.vc")
+        plugin_registry.register_plugin("aries_cloudagent.vc.data_integrity")
         plugin_registry.register_plugin("aries_cloudagent.wallet")
         plugin_registry.register_plugin("aries_cloudagent.wallet.keys")
 

aries_cloudagent/vc/data_integrity/cryptosuites/__init__.py

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

aries_cloudagent/vc/data_integrity/cryptosuites/eddsa_jcs_2022.py

@@ -0,0 +1,201 @@
+"""EddsaJcs2022 cryptosuite."""
+
+from hashlib import sha256
+import canonicaljson
+
+from ....wallet.base import BaseWallet
+from ....wallet.keys.manager import MultikeyManager
+from ....utils.multiformats import multibase
+from ....core.profile import ProfileSession
+from ....resolver.did_resolver import DIDResolver
+from ..errors import PROBLEM_DETAILS
+
+
+class CryptosuiteError(Exception):
+    """Generic Cryptosuite Error."""
+
+
+class EddsaJcs2022:
+    """EddsaJcs2022 cryptosuite.
+
+    https://www.w3.org/TR/vc-di-eddsa/#eddsa-jcs-2022.
+    """
+
+    def __init__(self, *, session: ProfileSession):
+        """Create new EddsaJcs2022 Cryptosuite instance.
+
+        Args:
+            session: ProfileSession to use during crypto operations.
+
+        """
+        super().__init__()
+        self.session = session
+        self.wallet = session.inject(BaseWallet)
+        self.key_manager = MultikeyManager(session)
+
+    async def create_proof(self, unsecured_data_document: dict, options: dict):
+        """Create proof algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#create-proof-eddsa-jcs-2022.
+        """
+        proof = options.copy()
+
+        # Spec says to copy document context to the proof but it's unecessary IMO,
+        # commenting out for the time being...
+
+        # if '@context' in unsecured_data_document:
+        #     proof['@context'] = unsecured_data_document['@context']
+
+        proof_config = self.proof_configuration(proof)
+        transformed_data = self.transformation(unsecured_data_document, options)
+        hash_data = self.hashing(transformed_data, proof_config)
+        proof_bytes = await self.proof_serialization(hash_data, options)
+
+        proof["proofValue"] = multibase.encode(proof_bytes, "base58btc")
+
+        return proof
+
+    def proof_configuration(self, options: dict):
+        """Proof configuration algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#proof-configuration-eddsa-jcs-2022.
+        """
+        proof_config = options.copy()
+
+        assert (
+            proof_config["type"] == "DataIntegrityProof"
+        ), 'Expected proof.type to be "DataIntegrityProof'
+        assert (
+            proof_config["cryptosuite"] == "eddsa-jcs-2022"
+        ), 'Expected proof.cryptosuite to be "eddsa-jcs-2022'
+
+        if "created" in proof_config:
+            # TODO assert proper [XMLSCHEMA11-2] dateTimeStamp string
+            assert proof_config[
+                "created"
+            ], "Expected proof.created to be a [XMLSCHEMA11-2] dateTimeStamp string."
+
+        if "expires" in proof_config:
+            # TODO assert proper [XMLSCHEMA11-2] dateTimeStamp string
+            assert proof_config[
+                "expires"
+            ], "Expected proof.expires to be a [XMLSCHEMA11-2] dateTimeStamp string."
+
+        return self._canonicalize(proof_config)
+
+    def transformation(self, unsecured_document: dict, options: dict):
+        """Transformation algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#transformation-eddsa-jcs-2022.
+        """
+        assert (
+            options["type"] == "DataIntegrityProof"
+        ), "Expected proof.type to be `DataIntegrityProof`"
+        assert (
+            options["cryptosuite"] == "eddsa-jcs-2022"
+        ), "Expected proof.cryptosuite to be `eddsa-jcs-2022`"
+
+        return self._canonicalize(unsecured_document)
+
+    def hashing(self, transformed_document, canonical_proof_config):
+        """Hashing algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#hashing-eddsa-jcs-2022.
+        """
+        return (
+            sha256(canonical_proof_config).digest()
+            + sha256(transformed_document).digest()
+        )
+
+    async def proof_serialization(self, hash_data: bytes, options: dict):
+        """Proof Serialization Algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#proof-serialization-eddsa-jcs-2022.
+        """
+        # If the verification method is a did:key: URI,
+        # we derive the signing key from a multikey value
+        if options["verificationMethod"].startswith("did:key:"):
+            multikey = options["verificationMethod"].split("#")[-1]
+            key_info = await self.key_manager.from_multikey(multikey)
+
+        # Otherwise we derive the signing key from a kid
+        else:
+            key_info = await self.key_manager.from_kid(options["verificationMethod"])
+
+        return await self.wallet.sign_message(
+            message=hash_data,
+            from_verkey=self.key_manager._multikey_to_verkey(key_info["multikey"]),
+        )
+
+    def _canonicalize(self, data: dict):
+        """Json canonicalization."""
+        return canonicaljson.encode_canonical_json(data)
+
+    async def _get_multikey(self, kid: str):
+        """Derive a multikey from the verification method."""
+
+        # If verification method is a did:key URI,
+        # we derive the multikey directly from the value.
+        if kid.startswith("did:key:"):
+            return kid.split("#")[-1]
+
+        # Otherwise we resolve the verification method and extract the multikey.
+        else:
+            verification_method = await DIDResolver().dereference(
+                profile=self.session.profile, did_url=kid
+            )
+            assert (
+                verification_method["type"] == "Multikey"
+            ), "Expected Multikey verificationMethod type"
+
+            return verification_method["publicKeyMultibase"]
+
+    async def verify_proof(self, secured_document: dict):
+        """Verify proof algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#verify-proof-eddsa-jcs-2022.
+        """
+        unsecured_document = secured_document.copy()
+        proof = unsecured_document.pop("proof")
+        proof_options = proof.copy()
+        proof_bytes = multibase.decode(proof_options.pop("proofValue"))
+
+        try:
+            # Currently leaving context processing out of scope,
+            # leaving code commented as it's technically an algorithm step.
+            # Due to the cryptosuite being based on JSON canonicalization,
+            # the integrity of the document is protected without RDF processing.
+
+            # https://www.w3.org/TR/vc-data-integrity/#validating-contexts
+
+            # assert secured_document['@context'] == proof_options['@context']
+            # unsecured_document['@context'] = proof_options['@context']
+
+            transformed_data = self.transformation(unsecured_document, proof_options)
+            proof_config = self.proof_configuration(proof_options)
+            hash_data = self.hashing(transformed_data, proof_config)
+            if not await self.proof_verification(hash_data, proof_bytes, proof_options):
+                raise CryptosuiteError("Invalid signature.")
+
+        except (AssertionError, CryptosuiteError) as err:
+            problem_detail = PROBLEM_DETAILS["PROOF_VERIFICATION_ERROR"] | {
+                "message": str(err)
+            }
+            return {"verified": False, "proof": proof, "problemDetails": [problem_detail]}
+
+        return {"verified": True, "proof": proof, "problemDetails": []}
+
+    async def proof_verification(
+        self, hash_data: bytes, proof_bytes: bytes, options: dict
+    ):
+        """Proof verification algorithm.
+
+        https://www.w3.org/TR/vc-di-eddsa/#proof-verification-eddsa-jcs-2022.
+        """
+        multikey = await self._get_multikey(options["verificationMethod"])
+        return await self.wallet.verify_message(
+            message=hash_data,
+            signature=proof_bytes,
+            from_verkey=self.key_manager._multikey_to_verkey(multikey),
+            key_type=self.key_manager.key_type_from_multikey(multikey),
+        )

aries_cloudagent/vc/data_integrity/errors.py

@@ -0,0 +1,18 @@
+"""Problem Details objects for error handling."""
+
+# https://www.w3.org/TR/vc-data-integrity/#processing-errors
+PROBLEM_DETAILS = {
+    "PROOF_GENERATION_ERROR": {
+        "type": "https://w3id.org/security#PROOF_GENERATION_ERROR"
+    },
+    "PROOF_VERIFICATION_ERROR": {
+        "type": "https://w3id.org/security#PROOF_VERIFICATION_ERROR"
+    },
+    "PROOF_TRANSFORMATION_ERROR": {
+        "type": "https://w3id.org/security#PROOF_TRANSFORMATION_ERROR"
+    },
+    "INVALID_DOMAIN_ERROR": {"type": "https://w3id.org/security#INVALID_DOMAIN_ERROR"},
+    "INVALID_CHALLENGE_ERROR": {
+        "type": "https://w3id.org/security#INVALID_CHALLENGE_ERROR"
+    },
+}

aries_cloudagent/vc/data_integrity/manager.py

@@ -0,0 +1,78 @@
+"""DataIntegrity class."""
+
+from ...core.profile import ProfileSession
+from .cryptosuites import CRYPTOSUITES
+
+
+class DataIntegrityManagerError(Exception):
+    """Generic DataIntegrityManager Error."""
+
+
+class DataIntegrityManager:
+    """Class for managing data integrity proofs."""
+
+    def __init__(self, session: ProfileSession):
+        """Initialize the DataIntegrityManager."""
+        self.session = session
+
+    async def add_proof(self, document, options):
+        """Data integrity add proof algorithm.
+
+        https://www.w3.org/TR/vc-data-integrity/#add-proof.
+        """
+
+        # Instanciate a cryptosuite
+        suite = CRYPTOSUITES[options["cryptosuite"]](session=self.session)
+
+        # Capture existing proofs if any
+        all_proofs = document.pop("proof", [])
+        assert isinstance(all_proofs, list) or isinstance(all_proofs, dict)
+        all_proofs = [all_proofs] if isinstance(all_proofs, dict) else all_proofs
+
+        # Create secured document and create new proof
+        secured_document = document.copy()
+        secured_document["proof"] = all_proofs
+        secured_document["proof"].append(await suite.create_proof(document, options))
+        return secured_document
+
+    async def verify_proof(self, secured_document):
+        """Verify a proof attached to a secured document.
+
+        https://www.w3.org/TR/vc-data-integrity/#verify-proof.
+        """
+        unsecured_document = secured_document.copy()
+        all_proofs = unsecured_document.pop("proof")
+        all_proofs = all_proofs if isinstance(all_proofs, list) else [all_proofs]
+        verification_results = {}
+        verification_results["verifiedDocument"] = unsecured_document
+        verification_results["results"] = []
+        for proof in all_proofs:
+            try:
+                self.assert_proof(proof)
+                # Instanciate a cryptosuite
+                suite = CRYPTOSUITES[proof["cryptosuite"]](session=self.session)
+                input_document = unsecured_document.copy()
+                input_document["proof"] = proof
+                verification_result = await suite.verify_proof(input_document)
+                verification_results["results"].append(verification_result)
+            except AssertionError as err:
+                verification_result = {
+                    "verified": False,
+                    "problemDetails": [{"type": "", "message": str(err)}],
+                }
+                verification_results["results"].append(verification_result)
+        verification_results["verified"] = (
+            True
+            if all(result["verified"] for result in verification_results["results"])
+            else False
+        )
+        return verification_results
+
+    def assert_proof(self, proof):
+        """Generic proof assertions for a data integrity proof."""
+        assert proof["cryptosuite"] in CRYPTOSUITES, "Unsupported cryptosuite."
+        assert proof["proofValue"], "Missing proof value."
+        assert proof["proofPurpose"] in [
+            "authentication",
+            "assertionMethod",
+        ], "Unknown proofPurpose."

aries_cloudagent/vc/data_integrity/models/__init__.py

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