Add replay attack protection.

Author: jeffFranklin

README.md

@@ -59,6 +59,8 @@ def login():
 
 ## Considerations
 
+### Sessions
+
 Give some consideration to session lifetime. The session in this example lives as a
 signed cookie. Ideally the cookie would expire at browser close, along with
 some time limit appropriate for your application. An example again with flask
@@ -71,3 +73,18 @@ app.config.update(
     PERMANENT_SESSION_LIFETIME=timedelta(minutes=10)
 )
 ```
+
+### Replay attack prevention
+
+By default this package uses an in-memory cache to check for replay attacks.
+To use a distributed cache such as redis or memcached you would inject a
+cache object into `uw_saml2.auth.CACHE`. Here's an example of how to do it...
+
+```python
+import werkzeug.contrib.cache
+import uw_saml2.auth
+
+uw_saml2.auth.CACHE = werkzeug.contrib.cache.RedisCache()
+```
+
+Django's cache backend uses the same methods so that could be injected as well.

setup.py

@@ -24,5 +24,6 @@
       license='Apache License, Version 2.0',
       packages=find_packages(),
       include_package_data=True,
+      install_requires=['Werkzeug'],
       extras_require={'python3-saml': saml_requires, 'test': tests_require}
       )

tests/test_auth.py

@@ -1,4 +1,5 @@
 import uw_saml2
+import uw_saml2.auth
 from onelogin.saml2.utils import OneLogin_Saml2_Utils
 import pytest
 import mock
@@ -52,6 +53,21 @@ def test_process_response(mock_time_and_id):
     assert attributes == expected_attributes
 
 
+def test_process_response_replay(mock_time_and_id):
+    with open(os.path.join(BASEDIR, 'samlresponse.txt')) as fd:
+        post = dict(urllib.parse.parse_qsl(fd.read()))
+    # set a time when this response was still valid
+    mock_time_and_id.return_value = 1549304000
+    entity_id = 'https://samldemo.iamdev.s.uw.edu/saml'
+    acs_url = 'https://samldemo.iamdev.s.uw.edu/saml/login'
+    attributes = uw_saml2.process_response(post, entity_id=entity_id,
+                                           acs_url=acs_url)
+    with pytest.raises(uw_saml2.auth.SamlResponseError) as excinfo:
+        attributes = uw_saml2.process_response(post, entity_id=entity_id,
+                                               acs_url=acs_url)
+    assert 'replay' in str(excinfo.value).lower()
+
+
 @pytest.fixture
 def mock_time_and_id(monkeypatch):
     """Mock the two things guaranteed to be nondeterministic - date and id."""
@@ -60,4 +76,5 @@ def mock_time_and_id(monkeypatch):
     mock_time.return_value = 1549302384
     monkeypatch.setattr(f'{utils}.now', mock_time)
     monkeypatch.setattr(f'{utils}.generate_unique_id', lambda: 'FOOBAR123')
+    uw_saml2.auth.CACHE.clear()
     return mock_time

uw_saml2/auth.py

@@ -1,11 +1,15 @@
 """UW-specific adapter for the python3-saml package."""
+import werkzeug.contrib.cache
 from .python3_saml import get_saml_authenticator
 from .idp.uw import UwIdp
 from .sp import Config, TWO_FACTOR_CONTEXT
 from .idp import attribute
 from logging import getLogger
 logger = getLogger(__name__)
 
+# For distributed environments, inject a distributed cache.
+CACHE = werkzeug.contrib.cache.SimpleCache()
+
 
 def login_redirect(entity_id=None, acs_url=None, return_to='/',
                    force_authn=False, idp=UwIdp, two_factor=False):
@@ -55,6 +59,11 @@ def process_response(post, entity_id=None, acs_url=None, idp=UwIdp,
     errors = auth.get_errors()
     if errors:
         raise SamlResponseError(auth.get_last_error_reason())
+
+    message_id = auth.get_last_message_id()
+    if not CACHE.add(f'uw_saml2:response:message_id:{message_id}', True):
+        raise SamlResponseError(f'SAML Replay of {message_id}')
+
     attribute_data = dict(attribute.map(auth.get_attributes(), idp=idp))
 
     authn_contexts = auth.get_last_authn_contexts()

uw_saml2/mock.py

@@ -1,4 +1,5 @@
 import urllib.parse
+from random import random
 from .idp import uw, federated
 MOCK_LOGIN_URL = '/mock-login'
 
@@ -18,7 +19,7 @@ def process_response(self):
         idp = self.request['post_data'].get('idp')
         if idp != self.idp:
             self.errors.append(f'idp mismatch {idp} != {self.idp}')
-        return
+        self.message_id = f'MOCKSAML{int(random() * 10**8)}'
 
     def get_attributes(self):
         """Just reflect what's posted right back."""
@@ -41,3 +42,6 @@ def get_last_error_reason(self):
 
     def get_last_authn_contexts(self):
         return []
+
+    def get_last_message_id(self):
+        return self.message_id