Saml2 patch module

README.md

@@ -55,18 +55,9 @@ cd example/
 virtualenv -ppython3 env
 source env/bin/activate
 
-pip install --no-deps djangosaml2-spid
 pip install djangosaml2-spid
 ````
 
-⚠️ Why `pip install` have beed executed twice? spid-django needs a fork of PySAML2 that's not distribuited though pypi.
-This way to install it prevents the following error:
-
-````
-ERROR: Packages installed from PyPI cannot depend on packages which are not also hosted on PyPI.
-djangosaml2-spid depends on pysaml2@ git+https://github.com/peppelinux/pysaml2.git@pplnx-7.0.1#pysaml2
-````
-
 Your example saml2 configuration is in `spid_config/spid_settings.py`.
 See djangosaml2 and pysaml2 official docs for clarifications.
 

requirements.txt

@@ -1,7 +1,6 @@
 django>=2.2.24,<4.0
 
 # hint before: pip install -U setuptools
-pysaml2 @ git+https://github.com/peppelinux/pysaml2.git@pplnx-7.0.1#pysaml2
 cffi
 
 # django saml2 SP

src/djangosaml2_spid/__init__.py

@@ -0,0 +1,2 @@
+# For Django < 3.2
+default_app_config = 'djangosaml2_spid.apps.Djangosaml2SpidConfig'

src/djangosaml2_spid/_saml2.py

@@ -0,0 +1,224 @@
+#
+# Patch pysaml2 library in order to be used within spid-django.
+#
+DISABLE_WEAK_XMLSEC_ALGORITHMS = True  # https://github.com/IdentityPython/pysaml2/pull/628
+ADD_XSD_DATE_TYPE = True  # https://github.com/IdentityPython/pysaml2/pull/602
+PATCH_RESPONSE_VERIFY = True  # https://github.com/peppelinux/pysaml2/commit/8bdbbdf41ce63a37d3ba02c8f48a3dba0217d463
+
+
+def pysaml2_patch():
+    import base64
+    import datetime
+    import logging
+
+    import saml2.metadata
+
+    from saml2 import SamlBase
+    from saml2.algsupport import get_algorithm_support, DigestMethod, \
+        DIGEST_METHODS, SigningMethod, SIGNING_METHODS
+    from saml2.response import StatusResponse, RequestVersionTooLow, RequestVersionTooHigh
+    from saml2.saml import AttributeValueBase
+
+    if DISABLE_WEAK_XMLSEC_ALGORITHMS:
+        from django.conf import settings
+
+        # The additional parameter 'xmlsec_disabled_algs' is replaced with a setting
+        # that is checked in a patched saml2.algsupport.algorithm_support_in_metadata.
+        settings.SAML_XMLSEC_DISABLED_ALGS = getattr(settings, "SAML_XMLSEC_DISABLED_ALGS", [])
+
+        def algorithm_support_in_metadata(xmlsec):
+            if xmlsec is None:
+                return []
+
+            support = get_algorithm_support(xmlsec)
+            element_list = []
+            for alg in support["digest"]:
+                if alg in settings.SAML_XMLSEC_DISABLED_ALGS:
+                    continue
+                element_list.append(DigestMethod(algorithm=DIGEST_METHODS[alg]))
+            for alg in support["signing"]:
+                if alg in settings.SAML_XMLSEC_DISABLED_ALGS:
+                    continue
+                element_list.append(SigningMethod(algorithm=SIGNING_METHODS[alg]))
+            return element_list
+
+        saml2.metadata.algorithm_support_in_metadata = algorithm_support_in_metadata
+
+    if ADD_XSD_DATE_TYPE:
+        def set_text(self, value, base64encode=False):
+            def _wrong_type_value(xsd, value):
+                msg = 'Type and value do not match: {xsd}:{type}:{value}'
+                msg = msg.format(xsd=xsd, type=type(value), value=value)
+                raise ValueError(msg)
+
+            if isinstance(value, bytes):
+                value = value.decode('utf-8')
+            type_to_xsd = {
+                str: 'string',
+                int: 'integer',
+                float: 'float',
+                bool: 'boolean',
+                type(None): '',
+            }
+            # entries of xsd-types each declaring:
+            # - a corresponding python type
+            # - a function to turn a string into that type
+            # - a function to turn that type into a text-value
+            xsd_types_props = {
+                'string': {
+                    'type': str,
+                    'to_type': str,
+                    'to_text': str,
+                },
+                'date': {
+                    'type': datetime.date,
+                    'to_type': lambda x: datetime.datetime.strptime(x, '%Y-%m-%d').date(),
+                    'to_text': str,
+                },
+                'integer': {
+                    'type': int,
+                    'to_type': int,
+                    'to_text': str,
+                },
+                'short': {
+                    'type': int,
+                    'to_type': int,
+                    'to_text': str,
+                },
+                'int': {
+                    'type': int,
+                    'to_type': int,
+                    'to_text': str,
+                },
+                'long': {
+                    'type': int,
+                    'to_type': int,
+                    'to_text': str,
+                },
+                'float': {
+                    'type': float,
+                    'to_type': float,
+                    'to_text': str,
+                },
+                'double': {
+                    'type': float,
+                    'to_type': float,
+                    'to_text': str,
+                },
+                'boolean': {
+                    'type': bool,
+                    'to_type': lambda x: {
+                        'true': True,
+                        'false': False,
+                    }[str(x).lower()],
+                    'to_text': lambda x: str(x).lower(),
+                },
+                'base64Binary': {
+                    'type': str,
+                    'to_type': str,
+                    'to_text': (
+                        lambda x: base64.encodebytes(x.encode()) if base64encode else x
+                    ),
+                },
+                'anyType': {
+                    'type': type(value),
+                    'to_type': lambda x: x,
+                    'to_text': lambda x: x,
+                },
+                '': {
+                    'type': type(None),
+                    'to_type': lambda x: None,
+                    'to_text': lambda x: '',
+                },
+            }
+            xsd_string = (
+                'base64Binary' if base64encode
+                else self.get_type()
+                or type_to_xsd.get(type(value)))
+            xsd_ns, xsd_type = (
+                ['', type(None)] if xsd_string is None
+                else ['', ''] if xsd_string == ''
+                else [
+                    'xs' if xsd_string in xsd_types_props else '',
+                    xsd_string
+                ] if ':' not in xsd_string
+                else xsd_string.split(':', 1))
+            xsd_type_props = xsd_types_props.get(xsd_type, {})
+            valid_type = xsd_type_props.get('type', type(None))
+            to_type = xsd_type_props.get('to_type', str)
+            to_text = xsd_type_props.get('to_text', str)
+            # cast to correct type before type-checking
+            if type(value) is str and valid_type is not str:
+                try:
+                    value = to_type(value)
+                except (TypeError, ValueError, KeyError):
+                    # the cast failed
+                    _wrong_type_value(xsd=xsd_type, value=value)
+            if type(value) is not valid_type:
+                _wrong_type_value(xsd=xsd_type, value=value)
+            text = to_text(value)
+            self.set_type(
+                '{ns}:{type}'.format(ns=xsd_ns, type=xsd_type) if xsd_ns
+                else xsd_type if xsd_type
+                else '')
+            SamlBase.__setattr__(self, 'text', text)
+            return self
+
+        AttributeValueBase.set_text = set_text
+
+    if PATCH_RESPONSE_VERIFY:
+        logger = logging.getLogger(__name__)
+
+        def _verify(self):
+            if self.request_id and self.in_response_to and \
+                    self.in_response_to != self.request_id:
+                logger.error("Not the id I expected: %s != %s",
+                             self.in_response_to, self.request_id)
+                return None
+            if self.response.version != "2.0":
+                _ver = float(self.response.version)
+                if _ver < 2.0:
+                    raise RequestVersionTooLow()
+                else:
+                    raise RequestVersionTooHigh()
+
+            if self.asynchop:
+                if not (
+                     getattr(self.response, 'destination')
+                ):
+                    logger.error(
+                        f"Invalid response destination in asynchop"
+                    )
+                    return None
+                elif self.response.destination not in self.return_addrs:
+                    logger.error(
+                        f"{self.response.destination} not in {self.return_addrs}"
+                    )
+                    return None
+
+            valid = self.issue_instant_ok() and self.status_ok()
+            return valid
+
+        StatusResponse._verify = _verify
+
+
+def register_oasis_default_nsmap():
+    """Register OASIS default prefix-namespace associations."""
+    from xml.etree import ElementTree
+
+    oasis_default_nsmap = {
+        'saml': 'urn:oasis:names:tc:SAML:2.0:assertion',
+        'samlp': 'urn:oasis:names:tc:SAML:2.0:protocol',
+        'ds': 'http://www.w3.org/2000/09/xmldsig#',
+        'xsi': 'http://www.w3.org/2001/XMLSchema-instance',
+        'xs': 'http://www.w3.org/2001/XMLSchema',
+        'mdui': 'urn:oasis:names:tc:SAML:metadata:ui',
+        'md': 'urn:oasis:names:tc:SAML:2.0:metadata',
+        'xenc': 'http://www.w3.org/2001/04/xmlenc#',
+        'alg': 'urn:oasis:names:tc:SAML:metadata:algsupport',
+        'mdattr': 'urn:oasis:names:tc:SAML:metadata:attribute',
+        'idpdisc': 'urn:oasis:names:tc:SAML:profiles:SSO:idp-discovery-protocol',
+    }
+
+    for prefix, uri in oasis_default_nsmap.items():
+        ElementTree.register_namespace(prefix, uri)

src/djangosaml2_spid/apps.py

@@ -3,3 +3,11 @@
 
 class Djangosaml2SpidConfig(AppConfig):
     name = "djangosaml2_spid"
+
+    def ready(self):
+        try:
+            from ._saml2 import pysaml2_patch, register_oasis_default_nsmap
+            pysaml2_patch()
+            register_oasis_default_nsmap()
+        except ImportError:
+            pass

src/djangosaml2_spid/conf.py

@@ -217,7 +217,6 @@
     },
 )
 
-
 # Attributes that this project need to identify a user
 settings.SPID_REQUIRED_ATTRIBUTES = getattr(
     settings,

src/djangosaml2_spid/spid_metadata.py

@@ -1,10 +1,13 @@
+from xml.etree import ElementTree
+
 import saml2
+import saml2.md
 from django.conf import settings
 from saml2.metadata import entity_descriptor, sign_entity_descriptor
 from saml2.sigver import security_context
 
 
-def italian_sp_metadata(conf, md_type:str="spid"):
+def italian_sp_metadata(conf, md_type: str = "spid"):
     metadata = entity_descriptor(conf)
 
     # this will renumber acs starting from 0 and set index=0 as is_default
@@ -60,7 +63,8 @@ def spid_contacts_29_v3(metadata):
     https://www.agid.gov.it/sites/default/files/repository_files/spid-avviso-n29v3-specifiche_sp_pubblici_e_privati_0.pdf
     """
 
-    saml2.md.SamlBase.register_prefix(settings.SPID_PREFIXES)
+    for prefix, uri in settings.SPID_PREFIXES.items():
+        ElementTree.register_namespace(prefix, uri)
 
     contact_map = settings.SPID_CONTACTS
     metadata.contact_person = []
@@ -84,7 +88,8 @@ def spid_contacts_29_v3(metadata):
                 ext = saml2.ExtensionElement(
                     k, namespace=settings.SPID_PREFIXES["spid"], text=v
                 )
-                # Avviso SPID n. 19 v.4 per enti AGGREGATORI il tag ContactPerson deve avere l’attributo spid:entityType valorizzato come spid:aggregator
+                # Avviso SPID n. 19 v.4 per enti AGGREGATORI il tag ContactPerson deve
+                # avere l’attributo spid:entityType valorizzato come spid:aggregator
                 if k == "PublicServicesFullOperator":
                     spid_contact.extension_attributes = {
                         "spid:entityType": "spid:aggregator"
@@ -155,7 +160,8 @@ def cie_contacts(metadata):
     """
     """
 
-    saml2.md.SamlBase.register_prefix(settings.CIE_PREFIXES)
+    for prefix, uri in settings.CIE_PREFIXES.items():
+        ElementTree.register_namespace(prefix, uri)
 
     contact_map = settings.CIE_CONTACTS
     metadata.contact_person = []
@@ -193,6 +199,5 @@ def cie_contacts(metadata):
                 )
                 elements[k] = ext
 
-
         cie_contact.extensions = cie_extensions
         metadata.contact_person.append(cie_contact)

src/djangosaml2_spid/tests.py

@@ -460,3 +460,40 @@ def test_echo_attributes(self):
             b"No active SAML identity found. Are you "
             b"sure you have logged in via SAML?",
         )
+
+
+class TestSaml2Patches(unittest.TestCase):
+
+    def test_default_namespaces(self):
+        oasis_default_nsmap = {
+            'saml': 'urn:oasis:names:tc:SAML:2.0:assertion',
+            'samlp': 'urn:oasis:names:tc:SAML:2.0:protocol',
+            'ds': 'http://www.w3.org/2000/09/xmldsig#',
+            'xsi': 'http://www.w3.org/2001/XMLSchema-instance',
+            'xs': 'http://www.w3.org/2001/XMLSchema',
+            'mdui': 'urn:oasis:names:tc:SAML:metadata:ui',
+            'md': 'urn:oasis:names:tc:SAML:2.0:metadata',
+            'xenc': 'http://www.w3.org/2001/04/xmlenc#',
+            'alg': 'urn:oasis:names:tc:SAML:metadata:algsupport',
+            'mdattr': 'urn:oasis:names:tc:SAML:metadata:attribute',
+            'idpdisc': 'urn:oasis:names:tc:SAML:profiles:SSO:idp-discovery-protocol',
+        }
+
+        for prefix, uri in oasis_default_nsmap.items():
+            self.assertIn(uri, ElementTree._namespace_map)
+            self.assertEqual(prefix, ElementTree._namespace_map[uri])
+
+    def test_disable_weak_xmlsec_algorithms(self):
+        import saml2.metadata
+        from saml2.algsupport import algorithm_support_in_metadata
+
+        self.assertIsNot(saml2.metadata.algorithm_support_in_metadata, algorithm_support_in_metadata)
+        self.assertEqual(saml2.metadata.algorithm_support_in_metadata.__module__, 'djangosaml2_spid._saml2')
+
+    def test_add_xsd_date_type(self):
+        from saml2.saml import AttributeValueBase
+        self.assertEqual(AttributeValueBase.set_text.__module__, 'djangosaml2_spid._saml2')
+
+    def test_patch_response_verify(self):
+        from saml2.response import StatusResponse
+        self.assertEqual(StatusResponse._verify.__module__, 'djangosaml2_spid._saml2')