Reformat and rearrange code

Signed-off-by: Ivan Kanakarakis <[email protected]>

Author: c00kiemon5ter

docs/howto/config.rst

@@ -227,8 +227,8 @@ a period for which the metadata fetched from the the MDQ server are considered
 fresh. After that period has passed the metadata are not valid anymore and must
 be fetched again. The period must be in the format defined in `iso 8601 <https://www.iso.org/iso-8601-date-and-time-format.html>`_.
 
-By default, if `freshness_period` is not defined, the metadata are fetched
-only the first time they are requested and never refreshed.
+By default, if `freshness_period` is not defined, the metadata are refreshed
+every 12 hours (`P0Y0M0DT12H0M0S`).
 
 organization
 ^^^^^^^^^^^^

src/saml2/mdstore.py

@@ -32,7 +32,11 @@
 from saml2.s_utils import UnknownSystemEntity
 from saml2.sigver import split_len
 from saml2.validate import valid_instance
-from saml2.time_util import valid, instant, add_duration, before, str_to_time
+from saml2.time_util import valid
+from saml2.time_util import instant
+from saml2.time_util import add_duration
+from saml2.time_util import before
+from saml2.time_util import str_to_time
 from saml2.validate import NotValid
 from saml2.sigver import security_context
 from saml2.extension.mdattr import NAMESPACE as NS_MDATTR
@@ -72,6 +76,11 @@
 ENTITY_CATEGORY_SUPPORT = "http://macedir.org/entity-category-support"
 ASSURANCE_CERTIFICATION = "urn:oasis:names:tc:SAML:attribute:assurance-certification"
 
+SAML_METADATA_CONTENT_TYPE = "application/samlmetadata+xml"
+DEFAULT_FRESHNESS_PERIOD = "P0Y0M0DT12H0M0S"
+
+
+
 REQ2SRV = {
     # IDP
     "authn_request": "single_sign_on_service",
@@ -664,22 +673,21 @@ def signed(self):
     def parse_and_check_signature(self, txt):
         self.parse(txt)
 
-        if self.cert:
-            if not self.signed():
-                return True
-
-            node_name = self.node_name \
-                        or "%s:%s" % (md.EntitiesDescriptor.c_namespace,
-                                      md.EntitiesDescriptor.c_tag)
+        if not self.cert:
+            return True
 
-            if self.security.verify_signature(
-                    txt, node_name=node_name, cert_file=self.cert):
-                return True
-            else:
-                return False
-        else:
+        if not self.signed():
             return True
 
+        fallback_name = "{ns}:{tag}".format(
+            ns=md.EntitiesDescriptor.c_namespace, tag=md.EntitiesDescriptor.c_tag
+        )
+        node_name = self.node_name or fallback_name
+
+        return self.security.verify_signature(
+            txt, node_name=node_name, cert_file=self.cert
+        )
+
 
 class MetaDataFile(InMemoryMetaData):
     """
@@ -805,9 +813,6 @@ def load(self, *args, **kwargs):
             self.entity[key] = item
 
 
-SAML_METADATA_CONTENT_TYPE = 'application/samlmetadata+xml'
-
-
 class MetaDataMDX(InMemoryMetaData):
     """
     Uses the MDQ protocol to fetch entity information.
@@ -817,8 +822,9 @@ class MetaDataMDX(InMemoryMetaData):
 
     @staticmethod
     def sha1_entity_transform(entity_id):
-        return "{{sha1}}{}".format(
-            hashlib.sha1(entity_id.encode("utf-8")).hexdigest())
+        entity_id_sha1 = hashlib.sha1(entity_id.encode("utf-8")).hexdigest()
+        transform = "{{sha1}}{digest}".format(digest=entity_id_sha1)
+        return transform
 
     def __init__(self, url=None, security=None, cert=None,
                  entity_transform=None, freshness_period=None, **kwargs):
@@ -847,53 +853,61 @@ def __init__(self, url=None, security=None, cert=None,
 
         self.cert = cert
         self.security = security
-        self.freshness_period = freshness_period
-        if freshness_period:
-            self.expiration_date = {}
+        self.freshness_period = freshness_period or DEFAULT_FRESHNESS_PERIOD
+        self.expiration_date = {}
 
         # We assume that the MDQ server will return a single entity
         # described by a single <EntityDescriptor> element. The protocol
         # does allow multiple entities to be returned in an
         # <EntitiesDescriptor> element but we will not currently support
         # that use case since it is unlikely to be leveraged for most
         # flows.
-        self.node_name = "%s:%s" % (md.EntityDescriptor.c_namespace,
-                                      md.EntityDescriptor.c_tag)
+        self.node_name = "{ns}:{tag}".format(
+            ns=md.EntityDescriptor.c_namespace, tag=md.EntityDescriptor.c_tag
+        )
 
     def load(self, *args, **kwargs):
         # Do nothing
         pass
 
-    def fetch_metadata(self, item):
-        mdx_url = "%s/entities/%s" % (self.url, self.entity_transform(item))
-        response = requests.get(mdx_url, headers={
-            'Accept': SAML_METADATA_CONTENT_TYPE})
-        if response.status_code == 200:
-            _txt = response.content
-            if self.parse_and_check_signature(_txt):
-                if self.freshness_period:
-                    curr_time = str_to_time(instant())
-                    self.expiration_date[item] = add_duration(
-                        curr_time, self.freshness_period)
-                return self.entity[item]
-        else:
-            logger.info("Response status: %s", response.status_code)
-        raise KeyError
+    def _fetch_metadata(self, item):
+        mdx_url = "{url}/entities/{id}".format(
+            url=self.url, id=self.entity_transform(item)
+        )
 
-    def _is_fresh(self, item):
-        return self.freshness_period and before(self.expiration_date[item])
+        response = requests.get(mdx_url, headers={"Accept": SAML_METADATA_CONTENT_TYPE})
+        if response.status_code != 200:
+            error_msg = "Fething {item}: Got response status {status}".format(
+                item=item, status=response.status_code
+            )
+            logger.info(error_msg)
+            raise KeyError(error_msg)
+
+        _txt = response.content
+        if not self.parse_and_check_signature(_txt):
+            error_msg = "Fething {item}: invalid signature".format(
+                item=item, status=response.status_code
+            )
+            logger.info(error_msg)
+            raise KeyError(error_msg)
+
+        curr_time = str_to_time(instant())
+        self.expiration_date[item] = add_duration(curr_time, self.freshness_period)
+        return self.entity[item]
+
+    def _is_metadata_fresh(self, item):
+        return before(self.expiration_date[item])
 
     def __getitem__(self, item):
-        if item in self.entity:
-            if self._is_fresh(item):
-                entity = self.entity[item]
-            else:
-                logger.info("Metadata for {} have expired, refreshing "
-                            "metadata".format(item))
-                self.entity.pop(item)
-                entity = self.fetch_metadata(item)
+        if item not in self.entity:
+            entity = self._fetch_metadata(item)
+        elif not self._is_metadata_fresh(item):
+            msg = "Metadata for {} have expired; refreshing metadata".format(item)
+            logger.info(msg)
+            old_entity = self.entity.pop(item)
+            entity = self._fetch_metadata(item)
         else:
-            entity = self.fetch_metadata(item)
+            entity = self.entity[item]
         return entity
 
     def single_sign_on_service(self, entity_id, binding=None, typ="idpsso"):

tests/test_30_mdstore.py

@@ -4,8 +4,11 @@
 import os
 import re
 from collections import OrderedDict
+from unittest.mock import Mock
 from unittest.mock import patch
 
+import responses
+
 from six.moves.urllib import parse
 
 from saml2.config import Config
@@ -26,9 +29,6 @@
 from saml2.s_utils import UnknownPrincipal
 from pathutils import full_path
 
-import responses
-import mock
-
 
 TESTS_DIR = os.path.dirname(__file__)
 
@@ -336,24 +336,55 @@ def test_mdx_single_sign_on_service():
 
 
 @responses.activate
-@mock.patch('saml2.mdstore.before')
-def test_mdx_metadata_freshness_period(mock_datetime):
-    """Ensure that metadata is refreshed only when they have expired."""
-    entity_id = \
-        "http://xenosmilus.umdc.umu.se/simplesaml/saml2/idp/metadata.php"
+def test_mdx_metadata_freshness_period_not_expired():
+    """Ensure that metadata is not refreshed if not expired."""
 
+    entity_id = "http://xenosmilus.umdc.umu.se/simplesaml/saml2/idp/metadata.php"
     url = "http://mdx.example.com/entities/{}".format(
-        parse.quote_plus(MetaDataMDX.sha1_entity_transform(entity_id)))
-    responses.add(responses.GET, url, body=TEST_METADATA_STRING, status=200,
-                  content_type=SAML_METADATA_CONTENT_TYPE)
+        parse.quote_plus(MetaDataMDX.sha1_entity_transform(entity_id))
+    )
+
+    responses.add(
+        responses.GET,
+        url,
+        body=TEST_METADATA_STRING,
+        status=200,
+        content_type=SAML_METADATA_CONTENT_TYPE,
+    )
+
+    mdx = MetaDataMDX("http://mdx.example.com", freshness_period="P0Y0M0DT0H2M0S")
+    mdx._is_metadata_fresh = Mock(return_value=True)
 
-    mock_datetime.return_value = True
-    mdx = MetaDataMDX("http://mdx.example.com",
-                      freshness_period="P0Y0M0DT0H2M0S")
     mdx.single_sign_on_service(entity_id, BINDING_HTTP_REDIRECT)
+    assert entity_id in mdx.entity
+
     mdx.single_sign_on_service(entity_id, BINDING_HTTP_REDIRECT)
     assert len(responses.calls) == 1
-    mock_datetime.return_value = False
+
+
+@responses.activate
+def test_mdx_metadata_freshness_period_expired():
+    """Ensure that metadata is not refreshed if not expired."""
+
+    entity_id = "http://xenosmilus.umdc.umu.se/simplesaml/saml2/idp/metadata.php"
+    url = "http://mdx.example.com/entities/{}".format(
+        parse.quote_plus(MetaDataMDX.sha1_entity_transform(entity_id))
+    )
+
+    responses.add(
+        responses.GET,
+        url,
+        body=TEST_METADATA_STRING,
+        status=200,
+        content_type=SAML_METADATA_CONTENT_TYPE,
+    )
+
+    mdx = MetaDataMDX("http://mdx.example.com", freshness_period="P0Y0M0DT0H2M0S")
+    mdx._is_metadata_fresh = Mock(return_value=False)
+
+    mdx.single_sign_on_service(entity_id, BINDING_HTTP_REDIRECT)
+    assert entity_id in mdx.entity
+
     mdx.single_sign_on_service(entity_id, BINDING_HTTP_REDIRECT)
     assert len(responses.calls) == 2