Okfn updates

.github/workflows/ci.yml

@@ -17,7 +17,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v4
         with:
-          python-version: '3.8'
+          python-version: '3.10'
 
       - name: Install flake8
         run: |
@@ -26,16 +26,16 @@ jobs:
 
       - name: Lint with flake8
         run: |
-          flake8 . --count --max-complexity=10 --max-line-length=127 --statistics --exclude ckan,ckanext-saml2auth
+          flake8 . --count --max-complexity=12 --max-line-length=127 --statistics --exclude ckan,ckanext-saml2auth
 
   test:
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
-        python-version: [ '3.7', '3.8', '3.9']
-        ckan-version: ["2.9", "2.10"]
-    name: Python ${{ matrix.python-version }} extension test
+        python-version: ['3.9', '3.10']  # TODO '3.11'
+        ckan-version: ["2.10", "2.11"]
+    name: Python ${{ matrix.python-version }} CKAN ${{ matrix.ckan-version }} extension test
 
     services:
       postgresql:
@@ -61,9 +61,7 @@ jobs:
           - 6379:6379
 
       ckan-solr:
-        # Workflow level env variables are not addressable on job level, only on steps level
-        # image: ghcr.io/keitaroinc/ckan-solr-dev:{{ env.CKANVERSION }}
-        image: ghcr.io/keitaroinc/ckan-solr-dev:2.9
+        image: ckan/ckan-solr:2.10
         ports:
           - 8983:8983
 
@@ -90,8 +88,9 @@ jobs:
 
       - name: Test with pytest
         run: |
+          echo "Running SAML2AUTH tests"
           pytest --ckan-ini=subdir/test.ini --cov=ckanext.saml2auth --disable-warnings ckanext/saml2auth/tests
-
+      
       - name: Coveralls
         uses: AndreMiras/coveralls-python-action@develop
         with:
@@ -132,4 +131,4 @@ jobs:
     - name: Coveralls Finished
       uses: AndreMiras/coveralls-python-action@develop
       with:
-        parallel-finished: true
+        parallel-finished: true

README.md

@@ -1,12 +1,14 @@
 [![CI][]][1] [![Coverage][]][2] [![Gitter][]][3] [![Pypi][]][4] [![Python][]][5] [![CKAN][]][6]
 
+
 # ckanext-saml2auth
 
 A [CKAN](https://ckan.org) extension to enable Single Sign-On (SSO) for CKAN data portals via SAML2 Authentication.
 
 ## Requirements
 
-This extension works with CKAN 2.9+.
+This extension works with CKAN 2.10+
+Note: For CKAN 2.9 or older use v1.3.0 or older versions.
 
 ## Installation
 
@@ -135,6 +137,8 @@ Optional:
     # Saml logout request preferred binding settings variable
     # Default: urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST
     ckanext.saml2auth.logout_expected_binding =  urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST
+    # If you don't want to logout from external source you can use
+    ckanext.saml2auth.logout_expected_binding = skip-external-logout
 
     # Default fallback endpoint to redirect to if no RelayState provided in the SAML Response
     # Default: user.me (ie /dashboard)

bin/setup-ckan.bash

@@ -46,9 +46,10 @@ cd ckan
 ckan -c test-core.ini db init
 cd -
 
-echo "Installing ckanext-saml2auth and its requirements..."
-python setup.py develop
+echo "Installing saml2 requirements..."
 pip install -r dev-requirements.txt
+echo "Installing ckanext-saml2auth..."
+pip install -e .
 
 echo "Moving test.ini into a subdir..."
 mkdir subdir

ckanext/saml2auth/cache.py

@@ -18,12 +18,16 @@
 import logging
 
 from saml2.ident import code, decode
+from saml2.saml import NameID
 
 log = logging.getLogger(__name__)
 
 
 def set_subject_id(session, subject_id):
-    session['_saml2_subject_id'] = code(subject_id)
+    if isinstance(subject_id, str):
+        session['_saml2_subject_id'] = subject_id
+    else:
+        session['_saml2_subject_id'] = code(subject_id)
 
 
 def get_subject_id(session):
@@ -34,11 +38,32 @@ def get_subject_id(session):
 
 
 def set_saml_session_info(session, saml_session_info):
+    """Adds information about pysaml2 AuthnResponse to CKAN's session.
+
+    `pysaml2` returns a NameID object in the session_info() call. Since we want
+    to serialize the object to write it into the cookie we need to convert it.
+    `name_id` is the same as `_saml2_subject_id` so we apply `code` as we do in
+    `set_subject_id`.
+
+    We are not sure if it always return an object, so we checking to be sure.
+    """
+    if isinstance(saml_session_info['name_id'], NameID):
+        saml_session_info['name_id'] = code(saml_session_info['name_id'])
     session['_saml_session_info'] = saml_session_info
 
 
 def get_saml_session_info(session):
+    """Returns the saml session info from the session object.
+
+    The session object is serializable but pysaml expect a NameID object as
+    name_id, so we are decoding it again as we do in get_subject_id.
+    """
     try:
-        return session['_saml_session_info']
+        session_info = session['_saml_session_info']
     except KeyError:
         return None
+
+    if isinstance(session_info['name_id'], str):
+        session_info['name_id'] = decode(session_info['name_id'])
+
+    return session_info

ckanext/saml2auth/helpers.py

@@ -121,3 +121,12 @@ def get_site_domain_for_cookie():
     parsed_url = urlparse(site_url)
     host = parsed_url.netloc.split(':')[0]
     return host if '.' in host else None
+
+
+def get_saml2auth_login_button_text():
+    """
+    Returns the configured text for the SAML2 login button.
+    Defaults to 'SSO' if not configured.
+    """
+    text = toolkit.config.get('ckanext.saml2auth.login_button_text', 'SSO')
+    return text

ckanext/saml2auth/plugin.py

@@ -48,8 +48,8 @@ class Saml2AuthPlugin(plugins.SingletonPlugin):
 
     def get_helpers(self):
         return {
-            'is_default_login_enabled':
-                h.is_default_login_enabled
+            'is_default_login_enabled': h.is_default_login_enabled,
+            'get_saml2auth_login_button_text': h.get_saml2auth_login_button_text,
         }
 
     # IConfigurable
@@ -120,9 +120,12 @@ def _perform_slo():
 
     response = None
 
-    client = h.saml_client(
-        sp_config()
-    )
+    config = sp_config()
+    if config.get('logout_expected_binding') == 'skip-external-logout':
+        log.debug('Skipping external logout')
+        return
+
+    client = h.saml_client(config)
     saml_session_info = get_saml_session_info(session)
     subject_id = get_subject_id(session)
 

ckanext/saml2auth/templates/user/snippets/login_form.html

@@ -18,5 +18,5 @@
 {% ckan_extends %}
 {% block login_button %}
   <button class="btn btn-primary" type="submit">{{ _('Login') }}</button>
-  <a class="btn btn-default" href="{{ h.url_for('saml2auth.saml2login') }}">{{ _('SSO') }}</a>
+  <a class="btn btn-default" href="{{ h.url_for('saml2auth.saml2login') }}">{{ h.get_saml2auth_login_button_text() }}</a>
 {% endblock %}

ckanext/saml2auth/tests/responses/unsigned0.xml

@@ -17,7 +17,7 @@
       <saml:NameID SPNameQualifier="{{ entity_id }}" Format="urn:oasis:names:tc:SAML:2.0:nameid-format:transient">_ce3d2948b4cf20146dee0a0b3dd6f69b6cf86f62d7</saml:NameID>
       <saml:SubjectConfirmation Method="urn:oasis:names:tc:SAML:2.0:cm:bearer">
         <saml:SubjectConfirmationData 
-          NotOnOrAfter="2024-01-18T06:21:48Z" 
+          NotOnOrAfter="2026-01-18T06:21:48Z" 
           Recipient="{{ recipient }}" 
           InResponseTo="ONELOGIN_4fee3b046395c4e751011e97f8900b5273d56685"/>
       </saml:SubjectConfirmation>

ckanext/saml2auth/tests/test_blueprint.py

@@ -61,50 +61,12 @@ 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):
-        if toolkit.check_ckan_version(min_version='2.10'):
-            # Remove when dropping support for 2.9
-            pytest.skip("auth_tkt cookie has been deprecated in 2.10")
-        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()
-
     @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_ckan_cookie_cleared_on_slo(self, app):
-        if not toolkit.check_ckan_version(min_version='2.10'):
-            # Remove when dropping support for 2.9
-            pytest.skip("This test logic introduced in CKAN 2.10")
+
         url = url_for('user.logout')
 
         import datetime
@@ -121,11 +83,17 @@ def test_ckan_cookie_cleared_on_slo(self, app):
         cookie_headers = [
             h[1] for h in response.headers
             if h[0].lower() == 'set-cookie']
+        # Sample for CKAN 2.11
+        # ['ckan=; Domain=test.ckan.net; Expires=Thu, 01 Jan 1970 00:00:00 GMT; Path=/']
 
         # Starting 2.10, CKAN's SessionMiddleware will append a
         # new Set-cookie header on every first response from the server.
         # This includes test requests.
-        assert len(cookie_headers) == 2
+        # For CKAN 2.11, we only get the session cookie is named 'ckan'
+        if toolkit.check_ckan_version(min_version='2.11'):
+            assert len(cookie_headers) == 1
+        else:
+            assert len(cookie_headers) == 2
 
         first_cookie = cookie_headers[0]
 

ckanext/saml2auth/tests/test_blueprint_get_request.py

@@ -58,7 +58,7 @@ def _prepare_unsigned_response():
         'entity_id': 'urn:gov:gsa:SAML:2.0.profiles:sp:sso:test:entity',
         'destination': 'http://test.ckan.net/acs',
         'recipient': 'http://test.ckan.net/acs',
-        'issue_instant': datetime.now().isoformat()
+        'issue_instant': datetime.now().isoformat(),
     }
     t = Template(unsigned_response)
     final_response = t.render(**context)
@@ -117,7 +117,22 @@ def test_unsigned_request(self, app):
         data = {
             'SAMLResponse': encoded_response
         }
-        response = app.post(url=url, params=data)
+        try:
+            response = app.post(url=url, params=data)
+        except Exception as e:
+            # decode the response
+            import base64
+            decoded = base64.b64decode(encoded_response)
+            raise Exception(
+                f'Error test_unsigned_request: {e}\n'
+                f'encoded_response: {decoded}\n'
+                f'url: {url}\n'
+            )
+
+        if response.status_code != 200:
+            assert False, f'Failed test_unsigned_request: {response.body}'
+            # Can't use response, too old (now=2024-07-31T17:42:38Z + slack=0 > not_on_or_after=2024-01-18T06:21:48Z
+
         assert 200 == response.status_code
 
     def render_file(self, path, context, save_as=None):

ckanext/saml2auth/tests/test_cache.py

@@ -0,0 +1,86 @@
+from ckanext.saml2auth.cache import set_subject_id, get_subject_id, get_saml_session_info
+from types import SimpleNamespace
+from saml2.ident import code
+
+nameid = SimpleNamespace(
+    name_qualifier="issuer.example.com",
+    sp_name_qualifier="sp.example.com",
+    text="user123",
+    format="urn:oasis:names:tc:SAML:2.0:nameid-format:persistent",
+    sp_provided_id=None
+)
+
+
+def test_set_subject_id_with_string():
+    session = {}
+    set_subject_id(session, "user123")
+    assert session['_saml2_subject_id'] == "user123"
+
+
+def test_set_subject_id_with_nameid_like_object():
+    session = {}
+    expected_code = code(nameid)
+    set_subject_id(session, nameid)
+    assert session['_saml2_subject_id'] == expected_code
+
+
+def test_get_subject_id_missing():
+    session = {}
+    result = get_subject_id(session)
+    assert result is None
+
+
+def test_get_subject_id_with_real_code_decode():
+
+    encoded = code(nameid)
+    session = {'_saml2_subject_id': encoded}
+
+    result = get_subject_id(session)
+
+    assert result.name_qualifier == nameid.name_qualifier
+    assert result.sp_name_qualifier == nameid.sp_name_qualifier
+    assert result.format == nameid.format
+    assert result.text == nameid.text
+    assert result.sp_provided_id == nameid.sp_provided_id
+
+
+def test_get_saml_session_info_with_encoded_nameid():
+
+    encoded_nameid = code(nameid)
+    session = {
+        '_saml_session_info': {
+            'name_id': encoded_nameid,
+            'other_data': 'example'
+        }
+    }
+
+    result = get_saml_session_info(session)
+
+    assert isinstance(result['name_id'], object)
+    assert result['name_id'].name_qualifier == nameid.name_qualifier
+    assert result['name_id'].sp_name_qualifier == nameid.sp_name_qualifier
+    assert result['name_id'].text == nameid.text
+    assert result['name_id'].format == nameid.format
+    assert result['name_id'].sp_provided_id == nameid.sp_provided_id
+    assert result['other_data'] == 'example'
+
+
+def test_get_saml_session_info_with_decoded_nameid():
+
+    session = {
+        '_saml_session_info': {
+            'name_id': nameid,
+            'foo': 'bar'
+        }
+    }
+
+    result = get_saml_session_info(session)
+
+    assert result['name_id'] == nameid
+    assert result['foo'] == 'bar'
+
+
+def test_get_saml_session_info_missing():
+    session = {}
+    result = get_saml_session_info(session)
+    assert result is None