0001-views-use-a-view-to-show-all-error-messages-56354.patch

Benjamin Dauvergne, 12 janvier 2022 23:24

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

Subject: [PATCH 1/2] views: use a view to show all error messages (#56354)

 .../mellon/authentication_failed.html         |  20 +-
 mellon/views.py                               | 227 +++++++++---------
 tests/test_sso_slo.py                         |  22 +-
 tests/test_views.py                           |  16 +-
 4 files changed, 155 insertions(+), 130 deletions(-)

       <h2 class="mellon-message-header">{% trans "Authentication failed" %}</h2>
       <p class="mellon-message-body">
         {% blocktrans %}The authentication has failed.{% endblocktrans %}
         {% if reason %}<p class="mellon-reason">{% trans "Reason" %}&nbsp;: {{ reason }}</p>{% endif %}
         {% for reason in reasons %}
           <p class="mellon-reason">{% if loop.first %}{% trans "Reason" %}&nbsp;: {% endif %}{{ reason }}</p>
         {% endfor %}
       </p>
       <p class="mellon-message-continue">
         <a class="mellon-link" href="{{ next_url }}">{% trans "Continue" %}</a>
       </p>
       {% if debug %}
       <!-- DEBUG INFO:
            Issuer: {{ issuer }}
            Message: {{ status_message }}
            Status codes: {{ status_codes|join:", " }}
            Relaystate: {{ relaystate }}
       -->
       <p>{% trans "Information shown because DEBUG is True" %}</p>
       <pre>
     Issuer: {{ issuer }}
     Message: {{ status_message }}
     Status codes: {{ status_codes|join:", " }}
     Relaystate: {{ relaystate }}
     {% for key, value in debug_details.items %}
     {{key}}: {{value|pprint}}
     {% endfor %}
       </pre>
       {% endif %}
     </div>
     {% endblock %}

     from io import StringIO
     from xml.sax.saxutils import escape
     import django.http
     import lasso
     import requests
     from django.conf import settings
-...
     User = get_user_model()
     class HttpResponseBadRequest(django.http.HttpResponseBadRequest):
         def __init__(self, *args, **kwargs):
             kwargs['content_type'] = kwargs.get('content_type', 'text/plain')
             super().__init__(*args, **kwargs)
             self['X-Content-Type-Options'] = 'nosniff'
     class LogMixin:
         """Initialize a module logger in new objects"""
-...
     class ProfileMixin:
         profile = None
         def dispatch(self, request, *args, **kwargs):
             try:
                 return super().dispatch(request, *args, **kwargs)
             # handle common errors
             except utils.CreateServerError:
                 return self.failure(
                     request,
                     reason=_(
                         'Unable to initialize a SAML server object, the private key '
                         'is maybe invalid or unreadable, please check its access '
                         'rights and content.'
                     ),
+                )
         def set_next_url(self, next_url):
             if not check_next_url(self.request, next_url):
                 return
-...
         def get_next_url(self, default=None):
             return self.get_state('next_url', default=default)
         def show_message_status_is_not_success(self, profile, prefix):
         @property
         def template_base(self):
             return self.kwargs.get('template_base', 'base.html')
         def render(self, request, template_names, context, status=None):
             context['template_base'] = self.template_base
             if 'context_hook' in self.kwargs:
                 self.kwargs['context_hook'](context)
             return render(request, template_names, context, status=status)
         def failure_status_is_not_success(self, request, profile, prefix):
             status_codes, idp_message = utils.get_status_codes_and_message(profile)
             args = ['%s: status is not success codes: %r', prefix, status_codes]
             if idp_message:
                 args[0] += ' message: %s'
                 args.append(idp_message)
             self.log.warning(*args)
             return self.failure(
                 request,
                 reasons=[
                     _('%s refused by identity provider') % prefix,
                     idp_message,
                     _('Status codes: %s') % ', '.join([str(code) for code in status_codes]),
                 ],
                 status_codes=status_codes,
+            )
         def dispatch(self, request, *args, **kwargs):
             try:
                 return super().dispatch(request, *args, **kwargs)
             except utils.CreateServerError:
                 return self.failure(
                     request,
                     reason=_(
                         'Unable to initialize a SAML server object, the private key '
                         'is maybe invalid or unreadable, please check its access '
                         'rights and content.'
                     ),
+                )
         def failure(self, request, reason='', status_codes=()):
         def failure(self, request, reason='', reasons=None, status_codes=(), exception=None, debug_details=None):
             '''show error message to user after a login failure'''
             login = self.profile
             idp = utils.get_idp(login and login.remoteProviderId)
             if not idp and login:
                 self.log.warning('entity id %r is unknown', login.remoteProviderId)
                 return HttpResponseBadRequest('entity id %r is unknown' % login.remoteProviderId)
             reasons = reasons or []
             reasons.append(reason)
             profile = self.profile
             idp = utils.get_idp(profile.remoteProviderId) if profile else {}
             if not idp and profile and not isinstance(exception, lasso.ServerProviderNotFoundError):
                 self.log.warning('entity id %r is unknown', profile.remoteProviderId)
                 reasons.insert(
 ,
                     _('Other error: %s')
                     % (_('the received message EntityID (%r) is unknown.') % profile.remoteProviderId),
+                )
             if exception and isinstance(
                 exception,
+                (
                     lasso.DsError,
                     lasso.ProfileCannotVerifySignatureError,
                     lasso.LoginInvalidSignatureError,
                     lasso.LoginInvalidAssertionSignatureError,
                 ),
             ):
                 reasons.append(
                     _(
                         'There was a problem with a signature. It usually means '
                         'that signature key has changed or is unknown. '
                         'The metadata of the identity provider may be out of date.'
+                    )
+                )
                 if 'METADATA_URL' not in idp:
                     reasons.append(
                         _(
                             'Possible solution: you currently do not use METADATA_URL '
                             'to reference the metadata, use it to always keep the metadata up to date.'
+                        )
+                    )
             error_url = utils.get_setting(idp, 'ERROR_URL')
             error_redirect_after_timeout = utils.get_setting(idp, 'ERROR_REDIRECT_AFTER_TIMEOUT')
             if error_url:
-...
                 'mellon/authentication_failed.html',
+                {
                     'debug': settings.DEBUG,
                     'reason': reason,
                     'reasons': reasons,
                     'status_codes': status_codes,
                     'issuer': login and login.remoteProviderId,
                     'issuer': profile.remoteProviderId if profile else None,
                     'next_url': next_url,
                     'relaystate': login and login.msgRelayState,
                     'relaystate': profile.msgRelayState if profile else None,
                     'error_redirect_after_timeout': error_redirect_after_timeout,
                     'debug_details': debug_details or {},
                 },
                 status=400,
+            )
-...
             with self.capture_logs() if self.debug_login else nullcontext():
                 return super().dispatch(request, *args, **kwargs)
         @property
         def template_base(self):
             return self.kwargs.get('template_base', 'base.html')
         def render(self, request, template_names, context):
             context['template_base'] = self.template_base
             if 'context_hook' in self.kwargs:
                 self.kwargs['context_hook'](context)
             return render(request, template_names, context)
         def get_idp(self, request):
             entity_id = request.POST.get('entityID') or request.GET.get('entityID')
         def get_idp(self, entity_id):
             if not entity_id:
                 for idp in utils.get_idps():
                     return idp
-...
             '''Assertion consumer'''
             if 'SAMLart' in request.POST:
                 if 'artifact' not in app_settings.ASSERTION_CONSUMER_BINDINGS:
                     raise Http404('artifact binding is not supported')
                     raise self.failure(request, _('Artifact binding is not supported'))
                 return self.continue_sso_artifact(request, lasso.HTTP_METHOD_ARTIFACT_POST)
             if 'SAMLResponse' not in request.POST:
                 if 'post' not in app_settings.ASSERTION_CONSUMER_BINDINGS:
                     raise Http404('post binding is not supported')
                     raise self.failure(request, _('Post binding is not supported'))
                 return self.get(request, *args, **kwargs)
             # prevent null characters in SAMLResponse
             if not utils.is_nonnull(request.POST['SAMLResponse']):
                 return HttpResponseBadRequest('SAMLResponse contains a null character')
                 return self.failure(_('SAMLResponse contains a null character'))
             self.log.info('Got SAML Response', extra={'saml_response': request.POST['SAMLResponse']})
             self.profile = login = utils.create_login(request)
             idp_message = None
             status_codes = []
             # prevent null characters in SAMLResponse
             try:
                 login.processAuthnResponseMsg(request.POST['SAMLResponse'])
                 login.acceptSso()
             except lasso.ProfileCannotVerifySignatureError:
                 self.log.warning(
                     'SAML authentication failed: signature validation failed for %r', login.remoteProviderId
+                )
             except lasso.ParamError:
                 self.log.exception('lasso param error')
             except (
                 lasso.LoginStatusNotSuccessError,
                 lasso.ProfileStatusNotSuccessError,
                 lasso.ProfileRequestDeniedError,
             ):
                 self.show_message_status_is_not_success(login, 'SAML authentication failed')
             except lasso.Error as e:
                 return HttpResponseBadRequest('error processing the authentication response: %r' % e)
             else:
                 if 'RelayState' in request.POST and utils.is_nonnull(request.POST['RelayState']):
                     login.msgRelayState = request.POST['RelayState']
                 return self.sso_success(request, login)
             return self.failure(request, reason=idp_message, status_codes=status_codes)
                 return self.failure_status_is_not_success(request, login, _('Login'))
             except Exception as e:
                 return self.failure(request, _('Technical error'), exception=e)
             if 'RelayState' in request.POST and utils.is_nonnull(request.POST['RelayState']):
                 login.msgRelayState = request.POST['RelayState']
             return self.sso_success(request, login)
         def get_attribute_value(self, attribute, attribute_value):
             # check attribute_value contains only text
-...
             return response
         def continue_sso_artifact(self, request, method):
             idp_message = None
             status_codes = []
             if method == lasso.HTTP_METHOD_ARTIFACT_GET:
                 message = request.META['QUERY_STRING']
                 artifact = request.GET['SAMLart']
-...
                 login.msgRelayState = relay_state
             try:
                 login.initRequest(message, method)
             except lasso.ProfileInvalidArtifactError:
             except lasso.ProfileInvalidArtifactError as e:
                 self.log.warning('artifact is malformed %r', artifact)
                 return HttpResponseBadRequest('artifact is malformed %r' % artifact)
             except lasso.ServerProviderNotFoundError:
                 return self.failure(request, _('Artifact is invalid.'), exception=e)
             except lasso.ServerProviderNotFoundError as e:
                 self.log.warning('no entity id found for artifact %s', artifact)
                 return HttpResponseBadRequest('no entity id found for this artifact %r' % artifact)
                 return self.failure(
                     request,
                     _('the received message EntityID (%r) is unknown.') % login.remoteProviderId,
                     exception=e,
+                )
             idp = utils.get_idp(login.remoteProviderId)
             if not idp:
                 return HttpResponseBadRequest('entity id %r is unknown' % login.remoteProviderId)
                 return self.failure(
                     request,
                     _('the received message EntityID (%r) is unknown.') % login.remoteProviderId,
+                )
             verify_ssl_certificate = utils.get_setting(idp, 'VERIFY_SSL_CERTIFICATE')
             login.buildRequestMsg()
             try:
-...
+                )
             except RequestException as e:
                 self.log.warning('unable to reach %r: %s', login.msgUrl, e)
                 return self.failure(
                     request,
                     reason=_('IdP is temporarily down, please try again ' 'later.'),
                     status_codes=status_codes,
+                )
                 return self.failure(request, reason=_('IdP is temporarily down, please try again later.'))
             if result.status_code != 200:
                 self.log.warning(
                     'SAML authentication failed: IdP returned %s when given artifact: %r',
                     result.status_code,
                     result.content,
+                )
                 return self.failure(request, reason=idp_message, status_codes=status_codes)
                 return self.failure(request, reason=_('IdP is temporarily down, please try again later.'))
             self.log.info('Got SAML Artifact Response', extra={'saml_response': result.content})
             result.encoding = utils.get_xml_encoding(result.content)
-...
                 # artifact is invalid, idp returned no response
                 self.log.warning('ArtifactResolveResponse is empty: dead artifact %r', artifact)
                 return self.retry_login()
             except lasso.ProfileInvalidMsgError:
             except lasso.ProfileInvalidMsgError as e:
                 self.log.warning('ArtifactResolveResponse is malformed %r', result.content[:200])
                 if settings.DEBUG:
                     return HttpResponseBadRequest('ArtififactResolveResponse is malformed\n%r' % result.content)
                 else:
                     return HttpResponseBadRequest('ArtififactResolveResponse is malformed')
             except lasso.ProfileCannotVerifySignatureError:
                 self.log.warning(
                     'SAML authentication failed: signature validation failed for %r', login.remoteProviderId
                 return self.failure(
                     request,
                     _('Could not process the ArtifactResolveResponse'),
                     exception=e,
+                )
             except lasso.ParamError:
                 self.log.exception('lasso param error')
             except (
                 lasso.LoginStatusNotSuccessError,
                 lasso.ProfileStatusNotSuccessError,
                 lasso.ProfileRequestDeniedError,
             ):
                 status = login.response.status
                 a = status
                 while a.statusCode:
                     status_codes.append(a.statusCode.value)
                     a = a.statusCode
                 args = ['SAML authentication failed: status is not success codes: %r', status_codes]
                 if status.statusMessage:
                     idp_message = lasso_decode(status.statusMessage)
                     args[0] += ' message: %r'
                     args.append(status.statusMessage)
                 self.log.warning(*args)
             except lasso.Error as e:
                 self.log.exception('unexpected lasso error')
                 return HttpResponseBadRequest('error processing the authentication response: %r' % e)
                 return self.failure_status_is_not_success(request, login, _('Login'))
             except Exception as e:
                 return self.failure(
                     request,
                     _('Technical error'),
                     exception=e,
                     debug_details={'ArtifactResolveResponse_content': result.content},
+                )
             else:
                 return self.sso_success(request, login)
             return self.failure(request, reason=idp_message, status_codes=status_codes)
         def request_discovery_service(self, request, is_passive=False):
             return_url = request.build_absolute_uri()
-...
                 return self.request_discovery_service(request, is_passive=request.GET.get('passive') == '1')
             next_url = check_next_url(self.request, request.GET.get(REDIRECT_FIELD_NAME))
             idp = self.get_idp(request)
             entity_id = request.POST.get('entityID') or request.GET.get('entityID')
             idp = self.get_idp(entity_id)
             if not idp:
                 return HttpResponseBadRequest('no idp found')
                 return self.failure(request, _('No idp found for "%s"') % (entity_id or ''))
             self.profile = login = utils.create_login(request)
             self.log.debug('authenticating to %r', idp['ENTITY_ID'])
             try:
-...
                 self.add_login_hints(idp, authn_request, request=request, next_url=next_url or '/')
                 login.buildAuthnRequestMsg()
             except lasso.Error as e:
                 return HttpResponseBadRequest('error initializing the authentication request: %r' % e)
                 return self.failure(request, _('Could not initialize the authentication request'), exception=e)
             self.log.debug('sending authn request %r', authn_request.dump())
             self.log.debug('to url %r', login.msgUrl)
             return HttpResponseRedirect(login.msgUrl)
-...
             try:
                 logout.processRequestMsg(msg)
             except lasso.Error as e:
                 return HttpResponseBadRequest('error processing logout request: %r' % e)
                 return self.failure(request, _('Could not process the logout request'), exception=e)
             entity_id = force_text(logout.remoteProviderId)
             session_indexes = {force_text(sessionIndex) for sessionIndex in logout.request.sessionIndexes}
-...
             try:
                 logout.buildResponseMsg()
             except lasso.Error as e:
                 return HttpResponseBadRequest('error processing logout request: %r' % e)
                 return self.failure(request, _('Could not process the logout response'), exception=e)
             if logout.msgBody:
                 return HttpResponse(force_text(logout.msgBody), content_type='text/xml')
             else:
-...
             try:
                 logout.processResponseMsg(request.META['QUERY_STRING'])
             except lasso.ProfileStatusNotSuccessError:
                 self.show_message_status_is_not_success(logout, 'SAML logout failed')
                 return self.failure_status_is_not_success(request, logout, _('Logout'))
             except lasso.LogoutPartialLogoutError:
                 self.log.warning('partial logout')
             except lasso.Error as e:

+        )
         assert not relay_state
         assert url.endswith(reverse('mellon_login'))
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         response = app.post(
             reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state}, status=400
+        )
         assert 'User is not allowed to login' in response
         assert (
             "status is not success codes: ['urn:oasis:names:tc:SAML:2.0:status:Responder',\
      'urn:oasis:names:tc:SAML:2.0:status:RequestDenied']"
             in caplog.text
+        )
         assert 'User is not allowed to login' in caplog.text
     @pytest.mark.urls('urls_tests_template_base')
-...
         url, body, relay_state = idp.process_authn_request_redirect(
             response['Location'], auth_result=False, msg='User is not allowed to login'
+        )
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         response = app.post(
             reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state}, status=400
+        )
         assert 'Theme is ok' in response.text
         response = app.get(reverse('mellon_login'))
-...
         url, body, relay_state = idp.process_authn_request_redirect(
             response['Location'], auth_result=False, msg='User is not allowed to login'
+        )
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         response = app.post(
             reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state}, status=400
+        )
         assert 'Theme is ok' in response.text
         assert 'HOOK' in response.text
-...
         url, body, relay_state = idp.process_authn_request_redirect(
             response['Location'], auth_result=False, msg='User is not allowed to login'
+        )
         response = app.post(reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state})
         response = app.post(
             reverse('mellon_login'), params={'SAMLResponse': body, 'RelayState': relay_state}, status=400
+        )
         assert 'Theme is ok' not in response.text
-...
         assert 'SAMLart' in url
         acs_artifact_url = url.split('testserver', 1)[1]
         with HTTMock(idp.mock_artifact_resolver()):
             response = app.get(acs_artifact_url, params={'RelayState': relay_state})
             response = app.get(acs_artifact_url, params={'RelayState': relay_state}, status=400)
         assert (
             "status is not success codes: ['urn:oasis:names:tc:SAML:2.0:status:Responder',\
      'urn:oasis:names:tc:SAML:2.0:status:RequestDenied']"
-...
         # forget the artifact
         idp.artifact = ''
         with HTTMock(idp.mock_artifact_resolver()):
             response = app.get(acs_artifact_url)
             response = app.get(acs_artifact_url, status=400)
         # check cookie is deleted after failed retry
         # Py3-Dj111 variation

         private_settings.MELLON_IDENTITY_PROVIDERS = []
         response = client.get('/login/')
         assert response.status_code == 400
         assert b'no idp found' in response.content
         assert 'No idp found for' in response.content.decode()
     def test_sp_initiated_login_discovery_service(private_settings, client):
-...
         private_settings.MELLON_IDENTITY_PROVIDERS = []
         private_settings.MELLON_DISCOVERY_SERVICE_URL = 'https://disco'
         response = client.get('/login/?nodisco=1')
         assert response.status_code == 400
         assert b'no idp found' in response.content
         assert 'No idp found for' in response.content.decode()
     def test_sp_initiated_login(private_settings, client):
-...
+        )
     def test_malfortmed_artifact(private_settings, client, caplog):
     def test_malfortmed_artifact(private_settings, app, caplog):
         private_settings.MELLON_IDENTITY_PROVIDERS = [
+            {
                 'METADATA': open('tests/metadata.xml').read(),
+            }
+        ]
         response = client.get('/login/?SAMLart=xxx', status=400)
         assert response['Content-Type'] == 'text/plain'
         assert response['X-Content-Type-Options'] == 'nosniff'
         assert b'artifact is malformed' in response.content
         response = app.get('/login/?SAMLart=xxx', status=400)
         assert response['Content-Type'].split(';')[0] == 'text/html'
         assert 'Artifact is invalid' in response
         assert 'artifact is malformed' in caplog.text
-...
             fd.write('1')
         private_key.chmod(0o000)
         private_settings.MELLON_PRIVATE_KEY = str(private_key)
         response = app.get('/login/?next=%2Fwhatever')
         response = app.get('/login/?next=%2Fwhatever', status=400)
         assert 'Unable to initialize a SAML server object' in response
+    -

Projet

Général

Profil

Produits Entr'ouvert » django-mellon

0001-views-use-a-view-to-show-all-error-messages-56354.patch