fix: take trust_anchors from parameters

Author: PascalDR

pyeudiw/tests/satosa/test_backend.py

@@ -1,5 +1,6 @@
 import os
 import uuid
+import copy
 import base64
 import datetime
 import json
@@ -108,12 +109,15 @@ def issue_sd_jwt(specification: dict, settings: dict, issuer_key: JWK, holder_ke
 def _mock_auth_callback_function(context: Context, internal_data: InternalData):
     return JsonResponse({"response": "Authentication successful"}, status="200")
 
+def base64url_to_int(val):
+    import base64
+    import binascii
+    return int.from_bytes(base64.urlsafe_b64decode(val + '=='), 'big')
 class TestOpenID4VPBackend:
     @pytest.fixture(autouse=True)
     def create_backend(self):
-
         db_engine_inst = DBEngine(CONFIG["storage"])
-
+        
         jwk = {
             "kty": "RSA",
             "use": "sig",
@@ -129,42 +133,31 @@ def create_backend(self):
             "qi": "kn9Etj4a2erCUmoZUQalPjHxCRYm5Q3wAkFIRGSQADA51mkwQHyTYqXbHcmXn2ZgXBVI6XDWJB51Me-NCPfITTlusqxvATF7Q-QJtdK_FbgNtcVRNc1FMq_M7VBHA1i9wJR7T4t57aywfXPmlsA5TToTDRe-ybdw0C3ys4KQATs"
         }
 
-        def base64url_to_int(val):
-            import base64
-            import binascii
-            return int.from_bytes(base64.urlsafe_b64decode(val + '=='), 'big')
-
         # Extract components from JWK
-        n = base64url_to_int(jwk['n'])
-        e = base64url_to_int(jwk['e'])
-        d = base64url_to_int(jwk['d'])
-        p = base64url_to_int(jwk['p'])
-        q = base64url_to_int(jwk['q'])
-        dp = base64url_to_int(jwk['dp'])
-        dq = base64url_to_int(jwk['dq'])
-        qi = base64url_to_int(jwk['qi'])
+        _n = base64url_to_int(jwk['n'])
+        _e = base64url_to_int(jwk['e'])
+        _d = base64url_to_int(jwk['d'])
+        _p = base64url_to_int(jwk['p'])
+        _q = base64url_to_int(jwk['q'])
+        _dp = base64url_to_int(jwk['dp'])
+        _dq = base64url_to_int(jwk['dq'])
+        _qi = base64url_to_int(jwk['qi'])
 
         # Create RSA private key
         private_key = rsa.RSAPrivateNumbers(
-            p=p,
-            q=q,
-            d=d,
-            dmp1=dp,
-            dmq1=dq,
-            iqmp=qi,
-            public_numbers=rsa.RSAPublicNumbers(e=e, n=n)
+            p=_p,
+            q=_q,
+            d=_d,
+            dmp1=_dp,
+            dmq1=_dq,
+            iqmp=_qi,
+            public_numbers=rsa.RSAPublicNumbers(e=_e, n=_n)
         ).private_key()
 
         self.chain = der_list_to_pem_list(gen_chain(leaf_private_key=private_key))
         issuer_pem = self.chain[-1]
         self.x509_leaf_private_key = jwk
 
-        db_engine_inst.add_trust_anchor(
-            entity_id=ta_ec["iss"],
-            entity_configuration=ta_ec_signed,
-            exp=EXP,
-        )
-
         db_engine_inst.add_trust_anchor(
             entity_id="ca.example.com",
             entity_configuration=issuer_pem,

pyeudiw/tests/settings.py

@@ -6,6 +6,8 @@
 from pyeudiw.tools.utils import exp_from_now, iat_now
 from pyeudiw.tests.x509.test_x509 import gen_chain
 
+from pyeudiw.tests.federation.base import ta_jwk
+
 BASE_URL = "https://example.com"
 AUTHZ_PAGE = "example.com"
 AUTH_ENDPOINT = "https://example.com/auth"
@@ -216,7 +218,11 @@
                 "metadata": _METADATA,
                 "metadata_type": "openid_credential_verifier",
                 "authority_hints": ["https://trust-anchor.example.org"],
-                "trust_anchors": ["https://trust-anchor.example.org"],
+                "trust_anchors": {
+                    "https://trust-anchor.example.org": [
+                        ta_jwk.serialize(private=False),
+                    ]
+                },
                 "default_sig_alg": "RS256",
                 "federation_jwks": [
                     {

pyeudiw/trust/handler/federation.py

@@ -18,6 +18,10 @@
 from pyeudiw.trust.exceptions import MissingProtocolSpecificJwks, UnknownTrustAnchor
 from pyeudiw.trust.handler.interface import TrustHandlerInterface
 from pyeudiw.trust.model.trust_source import TrustSourceData, TrustEvaluationType
+from pyeudiw.federation.statements import (
+    get_entity_configurations,
+    get_entity_statements,
+)
 
 from .commons import DEFAULT_HTTPC_PARAMS
 
@@ -31,7 +35,7 @@ def __init__(
         self,
         metadata: List[dict],
         authority_hints: List[str],
-        trust_anchors: List[str],
+        trust_anchors: dict[str, dict[str, str]],
         default_sig_alg: str,
         federation_jwks: List[dict[str, Union[str, List[str]]]],
         trust_marks: List[dict],
@@ -53,7 +57,7 @@ def __init__(
         self.metadata_type = metadata_type
         self.metadata: dict = metadata
         self.authority_hints: List[str] = authority_hints
-        self.trust_anchors: List[str] = trust_anchors
+        self.trust_anchors: dict[str, dict[str, str]] = trust_anchors
         self.default_sig_alg: str = default_sig_alg
         self.federation_jwks: List[dict[str, Union[str, List[str]]]] = federation_jwks
         self.trust_marks: List[dict] = trust_marks
@@ -189,23 +193,30 @@ def validate_trust_material(
                 "Unknown Trust Anchor: can't find 'iss' in the "
                 f"first entity statement: {_first_statement} "
             )
-
-        try:
-            trust_anchor = db_engine.get_trust_anchor(trust_anchor_eid)
-        except EntryNotFound:
+        
+        if not trust_anchor_eid in self.trust_anchors:
             raise UnknownTrustAnchor(
                 f"Unknown Trust Anchor: '{trust_anchor_eid}' is not "
                 "a recognizable Trust Anchor."
             )
-
-        decoded_ec = decode_jwt_payload(
-            trust_anchor['federation']['entity_configuration']
-        )
-        jwks = decoded_ec.get('jwks', {}).get('keys', [])
+        
+        if len(self.trust_anchors[trust_anchor_eid]) != 0:
+            jwks = self.trust_anchors[trust_anchor_eid]
+        else:
+            try:
+                trust_anchor = get_entity_configurations(trust_anchor_eid, self.httpc_params, False)
+                decoded_ec = decode_jwt_payload(
+                    trust_anchor['federation']['entity_configuration']
+                )
+                jwks = decoded_ec.get('jwks', {}).get('keys', [])
+            except Exception as e:
+                raise UnknownTrustAnchor(
+                    f"Cannot fetch Trust Anchor '{trust_anchor_eid}' entity configuration: {e}"
+                ) from e
 
         if not jwks:
             raise MissingProtocolSpecificJwks(
-                f"Cannot find any jwks in {decoded_ec}"
+                f"Cannot find any jwks in for the Trust Anchor '{trust_anchor_eid}'"
             )
 
         tc = StaticTrustChainValidator(