fix: added PoP signature validation on vp token with holder key binding

italia · Sep 19, 2023 · 5472a9c · 5472a9c
1 parent e46b890
commit 5472a9c
Show file tree

Hide file tree

Showing 4 changed files with 35 additions and 11 deletions.
diff --git a/example/satosa/integration_test/main.py b/example/satosa/integration_test/main.py
@@ -187,8 +187,10 @@
     aud=str(uuid.uuid4()),
     sign_alg=DEFAULT_SIG_KTY_MAP[WALLET_PRIVATE_JWK.key.kty],
     holder_key=(
-        import_pyca_pri_rsa(WALLET_PRIVATE_JWK.key.priv_key,
-                            kid=WALLET_PRIVATE_JWK.kid)
+        import_pyca_pri_rsa(
+            WALLET_PRIVATE_JWK.key.priv_key,
+            kid=WALLET_PRIVATE_JWK.kid
+        )
         if sd_specification.get("key_binding", False)
         else None
     )

diff --git a/pyeudiw/openid4vp/vp_sd_jwt.py b/pyeudiw/openid4vp/vp_sd_jwt.py
@@ -1,4 +1,5 @@
 from pyeudiw.jwk import JWK
+from pyeudiw.jwt import JWSHelper
 from pyeudiw.sd_jwt import verify_sd_jwt
 
 from pyeudiw.jwk.exceptions import KidNotFoundError
@@ -20,6 +21,11 @@ def verify_sdjwt(
         issuer_jwk = JWK(issuer_jwks_by_kid[self.credential_headers["kid"]])
         holder_jwk = JWK(self.credential_payload["cnf"]["jwk"])
 
+        # verify PoP
+        jws = JWSHelper(holder_jwk)
+        if not jws.verify(self.jwt):
+            return False
+
         result = verify_sd_jwt(
             sd_jwt_presentation=self.payload["vp"],
             issuer_key=issuer_jwk,

diff --git a/pyeudiw/satosa/backend.py b/pyeudiw/satosa/backend.py
@@ -701,9 +701,14 @@ def get_response_endpoint(self, context):
         finalized_session = None
 
         try:
-            finalized_session = self.db_engine.get_by_state_and_session_id(
-                state=state, session_id=session_id
-            )
+            if state:
+                finalized_session = self.db_engine.get_by_state_and_session_id(
+                    state=state, session_id=session_id
+                )
+            else:
+                finalized_session = self.db_engine.get_by_session_id(
+                    session_id=session_id
+                )
         except Exception as e:
             _msg = f"Error while retrieving session by state {state} and session_id {session_id}: {e}"
             return self.handle_error(

diff --git a/pyeudiw/sd_jwt/__init__.py b/pyeudiw/sd_jwt/__init__.py
@@ -158,20 +158,31 @@ def _cb_get_issuer_key(issuer: str, settings: dict, adapted_keys: dict):
         raise Exception(f"Unknown issuer: {issuer}")
 
 
-def verify_sd_jwt(sd_jwt_presentation: str, issuer_key: JWK, holder_key: JWK, settings: dict = {'default_exp': 60, 'key_binding': True}) -> dict:
-    settings.update({"issuer": unpad_jwt_payload(sd_jwt_presentation)["iss"]})
+def verify_sd_jwt(
+    sd_jwt_presentation: str, 
+    issuer_key: JWK, 
+    holder_key: JWK, 
+    settings: dict = {'default_exp': 60, 'key_binding': True}
+) -> dict:
+
+    settings.update(
+        {
+            "issuer": unpad_jwt_payload(sd_jwt_presentation)["iss"]
+        }
+    )
     adapted_keys = {
         "issuer_key": jwcrypto.jwk.JWK(**issuer_key.as_dict()),
         "holder_key": jwcrypto.jwk.JWK(**holder_key.as_dict()),
         "issuer_public_key": jwcrypto.jwk.JWK(**issuer_key.as_dict())
     }
+
     serialization_format = "compact"
     sdjwt_at_verifier = SDJWTVerifier(
         sd_jwt_presentation,
-        (lambda x: _cb_get_issuer_key(x, settings, adapted_keys)),
-        None,
-        None,
-        serialization_format=serialization_format,
+        cb_get_issuer_key = (lambda x: _cb_get_issuer_key(x, settings, adapted_keys)),
+        expected_aud = None,
+        expected_nonce = None,
+        serialization_format = serialization_format,
     )
 
     return sdjwt_at_verifier.get_verified_payload()