feat: added initial mdoc support

pyeudiw/satosa/default/response_handler.py

@@ -41,6 +41,9 @@
 from pyeudiw.trust.exceptions import InvalidJwkMetadataException
 from pyeudiw.jwt.exceptions import JWSVerificationError
 from pyeudiw.sd_jwt.exceptions import UnsupportedSdAlg, InvalidKeyBinding
+from pyeudiw.sd_jwt.schema import is_sd_jwt_kb_format
+from pyeudiw.openid4vp.vp_mdoc_cbor import VpMDocCbor
+from pyeudiw.openid4vp.exceptions import MdocCborValidationError, VPExpired
 
 
 
@@ -49,11 +52,11 @@ class ResponseHandler(ResponseHandlerInterface):
     _SUPPORTED_RESPONSE_CONTENT_TYPE = "application/x-www-form-urlencoded"
     _ACCEPTED_ISSUER_METADATA_TYPE = "openid_credential_issuer"
 
-    def _extract_all_user_attributes(self, attributes_by_issuers: dict) -> dict:
+    def _extract_all_user_attributes(self, extracted_attributess: dict) -> dict:
         # for all the valid credentials, take the payload and the disclosure and disclose user attributes
         # returns the user attributes ...
         all_user_attributes = dict()
-        for i in attributes_by_issuers.values():
+        for i in extracted_attributess.values():
             all_user_attributes.update(**i)
         return all_user_attributes
 
@@ -158,7 +161,7 @@ def response_endpoint(
         # (1) we don't check that presentation submission matches definition (yet)
         # (2) we don't check that vp tokens are aligned with information declared in the presentation submission
         # (3) we use all disclosed claims in vp tokens to build the user identity
-        attributes_by_issuer: dict[str, dict[str, Any]] = {}
+        extracted_attributes: dict[str, dict[str, Any]] = {}
         credential_issuers: list[str] = []
         encoded_vps: list[str] = (
             [authz_payload.vp_token]
@@ -168,71 +171,115 @@ def response_endpoint(
 
         for vp_token in encoded_vps:
             # verify vp token and extract user information
-            try:
-                token_parser, token_verifier = self._vp_verifier_factory(
-                    authz_payload.presentation_submission, vp_token, request_session
-                )
-            except NotKBJWT as e400:
-                return self._handle_400(
-                    context, 
-                    "invalid vp token: not a key-bound jwt",
-                    e400
-                )
-
-            try:
-                token_issuer = token_parser.get_issuer_name()
-                whitelisted_keys = self.trust_evaluator.get_public_keys(token_issuer)
-                token_verifier.verify_signature(whitelisted_keys)
-                token_verifier.verify_challenge()
-            except MissingIssuer as e400:
-                return self._handle_400(
-                    context, 
-                    "invalid vp token: missing issuer information",
-                    e400
-                )
-            except InvalidJwkMetadataException as e500:
-                return self._handle_500(
-                    context, 
-                    "trust error: cannot fetch public keys",
-                    e500
-                )
-            except JWSVerificationError as e400:
-                return self._handle_400(
-                    context, 
-                    "invalid vp token: invalid signature",
-                    e400
-                )
-            except UnsupportedSdAlg as e400:
-                return self._handle_400(
-                    context, 
-                    "invalid vp token: unsupported signature algorithm",
-                    e400
-                )
-            except InvalidKeyBinding as e400:
-                return self._handle_400(
-                    context, 
-                    "invalid vp token: nonce or aud mismatch",
-                    e400
-                )
-            except ValueError as e400:
-                return self._handle_400(
-                    context, 
-                    "invalid vp token: missing or invalid iat claim",
-                    e400
-                )
-            except Exception as e400:
-                return self._handle_400(
-                    context, 
-                    "trust error: cannot verify vp token",
-                    e400
-                )
-
-            claims = token_parser.get_credentials()
-            iss = token_parser.get_issuer_name()
-            attributes_by_issuer[iss] = claims
-            self._log_debug(context, f"disclosed claims {claims} from issuer {iss}")
-
-        all_attributes = self._extract_all_user_attributes(attributes_by_issuer)
+            if is_sd_jwt_kb_format(vp_token):
+                try:
+                    challenge = self._get_verifier_challenge(request_session)
+
+                    token_processor = VpVcSdJwtParserVerifier(
+                        vp_token, challenge["aud"], challenge["nonce"]
+                    )
+
+                    token_issuer = token_processor.get_issuer_name()
+
+                    whitelisted_keys = self.trust_evaluator.get_public_keys(token_issuer)
+                    token_processor.verify_signature(whitelisted_keys)
+                    token_processor.verify_challenge()
+
+                    if token_processor.is_expired() == True:
+                        raise VPExpired("VP is expired")
+
+                    claims = token_processor.get_credentials()
+                    iss = token_processor.get_issuer_name()
+
+                    extracted_attributes[iss] = claims
+                except MissingIssuer as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: missing issuer information",
+                        e400
+                    )
+                except InvalidJwkMetadataException as e500:
+                    return self._handle_500(
+                        context, 
+                        "trust error: cannot fetch public keys",
+                        e500
+                    )
+                except JWSVerificationError as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: invalid signature",
+                        e400
+                    )
+                except UnsupportedSdAlg as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: unsupported signature algorithm",
+                        e400
+                    )
+                except InvalidKeyBinding as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: nonce or aud mismatch",
+                        e400
+                    )
+                except ValueError as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: missing or invalid iat claim",
+                        e400
+                    )
+                except VPExpired as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: expired",
+                        e400
+                    )
+                except Exception as e400:
+                    breakpoint()
+                    return self._handle_400(
+                        context, 
+                        "trust error: cannot verify vp token",
+                        e400
+                    )
+            else:
+                try:
+                    token_processor = VpMDocCbor(vp_token)
+                except Exception as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: cannot parse vp token",
+                        e400
+                    )
+
+                try:
+                    token_processor.verify_signature()
+                    if token_processor.is_expired() == True:
+                        raise VPExpired("VP is expired")
+
+                    claims = token_processor.get_credentials()
+                    doc_type = token_processor.get_doc_type()
+
+                    extracted_attributes[doc_type] = claims
+                except MdocCborValidationError as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: invalid signature",
+                        e400
+                    )
+                except VPExpired as e400:
+                    return self._handle_400(
+                        context, 
+                        "invalid vp token: expired",
+                        e400
+                    )
+                except Exception as e400:
+                    return self._handle_400(
+                        context, 
+                        "trust error: cannot verify vp token",
+                        e400
+                    )
+
+        all_attributes = self._extract_all_user_attributes(extracted_attributes)
         iss_list_serialized = ";".join(credential_issuers)  # marshaling is whatever
         internal_resp = self._translate_response(
             all_attributes, iss_list_serialized, context
@@ -358,17 +405,6 @@ def _parse_authorization_response(
                     Exception("invalid program state"),
                 )
 
-    def _vp_verifier_factory(
-        self, presentation_submission: dict, token: str, session_data: dict
-    ) -> tuple[VpTokenParser, VpTokenVerifier]:
-        # TODO: la funzione dovrebbe consumare la presentation submission per sapere quale token
-        # ritornare - per ora viene ritornata l'unica implementazione possibile
-        challenge = self._get_verifier_challenge(session_data)
-        token_processor = VpVcSdJwtParserVerifier(
-            token, challenge["aud"], challenge["nonce"]
-        )
-        return (token_processor, deepcopy(token_processor))
-
     def _get_verifier_challenge(self, session_data: dict) -> VerifierChallenge:
         # TODO: check aud according to the LSP Potential singularities ...
         return {"aud": self.client_id, "nonce": session_data["nonce"]}