[WIP] Add eIDAS support

setup.cfg

@@ -51,6 +51,7 @@ install_requires =
     pytz
     requests >= 1.0.0
     six
+    iso3166
 
 
 [options.packages.find]

src/saml2/config.py

@@ -7,6 +7,10 @@
 import os
 import re
 import sys
+from functools import partial
+import re
+from urllib import parse
+from iso3166 import countries
 
 import six
 
@@ -21,6 +25,8 @@
 from saml2.mdstore import MetadataStore
 from saml2.saml import NAME_FORMAT_URI
 from saml2.virtual_org import VirtualOrg
+from saml2.utility import not_empty
+from saml2.utility.config import ConfigValidationError
 
 logger = logging.getLogger(__name__)
 
@@ -97,6 +103,9 @@
     "sp_type",
     "sp_type_in_metadata",
     "requested_attributes",
+    "node_country",
+    "application_identifier",
+    "protocol_version"
 ]
 
 AA_IDP_ARGS = [
@@ -118,6 +127,10 @@
     "domain",
     "name_qualifier",
     "edu_person_targeted_id",
+    "node_country",
+    "application_identifier",
+    "protocol_version",
+    "supported_loa"
 ]
 
 PDP_ARGS = ["endpoints", "name_form", "name_id_format"]
@@ -540,6 +553,9 @@ def service_per_endpoint(self, context=None):
                 res[endp] = (service, binding)
         return res
 
+    def validate(self):
+        pass
+
 
 class SPConfig(Config):
     def_context = "sp"
@@ -569,13 +585,198 @@ def ecp_endpoint(self, ipaddress):
         return None
 
 
+class eIDASConfig(Config):
+    def get_endpoint_element(self, element):
+        pass
+
+    def get_protocol_version(self):
+        pass
+
+    def get_application_identifier(self):
+        pass
+
+    def get_node_country(self):
+        pass
+
+    @staticmethod
+    def validate_node_country_format(node_country):
+        try:
+            return countries.get(node_country).alpha2 == node_country
+        except KeyError:
+            return False
+
+    @staticmethod
+    def validate_application_identifier_format(application_identifier):
+        pattern_match = re.search(r"([a-zA-Z0-9])+:([a-zA-Z0-9():_\-])+:([0-9])+"
+                                  r"(\.([0-9])+){1,2}", application_identifier)
+        if not application_identifier or pattern_match:
+            return True
+        return False
+
+    @staticmethod
+    def get_type_contact_person(contacts, ctype):
+        return [contact for contact in contacts
+                if contact.get("contact_type") == ctype]
+
+    @staticmethod
+    def contact_has_email_address(contact):
+        return not_empty(contact.get("email_address"))
+
+    @property
+    def warning_validators(self):
+        return {
+            "single_logout_service SHOULD NOT be declared":
+                self.get_endpoint_element("single_logout_service") is None,
+            "artifact_resolution_service SHOULD NOT be declared":
+                self.get_endpoint_element("artifact_resolution_service") is None,
+            "manage_name_id_service SHOULD NOT be declared":
+                self.get_endpoint_element("manage_name_id_service") is None,
+            "application_identifier SHOULD be declared":
+                not_empty(self.get_application_identifier()),
+            "protocol_version SHOULD be declared":
+                not_empty(self.get_protocol_version()),
+            "minimal organization info (name/dname/url) SHOULD be declared":
+                not_empty(self.organization),
+            "contact_person with contact_type 'technical' and at least one "
+            "email_address SHOULD be declared":
+                any(filter(self.contact_has_email_address,
+                           self.get_type_contact_person(self.contact_person,
+                                                        ctype="technical"))),
+            "contact_person with contact_type 'support' and at least one "
+            "email_address SHOULD be declared":
+                any(filter(self.contact_has_email_address,
+                           self.get_type_contact_person(self.contact_person,
+                                                        ctype="support")))
+        }
+
+    @property
+    def error_validators(self):
+        return {
+            "KeyDescriptor MUST be declared":
+                not_empty(self.cert_file or self.encryption_keypairs),
+            "node_country MUST be declared in ISO 3166-1 alpha-2 format":
+                self.validate_node_country_format(self.get_node_country()),
+            "application_identifier MUST be in the form <vendor name>:<software "
+            "identifier>:<major-version>.<minor-version>[.<patch-version>]":
+                self.validate_application_identifier_format(
+                    self.get_application_identifier()),
+            "entityid MUST be an HTTPS URL pointing to the location of its published "
+            "metadata":
+                parse.urlparse(self.entityid).scheme == "https"
+        }
+
+    def validate(self):
+        if not all(self.warning_validators.values()):
+            logger.warning(
+                "Configuration validation warnings occurred: {}".format(
+                    [msg for msg, check in self.warning_validators.items()
+                     if check is not True]
+                )
+            )
+
+        if not all(self.error_validators.values()):
+            error = "Configuration validation errors occurred {}:".format(
+                [msg for msg, check in self.error_validators.items()
+                 if check is not True])
+            logger.error(error)
+            raise ConfigValidationError(error)
+
+
+class eIDASSPConfig(SPConfig, eIDASConfig):
+    def get_endpoint_element(self, element):
+        return getattr(self, "_sp_endpoints", {}).get(element, None)
+
+    def get_application_identifier(self):
+        return getattr(self, "_sp_application_identifier", None)
+
+    def get_protocol_version(self):
+        return getattr(self, "_sp_protocol_version", None)
+
+    def get_node_country(self):
+        return getattr(self, "_sp_node_country", None)
+
+    @property
+    def warning_validators(self):
+        sp_warning_validators = {
+            "hide_assertion_consumer_service SHOULD be set to True":
+                getattr(self, "_sp_hide_assertion_consumer_service", None) is True
+        }
+        return {**super().warning_validators, **sp_warning_validators}
+
+    @property
+    def error_validators(self):
+        sp_error_validators = {
+            "authn_requests_signed MUST be set to True":
+                getattr(self, "_sp_authn_requests_signed", None) is True,
+            "sp_type MUST be set to 'public' or 'private'":
+                getattr(self, "_sp_sp_type", None) in ("public", "private")
+        }
+        return {**super().error_validators, **sp_error_validators}
+
+
 class IdPConfig(Config):
     def_context = "idp"
 
     def __init__(self):
         Config.__init__(self)
 
 
+class eIDASIdPConfig(IdPConfig, eIDASConfig):
+    def get_endpoint_element(self, element):
+        return getattr(self, "_idp_endpoints", {}).get(element, None)
+
+    def get_application_identifier(self):
+        return getattr(self, "_idp_application_identifier", None)
+
+    def get_protocol_version(self):
+        return getattr(self, "_idp_protocol_version", None)
+
+    def get_node_country(self):
+        return getattr(self, "_idp_node_country", None)
+
+    def verify_non_notified_loa(self):
+        return not any(
+            [x.startswith("http://eidas.europa.eu/LoA/")
+             for x in getattr(self, "_idp_supported_loa", {}).get("non_notified")]
+        )
+
+    def verify_notified_loa(self):
+        return all(
+            [x in ["http://eidas.europa.eu/LoA/low",
+                   "http://eidas.europa.eu/LoA/substantial",
+                   "http://eidas.europa.eu/LoA/high"]
+             for x in getattr(self, "_idp_supported_loa", {}).get("notified")]
+        )
+
+    @property
+    def warning_validators(self):
+        idp_warning_validators = {}
+        return {**super().warning_validators, **idp_warning_validators}
+
+    @property
+    def error_validators(self):
+        idp_error_validators = {
+            "want_authn_requests_signed MUST be set to True":
+            getattr(self, "_idp_want_authn_requests_signed", None) is True,
+            "provided_attributes MUST be set to denote the supported attributes by "
+            "the IdP":
+            not_empty(getattr(self, "_idp_provided_attributes", None)),
+            "supported_loa for non-notified eIDs MUST NOT use an "
+            "http://eidas.europa.eu/LoA/ prefix":
+            self.verify_non_notified_loa(),
+            "supported_loa for notified eID MUST be (at least) one of "
+            "[http://eidas.europa.eu/LoA/low, "
+            "http://eidas.europa.eu/LoA/substantial, "
+            "http://eidas.europa.eu/LoA/high]":
+            self.verify_notified_loa(),
+            "sign_response MUST be set to True":
+            getattr(self, "_idp_sign_response", None) is True,
+            "encrypt_assertion MUST be set to True":
+            getattr(self, "_idp_encrypt_assertion", None) is True,
+        }
+        return {**super().error_validators, **idp_error_validators}
+
+
 def config_factory(_type, config):
     """
 

src/saml2/extension/node_country.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python
+
+#
+# Generated Thu Dec 12 18:16:51 2019 by parse_xsd.py version 0.5.
+#
+
+import saml2
+from saml2 import SamlBase
+
+
+NAMESPACE = 'http://eidas.europa.eu/saml-extensions'
+class NodeCountryType_(SamlBase):
+    """The http://eidas.europa.eu/saml-extensions:NodeCountryType element """
+
+    c_tag = 'NodeCountryType'
+    c_namespace = NAMESPACE
+    c_children = SamlBase.c_children.copy()
+    c_attributes = SamlBase.c_attributes.copy()
+    c_child_order = SamlBase.c_child_order[:]
+    c_cardinality = SamlBase.c_cardinality.copy()
+
+def node_country_type__from_string(xml_string):
+    return saml2.create_class_from_xml_string(NodeCountryType_, xml_string)
+
+
+class NodeCountry(NodeCountryType_):
+    """The http://eidas.europa.eu/saml-extensions:NodeCountry element """
+
+    c_tag = 'NodeCountry'
+    c_namespace = NAMESPACE
+    c_children = NodeCountryType_.c_children.copy()
+    c_attributes = NodeCountryType_.c_attributes.copy()
+    c_child_order = NodeCountryType_.c_child_order[:]
+    c_cardinality = NodeCountryType_.c_cardinality.copy()
+
+def node_country_from_string(xml_string):
+    return saml2.create_class_from_xml_string(NodeCountry, xml_string)
+
+
+ELEMENT_FROM_STRING = {
+    NodeCountry.c_tag: node_country_from_string,
+    NodeCountryType_.c_tag: node_country_type__from_string,
+}
+
+ELEMENT_BY_TAG = {
+    'NodeCountry': NodeCountry,
+    'NodeCountryType': NodeCountryType_,
+}
+def factory(tag, **kwargs):
+    return ELEMENT_BY_TAG[tag](**kwargs)

src/saml2/metadata.py

@@ -10,6 +10,7 @@
 from saml2.extension import shibmd
 from saml2.extension import mdattr
 from saml2.extension import sp_type
+from saml2.extension import node_country
 from saml2.saml import NAME_FORMAT_URI
 from saml2.saml import AttributeValue
 from saml2.saml import Attribute
@@ -20,6 +21,7 @@
 from saml2 import BINDING_SOAP
 from saml2 import samlp
 from saml2 import class_name
+from saml2.utility import make_list
 
 from saml2 import xmldsig as ds
 import six
@@ -592,6 +594,14 @@ def do_idpsso_descriptor(conf, cert=None, enc_cert=None):
         except KeyError:
             setattr(idpsso, key, DEFAULTS[key])
 
+    attributes = [
+        Attribute(name=attribute.get("name", None),
+                  name_format=attribute.get("name_format", None),
+                  friendly_name=attribute.get("friendly_name", None))
+        for attribute in conf.getattr("provided_attributes", "idp")
+    ]
+    idpsso.attribute = attributes
+
     return idpsso
 
 
@@ -770,6 +780,45 @@ def entity_descriptor(confd):
         entd.authn_authority_descriptor = do_aq_descriptor(confd, mycert,
                                                            enc_cert)
 
+    conf_node_country = confd.getattr('node_country', confd.context)
+    if conf_node_country:
+        if not entd.extensions:
+            entd.extensions = md.Extensions()
+        item = node_country.NodeCountry(text=conf_node_country)
+        entd.extensions.add_extension_element(item)
+
+    app_identifer = confd.getattr("application_identifier", confd.context)
+    if app_identifer:
+        entd.extensions = entd.extensions or md.Extensions()
+        ava = AttributeValue(text=app_identifer)
+        attr = Attribute(
+            attribute_value=ava,
+            name="http://eidas.europa.eu/entity-attributes/application-identifier"
+        )
+        _add_attr_to_entity_attributes(entd.extensions, attr)
+
+    protocol_version = confd.getattr("protocol_version", confd.context)
+    if protocol_version:
+        entd.extensions = entd.extensions or md.Extensions()
+        ava = [AttributeValue(text=str(c)) for c in make_list(protocol_version)]
+        attr = Attribute(
+            attribute_value=ava,
+            name="http://eidas.europa.eu/entity-attributes/protocol-version"
+        )
+        _add_attr_to_entity_attributes(entd.extensions, attr)
+
+    loa = confd.getattr("supported_loa", confd.context)
+    if loa:
+        entd.extensions = entd.extensions or md.Extensions()
+        ava = [AttributeValue(text=str(c)) for c in make_list(
+            loa.get("notified", []), loa.get("non_notified", []))]
+        attr = Attribute(
+            attribute_value=ava,
+            name="urn:oasis:names:tc:SAML:attribute:assurance-certification",
+            name_format="urn:oasis:names:tc:saml2:2.0:attrname-format:uri"
+        )
+        _add_attr_to_entity_attributes(entd.extensions, attr)
+
     return entd