Coverage for cas_server/views.py : 99%
Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
|
# -*- coding: utf-8 -*- # This program is distributed in the hope that it will be useful, but WITHOUT # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS # FOR A PARTICULAR PURPOSE. See the GNU General Public License version 3 for # more details. # # You should have received a copy of the GNU General Public License version 3 # along with this program; if not, write to the Free Software Foundation, Inc., 51 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. # # (c) 2015-2016 Valentin Samir
except ImportError: from django.core.urlresolvers import reverse
"""destroy CAS session utils"""
""" effectively destroy a CAS session
:param boolean all_session: If ``True`` destroy all the user sessions, otherwise destroy the current user session. :return: The number of destroyed sessions :rtype: int """ # initialize the counter of the number of destroyed sesisons # save the current user username before flushing the session else: # try to get the user from the current session models.User.objects.get( username=username, session_key=self.request.session.session_key ) ) # if user not found in database, flush the session anyway
# If all_session is set, search all of the user sessions models.User.objects.filter( username=username ).exclude( session_key=self.request.session.session_key ) )
# Iterate over all user sessions that have to be logged out # get the user session # flush the session # send SLO requests # delete the user # increment the destroyed session counter
"""base class for csrf exempt class views"""
def dispatch(self, request, *args, **kwargs): """ dispatch different http request to the methods of the same name
:param django.http.HttpRequest request: The current request object """
"""destroy CAS session (logout) view"""
#: current :class:`django.http.HttpRequest` object #: service GET parameter #: url GET paramet #: ``True`` if the HTTP_X_AJAX http header is sent and ``settings.CAS_ENABLE_AJAX_AUTH`` #: is ``True``, ``False`` otherwise.
""" Initialize the :class:`LogoutView` attributes on GET request
:param django.http.HttpRequest request: The current request object """
""" method called on GET request on this view
:param django.http.HttpRequest request: The current request object """ # initialize the class attributes # if CAS federation mode is enable, bakup the provider before flushing the sessions self.request.session.get("username") ) # if CAS federation mode is enable, redirect to user CAS logout page, appending the # current querystring # if service is set, redirect to service after logout # if service is not set but url is set, redirect to url after logout else: # build logout message depending of the number of sessions the user logs out "<h3>Logout successful</h3>" "You have successfully logged out from the Central Authentication Service. " "For security reasons, close your web browser." )) "<h3>Logout successful</h3>" "You have successfully logged out from %d sessions of the Central " "Authentication Service. " "For security reasons, close your web browser." ) % session_nb) else: "<h3>Logout successful</h3>" "You were already logged out from the Central Authentication Service. " "For security reasons, close your web browser." ))
# depending of settings, redirect to the login page with a logout message or display # the logout page. The default is to display tge logout page. 'status': 'success', 'detail': 'logout', 'url': url, 'session_nb': session_nb } else: else: else: request, settings.CAS_LOGOUT_TEMPLATE, utils.context({'logout_msg': logout_msg}) )
""" view to authenticated user against a backend CAS then CAS_FEDERATE is True
csrf is disabled for allowing SLO requests reception. """
#: current URL used as service URL by the CAS client
""" return a CAS client object matching provider
:param django.http.HttpRequest request: The current request object :param cas_server.models.FederatedIendityProvider provider: the user identity provider :return: The user CAS client object :rtype: :class:`federate.CASFederateValidateUser <cas_server.federate.CASFederateValidateUser>` """ # compute the current url, ignoring ticket dans provider GET parameters
""" method called on POST request
:param django.http.HttpRequest request: The current request object :param unicode provider: Optional parameter. The user provider suffix. """ # if settings.CAS_FEDERATE is not True redirect to the login page # POST with a provider suffix, this is probably an SLO request. csrf is disabled for # allowing SLO requests reception # else, a User is trying to log in using an identity provider # Manually checking for csrf to protect the code below if reason is not None: # pragma: no cover (csrf checks are disabled during tests) return reason # Failed the test, stop here. request.POST, ignore={"provider", "csrfmiddlewaretoken", "ticket", "lt"} ) "cas_server:federateAuth", kwargs=dict(provider=form.cleaned_data["provider"].suffix), params=params ) else:
""" method called on GET request
:param django.http.HttpRequestself. request: The current request object :param unicode provider: Optional parameter. The user provider suffix. """ # if settings.CAS_FEDERATE is not True redirect to the login page # Is the user is already authenticated, no need to request authentication to the user # identity provider. # get the identity provider from its suffix # get a CAS client for the user identity provider # if no ticket submited, redirect to the identity provider CAS login page else: # if the ticket validation succeed "Got a valid ticket for %s from %s" % ( auth.username, auth.provider.server_url ) ) auth.federated_username, request.session.session_key, ticket ) # redirect to the the login page for the user to become authenticated # thanks to the `federate_username` and `federate_ticket` session parameters # If the user has checked "remember my identity provider" store it in a # cookie response, "remember_provider", provider.suffix, max_age ) # else redirect to the identity provider CAS login page else: ( "Got an invalid ticket %s from %s for service %s. " "Retrying authentication" ) % ( ticket, auth.provider.server_url, self.service_url ) ) # both xml.etree.ElementTree and lxml.etree exceptions inherit from SyntaxError request, messages.ERROR, _( u"Invalid response from your identity provider CAS upon " u"ticket %(ticket)s validation: %(error)r" ) % {'ticket': ticket, 'error': error} ) # if the identity provider is not found, redirect to the login page
"""credential requestor / acceptor"""
# pylint: disable=too-many-instance-attributes # Nine is reasonable in this case.
#: The current :class:`models.User<cas_server.models.User>` object #: The form to display to the user
#: current :class:`django.http.HttpRequest` object #: service GET/POST parameter #: ``True`` if renew GET/POST parameter is present and not "False" #: the warn GET/POST parameter #: the gateway GET/POST parameter #: the method GET/POST parameter
#: ``True`` if the HTTP_X_AJAX http header is sent and ``settings.CAS_ENABLE_AJAX_AUTH`` #: is ``True``, ``False`` otherwise.
#: ``True`` if the user has just authenticated #: ``True`` if renew GET/POST parameter is present and not "False"
#: The :class:`FederateAuth` transmited username (only used if ``settings.CAS_FEDERATE`` #: is ``True``) #: The :class:`FederateAuth` transmited ticket (only used if ``settings.CAS_FEDERATE`` is #: ``True``)
""" Initialize POST received parameters
:param django.http.HttpRequest request: The current request object """ # in federated mode, the valdated indentity provider CAS ticket is used as password
"""Generate a new LoginTicket and add it to the list of valid LT for the user"""
""" Check is the POSTed LoginTicket is valid, if yes invalide it
:return: ``True`` if the LoginTicket is valid, ``False`` otherwise :rtype: bool """ # save LT for later check # generate a new LT (by posting the LT has been consumed) # check if send LT is valid else: # we need to redo the affectation for django to detect that the list has changed # and for its new value to be store in the session
""" method called on POST request on this view
:param django.http.HttpRequest request: The current request object """ # initialize class parameters # process the POST request self.request, messages.ERROR, _(u"Invalid login ticket, please try to log in again") ) # On successful login, update the :class:`models.User<cas_server.models.User>` ``date`` # attribute by saving it. (``auto_now=True``) username=self.request.session['username'], session_key=self.request.session.session_key )[0] # preserve valid LoginTickets from session flush # On login failure, flush the session # restore valid LoginTickets else: # pragma: no cover (should no happen) raise EnvironmentError("invalid output for LoginView.process_post") # call the GET/POST common part response, "warn", "on", 10 * 365 * 24 * 3600 ) else:
""" Analyse the POST request:
* check that the LoginTicket is valid * check that the user sumited credentials are valid
:return: * :attr:`INVALID_LOGIN_TICKET` if the POSTed LoginTicket is not valid * :attr:`USER_ALREADY_LOGGED` if the user is already logged and do no request reauthentication. * :attr:`USER_LOGIN_FAILURE` if the user is not logged or request for reauthentication and his credentials are not valid * :attr:`USER_LOGIN_OK` if the user is not logged or request for reauthentication and his credentials are valid :rtype: int """ # authentication request receive, initialize the form to use else: else:
""" Initialize GET received parameters
:param django.http.HttpRequest request: The current request object """ # here username and ticket are fetch from the session after a redirection from # FederateAuth.get
""" method called on GET request on this view
:param django.http.HttpRequest request: The current request object """ # initialize class parameters # process the GET request # call the GET/POST common part
""" Analyse the GET request
:return: * :attr:`USER_NOT_AUTHENTICATED` if the user is not authenticated or is requesting for authentication renewal * :attr:`USER_AUTHENTICATED` if the user is authenticated and is not requesting for authentication renewal :rtype: int """ # generate a new LT # authentication will be needed, initialize the form to use
""" Initialization of the good form depending of POST and GET parameters
:param django.http.QueryDict values: A POST or GET QueryDict """ 'service': self.service, 'method': self.method, 'warn': ( self.warn or self.request.session.get("warn") or self.request.COOKIES.get('warn') ), 'lt': self.request.session['lt'][-1], 'renew': self.renew } values, initial=form_initial ) else: else: values, initial=form_initial )
""" Perform login against a service
:return: * The rendering of the ``settings.CAS_WARN_TEMPLATE`` if the user asked to be warned before ticket emission and has not yep been warned. * The redirection to the service URL with a ticket GET parameter * The redirection to the service URL without a ticket if ticket generation failed and the :attr:`gateway` attribute is set * The rendering of the ``settings.CAS_LOGGED_TEMPLATE`` template with some error messages if the ticket generation failed (e.g: user not allowed). :rtype: django.http.HttpResponse """ # is the service allowed # is the current user allowed on this service # if the user has asked to be warned before any login to a service self.request, messages.WARNING, _(u"Authentication has been required by service %(name)s (%(url)s)") % {'name': service_pattern.name, 'url': self.service} ) else: 'service': self.service, 'renew': self.renew, 'gateway': self.gateway, 'method': self.method, 'warned': True, 'lt': self.request.session['lt'][-1] }) self.request, settings.CAS_WARN_TEMPLATE, utils.context({'form': warn_form}) ) else: # redirect, using method ? self.service, service_pattern, renew=self.renewed ) else: self.request, messages.ERROR, _(u'Service %(url)s not allowed.') % {'url': self.service} ) self.request, messages.ERROR, _(u"Username not allowed") ) self.request, messages.ERROR, _(u"User characteristics not allowed") ) self.request, messages.ERROR, _(u"The attribute %(field)s is needed to use" u" that service") % {'field': service_pattern.user_field} )
# if gateway is set and auth failed redirect to the service without authentication
self.request, settings.CAS_LOGGED_TEMPLATE, utils.context({'session': self.request.session}) ) else:
""" Processing authenticated users
:return: * The returned value of :meth:`service_login` if :attr:`service` is defined * The rendering of ``settings.CAS_LOGGED_TEMPLATE`` otherwise :rtype: django.http.HttpResponse """ # Try to get the current :class:`models.User<cas_server.models.User>` object for the current # session username=self.request.session.get("username"), session_key=self.request.session.session_key ) # if not found, flush the session and redirect to the login page "User %s seems authenticated but is not found in the database." % ( self.request.session.get("username"), ) ) "status": "error", "detail": "login required", "url": utils.reverse_params("cas_server:login", params=self.request.GET) } else:
# if login against a service # else display the logged template else: else: self.request, settings.CAS_LOGGED_TEMPLATE, utils.context({'session': self.request.session}) )
""" Processing non authenticated users
:return: * The rendering of ``settings.CAS_LOGIN_TEMPLATE`` with various messages depending of GET/POST parameters * The redirection to :class:`FederateAuth` if ``settings.CAS_FEDERATE`` is ``True`` and the "remember my identity provider" cookie is found :rtype: django.http.HttpResponse """ # clean messages before leaving django
self.request, messages.WARNING, _(u"Authentication renewal required by service %(name)s (%(url)s).") % {'name': service_pattern.name, 'url': self.service} ) else: self.request, messages.WARNING, _(u"Authentication required by service %(name)s (%(url)s).") % {'name': service_pattern.name, 'url': self.service} ) self.request, messages.ERROR, _(u'Service %s not allowed') % self.service ) "status": "error", "detail": "login required", "url": utils.reverse_params("cas_server:login", params=self.request.GET) } else: self.request, settings.CAS_LOGIN_TEMPLATE, utils.context({ 'form': self.form, 'auto_submit': True, 'post_url': reverse("cas_server:login") }) ) else: self.request.COOKIES.get('remember_provider') and FederatedIendityProvider.objects.filter( suffix=self.request.COOKIES['remember_provider'] ) ): "cas_server:federateAuth", params=params, kwargs=dict(provider=self.request.COOKIES['remember_provider']) ) else: # if user is authenticated and auth renewal is requested, redirect directly # to the user identity provider self.request.session.get("username") ) "cas_server:federateAuth", params=params, kwargs=dict(provider=user.provider.suffix) ) # Should normally not happen: if the user is logged, it exists in the # database. except FederatedUser.DoesNotExist: # pragma: no cover pass self.request, settings.CAS_LOGIN_TEMPLATE, utils.context({ 'form': self.form, 'post_url': reverse("cas_server:federateAuth") }) ) else: self.request, settings.CAS_LOGIN_TEMPLATE, utils.context({'form': self.form}) )
""" Common part execute uppon GET and POST request
:return: * The returned value of :meth:`authenticated` if the user is authenticated and not requesting for authentication or if the authentication has just been renewed * The returned value of :meth:`not_authenticated` otherwise :rtype: django.http.HttpResponse """ # if authenticated and successfully renewed authentication if needed else:
""" A simple view to validate username/password/service tuple
csrf is disable as it is intended to be used by programs. Security is assured by a shared secret between the programs dans django-cas-server. """
def post(request): """ method called on POST request on this view
:param django.http.HttpRequest request: The current request object :return: ``HttpResponse(u"yes\\n")`` if the POSTed tuple (username, password, service) if valid (i.e. (username, password) is valid dans username is allowed on service). ``HttpResponse(u"no\\n…")`` otherwise, with possibly an error message on the second line. :rtype: django.http.HttpResponse """
"no\nplease set CAS_AUTH_SHARED_SECRET", content_type="text/plain; charset=utf-8" ) request.POST, initial={ 'service': service, 'method': 'POST', 'warn': False } ) username=form.cleaned_data['username'], session_key=request.session.session_key )[0] # is the service allowed # is the current user allowed on this service else:
"""service ticket validation""" def get(request): """ method called on GET request on this view
:param django.http.HttpRequest request: The current request object :return: * ``HttpResponse("yes\\nusername")`` if submited (service, ticket) is valid * else ``HttpResponse("no\\n")`` :rtype: django.http.HttpResponse """ # store wanted GET parameters # service and ticket parameters are mandatory # search for the ticket, associated at service that is not yet validated but is # still valid "Validate: Service ticket %s validated, user %s authenticated on service %s" % ( ticket.value, ticket.user.username, ticket.service ) ) u"yes\n%s\n" % ticket.username(), content_type="text/plain; charset=utf-8" ) ( "Validate: Service ticket %s not found or " "already validated, auth to %s failed" ) % ( ticket, service ) ) else:
"""Base class for both saml and cas validation error"""
#: The error code #: The error message
def __str__(self): return u"%s" % self.msg
""" render the error template for the exception
:param django.http.HttpRequest request: The current request object: :return: the rendered ``cas_server/serviceValidateError.xml`` template :rtype: django.http.HttpResponse """ request, self.template, self.context(), content_type="text/xml; charset=utf-8" )
"""handle service validation error"""
#: template to be render for the error
""" content to use to render :attr:`template`
:return: A dictionary to contextualize :attr:`template` :rtype: dict """
"""service ticket validation [CAS 2.0] and [CAS 3.0]""" #: Current :class:`django.http.HttpRequest` object #: The service GET parameter #: the ticket GET parameter #: the pgtUrl GET parameter #: the renew GET parameter #: specify if ProxyTicket are allowed by the view. Hence we user the same view for #: ``/serviceValidate`` and ``/proxyValidate`` juste changing the parameter.
""" method called on GET request on this view
:param django.http.HttpRequest request: The current request object: :return: The rendering of ``cas_server/serviceValidate.xml`` if no errors is raised, the rendering or ``cas_server/serviceValidateError.xml`` otherwise. :rtype: django.http.HttpResponse """ # define the class parameters
# service and ticket parameter are mandatory u'INVALID_REQUEST', u"you must specify a service and a ticket" ).render(request) else: # search the ticket in the database # prepare template rendering context 'username': self.ticket.username(), 'attributes': self.ticket.attributs_flat(), 'proxies': proxies, 'auth_date': self.ticket.user.last_login.replace(microsecond=0).isoformat(), 'is_new_login': 'true' if self.ticket.renew else 'false' } # if pgtUrl is set, require https or localhost self.pgt_url.startswith("https://") or re.match(r"^http://(127\.0\.0\.1|localhost)(:[0-9]+)?(/.*)?$", self.pgt_url) ): else: "ValidateService: ticket %s validated for user %s on service %s." % ( self.ticket.value, self.ticket.user.username, self.ticket.service ) ) "ValidateService: User attributs are:\n%s" % ( pprint.pformat(self.ticket.attributs), ) ) request, "cas_server/serviceValidate.xml", params, content_type="text/xml; charset=utf-8" ) "ValidateService: validation error: %s %s" % (error.code, error.msg) )
""" fetch the ticket against the database and check its validity
:raises ValidateError: if the ticket is not found or not valid, potentially for that service :returns: A couple (ticket, proxies list) :rtype: :obj:`tuple` """ else:
""" Handle PGT request
:param dict params: A template context dict :raises ValidateError: if pgtUrl is invalid or if TLS validation of the pgtUrl fails :return: The rendering of ``cas_server/serviceValidate.xml``, using ``params`` :rtype: django.http.HttpResponse """ user=self.ticket.user, service=self.pgt_url, service_pattern=pattern, single_log_out=pattern.single_log_out ) else: ( "ValidateService: ticket %s validated for user %s on service %s. " "Proxy Granting Ticket transmited to %s." ) % ( self.ticket.value, self.ticket.user.username, self.ticket.service, self.pgt_url ) ) "ValidateService: User attributs are:\n%s" % ( pprint.pformat(self.ticket.attributs), ) ) self.request, "cas_server/serviceValidate.xml", params, content_type="text/xml; charset=utf-8" ) u'INVALID_PROXY_CALLBACK', u"%s: %s" % (type(error), str(error)) ) else: u'INVALID_PROXY_CALLBACK', u"callback url not allowed by configuration" ) u'INVALID_PROXY_CALLBACK', u'callback url not allowed by configuration' )
"""proxy ticket service"""
#: Current :class:`django.http.HttpRequest` object #: A ProxyGrantingTicket from the pgt GET parameter #: the targetService GET parameter
""" method called on GET request on this view
:param django.http.HttpRequest request: The current request object: :return: The returned value of :meth:`process_proxy` if no error is raised, else the rendering of ``cas_server/serviceValidateError.xml``. :rtype: django.http.HttpResponse """ # pgt and targetService parameters are mandatory else: u'INVALID_REQUEST', u"you must specify and pgt and targetService" )
""" handle PT request
:raises ValidateError: if the PGT is not found, or the target service not allowed or the user not allowed on the tardet service. :return: The rendering of ``cas_server/proxy.xml`` :rtype: django.http.HttpResponse """ # is the target service allowed # to get a proxy ticket require that the service allow it u'UNAUTHORIZED_SERVICE', u'the service %s does not allow proxy tickets' % self.target_service ) # is the proxy granting ticket valid # is the pgt user allowed on the target service ProxyTicket, self.target_service, pattern, renew=False ) "Proxy ticket created for user %s on service %s." % ( ticket.user.username, self.target_service ) ) self.request, "cas_server/proxy.xml", {'ticket': pticket.value}, content_type="text/xml; charset=utf-8" ) u'UNAUTHORIZED_USER', u'User %s not allowed on %s' % (ticket.user.username, self.target_service) )
"""handle saml validation error"""
#: template to be render for the error
""" :return: A dictionary to contextualize :attr:`template` :rtype: dict """ 'code': self.code, 'msg': self.msg, 'IssueInstant': timezone.now().isoformat(), 'ResponseID': utils.gen_saml_id() }
"""SAML ticket validation"""
""" method called on POST request on this view
:param django.http.HttpRequest request: The current request object :return: the rendering of ``cas_server/samlValidate.xml`` if no error is raised, else the rendering of ``cas_server/samlValidateError.xml``. :rtype: django.http.HttpResponse """ timedelta(seconds=self.ticket.VALIDITY)).isoformat() 'IssueInstant': timezone.now().isoformat(), 'expireInstant': expire_instant, 'Recipient': self.target, 'ResponseID': utils.gen_saml_id(), 'username': self.ticket.username(), 'attributes': self.ticket.attributs_flat(), 'auth_date': self.ticket.user.last_login.replace(microsecond=0).isoformat(), 'is_new_login': 'true' if self.ticket.renew else 'false'
} "SamlValidate: ticket %s validated for user %s on service %s." % ( self.ticket.value, self.ticket.user.username, self.ticket.service ) ) "SamlValidate: User attributes are:\n%s" % pprint.pformat(self.ticket.attributs) )
request, "cas_server/samlValidate.xml", params, content_type="text/xml; charset=utf-8" )
""" validate ticket from SAML XML body
:raises: SamlValidateError: if the ticket is not found or not valid, or if we fail to parse the posted XML. :return: a ticket object :rtype: :class:`models.Ticket<cas_server.models.Ticket>` """ u'AuthnFailed', u'TARGET %s does not match ticket service' % self.target ) u'AuthnFailed', u'ticket %s should begin with PT- or ST-' % ticket ) |