wip: storage layer + tests fixes

example/satosa/pyeudiw_backend.yaml

@@ -81,16 +81,11 @@ config:
         timeout: 6
 
   trust:
-    direct_trust:
-      module: pyeudiw.trust.default.direct_trust
+    direct_trust_sd_jwt_vc:
+      module: pyeudiw.trust.default.direct_trust_sd_jwt_vc
       class: DirectTrustSdJwtVc
       config:
         jwk_endpoint: /.well-known/jwt-vc-issuer
-        httpc_params:
-          connection:
-            ssl: true
-          session:
-            timeout: 6
     federation:
       module: pyeudiw.trust.default.federation
       class: FederationTrustModel

pyeudiw/jwt/parse.py

@@ -48,10 +48,10 @@ def unsafe_parse_jws(token: str) -> DecodedJwt:
 
 def extract_key_identifier(token_header: dict) -> JWK | KeyIdentifier_T:
     # TODO: the trust evaluation order might be mapped on the same configuration ordering
+    if "kid" in token_header.keys():
+        return KeyIdentifier_T(token_header["kid"])
     if "trust_chain" in token_header.keys():
-        return get_public_key_from_trust_chain(token_header["key"])
+        return get_public_key_from_trust_chain(token_header["kid"])
     if "x5c" in token_header.keys():
         return get_public_key_from_x509_chain(token_header["x5c"])
-    if "kid" in token_header.keys():
-        return KeyIdentifier_T(token_header["kid"])
     raise ValueError(f"unable to infer identifying key from token head: searched among keys {token_header.keys()}")

pyeudiw/openid4vp/interface.py

@@ -44,5 +44,12 @@ def verify_signature(self, public_key: JWK) -> None:
         :raises [InvalidSignatureException]:
         """
         raise NotImplementedError
+
+    def verify_challenge(self) -> None:
+        """
+        :raises []:
+        """
+        raise NotImplementedError
+
 
     # TODO: VP proof of possession verification method should be implemented

pyeudiw/openid4vp/vp_sd_jwt_vc.py

@@ -3,8 +3,10 @@
 from pyeudiw.jwk import JWK
 from pyeudiw.jwt.parse import KeyIdentifier_T, extract_key_identifier
 from pyeudiw.jwt.verification import is_jwt_expired
+from pyeudiw.openid4vp.exceptions import InvalidVPKeyBinding
 from pyeudiw.openid4vp.interface import VpTokenParser, VpTokenVerifier
-from pyeudiw.sd_jwt.schema import is_sd_jwt_kb_format
+from pyeudiw.sd_jwt.exceptions import InvalidKeyBinding, UnsupportedSdAlg
+from pyeudiw.sd_jwt.schema import VerifierChallenge, is_sd_jwt_kb_format
 from pyeudiw.sd_jwt.sd_jwt import SdJwt
 
 
@@ -39,3 +41,12 @@ def is_expired(self) -> bool:
 
     def verify_signature(self, public_key: JWK) -> None:
         return self.sdjwt.verify_issuer_jwt_signature(public_key)
+
+    def verify_challenge(self) -> None:
+        challenge : VerifierChallenge = {}
+        challenge["aud"] = self.verifier_id
+        challenge["nonce"] = self.verifier_nonce
+        try:
+            self.sdjwt.verify_holder_kb_jwt(challenge)
+        except (UnsupportedSdAlg, InvalidKeyBinding):
+            raise InvalidVPKeyBinding

pyeudiw/satosa/default/openid4vp_backend.py

@@ -95,7 +95,7 @@ def __init__(
 
         self.response_code_helper = ResponseCodeSource(self.config["response_code"]["sym_key"])
         trust_configuration = self.config.get("trust", {})
-        self.trust_evaluator = CombinedTrustEvaluator(dynamic_trust_evaluators_loader(trust_configuration))
+        self.trust_evaluator = CombinedTrustEvaluator(dynamic_trust_evaluators_loader(trust_configuration), self.db_engine)
         self.init_trust_resources()  # Questo carica risorse, metadata endpoint (sotto formate di attributi con pattern *_endpoint) etc, che satosa deve pubblicare
 
     def register_endpoints(self) -> list[tuple[str, Callable[[Context], Response]]]:

pyeudiw/satosa/default/response_handler.py

@@ -12,7 +12,7 @@
 
 from pyeudiw.jwk import JWK
 from pyeudiw.openid4vp.authorization_response import AuthorizeResponseDirectPost, AuthorizeResponsePayload
-from pyeudiw.openid4vp.exceptions import InvalidVPToken, KIDNotFound
+from pyeudiw.openid4vp.exceptions import InvalidVPKeyBinding, InvalidVPToken, KIDNotFound
 from pyeudiw.openid4vp.interface import VpTokenParser, VpTokenVerifier
 from pyeudiw.openid4vp.vp import Vp
 from pyeudiw.openid4vp.vp_sd_jwt_vc import VpVcSdJwtParserVerifier
@@ -164,11 +164,11 @@ def response_endpoint(self, context: Context, *args: tuple) -> Redirect | JsonRe
         try:
             request_session = self._retrieve_session_from_state(authz_payload.state)
         except AuthorizeUnmatchedResponse as e400:
-            self._handle_400(context, e400.args[0], e400.args[1])
+            return self._handle_400(context, e400.args[0], e400.args[1])
         except InvalidInternalStateError as e500:
-            self._handle_500(context, e500.args[0], "invalid state")
+            return self._handle_500(context, e500.args[0], "invalid state")
         except FinalizedSessionError as e400:
-            self._handle_400(context, e400.args[0], HTTPError(e400.args[0]))
+            return self._handle_400(context, e400.args[0], HTTPError(e400.args[0]))
 
         # the flow below is a simplified algorithm of authentication response processing, where:
         # (1) we don't check that presentation submission matches definition (yet)
@@ -180,13 +180,21 @@ def response_endpoint(self, context: Context, *args: tuple) -> Redirect | JsonRe
         for vp_token in encoded_vps:
             # verify vp token and extract user information
             # TODO: specialized try/except for each call, from line 182 to line 187
-            token_parser, token_verifier = self._vp_verifier_factory(authz_payload.presentation_submission, vp_token, request_session)
+            try:
+                token_parser, token_verifier = self._vp_verifier_factory(authz_payload.presentation_submission, vp_token, request_session)
+            except ValueError as e:
+                return self._handle_400(context, f"VP parsing error: {e}")
             pub_jwk = _find_vp_token_key(token_parser, self.trust_evaluator)
             token_verifier.verify_signature(pub_jwk)
+            try:
+                token_verifier.verify_challenge()
+            except InvalidVPKeyBinding as e:
+                return self._handle_400(context, f"VP parsing error: {e}")
             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)
         iss_list_serialized = ";".join(credential_issuers)  # marshaling is whatever
         internal_resp = self._translate_response(all_attributes, iss_list_serialized, context)

pyeudiw/storage/base_storage.py

@@ -7,13 +7,15 @@
 
 
 class TrustType(Enum):
-    X509 = 0
-    FEDERATION = 1
+    X509 = "x509"
+    FEDERATION = "federation"
+    DIRECT_TRUST_SD_JWT_VC = "direct_trust_sd_jwt_vc"
 
 
 trust_type_map: dict = {
     TrustType.X509: "x509",
-    TrustType.FEDERATION: "federation"
+    TrustType.FEDERATION: "federation",
+    TrustType.DIRECT_TRUST_SD_JWT_VC: "direct_trust_sd_jwt_vc"
 }
 
 trust_attestation_field_map: dict = {
@@ -170,7 +172,7 @@ def has_trust_anchor(self, entity_id: str) -> bool:
         """
         raise NotImplementedError()
 
-    def add_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType) -> str:
+    def add_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType, jwks: dict) -> str:
         """
         Add a trust attestation.
 
@@ -182,6 +184,8 @@ def add_trust_attestation(self, entity_id: str, attestation: list[str], exp: dat
         :type exp: datetime
         :param trust_type: the trust type.
         :type trust_type: TrustType
+        :param jwks: cached jwks
+        :type jwks: dict
 
         :returns: the document id.
         :rtype: str
@@ -220,7 +224,7 @@ def add_trust_anchor(self, entity_id: str, entity_configuration: str, exp: datet
         """
         raise NotImplementedError()
 
-    def update_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType) -> str:
+    def update_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType, jwks: dict) -> str:
         """
         Update a trust attestation.
 
@@ -232,6 +236,8 @@ def update_trust_attestation(self, entity_id: str, attestation: list[str], exp:
         :type exp: datetime
         :param trust_type: the trust type.
         :type trust_type: TrustType
+        :param jwks: cached jwks
+        :type jwks: dict
 
         :returns: the document id.
         :rtype: str

pyeudiw/storage/db_engine.py

@@ -156,24 +156,24 @@ def has_trust_attestation(self, entity_id: str) -> bool:
     def has_trust_anchor(self, entity_id: str) -> bool:
         return self.get_trust_anchor(entity_id) is not None
 
-    def add_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType = TrustType.FEDERATION) -> str:
-        return self.write("add_trust_attestation", entity_id, attestation, exp, trust_type)
+    def add_trust_attestation(self, entity_id: str, attestation: list[str] = [], exp: datetime = None, trust_type: TrustType = TrustType.FEDERATION, jwks: dict = None) -> str:
+        return self.write("add_trust_attestation", entity_id, attestation, exp, trust_type, jwks)
 
     def add_trust_attestation_metadata(self, entity_id: str, metadat_type: str, metadata: dict) -> str:
         return self.write("add_trust_attestation_metadata", entity_id, metadat_type, metadata)
 
     def add_trust_anchor(self, entity_id: str, entity_configuration: str, exp: datetime, trust_type: TrustType = TrustType.FEDERATION) -> str:
         return self.write("add_trust_anchor", entity_id, entity_configuration, exp, trust_type)
 
-    def update_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType = TrustType.FEDERATION) -> str:
-        return self.write("update_trust_attestation", entity_id, attestation, exp, trust_type)
+    def update_trust_attestation(self, entity_id: str, attestation: list[str] = [], exp: datetime = None, trust_type: TrustType = TrustType.FEDERATION, jwks: dict = None) -> str:
+        return self.write("update_trust_attestation", entity_id, attestation, exp, trust_type, jwks)
 
-    def add_or_update_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType = TrustType.FEDERATION) -> str:
+    def add_or_update_trust_attestation(self, entity_id: str, attestation: list[str] = [], exp: datetime = None, trust_type: TrustType = TrustType.FEDERATION, jwks: dict = None) -> str:
         try:
             self.get_trust_attestation(entity_id)
-            return self.write("update_trust_attestation", entity_id, attestation, exp, trust_type)
+            return self.write("update_trust_attestation", entity_id, attestation, exp, trust_type, jwks)
         except (EntryNotFound, ChainNotExist):
-            return self.write("add_trust_attestation", entity_id, attestation, exp, trust_type)
+            return self.write("add_trust_attestation", entity_id, attestation, exp, trust_type, jwks)
 
     def update_trust_anchor(self, entity_id: str, entity_configuration: dict, exp: datetime, trust_type: TrustType = TrustType.FEDERATION) -> str:
         return self.write("update_trust_anchor", entity_id, entity_configuration, exp, trust_type)

pyeudiw/storage/mongo_storage.py

@@ -12,6 +12,7 @@
     trust_anchor_field_map
 )
 from pyeudiw.storage.exceptions import (
+    ChainAlreadyExist,
     ChainNotExist,
     StorageEntryUpdateFailed
 )
@@ -238,53 +239,57 @@ def _add_entry(
 
         meth_suffix = collection[:-1]
         if getattr(self, f"has_{meth_suffix}")(entity_id):
-            # update it
-            getattr(self, f"update_{meth_suffix}")(entity_id, attestation, exp)
-            return entity_id
-            # raise ChainAlreadyExist(
-            # f"Chain with entity id {entity_id} already exist"
-            # )
+            # TODO: controllare funzionamento
+            # l'attestation passata come parametro non è quello che si aspetta il metodo di update
+            # bensì è l'intero oggetto trust_attestation
+
+            # # update it
+            # getattr(self, f"update_{meth_suffix}")(entity_id, attestation, exp)
+            # return entity_id
+            raise ChainAlreadyExist(f"Chain with entity id {entity_id} already exists")
 
         db_collection = getattr(self, collection)
         db_collection.insert_one(attestation)
         return entity_id
 
-    def _update_attestation_metadata(self, entity: dict, attestation: list[str], exp: datetime, trust_type: TrustType):
+    def _update_attestation_metadata(self, entity: dict, attestation: list[str], exp: datetime, trust_type: TrustType, jwks: dict):
         trust_name = trust_type_map[trust_type]
-        trust_field = trust_attestation_field_map[trust_type]
+        trust_field = trust_attestation_field_map.get(trust_type, None)
 
         trust_entity = entity.get(trust_name, {})
 
-        trust_entity[trust_field] = attestation
-        trust_entity["exp"] = exp
+        if trust_field and attestation: trust_entity[trust_field] = attestation
+        if exp: trust_entity["exp"] = exp
+        if jwks: trust_entity["jwks"] = jwks
 
         entity[trust_name] = trust_entity
 
         return entity
 
     def _update_anchor_metadata(self, entity: dict, attestation: list[str], exp: datetime, trust_type: TrustType):
         trust_name = trust_type_map[trust_type]
-        trust_field = trust_anchor_field_map[trust_type]
+        trust_field = trust_anchor_field_map.get(trust_type, None)
 
         trust_entity = entity.get(trust_name, {})
 
-        trust_entity[trust_field] = attestation
+        if trust_field and attestation: trust_entity[trust_field] = attestation
         trust_entity["exp"] = exp
 
         entity[trust_name] = trust_entity
 
         return entity
 
-    def add_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType) -> str:
+    def add_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType, jwks: dict) -> str:
         entity = {
             "entity_id": entity_id,
             "federation": {},
             "x509": {},
+            "direct_trust_sd_jwt_vc": {},
             "metadata": {}
         }
 
         updated_entity = self._update_attestation_metadata(
-            entity, attestation, exp, trust_type)
+            entity, attestation, exp, trust_type, jwks)
 
         return self._add_entry(
             "trust_attestations", entity_id, updated_entity, exp
@@ -326,11 +331,11 @@ def _update_trust_attestation(self, collection: str, entity_id: str, entity: dic
         )
         return documentStatus
 
-    def update_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType) -> str:
+    def update_trust_attestation(self, entity_id: str, attestation: list[str], exp: datetime, trust_type: TrustType, jwks: dict) -> str:
         old_entity = self._get_trust_attestation(
             "trust_attestations", entity_id) or {}
         upd_entity = self._update_attestation_metadata(
-            old_entity, attestation, exp, trust_type)
+            old_entity, attestation, exp, trust_type, jwks)
 
         return self._update_trust_attestation("trust_attestations", entity_id, upd_entity)
 

pyeudiw/tests/satosa/test_backend.py

@@ -24,6 +24,7 @@
     load_specification_from_yaml_string,
     import_ec
 )
+from pyeudiw.storage.base_storage import TrustType
 from pyeudiw.storage.db_engine import DBEngine
 from pyeudiw.tests.federation.base import (
     trust_chain_wallet,
@@ -58,6 +59,12 @@ def create_backend(self):
             exp=EXP,
         )
 
+        issuer_jwk = leaf_cred_jwk_prot.serialize(private=True)
+        db_engine_inst.add_or_update_trust_attestation(
+            entity_id=CREDENTIAL_ISSUER_ENTITY_ID,
+            trust_type=TrustType.DIRECT_TRUST_SD_JWT_VC,
+            jwks=issuer_jwk)
+
         self.backend = OpenID4VPBackend(
             Mock(), INTERNAL_ATTRIBUTES, CONFIG, BASE_URL, "name")
 
@@ -259,7 +266,7 @@ def test_vp_validation_in_response_endpoint(self, context):
         assert request_endpoint.status == "400"
         msg = json.loads(request_endpoint.message)
         assert msg["error"] == "invalid_request"
-        assert msg["error_description"] == "DirectPostResponse content parse and validation error. Single VPs are faulty."
+        assert msg["error_description"]
 
     def test_response_endpoint(self, context):
         self.backend.register_endpoints()

pyeudiw/tests/satosa/test_backend_trust.py

@@ -6,7 +6,7 @@
 
 from pyeudiw.tests.settings import (
     BASE_URL,
-    CONFIG_DIRECT_TRUST,
+    CONFIG,
     INTERNAL_ATTRIBUTES,
 )
 
@@ -18,7 +18,7 @@ def setup_direct_trust(self):
         self.backend = OpenID4VPBackend(
             Mock(),
             INTERNAL_ATTRIBUTES,
-            CONFIG_DIRECT_TRUST,
+            CONFIG,
             BASE_URL,
             "name"
         )

pyeudiw/tests/sd_jwt/test_sdjwt.py

@@ -101,8 +101,8 @@
 
 DISCLOSURES = [
     "WyJlbHVWNU9nM2dTTklJOEVZbnN4QV9BIiwgImZhbWlseV9uYW1lIiwgIkRvZSJd",
-    "WyJBSngtMDk1VlBycFR0TjRRTU9xUk9BIiwgImFkZHJlc3MiLCB7InN0cmVldF9hZGRy", \
-    "ZXNzIjogIjEyMyBNYWluIFN0IiwgImxvY2FsaXR5IjogIkFueXRvd24iLCAicmVnaW9u", \
+    "WyJBSngtMDk1VlBycFR0TjRRTU9xUk9BIiwgImFkZHJlc3MiLCB7InN0cmVldF9hZGRy" +
+    "ZXNzIjogIjEyMyBNYWluIFN0IiwgImxvY2FsaXR5IjogIkFueXRvd24iLCAicmVnaW9u" +
     "IjogIkFueXN0YXRlIiwgImNvdW50cnkiOiAiVVMifV0",
     "WyIyR0xDNDJzS1F2ZUNmR2ZyeU5STjl3IiwgImdpdmVuX25hbWUiLCAiSm9obiJd",
     "WyJsa2x4RjVqTVlsR1RQVW92TU5JdkNBIiwgIlVTIl0",