0004-misc-support-asynchronous-logout-41949.patch

Benjamin Dauvergne, 24 avril 2020 13:22

Voir les différences: en ligne côte à côte

Subject: [PATCH 4/7] misc: support asynchronous logout (#41949)

It means that will lookup for other Django sessions linked to the
received logout request; logout request can specify session indexes or
ask for logout of all sessions of the user targeted by the NameID.

 mellon/templates/mellon/session_dump.xml |  14 +--
 mellon/utils.py                          |  42 ++++----
 mellon/views.py                          | 109 +++++++++++++++++---
 tests/test_sso_slo.py                    | 122 ++++++++++++++++++++++-
 4 files changed, 246 insertions(+), 41 deletions(-)

     <ns0:Session xmlns:ns0="http://www.entrouvert.org/namespaces/lasso/0.0" xmlns:ns1="urn:oasis:names:tc:SAML:2.0:assertion" Version="2">
         {% for session_info in session_infos %}
         <ns0:NidAndSessionIndex AssertionID=""
                                 ProviderID="{{ entity_id }}"
                                 SessionIndex="{{ session_index }}">
                 <ns1:NameID Format="{{ name_id_format }}"
                         {% if name_id_name_qualifier %}NameQualifier="{{ name_id_name_qualifier }}"{% endif %}
                         {% if name_id_sp_name_qualifier %}SPNameQualifier="{{ name_id_sp_name_qualifier }}"{% endif %}
                         >{{ name_id_content }}</ns1:NameID>
                                 ProviderID="{{ session_info.entity_id }}"
                                 SessionIndex="{{ session_info.session_index }}">
                 <ns1:NameID Format="{{ session_info.name_id_format }}"
                         {% if session_info.name_id_name_qualifier %}NameQualifier="{{ session_info.name_id_name_qualifier }}"{% endif %}
                         {% if session_info.name_id_sp_name_qualifier %}SPNameQualifier="{{ session_info.name_id_sp_name_qualifier }}"{% endif %}
                         >{{ session_info.name_id_content }}</ns1:NameID>
         </ns0:NidAndSessionIndex>
         {% endfor %}
     </ns0:Session>

     import isodate
     from xml.parsers import expat
     import django
     from django.contrib import auth
     from django.template.loader import render_to_string
     from django.urls import reverse
     from django.utils.encoding import force_text
     from django.utils.timezone import make_aware, now, make_naive, is_aware, get_default_timezone
     from django.conf import settings
     from django.utils.six.moves.urllib.parse import urlparse
-...
     from . import app_settings
     logger = logging.getLogger(__name__)
     def create_metadata(request):
         entity_id = reverse('mellon_metadata')
-...
     def create_server(request):
         logger = logging.getLogger(__name__)
         root = request.build_absolute_uri('/')
         cache = getattr(settings, '_MELLON_SERVER_CACHE', {})
         if root not in cache:
-...
         return idp.get(name) or getattr(app_settings, name, default)
     def make_session_dump(lasso_name_id, indexes):
         session_infos = []
         name_id = force_text(lasso_name_id.content)
         name_id_format = force_text(lasso_name_id.format)
         name_qualifier = lasso_name_id.nameQualifier and force_text(lasso_name_id.nameQualifier)
         sp_name_qualifier = lasso_name_id.spNameQualifier and force_text(lasso_name_id.spNameQualifier)
         for index in indexes:
             issuer = index.saml_identifier.issuer
             session_infos.append({
                 'entity_id': issuer,
                 'session_index': index.session_index,
                 'name_id_content': name_id,
                 'name_id_format': name_id_format,
                 'name_id_name_qualifier': name_qualifier,
                 'name_id_sp_name_qualifier': sp_name_qualifier,
             })
         session_dump = render_to_string('mellon/session_dump.xml', {'session_infos': session_infos})
         return session_dump
     def create_logout(request):
         logger = logging.getLogger(__name__)
         server = create_server(request)
         mellon_session = request.session.get('mellon_session', {})
         entity_id = mellon_session.get('issuer')
         session_index = mellon_session.get('session_index')
         name_id_format = mellon_session.get('name_id_format')
         name_id_content = mellon_session.get('name_id_content')
         name_id_name_qualifier = mellon_session.get('name_id_name_qualifier')
         name_id_sp_name_qualifier = mellon_session.get('name_id_sp_name_qualifier')
         session_dump = render_to_string('mellon/session_dump.xml', {
             'entity_id': entity_id,
             'session_index': session_index,
             'name_id_format': name_id_format,
             'name_id_content': name_id_content,
             'name_id_name_qualifier': name_id_name_qualifier,
             'name_id_sp_name_qualifier': name_id_sp_name_qualifier,
         })
         logger.debug('session_dump %s', session_dump)
         logout = lasso.Logout(server)
         if not app_settings.PRIVATE_KEY and not app_settings.PRIVATE_KEYS:
             logout.setSignatureHint(lasso.PROFILE_SIGNATURE_HINT_FORBID)
         logout.setSessionFromDump(session_dump)
         return logout

     from __future__ import unicode_literals
     from importlib import import_module
     import logging
     import requests
     import lasso
-...
     from django.views.generic import View
     from django.http import HttpResponseRedirect, HttpResponse
     from django.contrib import auth
     from django.contrib.auth import get_user_model
     from django.conf import settings
     from django.views.decorators.csrf import csrf_exempt
     from django.shortcuts import render, resolve_url
-...
     from django.db import transaction
     from django.utils.translation import ugettext as _
     from . import app_settings, utils
     from . import app_settings, utils, models
     RETRY_LOGIN_COOKIE = 'MELLON_RETRY_LOGIN'
-...
     EO_NS = 'https://www.entrouvert.com/'
     LOGIN_HINT = '{%s}login-hint' % EO_NS
     User = get_user_model()
     class HttpResponseBadRequest(django.http.HttpResponseBadRequest):
         def __init__(self, *args, **kwargs):
-...
             if user is not None:
                 if user.is_active:
                     utils.login(request, user)
                     session_index = attributes['session_index']
                     if session_index:
                         models.SessionIndex.objects.get_or_create(
                             saml_identifier=user.saml_identifier,
                             session_key=request.session.session_key,
                             session_index=session_index)
                     self.log.info('user %s (NameID is %r) logged in using SAML', user,
                                   attributes['name_id_content'])
                     request.session['mellon_session'] = utils.flatten_datetime(attributes)
-...
     class LogoutView(ProfileMixin, LogMixin, View):
         def get(self, request, *args, **kwargs):
             if 'SAMLRequest' in request.GET:
                 return self.idp_logout(request)
                 return self.idp_logout(request, request.META['QUERY_STRING'])
             elif 'SAMLResponse' in request.GET:
                 return self.sp_logout_response(request)
             else:
                 return self.sp_logout_request(request)
         def idp_logout(self, request):
         def logout(self, request, issuer, saml_user, session_indexes, indexes):
             session_keys = set(indexes.values_list('session_key', flat=True))
             indexes.delete()
             synchronous_logout = request.user == saml_user
             asynchronous_logout = (
                 # the current session is not the only killed
                 len(session_keys) != 1
                 or (
                     # there is not current session
                     not request.user.is_authenticated()
                     # or the current session is not part of the list
                     or request.session.session_key not in session_keys))
             if asynchronous_logout:
                 current_session_key = request.session.session_key if request.user.is_authenticated() else None
                 session_engine = import_module(settings.SESSION_ENGINE)
                 store = session_engine.SessionStore()
                 count = 0
                 for session_key in session_keys:
                     if session_key != current_session_key:
                         try:
                             store.delete(session_key)
                             count += 1
                         except Exception:
                             self.log.warning('could not delete session_key %s', session_key, exc_info=True)
                 if not session_indexes:
                     self.log.info('asynchronous logout of all sessions of user %s', saml_user)
                 elif count:
                     self.log.info('asynchronous logout of %d sessions of user %s', len(session_keys), saml_user)
             if synchronous_logout:
                 user = request.user
                 auth.logout(request)
                 self.log.info('synchronous logout of %s', user)
         def idp_logout(self, request, msg):
             '''Handle logout request emitted by the IdP'''
             self.profile = logout = utils.create_logout(request)
             try:
                 logout.processRequestMsg(request.META['QUERY_STRING'])
                 logout.processRequestMsg(msg)
             except lasso.Error as e:
                 return HttpResponseBadRequest('error processing logout request: %r' % e)
             try:
                 logout.validateRequest()
             except lasso.Error as e:
                 self.log.warning('error validating logout request: %r' % e)
             issuer = request.session.get('mellon_session', {}).get('issuer')
             if issuer == logout.remoteProviderId:
                 self.log.info('user logged out by IdP SLO request')
                 auth.logout(request)
             issuer = force_text(logout.remoteProviderId)
             session_indexes = set(force_text(sessionIndex) for sessionIndex in logout.request.sessionIndexes)
             saml_identifier = models.UserSAMLIdentifier.objects.filter(
                 name_id=force_text(logout.nameIdentifier.content),
                 issuer=issuer).select_related('user').first()
             if saml_identifier:
                 name_id_user = saml_identifier.user
                 indexes = models.SessionIndex.objects.select_related(
                     'saml_identifier').filter(
                         saml_identifier=saml_identifier)
                 if session_indexes:
                     indexes = indexes.filter(session_index__in=session_indexes)
                 # lasso has too much state :/
                 logout.setSessionFromDump(
                     utils.make_session_dump(
                         logout.nameIdentifier,
                         indexes))
                 try:
                     logout.validateRequest()
                 except lasso.Error as e:
                     self.log.warning('error validating logout request: %r' % e)
                 else:
                     if session_indexes:
                         self.log.info('logout requested for sessionIndexes %s', session_indexes)
                     else:
                         self.log.info('full logout requested, no sessionIndexes')
                     self.logout(
                         request,
                         issuer=issuer,
                         saml_user=name_id_user,
                         session_indexes=session_indexes,
                         indexes=indexes)
             try:
                 logout.buildResponseMsg()
             except lasso.Error as e:
                 return HttpResponseBadRequest('error processing logout request: %r' % e)
             return HttpResponseRedirect(logout.msgUrl)
             if logout.msgBody:
                 return HttpResponse(force_text(logout.msgBody), content_type='text/xml')
             else:
                 return HttpResponseRedirect(logout.msgUrl)
         def sp_logout_request(self, request):
             '''Launch a logout request to the identity provider'''
-...
             return HttpResponseRedirect(next_url)
     logout = LogoutView.as_view()
     logout = csrf_exempt(LogoutView.as_view())
     def metadata(request, **kwargs):

     import pytest
     from pytest import fixture
     from django.contrib.sessions.models import Session
     from django.contrib.auth.models import User
     from django.urls import reverse
     from django.utils import six
     from django.utils.six.moves.urllib import parse as urlparse
-...
     class MockIdp(object):
         session_dump = None
         identity_dump = None
         def __init__(self, idp_metadata, private_key, sp_metadata):
             self.server = server = lasso.Server.newFromBuffers(idp_metadata, private_key)
             self.server.signatureMethod = lasso.SIGNATURE_METHOD_RSA_SHA256
             server.addProviderFromBuffer(lasso.PROVIDER_ROLE_SP, sp_metadata)
         def reset_session_dump(self):
             self.session_dump = None
         def process_authn_request_redirect(self, url, auth_result=True, consent=True, msg=None):
             login = lasso.Login(self.server)
             if self.identity_dump:
                 login.setIdentityFromDump(self.identity_dump)
             if self.session_dump:
                 login.setSessionFromDump(self.session_dump)
             login.processAuthnRequestMsg(url.split('?', 1)[1])
             # See
             # https://docs.python.org/2/library/zlib.html#zlib.decompress
-...
                 raise NotImplementedError
             if login.msgBody:
                 assert b'rsa-sha256' in base64.b64decode(login.msgBody)
             if login.identity:
                 self.identity_dump = login.identity.dump()
             else:
                 self.identity_dump = None
             if login.session:
                 self.session_dump = login.session.dump()
             else:
                 self.session_dump = None
             return login.msgUrl, login.msgBody, login.msgRelayState
         def resolve_artifact(self, soap_message):
-...
             assert 'rsa-sha256' in login.msgBody
             return '<?xml version="1.0"?>\n' + login.msgBody
         def init_slo(self, full=False, method=lasso.HTTP_METHOD_REDIRECT, relay_state=None):
             logout = lasso.Logout(self.server)
             logout.setIdentityFromDump(self.identity_dump)
             logout.setSessionFromDump(self.session_dump)
             logout.initRequest(None, method)
             logout.msgRelayState = relay_state
             if full:
                 logout.request.sessionIndexes = ()
                 logout.request.sessionIndex = None
             logout.buildRequestMsg()
             return logout.msgUrl, logout.msgBody, logout.msgRelayState
         def check_slo_return(self, url=None, body=None):
             logout = lasso.Logout(self.server)
             logout.setIdentityFromDump(self.identity_dump)
             logout.setSessionFromDump(self.session_dump)
             if body:
                 logout.processResponseMsg(force_str(body))
             else:
                 logout.processResponseMsg(force_str(url.split('?', 1)[-1]))
         def mock_artifact_resolver(self):
             @all_requests
             def f(url, request):
-...
         assert urlparse.urlparse(response['Location']).path == '/singleLogout'
     def test_sso_idp_slo(db, app, idp, caplog, sp_settings):
         assert Session.objects.count() == 0
         assert User.objects.count() == 0
         # first session
         response = app.get(reverse('mellon_login') + '?next=/whatever/')
         url, body, relay_state = idp.process_authn_request_redirect(response['Location'])
         assert relay_state
         assert 'eo:next_url' not in str(idp.request)
         assert url.endswith(reverse('mellon_login'))
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         assert 'created new user' in caplog.text
         assert 'logged in using SAML' in caplog.text
         assert urlparse.urlparse(response['Location']).path == '/whatever/'
         # start a new Lasso session
         idp.reset_session_dump()
         # second session
         app.cookiejar.clear()
         response = app.get(reverse('mellon_login') + '?next=/whatever/')
         url, body, relay_state = idp.process_authn_request_redirect(response['Location'])
         assert relay_state
         assert 'eo:next_url' not in str(idp.request)
         assert url.endswith(reverse('mellon_login'))
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         assert 'created new user' in caplog.text
         assert 'logged in using SAML' in caplog.text
         assert urlparse.urlparse(response['Location']).path == '/whatever/'
         assert Session.objects.count() == 2
         assert User.objects.count() == 1
         # idp logout
         url, body, relay_state = idp.init_slo()
         response = app.get(url)
         assert response.location.startswith('http://idp5/singleLogoutReturn?')
         assert Session.objects.count() == 1
         idp.check_slo_return(response.location)
     def test_sso_idp_slo_full(db, app, idp, caplog, sp_settings):
         assert Session.objects.count() == 0
         assert User.objects.count() == 0
         # first session
         response = app.get(reverse('mellon_login') + '?next=/whatever/')
         url, body, relay_state = idp.process_authn_request_redirect(response['Location'])
         assert relay_state
         assert 'eo:next_url' not in str(idp.request)
         assert url.endswith(reverse('mellon_login'))
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         assert 'created new user' in caplog.text
         assert 'logged in using SAML' in caplog.text
         assert urlparse.urlparse(response['Location']).path == '/whatever/'
         # second session
         app.cookiejar.clear()
         response = app.get(reverse('mellon_login') + '?next=/whatever/')
         url, body, relay_state = idp.process_authn_request_redirect(response['Location'])
         assert relay_state
         assert 'eo:next_url' not in str(idp.request)
         assert url.endswith(reverse('mellon_login'))
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         assert 'created new user' in caplog.text
         assert 'logged in using SAML' in caplog.text
         assert urlparse.urlparse(response['Location']).path == '/whatever/'
         assert Session.objects.count() == 2
         assert User.objects.count() == 1
         # idp logout
         url, body, relay_state = idp.init_slo(full=True)
         response = app.get(url)
         assert response.location.startswith('http://idp5/singleLogoutReturn?')
         assert Session.objects.count() == 0
         idp.check_slo_return(url=response.location)
     def test_sso(db, app, idp, caplog, sp_settings):
         response = app.get(reverse('mellon_login'))
         url, body, relay_state = idp.process_authn_request_redirect(response['Location'])
-...
         acs_artifact_url = url.split('testserver', 1)[1]
         with HTTMock(idp.mock_artifact_resolver()):
             response = app.get(acs_artifact_url, params={'RelayState': relay_state})
         assert 'created new user' in caplog.text
         assert 'created new user' not in caplog.text
         assert 'logged in using SAML' in caplog.text
         assert urlparse.urlparse(response['Location']).path == '/whatever/'
+    -

Projet

Général

Profil

Produits Entr'ouvert » django-mellon

0004-misc-support-asynchronous-logout-41949.patch