Project

General

Profile

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

Benjamin Dauvergne, 12 January 2022 11:24 PM

Download (25.6 KB)

View differences: 

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(-)
mellon/templates/mellon/authentication_failed.html
<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 %}
mellon/views.py
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(
0,
_('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:
tests/test_sso_slo.py
)
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
tests/test_views.py
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