Updating fedservice to adhere to draft.43 of OpenID Federation.

Author: rohe

src/fedservice/entity/__init__.py

@@ -73,9 +73,15 @@ def __init__(self,
                 _kwargs.update(_args)
                 setattr(self, key, instantiate(val["class"], **_kwargs))
 
+        _args = {}
+        for claim in ["trust_mark_issuers", "trust_mark_owners"]:
+            _val = kwargs.get(claim)
+            if _val:
+                _args[claim] = _val
+
         self.context = FederationContext(entity_id=entity_id, upstream_get=self.unit_get,
                                          authority_hints=authority_hints, keyjar=self.keyjar,
-                                         preference=preference)
+                                         preference=preference, **_args)
 
         if client_authn_methods:
             self.context.client_authn_methods = client_auth_setup(client_authn_methods)
@@ -375,8 +381,7 @@ def verify_trust_mark(self, trust_mark: str, check_with_issuer: Optional[bool] =
             resp = self.do_request("trust_mark_status",
                                    request_args={
                                        'sub': verified_trust_mark['sub'],
-                                       'id': verified_trust_mark['id'],
-                                       'trust_mark_id': verified_trust_mark['id']
+                                       'trust_mark_id': verified_trust_mark['trust_mark_id']
                                    },
                                    fetch_endpoint=_tmi_trust_chain.metadata["federation_entity"][
                                        "federation_trust_mark_status_endpoint"]

src/fedservice/entity/client/__init__.py

@@ -40,6 +40,8 @@ def __init__(self,
                  keyjar: Optional[KeyJar] = None,
                  priority: Optional[List[str]] = None,
                  trust_marks: Optional[List[str]] = None,
+                 trust_mark_issuers: Optional[dict] = None,
+                 trust_mark_owners: Optional[dict] = None,
                  trusted_roots: Optional[dict] = None,
                  metadata: Optional[dict] = None,
                  ):
@@ -54,6 +56,8 @@ def __init__(self,
                                    keyjar=keyjar,
                                    metadata=metadata,
                                    trust_marks=trust_marks,
+                                   trust_mark_issuers=trust_mark_issuers,
+                                   trust_mark_owners=trust_mark_owners,
                                    tr_priority=priority
                                    )
 
@@ -456,7 +460,7 @@ def parse_request_response(self, service, reqresp, response_body_type="", state=
         elif 400 <= reqresp.status_code < 500:
             logger.error("Error response ({}): {}".format(reqresp.status_code, reqresp.text))
             # expecting an error response
-            ctype =  get_content_type(reqresp)
+            ctype = get_content_type(reqresp)
             _deser_method = get_deserialization_method(ctype)
             if not _deser_method:
                 _deser_method = "json"

src/fedservice/entity/context.py

@@ -42,6 +42,8 @@ def __init__(self,
                  authority_hints: Optional[Union[list, str, Callable]] = None,
                  keyjar: Optional[KeyJar] = None,
                  preference: Optional[dict] = None,
+                 trust_mark_issuers: Optional[dict] = None,
+                 trust_mark_owners: Optional[dict] = None,
                  **kwargs
                  ):
 
@@ -59,6 +61,9 @@ def __init__(self,
         self.trust_marks = trust_marks or config.get('trust_marks', [])
         self.trusted_roots = trusted_roots or config.get('trusted_roots', {})
         self.authority_hints = authority_hints or config.get('authority_hints', [])
+        self.trust_mark_issuers = trust_mark_issuers or config.get('trust_mark_issuers', {})
+        self.trust_mark_owners = trust_mark_owners or config.get('trust_mark_owners', {})
+        self.trusted_roots = trusted_roots or config.get('trusted_roots', {})
 
         self.trust_chain = {}
         # self.issuer = self.entity_id

src/fedservice/entity/function/policy_operator.py

@@ -65,6 +65,11 @@ class Add(PolicyOperator):
     def __call__(self, claim, metadata, metadata_policy):
         if claim in metadata:
             for val in metadata_policy[claim][self.name]:
+                # the metadata claim value must be a list otherwise append doesn't work
+                if isinstance(metadata, list):
+                    pass
+                else:
+                    metadata[claim] = [metadata[claim]]
                 if val not in metadata[claim]:
                     metadata[claim].append(val)
         else:

src/fedservice/entity/function/trust_chain_collector.py

@@ -181,6 +181,13 @@ def get_metadata(self, entity_id):
 
         return _ec['metadata']
 
+    def get_verified_self_signed_entity_configuration(self, entity_id: str) -> str:
+        signed_entity_config = self.get_entity_configuration(entity_id)
+        if signed_entity_config is None:
+            return ''
+
+        return verify_self_signed_signature(signed_entity_config)
+
     def get_federation_fetch_endpoint(self, intermediate: str) -> str:
         logger.debug(f'--get_federation_fetch_endpoint({intermediate})')
         # In cache ??

src/fedservice/entity/function/trust_mark_verifier.py

@@ -89,15 +89,15 @@ def verify_delegation(self, trust_mark, trust_anchor_id):
         _federation_entity = get_federation_entity(self)
         _collector = _federation_entity.function.trust_chain_collector
         # Deal with the delegation
-        ta_fe_metadata = _collector.get_metadata(trust_anchor_id)['federation_entity']
+        _entity_configuration = _collector.get_verified_self_signed_entity_configuration(trust_anchor_id)
 
-        if trust_mark['id'] not in ta_fe_metadata['trust_mark_issuers']:
+        if trust_mark['trust_mark_id'] not in _entity_configuration['trust_mark_issuers']:
             return None
-        if trust_mark['id'] not in ta_fe_metadata['trust_mark_owners']:
+        if trust_mark['trust_mark_id'] not in _entity_configuration['trust_mark_owners']:
             return None
 
         _delegation = factory(trust_mark['delegation'])
-        tm_owner_info = ta_fe_metadata['trust_mark_owners'][trust_mark['id']]
+        tm_owner_info = _entity_configuration['trust_mark_owners'][trust_mark['trust_mark_id']]
         _key_jar = KeyJar()
         _key_jar = import_jwks(_key_jar, tm_owner_info['jwks'], tm_owner_info['sub'])
         keys = _key_jar.get_jwt_verify_keys(_delegation.jwt)

src/fedservice/entity/server/entity_configuration.py

@@ -46,6 +46,15 @@ def process_request(self, request=None, **kwargs):
         else:
             args = {}
 
+        _trust_mark_issuers = _fed_entity.context.trust_mark_issuers
+        if _trust_mark_issuers:
+            args["trust_mark_issuers"] = _trust_mark_issuers
+
+        _trust_mark_owners = _fed_entity.context.trust_mark_owners
+        if _trust_mark_owners:
+            args["trust_mark_owners"] = _trust_mark_owners
+
+
         _ec = create_entity_statement(iss=_entity_id,
                                       sub=_entity_id,
                                       key_jar=_fed_entity.get_attribute('keyjar'),

src/fedservice/entity/server/list.py

@@ -42,7 +42,7 @@ def filter(self,
                 trust_marks = conf.get('trust_marks')
                 matched = False
                 for trust_mark in trust_marks:
-                    if trust_mark['id'] == trust_mark_id:
+                    if trust_mark['trust_mark_id'] == trust_mark_id:
                         matched = True
 
             if matched:

src/fedservice/entity/server/who.py

@@ -88,7 +88,7 @@ def process_request(self, request=None, trust_anchor: str = "" ,**kwargs):
                     for _mark in _ec["trust_marks"]:
                         _verified_trust_mark = _federation_entity.verify_trust_mark(
                             _mark, check_with_issuer=True)
-                        if _verified_trust_mark.get("id") == tm_id:
+                        if _verified_trust_mark.get("trust_mark_id") == tm_id:
                             server_to_use.append(eid)
         else:
             server_to_use = list(_srv.keys())

src/fedservice/message.py

@@ -2,6 +2,7 @@
 import logging
 
 from cryptojwt.exception import Expired
+from cryptojwt.jws.jws import factory
 from cryptojwt.jwt import utc_time_sans_frac
 from idpyoidc import message
 from idpyoidc.exception import MissingRequiredAttribute
@@ -93,29 +94,31 @@ def naming_constraints_deser(val, sformat="json"):
 
 SINGLE_OPTIONAL_NAMING_CONSTRAINTS = (Message, False, msg_ser, naming_constraints_deser, False)
 
+class InformationalMetadataExtensions(Message):
+    c_param = {
+        "organization_name": SINGLE_OPTIONAL_STRING,
+        "contacts": OPTIONAL_LIST_OF_STRINGS,
+        "logo_url": SINGLE_OPTIONAL_STRING,
+        "policy_url": SINGLE_OPTIONAL_STRING,
+        "homepage_uri": SINGLE_OPTIONAL_STRING,
+    }
 
-class FederationEntity(Message):
+class FederationEntity(InformationalMetadataExtensions):
     """Class representing Federation Entity metadata."""
-    c_param = {
-        "federation_fetch_endpoint": SINGLE_REQUIRED_STRING,
+    c_param = InformationalMetadataExtensions.c_param.copy()
+    c_param.update({
+        "federation_fetch_endpoint": SINGLE_OPTIONAL_STRING,
         "federation_list_endpoint": SINGLE_OPTIONAL_STRING,
-        # "federation_registration_endpoint": SINGLE_OPTIONAL_STRING,  part of OP metadata
         "federation_resolve_endpoint": SINGLE_OPTIONAL_STRING,
         "federation_trust_mark_status_endpoint": SINGLE_OPTIONAL_STRING,
-        "federation_trust_mark_list_endpoint":SINGLE_OPTIONAL_STRING,
+        "federation_trust_mark_list_endpoint": SINGLE_OPTIONAL_STRING,
         "federation_trust_mark_endpoint": SINGLE_OPTIONAL_STRING,
-        "federation_historical_keys_endpoint":SINGLE_OPTIONAL_STRING,
+        "federation_historical_keys_endpoint": SINGLE_OPTIONAL_STRING,
         "endpoint_auth_signing_alg_values_supported": SINGLE_OPTIONAL_JSON,
-        "name": SINGLE_OPTIONAL_STRING,
-        "contacts": OPTIONAL_LIST_OF_STRINGS,
-        "policy_url": SINGLE_OPTIONAL_STRING,
-        "homepage_uri": SINGLE_OPTIONAL_STRING,
-        # "federation_trust_marks": SINGLE_OPTIONAL_JSON,
-        "organization_name": SINGLE_OPTIONAL_STRING,
         # If it's a Trust Anchor
-        "trust_mark_owners": SINGLE_OPTIONAL_DICT,
-        "trust_mark_issuers": SINGLE_OPTIONAL_DICT,
-    }
+        # "trust_mark_owners": SINGLE_OPTIONAL_DICT,
+        # "trust_mark_issuers": SINGLE_OPTIONAL_DICT,
+    })
 
 
 def federation_entity_deser(val, sformat="json"):
@@ -255,6 +258,7 @@ class OPMetadata(ProviderConfigurationResponse):
         "signed_jwks_uri": SINGLE_OPTIONAL_STRING
     })
 
+
 class FedASConfigurationResponse(ASConfigurationResponse):
     c_param = ASConfigurationResponse.c_param.copy()
     c_param.update({
@@ -398,6 +402,52 @@ def constrains_deser(val, sformat="json"):
 SINGLE_OPTIONAL_CONSTRAINS = (Message, False, msg_ser, constrains_deser, False)
 
 
+class TrustMarks(Message):
+    c_param = {}
+
+    def verify(self, **kwargs):
+        for _id, spec in self.items():
+            _trust_mark = spec.get("trust_mark")
+            if _trust_mark:
+                _trust_mark_id = spec.get("trust_mark_id")
+                if _trust_mark_id:
+                    # Have to peek into the trust mark
+                    _jws = factory(_trust_mark)
+                    if not _jws:
+                        raise ValueError(f"Not a proper signed JWT: {_trust_mark}")
+                    _tm_id = _jws.jwt.payload().get("trust_mark_id")
+                    if _tm_id != _trust_mark_id:
+                        raise ValueError("The Trust Mark identifier MUST have the same value as the trust_mark_id "
+                                         "claim")
+                else:
+                    raise MissingRequiredAttribute("trust_mark_id")
+            else:
+                raise MissingRequiredAttribute("trust_mark")
+
+
+class TrustMarkIssuers(Message):
+    c_param = {}
+
+    def verify(self, **kwargs):
+        for owner_id, spec in self.items():
+            if not isinstance(spec, list):
+                raise ValueError("issuers MUST be a list")
+
+
+class TrustMarkOwners(Message):
+
+    def verify(self, **kwargs):
+        # Dictionary of Trust Mark Owner information
+        for owner_id, spec in self.items():
+            if "sub" in spec and "jwks" in spec:  # If there are other claims ignore them
+                continue
+            else:
+                if "sub" not in spec:
+                    raise MissingRequiredAttribute("sub")
+                elif "jwks" not in spec:
+                    raise MissingRequiredAttribute("jwks")
+
+
 class EntityStatement(JsonWebToken):
     """The Entity Statement"""
     c_param = JsonWebToken.c_param.copy()
@@ -412,10 +462,15 @@ class EntityStatement(JsonWebToken):
         'authority_hints': OPTIONAL_LIST_OF_STRINGS,
         'metadata': SINGLE_OPTIONAL_METADATA,
         'metadata_policy': SINGLE_OPTIONAL_METADATA_POLICY,
+        'metadata_policy_crit': OPTIONAL_LIST_OF_STRINGS,
         'constraints': SINGLE_OPTIONAL_CONSTRAINS,
         "crit": OPTIONAL_LIST_OF_STRINGS,
         "policy_language_crit": OPTIONAL_LIST_OF_STRINGS,
-        'trust_marks': OPTIONAL_LIST_OF_STRINGS,
+        "source_endpoint": SINGLE_OPTIONAL_STRING,
+        'trust_marks': SINGLE_OPTIONAL_JSON,
+        'trust_mark_owners': SINGLE_OPTIONAL_JSON,
+        'trust_mark_issuers': SINGLE_OPTIONAL_JSON,
+        #
         'trust_anchor_id': SINGLE_OPTIONAL_STRING
     })
 
@@ -444,14 +499,24 @@ def verify(self, **kwargs):
             if _crit:
                 _metadata_policy.verify(policy_language_crit=_crit, **kwargs)
 
+        _trust_mark_issuers = self.get("trust_mark_issuers")
+        if _trust_mark_issuers:
+            _tmi = TrustMarkIssuers(**_trust_mark_issuers)
+            _tmi.verify()
+
+        _trust_mark_owners = self.get("trust_mark_owners")
+        if _trust_mark_owners:
+            _tmi = TrustMarkOwners(**_trust_mark_owners)
+            _tmi.verify()
+
 
 class TrustMark(JsonWebToken):
     c_param = JsonWebToken.c_param.copy()
     c_param.update({
         "sub": SINGLE_REQUIRED_STRING,
         'iss': SINGLE_REQUIRED_STRING,
         'iat': SINGLE_REQUIRED_INT,
-        "id": SINGLE_REQUIRED_STRING,
+        "trust_mark_id": SINGLE_REQUIRED_STRING,
         "logo_uri": SINGLE_OPTIONAL_STRING,
         "exp": SINGLE_OPTIONAL_INT,
         "ref": SINGLE_OPTIONAL_STRING,
@@ -588,19 +653,22 @@ class HistoricalKeysResponse(Message):
         'jwks': SINGLE_REQUIRED_DICT
     }
 
+
 class TrustMarkRequest(Message):
     c_param = {
         "trust_mark_id": SINGLE_REQUIRED_STRING,
         "sub": SINGLE_REQUIRED_STRING
     }
 
+
 class WhoRequest(Message):
     c_param = {
         "entity_type": SINGLE_OPTIONAL_STRING,
         "credential_type": SINGLE_OPTIONAL_STRING,
         "trust_mark_id": SINGLE_OPTIONAL_STRING
     }
 
+
 class WhoResponse(Message):
     c_param = {
         "entities_to_use": REQUIRED_LIST_OF_STRINGS