Avoid throwing exception on invalid or incomplete received presentations

aries_cloudagent/verifier/indy.py

@@ -1,5 +1,6 @@
 """Indy verifier implementation."""
 
+from enum import Enum
 import json
 import logging
 
@@ -9,6 +10,14 @@
 from .base import BaseVerifier
 
 
+class PreVerifyResult(Enum):
+    """Represent the result of IndyVerifier.pre_verify."""
+
+    OK = "ok"
+    INCOMPLETE = "missing essential components"
+    ENCODING_MISMATCH = "demonstrates tampering with raw values"
+
+
 class IndyVerifier(BaseVerifier):
     """Indy holder class."""
 
@@ -24,9 +33,9 @@ def __init__(self, wallet):
         self.wallet = wallet
 
     @staticmethod
-    def check_encoding(pres_req: dict, pres: dict) -> bool:
+    def pre_verify(pres_req: dict, pres: dict) -> PreVerifyResult:
         """
-        Check for tampering in presentation.
+        Check for essential components and tampering in presentation.
 
         Visit encoded attribute values against raw, and predicate bounds,
         in presentation, cross-reference against presentation request.
@@ -36,55 +45,62 @@ def check_encoding(pres_req: dict, pres: dict) -> bool:
             pres: corresponding presentation
 
         Returns:
-            True for OK, False for tamper evidence
+            An instance of `PreVerifyResult` representing the validation result
 
         """
+        if not pres or "requested_proof" not in pres or "proof" not in pres:
+            return PreVerifyResult.INCOMPLETE
+
         for (uuid, req_pred) in pres_req["requested_predicates"].items():
             canon_attr = canon(req_pred["name"])
-            for ge_proof in pres["proof"]["proofs"][
-                pres["requested_proof"]["predicates"][
-                    uuid
-                ]["sub_proof_index"]
-            ]["primary_proof"]["ge_proofs"]:
-                pred = ge_proof["predicate"]
-                if pred["attr_name"] == canon_attr:
-                    if pred["value"] != req_pred["p_value"]:
-                        return False
-                    break
-            else:
-                return False  # missing predicate in proof
-
+            try:
+                for ge_proof in pres["proof"]["proofs"][
+                    pres["requested_proof"]["predicates"][uuid]["sub_proof_index"]
+                ]["primary_proof"]["ge_proofs"]:
+                    pred = ge_proof["predicate"]
+                    if pred["attr_name"] == canon_attr:
+                        if pred["value"] != req_pred["p_value"]:
+                            return PreVerifyResult.INCOMPLETE
+                        break
+                else:
+                    return PreVerifyResult.INCOMPLETE
+            except (KeyError, TypeError):
+                return PreVerifyResult.INCOMPLETE
+
+        revealed_attrs = pres["requested_proof"].get("revealed_attrs", {})
+        revealed_groups = pres["requested_proof"].get("revealed_attr_groups", {})
         for (uuid, req_attr) in pres_req["requested_attributes"].items():
             if "name" in req_attr:
-                pres_req_attr_spec = {
-                    req_attr["name"]: pres["requested_proof"]["revealed_attrs"][uuid]
-                }
+                pres_req_attr_spec = {req_attr["name"]: revealed_attrs.get(uuid)}
             else:
-                group_spec = pres["requested_proof"].get(
-                    "revealed_attr_groups",
-                    {}
-                ).get(uuid)
-                if group_spec is None:
-                    return False
+                group_spec = revealed_groups.get(uuid)
+                if (
+                    group_spec is None
+                    or "sub_proof_index" not in group_spec
+                    or "values" not in group_spec
+                ):
+                    return PreVerifyResult.INCOMPLETE
                 pres_req_attr_spec = {
                     attr: {
                         "sub_proof_index": group_spec["sub_proof_index"],
-                        **pres["requested_proof"]["revealed_attr_groups"][
-                            uuid
-                        ]["values"][attr]
-                    } for attr in req_attr["names"]
+                        **group_spec["values"].get(attr),
+                    }
+                    for attr in req_attr["names"]
                 }
 
             for (attr, spec) in pres_req_attr_spec.items():
-                primary_enco = pres["proof"]["proofs"][
-                    spec["sub_proof_index"]
-                ]["primary_proof"]["eq_proof"]["revealed_attrs"].get(canon(attr))
+                try:
+                    primary_enco = pres["proof"]["proofs"][spec["sub_proof_index"]][
+                        "primary_proof"
+                    ]["eq_proof"]["revealed_attrs"][canon(attr)]
+                except (KeyError, TypeError):
+                    return PreVerifyResult.INCOMPLETE
                 if primary_enco != spec["encoded"]:
-                    return False
+                    return PreVerifyResult.ENCODING_MISMATCH
                 if primary_enco != encode(spec["raw"]):
-                    return False
+                    return PreVerifyResult.ENCODING_MISMATCH
 
-            return True
+            return PreVerifyResult.OK
 
     async def verify_presentation(
         self, presentation_request, presentation, schemas, credential_definitions
@@ -99,10 +115,11 @@ async def verify_presentation(
             credential_definitions: credential definition data
         """
 
-        if not IndyVerifier.check_encoding(presentation_request, presentation):
+        pv_result = self.pre_verify(presentation_request, presentation)
+        if pv_result != PreVerifyResult.OK:
             self.logger.error(
                 f"Presentation on nonce={presentation_request['nonce']} "
-                "demonstrates tampering with raw values"
+                f"cannot be validated: {pv_result.value}"
             )
             return False