improve error handling

Signed-off-by: PatStLouis <[email protected]>

aries_cloudagent/vc/data_integrity/cryptosuites/eddsa_jcs_2022.py

@@ -11,13 +11,14 @@
 )
 from ....utils.multiformats import multibase
 from ....core.profile import ProfileSession
+from ....core.error import BaseError
 from ..models.options import DataIntegrityProofOptions
 from ..models.proof import DataIntegrityProof
 from ..models.verification_response import ProblemDetails, DataIntegrityVerificationResult
 from ..errors import PROBLEM_DETAILS
 
 
-class CryptosuiteError(Exception):
+class CryptosuiteError(BaseError):
     """Generic Cryptosuite Error."""
 
 

aries_cloudagent/vc/data_integrity/manager.py

@@ -1,7 +1,9 @@
 """DataIntegrity class."""
 
 from ...core.profile import ProfileSession
-from .cryptosuites import CRYPTOSUITES
+from ...core.error import BaseError
+from ...resolver.base import DIDNotFound
+from .cryptosuites import EddsaJcs2022
 from .models.proof import DataIntegrityProof
 from .models.options import DataIntegrityProofOptions
 from .models.verification_response import (
@@ -11,8 +13,21 @@
 )
 from .errors import PROBLEM_DETAILS
 
+from datetime import datetime
 
-class DataIntegrityManagerError(Exception):
+CRYPTOSUITES = {
+    "eddsa-jcs-2022": EddsaJcs2022,
+}
+
+PROOF_TYPES = ["DataIntegrityProof"]
+
+PROOF_PURPOSES = [
+    "authentication",
+    "assertionMethod",
+]
+
+
+class DataIntegrityManagerError(BaseError):
     """Generic DataIntegrityManager Error."""
 
 
@@ -28,13 +43,14 @@ async def add_proof(self, document: dict, options: DataIntegrityProofOptions):
 
         https://www.w3.org/TR/vc-data-integrity/#add-proof.
         """
-
-        # Instanciate a cryptosuite
-        suite = CRYPTOSUITES[options.cryptosuite](session=self.session)
+        self.validate_proof_options(options)
+        suite = self.select_suite(options)
 
         # Capture existing proofs if any
         all_proofs = document.pop("proof", [])
-        assert isinstance(all_proofs, list) or isinstance(all_proofs, dict)
+        if not isinstance(all_proofs, list) and not isinstance(all_proofs, dict):
+            raise DataIntegrityManagerError("Expected proof to be a list or an object.")
+
         all_proofs = [all_proofs] if isinstance(all_proofs, dict) else all_proofs
 
         # Create secured document and create new proof
@@ -54,22 +70,24 @@ async def verify_proof(self, secured_document: dict):
         all_proofs = all_proofs if isinstance(all_proofs, list) else [all_proofs]
         verification_results = []
         for proof in all_proofs:
+            proof_options = proof.copy()
+            proof_options.pop("proofValue")
+            proof_options = DataIntegrityProofOptions.deserialize(proof_options)
             try:
-                proof = DataIntegrityProof.deserialize(proof)
-                self.assert_proof(proof)
-                # Instanciate a cryptosuite
-                suite = CRYPTOSUITES[proof.cryptosuite](session=self.session)
+                self.validate_proof_options(proof_options)
+                suite = self.select_suite(proof_options)
                 input_document = unsecured_document.copy()
-                input_document["proof"] = proof.serialize()
+                input_document["proof"] = proof
                 verification_result = await suite.verify_proof(input_document)
-            except AssertionError as err:
+
+            except (AssertionError, DataIntegrityManagerError, DIDNotFound) as err:
                 problem_detail = ProblemDetails.deserialize(
                     PROBLEM_DETAILS["PROOF_VERIFICATION_ERROR"]
                 )
                 problem_detail.detail = str(err)
                 verification_result = DataIntegrityVerificationResult(
                     verified=False,
-                    proof=proof,
+                    proof=DataIntegrityProof.deserialize(proof),
                     problem_details=[problem_detail],
                 )
             verification_results.append(verification_result)
@@ -81,11 +99,52 @@ async def verify_proof(self, secured_document: dict):
             results=verification_results,
         )
 
-    def assert_proof(self, proof: DataIntegrityProof):
-        """Generic proof assertions for a data integrity proof."""
-        assert proof.cryptosuite in CRYPTOSUITES, "Unsupported cryptosuite."
-        assert proof.proof_value, "Missing proof value."
-        assert proof.proof_purpose in [
-            "authentication",
-            "assertionMethod",
-        ], "Unknown proofPurpose."
+    def select_suite(self, options: DataIntegrityProofOptions):
+        """Instanciate a cryptographic suite.
+
+        https://www.w3.org/TR/vc-data-integrity/#cryptographic-suites.
+        """
+        if options.type == "DataIntegrityProof":
+            suite = CRYPTOSUITES[options.cryptosuite](session=self.session)
+
+        elif options.type in PROOF_TYPES:
+            # TODO add support for Ed25519Signature2020
+            pass
+
+        else:
+            raise DataIntegrityManagerError(f"Unsupported proof type {options.type}")
+        return suite
+
+    def validate_proof_options(self, proof_options: DataIntegrityProofOptions):
+        """Generic proof assertions for a data integrity proof options."""
+        if proof_options.created:
+            try:
+                datetime.fromisoformat(proof_options.created)
+            except ValueError:
+                raise DataIntegrityManagerError(
+                    f"Invalid proof creation datetime format {proof_options.created}"
+                )
+        if proof_options.expires:
+            try:
+                datetime.fromisoformat(proof_options.expires)
+            except ValueError:
+                raise DataIntegrityManagerError(
+                    f"Invalid proof expiration datetime format {proof_options.expires}"
+                )
+        if proof_options.type not in PROOF_TYPES:
+            raise DataIntegrityManagerError(
+                f"Unsupported proof type {proof_options.type}"
+            )
+        if proof_options.type == "DataIntegrityProof":
+            if not proof_options.cryptosuite:
+                raise DataIntegrityManagerError(
+                    "DataIntegrityProof must specify a cryptosuite."
+                )
+            if proof_options.cryptosuite not in CRYPTOSUITES:
+                raise DataIntegrityManagerError(
+                    f"Unsupported cryptosuite {proof_options.cryptosuite}"
+                )
+        if proof_options.proof_purpose not in PROOF_PURPOSES:
+            raise DataIntegrityManagerError(
+                f"Unsupported proof purpose {proof_options.proof_purpose}"
+            )

aries_cloudagent/vc/data_integrity/models/verification_response.py

@@ -103,8 +103,8 @@ class Meta:
         metadata={},
     )
 
-    problem_details = fields.Nested(
-        ProblemDetailsSchema(),
+    problem_details = fields.List(
+        fields.Nested(ProblemDetailsSchema()),
         data_key="problemDetails",
         required=True,
         metadata={},
@@ -155,7 +155,7 @@ class Meta:
     )
 
     results = fields.List(
-        fields.Dict(required=False),
+        fields.Nested(DataIntegrityVerificationResultSchema()),
         required=False,
         metadata={},
     )

aries_cloudagent/vc/data_integrity/routes.py

@@ -101,7 +101,7 @@ async def add_di_proof(request: web.BaseRequest):
         return web.json_response({"securedDocument": secured_document}, status=201)
 
     except (WalletNotFoundError, WalletError, DataIntegrityManagerError) as err:
-        raise web.HTTPNotFound(reason=err.roll_up) from err
+        raise web.HTTPBadRequest(reason=err.roll_up) from err
 
 
 @docs(tags=["vc"], summary="Verify a document secured with a data integrity proof.")
@@ -125,17 +125,21 @@ async def verify_di_secured_document(request: web.BaseRequest):
             verification_response = await DataIntegrityManager(session).verify_proof(
                 secured_document
             )
-        response = {
-            "verified": verification_response.verified,
-            "verifiedDocument": verification_response.verified_document,
-            "results": [result.serialize() for result in verification_response.results],
-        }
+        # response = {
+        #     "verified": verification_response.verified,
+        #     "verifiedDocument": verification_response.verified_document,
+        #     "results": [result.serialize() for result in verification_response.results],
+        # }
         if verification_response.verified:
-            return web.json_response({"verificationResults": response}, status=200)
-        return web.json_response({"verificationResults": response}, status=400)
+            return web.json_response(
+                {"verificationResults": verification_response.serialize()}, status=200
+            )
+        return web.json_response(
+            {"verificationResults": verification_response.serialize()}, status=400
+        )
 
     except (WalletNotFoundError, WalletError, DataIntegrityManagerError) as err:
-        raise web.HTTPNotFound(reason=err.roll_up) from err
+        raise web.HTTPBadRequest(reason=err.roll_up) from err
 
 
 async def register(app: web.Application):