Merge pull request #56 from okfn/clear-cookies-on-logout

[#55] Clear cookies on logout

ckanext/saml2auth/helpers.py

@@ -23,14 +23,15 @@
 import random
 import secrets
 from six import text_type
+from six.moves.urllib.parse import urlparse
 
 from ckanext.saml2auth.client import Saml2Client
 
 from saml2.config import Config as Saml2Config
 
 import ckan.model as model
 import ckan.authz as authz
-from ckan.common import config, asbool, aslist
+from ckan.plugins import toolkit
 
 
 log = logging.getLogger(__name__)
@@ -50,14 +51,14 @@ def generate_password():
 
 
 def is_default_login_enabled():
-    return asbool(
-        config.get('ckanext.saml2auth.enable_ckan_internal_login',
-                   False))
+    return toolkit.asbool(
+        toolkit.config.get('ckanext.saml2auth.enable_ckan_internal_login')
+    )
 
 
 def update_user_sysadmin_status(username, email):
-    sysadmins_list = aslist(
-        config.get('ckanext.saml2auth.sysadmins_list'))
+    sysadmins_list = toolkit.aslist(
+        toolkit.config.get('ckanext.saml2auth.sysadmins_list'))
     user = model.User.by_name(text_type(username))
     sysadmin = authz.is_sysadmin(username)
 
@@ -107,3 +108,16 @@ def get_location(http_info):
         return headers['Location']
     except KeyError:
         return http_info['url']
+
+
+def get_site_domain_for_cookie():
+    '''Return the domain part of the site URL
+
+    When running on localhost (or any single word host), browsers will
+    ignore the `Domain` bit in the Set-Cookie header (and Werkzeug will
+    not allow you to set it), so we return None on this case.
+    '''
+    site_url = toolkit.config.get('ckan.site_url')
+    parsed_url = urlparse(site_url)
+    host = parsed_url.netloc.split(':')[0]
+    return host if '.' in host else None

ckanext/saml2auth/plugin.py

@@ -21,7 +21,7 @@
 from saml2.client_base import LogoutError
 from saml2 import entity
 
-from flask import session, redirect
+from flask import session, redirect, make_response
 
 import ckan.plugins as plugins
 import ckan.plugins.toolkit as toolkit
@@ -103,47 +103,73 @@ def login(self):
 
     def logout(self):
 
-        client = h.saml_client(
-            sp_config()
-        )
-        saml_session_info = get_saml_session_info(session)
-        subject_id = get_subject_id(session)
-
-        if subject_id is None:
-            log.warning(
-                'The session does not contain the subject id for user {}'.format(g.user))
-        else:
-            try:
-                client.users.add_information_about_person(saml_session_info)
-                result = client.global_logout(name_id=subject_id)
-            except LogoutError as e:
-                log.exception(
-                    'SLO not supported by IDP: {}'.format(e))
-                # clear session
-
-            if not result:
-                log.error(
-                    'Looks like the user {} is not logged in any IdP/AA'.format(subject_id))
+        response = _perform_slo()
+
+        if response:
+            domain = h.get_site_domain_for_cookie()
+
+            # Clear auth cookie in the browser
+            response.set_cookie('auth_tkt', domain=domain, expires=0)
+
+            # Clear session cookie in the browser
+            response.set_cookie('ckan', domain=domain, expires=0)
+
+        return response
+
+
+def _perform_slo():
+
+    response = None
 
-            if len(result) > 1:
+    client = h.saml_client(
+        sp_config()
+    )
+    saml_session_info = get_saml_session_info(session)
+    subject_id = get_subject_id(session)
+
+    if subject_id is None:
+        log.warning(
+            'The session does not contain the subject id for user {}'.format(g.user))
+        return
+
+    try:
+        client.users.add_information_about_person(saml_session_info)
+        result = client.global_logout(name_id=subject_id)
+    except LogoutError as e:
+        log.exception(
+            'SLO not supported by IDP: {}'.format(e))
+        # clear session
+
+    if not result:
+        log.error(
+            'Looks like the user {} is not logged in any IdP/AA'.format(subject_id))
+
+    if len(result) > 1:
+        log.error(
+            'Sorry, I do not know how to logout from several sources.'
+            ' I will logout just from the first one')
+
+    for entity_id, logout_info in result.items():
+        if isinstance(logout_info, tuple):
+            binding, http_info = logout_info
+            if binding == entity.BINDING_HTTP_POST:
+                log.debug(
+                    'Returning form to the IdP to continue the logout process')
+                body = ''.join(http_info['data'])
+                extra_vars = {
+                    u'body': body
+                }
+                response = make_response(
+                    base.render(u'saml2auth/idp_logout.html', extra_vars)
+                )
+
+            elif binding == entity.BINDING_HTTP_REDIRECT:
+                log.debug(
+                    'Redirecting to the IdP to continue the logout process')
+
+                response = redirect(h.get_location(http_info), code=302)
+            else:
                 log.error(
-                    'Sorry, I do not know how to logout from several sources.'
-                    ' I will logout just from the first one')
-
-            for entity_id, logout_info in result.items():
-                if isinstance(logout_info, tuple):
-                    binding, http_info = logout_info
-                    if binding == entity.BINDING_HTTP_POST:
-                        log.debug(
-                            'Returning form to the IdP to continue the logout process')
-                        body = ''.join(http_info['data'])
-                        extra_vars = {
-                            u'body': body
-                        }
-                        return base.render(u'saml2auth/idp_logout.html', extra_vars)
-                    elif binding == entity.BINDING_HTTP_REDIRECT:
-                        log.debug(
-                            'Redirecting to the IdP to continue the logout process')
-                        return redirect(h.get_location(http_info), code=302)
-                    else:
-                        log.error('Failed to log out from Idp. Unknown binding: {}'.format(binding))
+                    'Failed to log out from Idp. Unknown binding: {}'.format(binding))
+
+    return response

ckanext/saml2auth/tests/test_blueprint.py

@@ -59,3 +59,37 @@ def test_came_from_sent_as_relay_state(self, app):
 
         response = app.get(url=url, follow_redirects=False)
         assert 'RelayState=%2Fdataset%2Fmy-dataset' in response.headers['Location']
+
+    @pytest.mark.ckan_config(u'ckanext.saml2auth.idp_metadata.location', u'local')
+    @pytest.mark.ckan_config(u'ckanext.saml2auth.idp_metadata.local_path',
+                             os.path.join(extras_folder, 'provider2', 'idp.xml'))
+    @pytest.mark.usefixtures('with_request_context')
+    def test_cookies_cleared_on_slo(self, app):
+
+        url = url_for('user.logout')
+
+        import datetime
+        from unittest import mock
+        from http.cookies import SimpleCookie
+        from flask import make_response
+        from dateutil.parser import parse as date_parse
+
+        with mock.patch(
+            'ckanext.saml2auth.plugin._perform_slo',
+                return_value=make_response('')):
+            response = app.get(url=url, follow_redirects=False)
+
+        cookie_headers = [
+            h[1] for h in response.headers
+            if h[0].lower() == 'set-cookie']
+
+        assert len(cookie_headers) == 2
+
+        for cookie_header in cookie_headers:
+            cookie = SimpleCookie()
+            cookie.load(cookie_header)
+            cookie_name = [name for name in cookie.keys()][0]
+            assert cookie_name in ['auth_tkt', 'ckan']
+            assert cookie[cookie_name]['domain'] == 'test.ckan.net'
+            cookie_date = date_parse(cookie[cookie_name]['expires'], ignoretz=True)
+            assert cookie_date < datetime.datetime.now()