Merge pull request #1 from Zicchio/dev_vp_trust

wip: dev vp trust
italia · Oct 3, 2024 · cf6f2c4 · cf6f2c4
2 parents 5fd5100 + 6b5a6c1
commit cf6f2c4
Show file tree

Hide file tree

Showing 48 changed files with 2,005 additions and 416 deletions.
diff --git a/example/satosa/integration_test/main.py b/example/satosa/integration_test/main.py
@@ -4,6 +4,7 @@
 import datetime
 import base64
 from bs4 import BeautifulSoup
+import unittest.mock
 
 from pyeudiw.jwt import DEFAULT_SIG_KTY_MAP
 from pyeudiw.presentation_exchange.schemas.oid4vc_presentation_definition import PresentationDefinition
@@ -201,16 +202,29 @@
 # )
 # End deprecated footprint VP envelop
 
+# intercept call to issuer jwk
+issuer_vct_md = {
+    "issuer": settings['issuer'],
+    "jwks": {"keys": [ISSUER_PRIVATE_JWK.as_dict()]}
+}
+jwt_vc_issuer_endpoint_response = requests.Response()
+jwt_vc_issuer_endpoint_response.status_code = 200
+jwt_vc_issuer_endpoint_response.headers.update({"Content-Type": "application/json"})
+jwt_vc_issuer_endpoint_response._content = json.dumps(issuer_vct_md).encode('utf-8')
+patcher = unittest.mock.patch("pyeudiw.vci.jwks_provider.get_http_url", return_value=[issuer_vct_md])
+patcher.start()  # TODO: this does not works!!
+
 # take relevant information from RP's EC
-rp_ec_jwt = http_user_agent.get(
-    f'{IDP_BASEURL}/OpenID4VP/.well-known/openid-federation',
-    verify=False
-).content.decode()
-rp_ec = decode_jwt_payload(rp_ec_jwt)
+# rp_ec_jwt = http_user_agent.get(
+#     f'{IDP_BASEURL}/OpenID4VP/.well-known/openid-federation',
+#     verify=False
+# ).content.decode()
+# rp_ec = decode_jwt_payload(rp_ec_jwt)
 
-presentation_definition = rp_ec["metadata"]["wallet_relying_party"]["presentation_definition"]
-PresentationDefinition(**presentation_definition)
-assert response_uri == rp_ec["metadata"]['wallet_relying_party']["response_uris_supported"][0]
+# presentation_definition = rp_ec["metadata"]["wallet_relying_party"]["presentation_definition"]
+# encryption_key = rp_ec["metadata"]['wallet_relying_party']['jwks']['keys'][1]
+# PresentationDefinition(**presentation_definition)
+# assert response_uri == rp_ec["metadata"]['wallet_relying_party']["response_uris_supported"][0]
 
 response = {
     "state": red_data['state'],
@@ -228,9 +242,21 @@
         "aud": response_uri
     }
 }
+
+# This should be a public key provided somehow by the verifier/relying party, but as of now it is a private key granted by an angel
+encryption_key = {
+    "kty": "RSA",
+    "d": "QUZsh1NqvpueootsdSjFQz-BUvxwd3Qnzm5qNb-WeOsvt3rWMEv0Q8CZrla2tndHTJhwioo1U4NuQey7znijhZ177bUwPPxSW1r68dEnL2U74nKwwoYeeMdEXnUfZSPxzs7nY6b7vtyCoA-AjiVYFOlgKNAItspv1HxeyGCLhLYhKvS_YoTdAeLuegETU5D6K1xGQIuw0nS13Icjz79Y8jC10TX4FdZwdX-NmuIEDP5-s95V9DMENtVqJAVE3L-wO-NdDilyjyOmAbntgsCzYVGH9U3W_djh4t3qVFCv3r0S-DA2FD3THvlrFi655L0QHR3gu_Fbj3b9Ybtajpue_Q",
+    "e": "AQAB",
+    "use": "enc",
+    "kid": "9Cquk0X-fNPSdePQIgQcQZtD6J0IjIRrFigW2PPK_-w",
+    "n": "utqtxbs-jnK0cPsV7aRkkZKA9t4S-WSZa3nCZtYIKDpgLnR_qcpeF0diJZvKOqXmj2cXaKFUE-8uHKAHo7BL7T-Rj2x3vGESh7SG1pE0thDGlXj4yNsg0qNvCXtk703L2H3i1UXwx6nq1uFxD2EcOE4a6qDYBI16Zl71TUZktJwmOejoHl16CPWqDLGo9GUSk_MmHOV20m4wXWkB4qbvpWVY8H6b2a0rB1B1YPOs5ZLYarSYZgjDEg6DMtZ4NgiwZ-4N1aaLwyO-GLwt9Vf-NBKwoxeRyD3zWE2FXRFBbhKGksMrCGnFDsNl5JTlPjaM3kYyImE941ggcuc495m-Fw",
+    "p": "2zmGXIMCEHPphw778YjVTar1eycih6fFSJ4I4bl1iq167GqO0PjlOx6CZ1-OdBTVU7HfrYRiUK_BnGRdPDn-DQghwwkB79ZdHWL14wXnpB5y-boHz_LxvjsEqXtuQYcIkidOGaMG68XNT1nM4F9a8UKFr5hHYT5_UIQSwsxlRQ0",
+    "q": "2jMFt2iFrdaYabdXuB4QMboVjPvbLA-IVb6_0hSG_-EueGBvgcBxdFGIZaG6kqHqlB7qMsSzdptU0vn6IgmCZnX-Hlt6c5X7JB_q91PZMLTO01pbZ2Bk58GloalCHnw_mjPh0YPviH5jGoWM5RHyl_HDDMI-UeLkzP7ImxGizrM"
+}
 encrypted_response = JWEHelper(
     # RSA (EC is not fully supported todate)
-    JWK(rp_ec["metadata"]['wallet_relying_party']['jwks']['keys'][1])
+    JWK(encryption_key)
 ).encrypt(response)
 
 
@@ -280,4 +306,5 @@
     obt_att_value = attributes[result_index].contents[0].contents[0]
     assert exp_att_value == obt_att_value, f"wrong attrirbute parsing expected {exp_att_value}, obtained {obt_att_value}"
 
+patcher.stop()
 print('test passed')
diff --git a/example/satosa/pyeudiw_backend.yaml b/example/satosa/pyeudiw_backend.yaml
@@ -80,29 +80,74 @@ config:
       session:
         timeout: 6
 
-  federation:
-    metadata_type: "wallet_relying_party"
-    authority_hints:
-        - http://127.0.0.1:8000
-    trust_anchors:
-        - http://127.0.0.1:8000
-    default_sig_alg: "RS256"
-    federation_entity_metadata:
-        organization_name: Developers Italia SATOSA OpenID4VP backend
-        homepage_uri: https://developers.italia.it
-        policy_uri: https://developers.italia.it
-        tos_uri: https://developers.italia.it
-        logo_uri: https://developers.italia.it/assets/icons/logo-it.svg
-
-    # private jwk
-    federation_jwks:
-      - kty: RSA
-        d: QUZsh1NqvpueootsdSjFQz-BUvxwd3Qnzm5qNb-WeOsvt3rWMEv0Q8CZrla2tndHTJhwioo1U4NuQey7znijhZ177bUwPPxSW1r68dEnL2U74nKwwoYeeMdEXnUfZSPxzs7nY6b7vtyCoA-AjiVYFOlgKNAItspv1HxeyGCLhLYhKvS_YoTdAeLuegETU5D6K1xGQIuw0nS13Icjz79Y8jC10TX4FdZwdX-NmuIEDP5-s95V9DMENtVqJAVE3L-wO-NdDilyjyOmAbntgsCzYVGH9U3W_djh4t3qVFCv3r0S-DA2FD3THvlrFi655L0QHR3gu_Fbj3b9Ybtajpue_Q
-        e: AQAB
-        kid: 9Cquk0X-fNPSdePQIgQcQZtD6J0IjIRrFigW2PPK_-w
-        n: utqtxbs-jnK0cPsV7aRkkZKA9t4S-WSZa3nCZtYIKDpgLnR_qcpeF0diJZvKOqXmj2cXaKFUE-8uHKAHo7BL7T-Rj2x3vGESh7SG1pE0thDGlXj4yNsg0qNvCXtk703L2H3i1UXwx6nq1uFxD2EcOE4a6qDYBI16Zl71TUZktJwmOejoHl16CPWqDLGo9GUSk_MmHOV20m4wXWkB4qbvpWVY8H6b2a0rB1B1YPOs5ZLYarSYZgjDEg6DMtZ4NgiwZ-4N1aaLwyO-GLwt9Vf-NBKwoxeRyD3zWE2FXRFBbhKGksMrCGnFDsNl5JTlPjaM3kYyImE941ggcuc495m-Fw
-        p: 2zmGXIMCEHPphw778YjVTar1eycih6fFSJ4I4bl1iq167GqO0PjlOx6CZ1-OdBTVU7HfrYRiUK_BnGRdPDn-DQghwwkB79ZdHWL14wXnpB5y-boHz_LxvjsEqXtuQYcIkidOGaMG68XNT1nM4F9a8UKFr5hHYT5_UIQSwsxlRQ0
-        q: 2jMFt2iFrdaYabdXuB4QMboVjPvbLA-IVb6_0hSG_-EueGBvgcBxdFGIZaG6kqHqlB7qMsSzdptU0vn6IgmCZnX-Hlt6c5X7JB_q91PZMLTO01pbZ2Bk58GloalCHnw_mjPh0YPviH5jGoWM5RHyl_HDDMI-UeLkzP7ImxGizrM
+  trust:
+    direct_trust:
+      module: pyeudiw.trust.default.direct_trust
+      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
+      config:
+        metadata_type: "wallet_relying_party"
+        authority_hints:
+            - http://127.0.0.1:8000
+        trust_anchors:
+            - public_keys: [ ... ]
+            - http://127.0.0.1:8000
+        default_sig_alg: "RS256"
+        federation_entity_metadata:
+            organization_name: Developers Italia SATOSA OpenID4VP backend
+            homepage_uri: https://developers.italia.it
+            policy_uri: https://developers.italia.it
+            tos_uri: https://developers.italia.it
+            logo_uri: https://developers.italia.it/assets/icons/logo-it.svg
+        federation_jwks:
+          - kty: RSA
+            d: QUZsh1NqvpueootsdSjFQz-BUvxwd3Qnzm5qNb-WeOsvt3rWMEv0Q8CZrla2tndHTJhwioo1U4NuQey7znijhZ177bUwPPxSW1r68dEnL2U74nKwwoYeeMdEXnUfZSPxzs7nY6b7vtyCoA-AjiVYFOlgKNAItspv1HxeyGCLhLYhKvS_YoTdAeLuegETU5D6K1xGQIuw0nS13Icjz79Y8jC10TX4FdZwdX-NmuIEDP5-s95V9DMENtVqJAVE3L-wO-NdDilyjyOmAbntgsCzYVGH9U3W_djh4t3qVFCv3r0S-DA2FD3THvlrFi655L0QHR3gu_Fbj3b9Ybtajpue_Q
+            e: AQAB
+            kid: 9Cquk0X-fNPSdePQIgQcQZtD6J0IjIRrFigW2PPK_-w
+            n: utqtxbs-jnK0cPsV7aRkkZKA9t4S-WSZa3nCZtYIKDpgLnR_qcpeF0diJZvKOqXmj2cXaKFUE-8uHKAHo7BL7T-Rj2x3vGESh7SG1pE0thDGlXj4yNsg0qNvCXtk703L2H3i1UXwx6nq1uFxD2EcOE4a6qDYBI16Zl71TUZktJwmOejoHl16CPWqDLGo9GUSk_MmHOV20m4wXWkB4qbvpWVY8H6b2a0rB1B1YPOs5ZLYarSYZgjDEg6DMtZ4NgiwZ-4N1aaLwyO-GLwt9Vf-NBKwoxeRyD3zWE2FXRFBbhKGksMrCGnFDsNl5JTlPjaM3kYyImE941ggcuc495m-Fw
+            p: 2zmGXIMCEHPphw778YjVTar1eycih6fFSJ4I4bl1iq167GqO0PjlOx6CZ1-OdBTVU7HfrYRiUK_BnGRdPDn-DQghwwkB79ZdHWL14wXnpB5y-boHz_LxvjsEqXtuQYcIkidOGaMG68XNT1nM4F9a8UKFr5hHYT5_UIQSwsxlRQ0
+            q: 2jMFt2iFrdaYabdXuB4QMboVjPvbLA-IVb6_0hSG_-EueGBvgcBxdFGIZaG6kqHqlB7qMsSzdptU0vn6IgmCZnX-Hlt6c5X7JB_q91PZMLTO01pbZ2Bk58GloalCHnw_mjPh0YPviH5jGoWM5RHyl_HDDMI-UeLkzP7ImxGizrM
+    x509:
+      module: pyeudiw.trust.default.x509
+      class: X509TrustModel
+      config:
+        trust_anchor_certificates:
+            - "todo"
+        trust_anchors_cn: # we might mix CN and SAN together
+            - http://127.0.0.1:8000
+  # FORMER IMPLEMENTATION OF TRUST CONFIGURATION
+  # federation:
+  #   metadata_type: "wallet_relying_party"
+  #   authority_hints:
+  #       - http://127.0.0.1:8000
+  #   trust_anchors:
+  #       - http://127.0.0.1:8000
+  #   default_sig_alg: "RS256"
+  #   federation_entity_metadata:
+  #       organization_name: Developers Italia SATOSA OpenID4VP backend
+  #       homepage_uri: https://developers.italia.it
+  #       policy_uri: https://developers.italia.it
+  #       tos_uri: https://developers.italia.it
+  #       logo_uri: https://developers.italia.it/assets/icons/logo-it.svg
+
+  #   # private jwk
+  #   federation_jwks:
+  #     - kty: RSA
+  #       d: QUZsh1NqvpueootsdSjFQz-BUvxwd3Qnzm5qNb-WeOsvt3rWMEv0Q8CZrla2tndHTJhwioo1U4NuQey7znijhZ177bUwPPxSW1r68dEnL2U74nKwwoYeeMdEXnUfZSPxzs7nY6b7vtyCoA-AjiVYFOlgKNAItspv1HxeyGCLhLYhKvS_YoTdAeLuegETU5D6K1xGQIuw0nS13Icjz79Y8jC10TX4FdZwdX-NmuIEDP5-s95V9DMENtVqJAVE3L-wO-NdDilyjyOmAbntgsCzYVGH9U3W_djh4t3qVFCv3r0S-DA2FD3THvlrFi655L0QHR3gu_Fbj3b9Ybtajpue_Q
+  #       e: AQAB
+  #       kid: 9Cquk0X-fNPSdePQIgQcQZtD6J0IjIRrFigW2PPK_-w
+  #       n: utqtxbs-jnK0cPsV7aRkkZKA9t4S-WSZa3nCZtYIKDpgLnR_qcpeF0diJZvKOqXmj2cXaKFUE-8uHKAHo7BL7T-Rj2x3vGESh7SG1pE0thDGlXj4yNsg0qNvCXtk703L2H3i1UXwx6nq1uFxD2EcOE4a6qDYBI16Zl71TUZktJwmOejoHl16CPWqDLGo9GUSk_MmHOV20m4wXWkB4qbvpWVY8H6b2a0rB1B1YPOs5ZLYarSYZgjDEg6DMtZ4NgiwZ-4N1aaLwyO-GLwt9Vf-NBKwoxeRyD3zWE2FXRFBbhKGksMrCGnFDsNl5JTlPjaM3kYyImE941ggcuc495m-Fw
+  #       p: 2zmGXIMCEHPphw778YjVTar1eycih6fFSJ4I4bl1iq167GqO0PjlOx6CZ1-OdBTVU7HfrYRiUK_BnGRdPDn-DQghwwkB79ZdHWL14wXnpB5y-boHz_LxvjsEqXtuQYcIkidOGaMG68XNT1nM4F9a8UKFr5hHYT5_UIQSwsxlRQ0
+  #       q: 2jMFt2iFrdaYabdXuB4QMboVjPvbLA-IVb6_0hSG_-EueGBvgcBxdFGIZaG6kqHqlB7qMsSzdptU0vn6IgmCZnX-Hlt6c5X7JB_q91PZMLTO01pbZ2Bk58GloalCHnw_mjPh0YPviH5jGoWM5RHyl_HDDMI-UeLkzP7ImxGizrM
 
     # trust_marks: # todo
     # - ...

diff --git a/pyeudiw/federation/statements.py b/pyeudiw/federation/statements.py
@@ -16,7 +16,7 @@
 from pydantic import ValidationError
 from pyeudiw.jwt.utils import decode_jwt_payload, decode_jwt_header
 from pyeudiw.jwt import JWSHelper
-from pyeudiw.jwk import find_jwk
+from pyeudiw.jwk import find_jwk_by_kid
 from pyeudiw.tools.utils import get_http_url
 
 import logging
@@ -171,7 +171,7 @@ def validate_by(self, ec: dict) -> bool:
                 f"{self.header.get('kid')} not found in {ec.jwks}"
             )
 
-        _jwk = find_jwk(_kid, ec.jwks)
+        _jwk = find_jwk_by_kid(_kid, ec.jwks)
 
         # verify signature
         jwsh = JWSHelper(_jwk)
@@ -211,7 +211,7 @@ def validate_by_its_issuer(self) -> bool:
             return False
 
         # verify signature
-        _jwk = find_jwk(_kid, ec.jwks)
+        _jwk = find_jwk_by_kid(_kid, ec.jwks)
         jwsh = JWSHelper(_jwk)
         payload = jwsh.verify(self.jwt)
         self.is_valid = True
@@ -314,7 +314,7 @@ def validate_by_itself(self) -> bool:
                 f"{_kid} not found in {self.jwks}")  # pragma: no cover
 
         # verify signature
-        _jwk = find_jwk(_kid, self.jwks)
+        _jwk = find_jwk_by_kid(_kid, self.jwks)
         jwsh = JWSHelper(_jwk)
         jwsh.verify(self.jwt)
         self.is_valid = True
@@ -539,7 +539,7 @@ def validate_descendant_statement(self, jwt: str) -> bool:
                 f"{_kid} not found in {self.jwks}")
 
         # verify signature
-        _jwk = find_jwk(_kid, self.jwks)
+        _jwk = find_jwk_by_kid(_kid, self.jwks)
         jwsh = JWSHelper(_jwk)
         payload = jwsh.verify(jwt)
 
@@ -565,7 +565,7 @@ def validate_by_superior_statement(self, jwt: str, ec: 'EntityStatement') -> str
             ec.validate_by_itself()
             ec.validate_descendant_statement(jwt)
             _jwks = get_federation_jwks(payload)
-            _jwk = find_jwk(self.header["kid"], _jwks)
+            _jwk = find_jwk_by_kid(self.header["kid"], _jwks)
 
             jwsh = JWSHelper(_jwk)
             payload = jwsh.verify(self.jwt)

diff --git a/pyeudiw/openid4vp/redirect.py → pyeudiw/federation/trust_chain/__init__.py b/pyeudiw/openid4vp/redirect.py → pyeudiw/federation/trust_chain/__init__.py
diff --git a/pyeudiw/federation/trust_chain/builder.py b/pyeudiw/federation/trust_chain/builder.py
@@ -0,0 +1 @@
+# TODO: move trust_chain_builder.py here
diff --git a/pyeudiw/federation/trust_chain/parse.py b/pyeudiw/federation/trust_chain/parse.py
@@ -0,0 +1,5 @@
+from pyeudiw.jwk import JWK
+
+
+def get_public_key_from_trust_chain(trust_chain: list[str]) -> JWK:
+    raise NotImplementedError("TODO")
diff --git a/pyeudiw/federation/trust_chain/validator.py b/pyeudiw/federation/trust_chain/validator.py
@@ -0,0 +1 @@
+# TODO: move trust_chain_validator.py here
diff --git a/pyeudiw/federation/trust_chain_validator.py b/pyeudiw/federation/trust_chain_validator.py
@@ -15,7 +15,7 @@
     InvalidEntityStatement
 )
 
-from pyeudiw.jwk import find_jwk
+from pyeudiw.jwk import find_jwk_by_kid
 from pyeudiw.jwk.exceptions import KidNotFoundError, InvalidKid
 
 logger = logging.getLogger(__name__)
@@ -127,7 +127,7 @@ def validate(self) -> bool:
         es_header = decode_jwt_header(last_element)
         es_payload = decode_jwt_payload(last_element)
 
-        ta_jwk = find_jwk(
+        ta_jwk = find_jwk_by_kid(
             es_header.get("kid", None), self.trust_anchor_jwks
         )
 
@@ -165,7 +165,7 @@ def validate(self) -> bool:
             st_payload = decode_jwt_payload(st)
 
             try:
-                jwk = find_jwk(
+                jwk = find_jwk_by_kid(
                     st_header.get("kid", None), fed_jwks
                 )
             except (KidNotFoundError, InvalidKid):

diff --git a/pyeudiw/jwk/__init__.py b/pyeudiw/jwk/__init__.py
@@ -153,7 +153,7 @@ def jwk_form_dict(key: dict, hash_func: str = "SHA-256") -> RSAJWK | ECJWK:
         return ECJWK(key, hash_func, ec_crv)
 
 
-def find_jwk(kid: str, jwks: list[dict], as_dict: bool = True) -> dict | JWK:
+def find_jwk_by_kid(kid: str, jwks: list[dict], as_dict: bool = True) -> dict | JWK:
     """
     Find the JWK with the indicated kid in the jwks list.
 

diff --git a/pyeudiw/jwt/parse.py b/pyeudiw/jwt/parse.py
@@ -0,0 +1,56 @@
+from dataclasses import dataclass
+from jwcrypto.common import base64url_decode, json_decode
+from pyeudiw.federation.trust_chain.parse import get_public_key_from_trust_chain
+from pyeudiw.jwk import JWK
+from pyeudiw.jwt.utils import is_jwt_format
+from pyeudiw.x509.verify import get_public_key_from_x509_chain
+
+KeyIdentifier_T = str
+
+
+@dataclass(frozen=True)
+class DecodedJwt:
+    """
+    Schema class for a decoded jwt.
+    This class is not meant to be instantiated directly. Use instead
+    the static metho parse(str) -> UnverfiedJwt
+    """
+    jwt: str
+    header: dict
+    payload: dict
+    signature: str
+
+    def parse(jws: str) -> 'DecodedJwt':
+        return unsafe_parse_jws(jws)
+
+
+def _unsafe_decode_part(part: str) -> dict:
+    return json_decode(base64url_decode(part))
+
+
+def unsafe_parse_jws(token: str) -> DecodedJwt:
+    """Parse a token into it's component.
+    Correctness of this function  is not guaranteed when the token is in a
+    derived format, such as sd-jwt and jwe.
+    """
+    if not is_jwt_format(token):
+        raise ValueError(f"unable to parse {token}: not a jwt")
+    b64header, b64payload, signature, *_ = token.split(".")
+    head = {}
+    payload = {}
+    try:
+        head = _unsafe_decode_part(b64header)
+        payload = _unsafe_decode_part(b64payload)
+    except Exception as e:
+        raise ValueError(f"unable to decode JWS part: {e}")
+    return DecodedJwt(token, head, payload, signature=signature)
+
+
+def extract_key_identifier(token_header: dict) -> JWK | KeyIdentifier_T:
+    if "trust_chain" in token_header.keys():
+        return get_public_key_from_trust_chain(token_header["key"])
+    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()}")
diff --git a/pyeudiw/jwt/schemas/jwt.py b/pyeudiw/jwt/schemas/jwt.py
@@ -1,4 +1,3 @@
-from dataclasses import dataclass
 from pydantic import BaseModel, Field
 from pyeudiw.federation.schemas.wallet_relying_party import SigningAlgValuesSupported, EncryptionAlgValuesSupported, EncryptionEncValuesSupported
 
@@ -11,11 +10,3 @@ class JWTConfig(BaseModel):
     enc_alg_supported: list[EncryptionAlgValuesSupported]
     enc_enc_supported: list[EncryptionEncValuesSupported]
     sig_alg_supported: list[SigningAlgValuesSupported]
-
-
-@dataclass(frozen=True)
-class UnverfiedJwt:
-    jwt: str
-    header: dict
-    payload: dict
-    signature: str