ynerant
/
django-cas-server
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update version to 0.6.3 

Bugs fixes
----------
* typos in README.rst
* w3c validation

Cleaning
--------
* Code factorisation (models.py, views.py)
* Usage of the documented API for models _meta in auth.DjangoAuthUser

Whats new
---------
* Add powered by footer
* set warn cookie using javascript if possible
* Unfold many to many attributes in auth.DjangoAuthUser attributes
* Add a github version badge
* documents templatetags
This commit is contained in:
Valentin Samir 2016-08-14 07:55:45 +02:00 committed by GitHub
parent 1a04b5af9d 84d0d267ad
commit 07a537b403
17 changed files with 369 additions and 255 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
README.rst
View File

@ -1,7 +1,7 @@
CAS Server
##########
|travis| |version| |lisence| |codacy| |coverage| |doc|
|travis| |coverage| |licence| |github_version| |pypi_version| |codacy| |doc|
CAS Server is a Django application implementing the `CAS Protocol 3.0 Specification
<https://apereo.github.io/cas/4.2.x/protocol/CAS-Protocol-Specification.html>`_.
@ -206,6 +206,7 @@ Template settings
templates. Set it to ``False`` to disable it.
* ``CAS_FAVICON_URL``: URL to the favicon (shortcut icon) used by the default templates.
Default is a key icon. Set it to ``False`` to disable it.
* ``CAS_SHOW_POWERED``: Set it to ``False`` to hide the powered by footer. The default is ``True``.
* ``CAS_COMPONENT_URLS``: URLs to css and javascript external components. It is a dictionnary
and it must have the five following keys: ``"bootstrap3_css"``, ``"bootstrap3_js"``,
``"html5shiv"``, ``"respond"``, ``"jquery"``. The default is::
@ -603,10 +604,13 @@ You could for example do as bellow :
.. |travis| image:: https://badges.genua.fr/travis/nitmir/django-cas-server/master.svg
:target: https://travis-ci.org/nitmir/django-cas-server
.. |version| image:: https://badges.genua.fr/pypi/v/django-cas-server.svg
.. |pypi_version| image:: https://badges.genua.fr/pypi/v/django-cas-server.svg
:target: https://pypi.python.org/pypi/django-cas-server
.. |lisence| image:: https://badges.genua.fr/pypi/l/django-cas-server.svg
.. |github_version| image:: https://badges.genua.fr/github/tag/nitmir/django-cas-server.svg?label=github
:target: https://github.com/nitmir/django-cas-server/releases/latest
.. |licence| image:: https://badges.genua.fr/pypi/l/django-cas-server.svg
:target: https://www.gnu.org/licenses/gpl-3.0.html
.. |codacy| image:: https://badges.genua.fr/codacy/grade/255c21623d6946ef8802fa7995b61366/master.svg

2
cas_server/__init__.py
View File

@ -11,7 +11,7 @@
"""A django CAS server application"""
#: version of the application
VERSION = '0.6.2'
VERSION = '0.6.3'
#: path the the application configuration class
default_app_config = 'cas_server.apps.CasAppConfig'

30
cas_server/auth.py
View File

@ -369,8 +369,34 @@ class DjangoAuthUser(AuthUser): # pragma: no cover
"""
if self.user:
attr = {}
for field in self.user._meta.fields:
attr[field.attname] = getattr(self.user, field.attname)
# _meta.get_fields() is from the new documented _meta interface in django 1.8
try:
field_names = [
field.attname for field in self.user._meta.get_fields()
if hasattr(field, "attname")
]
# backward compatibility with django 1.7
except AttributeError: # pragma: no cover (only used by django 1.7)
field_names = self.user._meta.get_all_field_names()
for name in field_names:
attr[name] = getattr(self.user, name)
# unfold user_permissions many to many relation
if 'user_permissions' in attr:
attr['user_permissions'] = [
(
u"%s.%s" % (
perm.content_type.model_class().__module__,
perm.content_type.model_class().__name__
),
perm.codename
) for perm in attr['user_permissions'].filter()
]
# unfold group many to many relation
if 'groups' in attr:
attr['groups'] = [group.name for group in attr['groups'].filter()]
return attr
else:
return {}

2
cas_server/default_settings.py
View File

@ -20,6 +20,8 @@ from importlib import import_module
CAS_LOGO_URL = static("cas_server/logo.png")
#: URL to the favicon (shortcut icon) used by the default templates. Default is a key icon.
CAS_FAVICON_URL = static("cas_server/favicon.ico")
#: Show the powered by footer if set to ``True``
CAS_SHOW_POWERED = True
#: URLs to css and javascript external components.
CAS_COMPONENT_URLS = {
"bootstrap3_css": "//maxcdn.bootstrapcdn.com/bootstrap/3.3.6/css/bootstrap.min.css",

175
cas_server/models.py
View File

@ -28,13 +28,38 @@ from datetime import timedelta
from concurrent.futures import ThreadPoolExecutor
from requests_futures.sessions import FuturesSession
import cas_server.utils as utils
from cas_server import utils
from . import VERSION
#: logger facility
logger = logging.getLogger(__name__)
class JsonAttributes(models.Model):
"""
Bases: :class:`django.db.models.Model`
A base class for models storing attributes as a json
"""
class Meta:
abstract = True
#: The attributes json encoded
_attributs = models.TextField(default=None, null=True, blank=True)
@property
def attributs(self):
"""The attributes"""
if self._attributs is not None:
return utils.json.loads(self._attributs)
@attributs.setter
def attributs(self, value):
"""attributs property setter"""
self._attributs = utils.json_encode(value)
@python_2_unicode_compatible
class FederatedIendityProvider(models.Model):
"""
@ -130,9 +155,9 @@ class FederatedIendityProvider(models.Model):
@python_2_unicode_compatible
class FederatedUser(models.Model):
class FederatedUser(JsonAttributes):
"""
Bases: :class:`django.db.models.Model`
Bases: :class:`JsonAttributes`
A federated user as returner by a CAS provider (username and attributes)
"""
@ -142,8 +167,6 @@ class FederatedUser(models.Model):
username = models.CharField(max_length=124)
#: A foreign key to :class:`FederatedIendityProvider`
provider = models.ForeignKey(FederatedIendityProvider, on_delete=models.CASCADE)
#: The user attributes json encoded
_attributs = models.TextField(default=None, null=True, blank=True)
#: The last ticket used to authenticate :attr:`username` against :attr:`provider`
ticket = models.CharField(max_length=255)
#: Last update timespampt. Usually, the last time :attr:`ticket` has been set.
@ -152,17 +175,6 @@ class FederatedUser(models.Model):
def __str__(self):
return self.federated_username
@property
def attributs(self):
"""The user attributes returned by the CAS backend on successful ticket validation"""
if self._attributs is not None:
return utils.json.loads(self._attributs)
@attributs.setter
def attributs(self, value):
"""attributs property setter"""
self._attributs = utils.json_encode(value)
@property
def federated_username(self):
"""The federated username with a suffix for the current :class:`FederatedUser`."""
@ -290,35 +302,23 @@ class User(models.Model):
:param request: The current django HttpRequest to display possible failure to the user.
:type request: :class:`django.http.HttpRequest` or :obj:`NoneType<types.NoneType>`
"""
async_list = []
session = FuturesSession(
executor=ThreadPoolExecutor(max_workers=settings.CAS_SLO_MAX_PARALLEL_REQUESTS)
)
# first invalidate all Tickets
ticket_classes = [ProxyGrantingTicket, ServiceTicket, ProxyTicket]
for ticket_class in ticket_classes:
queryset = ticket_class.objects.filter(user=self)
for ticket in queryset:
ticket.logout(session, async_list)
queryset.delete()
for future in async_list:
if future: # pragma: no branch (should always be true)
try:
future.result()
except Exception as error:
logger.warning(
"Error during SLO for user %s: %s" % (
self.username,
error
)
)
if request is not None:
error = utils.unpack_nested_exception(error)
messages.add_message(
request,
messages.WARNING,
_(u'Error during service logout %s') % error
)
for error in Ticket.send_slos(
[ticket_class.objects.filter(user=self) for ticket_class in ticket_classes]
):
logger.warning(
"Error during SLO for user %s: %s" % (
self.username,
error
)
)
if request is not None:
error = utils.unpack_nested_exception(error)
messages.add_message(
request,
messages.WARNING,
_(u'Error during service logout %s') % error
)
def get_ticket(self, ticket_class, service, service_pattern, renew):
"""
@ -544,20 +544,13 @@ class ServicePattern(models.Model):
if re.match(filtre.pattern, str(value)):
break
else:
bad_filter = (filtre.pattern, filtre.attribut, user.attributs.get(filtre.attribut))
logger.warning(
"User constraint failed for %s, service %s: %s do not match %s %s." % (
user.username,
self.name,
filtre.pattern,
filtre.attribut,
user.attributs.get(filtre.attribut)
(user.username, self.name) + bad_filter
)
)
raise BadFilter('%s do not match %s %s' % (
filtre.pattern,
filtre.attribut,
user.attributs.get(filtre.attribut)
))
raise BadFilter('%s do not match %s %s' % bad_filter)
if self.user_field and not user.attributs.get(self.user_field):
logger.warning(
"Cannot use %s a loggin for user %s on service %s because it is absent" % (
@ -715,9 +708,9 @@ class ReplaceAttributValue(models.Model):
@python_2_unicode_compatible
class Ticket(models.Model):
class Ticket(JsonAttributes):
"""
Bases: :class:`django.db.models.Model`
Bases: :class:`JsonAttributes`
Generic class for a Ticket
"""
@ -725,8 +718,6 @@ class Ticket(models.Model):
abstract = True
#: ForeignKey to a :class:`User`.
user = models.ForeignKey(User, related_name="%(class)s")
#: The user attributes to transmit to the service json encoded
_attributs = models.TextField(default=None, null=True, blank=True)
#: A boolean. ``True`` if the ticket has been validated
validate = models.BooleanField(default=False)
#: The service url for the ticket
@ -749,17 +740,6 @@ class Ticket(models.Model):
#: requests.
TIMEOUT = settings.CAS_TICKET_TIMEOUT
@property
def attributs(self):
"""The user attributes to be transmited to the service on successful validation"""
if self._attributs is not None:
return utils.json.loads(self._attributs)
@attributs.setter
def attributs(self, value):
"""attributs property setter"""
self._attributs = utils.json_encode(value)
class DoesNotExist(Exception):
"""raised in :meth:`Ticket.get` then ticket prefix and ticket classes mismatch"""
pass
@ -767,6 +747,33 @@ class Ticket(models.Model):
def __str__(self):
return u"Ticket-%s" % self.pk
@staticmethod
def send_slos(queryset_list):
"""
Send SLO requests to each ticket of each queryset of ``queryset_list``
:param list queryset_list: A list a :class:`Ticket` queryset
:return: A list of possibly encoutered :class:`Exception`
:rtype: list
"""
# sending SLO to timed-out validated tickets
async_list = []
session = FuturesSession(
executor=ThreadPoolExecutor(max_workers=settings.CAS_SLO_MAX_PARALLEL_REQUESTS)
)
errors = []
for queryset in queryset_list:
for ticket in queryset:
ticket.logout(session, async_list)
queryset.delete()
for future in async_list:
if future: # pragma: no branch (should always be true)
try:
future.result()
except Exception as error:
errors.append(error)
return errors
@classmethod
def clean_old_entries(cls):
"""Remove old ticket and send SLO to timed-out services"""
@ -779,25 +786,12 @@ class Ticket(models.Model):
Q(creation__lt=(timezone.now() - timedelta(seconds=cls.VALIDITY)))
)
).delete()
# sending SLO to timed-out validated tickets
async_list = []
session = FuturesSession(
executor=ThreadPoolExecutor(max_workers=settings.CAS_SLO_MAX_PARALLEL_REQUESTS)
)
queryset = cls.objects.filter(
creation__lt=(timezone.now() - timedelta(seconds=cls.TIMEOUT))
)
for ticket in queryset:
ticket.logout(session, async_list)
queryset.delete()
for future in async_list:
if future: # pragma: no branch (should always be true)
try:
future.result()
except Exception as error:
logger.warning("Error durring SLO %s" % error)
sys.stderr.write("%r\n" % error)
for error in cls.send_slos([queryset]):
logger.warning("Error durring SLO %s" % error)
sys.stderr.write("%r\n" % error)
def logout(self, session, async_list=None):
"""Send a SLO request to the ticket service"""
@ -811,16 +805,7 @@ class Ticket(models.Model):
self.user.username
)
)
xml = u"""<samlp:LogoutRequest xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol"
ID="%(id)s" Version="2.0" IssueInstant="%(datetime)s">
<saml:NameID xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"></saml:NameID>
<samlp:SessionIndex>%(ticket)s</samlp:SessionIndex>
</samlp:LogoutRequest>""" % \
{
'id': utils.gen_saml_id(),
'datetime': timezone.now().isoformat(),
'ticket': self.value
}
xml = utils.logout_request(self.value)
if self.service_pattern.single_log_out_callback:
url = self.service_pattern.single_log_out_callback
else:

27
cas_server/static/cas_server/alert-version.js
View File

@ -1,27 +0,0 @@
function alert_version(last_version){
jQuery(function( $ ){
$("#alert-version").click(function( e ){
e.preventDefault();
var date = new Date();
date.setTime(date.getTime()+(10*365*24*60*60*1000));
var expires = "; expires="+date.toGMTString();
document.cookie = "cas-alert-version=" + last_version + expires + "; path=/";
});
var nameEQ="cas-alert-version=";
var ca = document.cookie.split(";");
var value;
for(var i=0;i < ca.length;i++) {
var c = ca[i];
while(c.charAt(0) === " "){
c = c.substring(1,c.length);
}
if(c.indexOf(nameEQ) === 0){
value = c.substring(nameEQ.length,c.length);
}
}
if(value === last_version){
$("#alert-version").parent().hide();
}
});
}

44
cas_server/static/cas_server/functions.js Normal file
View File

@ -0,0 +1,44 @@
function createCookie(name, value, days){
var expires;
var date;
if(days){
date = new Date();
date.setTime(date.getTime()+(days*24*60*60*1000));
expires = "; expires="+date.toGMTString();
}
else{
expires = "";
}
document.cookie = name + "=" + value + expires + "; path=/";
}
function readCookie(name){
var nameEQ = name + "=";
var ca = document.cookie.split(";");
for(var i=0;i < ca.length;i++) {
var c = ca[i];
while (c.charAt(0) === " "){
c = c.substring(1,c.length);
}
if (c.indexOf(nameEQ) === 0){
return c.substring(nameEQ.length,c.length);
}
}
return null;
}
function eraseCookie(name) {
createCookie(name,"",-1);
}
function alert_version(last_version){
jQuery(function( $ ){
$("#alert-version").click(function( e ){
e.preventDefault();
createCookie("cas-alert-version", last_version, 10*365);
});
if(readCookie("cas-alert-version") === last_version){
$("#alert-version").parent().hide();
}
});
}

21
cas_server/static/cas_server/styles.css
View File

@ -1,6 +1,9 @@
html, body {
height: 100%;
}
body {
padding-top: 40px;
padding-bottom: 40px;
padding-bottom: 0;
background-color: #eee;
}
@ -41,6 +44,22 @@ body {
width:110px;
}
/* Wrapper for page content to push down footer */
#wrap {
min-height: 100%;
height: auto;
height: 100%;
/* Negative indent footer by it's height */
margin: 0 auto -40px;
}
#footer {
height: 40px;
text-align: center;
}
#footer p {
padding-top: 10px;
}
@media screen and (max-width: 680px) {
#app-name {
margin: 0;

35
cas_server/templates/cas_server/base.html
View File

@ -1,6 +1,4 @@
{% load i18n %}
{% load staticfiles %}
<!DOCTYPE html>
{% load i18n %}{% load staticfiles %}<!DOCTYPE html>
<html{% if request.LANGUAGE_CODE %} lang="{{ request.LANGUAGE_CODE }}"{% endif %}>
<head>
<meta charset="utf-8">
@ -18,12 +16,13 @@
<link href="{% static "cas_server/styles.css" %}" rel="stylesheet">
</head>
<body>
<div id="wrap">
<div class="container">
{% if auto_submit %}<noscript>{% endif %}
<div class="row">
<div class="col-lg-12 col-md-12 col-sm-12 col-xs-12">
<h1 id="app-name">
{% if settings.CAS_LOGO_URL %}<img src="{{settings.CAS_LOGO_URL}}"></img> {% endif %}
{% if settings.CAS_LOGO_URL %}<img src="{{settings.CAS_LOGO_URL}}" alt="cas-logo" />{% endif %}
{% trans "Central Authentication Service" %}</h1>
</div>
</div>
@ -53,7 +52,7 @@
class="alert alert-danger"
{% endif %}
{% endspaceless %}>
{{ message }}
{{message|safe}}
</div>
{% endfor %}
{% if auto_submit %}</noscript>{% endif %}
@ -62,11 +61,25 @@
<div class="col-lg-3 col-md-3 col-sm-2 col-xs-0"></div>
</div>
</div> <!-- /container -->
<script src="{{settings.CAS_COMPONENT_URLS.jquery}}"></script>
<script src="{{settings.CAS_COMPONENT_URLS.bootstrap3_js}}"></script>
{% if settings.CAS_NEW_VERSION_HTML_WARNING and upgrade_available %}
<script src="{% static "cas_server/alert-version.js" %}"></script>
<script>alert_version("{{LAST_VERSION}}")</script>
{% endif %}
</div>
<div style="clear: both;"></div>
{% if settings.CAS_SHOW_POWERED %}
<div id="footer">
<p><a class="text-muted" href="https://pypi.python.org/pypi/django-cas-server">django-cas-server powered</a></p>
</div>
{% endif %}
<script src="{{settings.CAS_COMPONENT_URLS.jquery}}"></script>
<script src="{{settings.CAS_COMPONENT_URLS.bootstrap3_js}}"></script>
<script src="{% static "cas_server/functions.js" %}"></script>
{% if settings.CAS_NEW_VERSION_HTML_WARNING and upgrade_available %}
<script type="text/javascript">alert_version("{{LAST_VERSION}}")</script>
{% endif %}
{% block javascript %}{% endblock %}
</body>
</html>
<!--
Powered by django-cas-server version {{VERSION}}
Pypi: https://pypi.python.org/pypi/django-cas-server
github: https://github.com/nitmir/django-cas-server
-->

2
cas_server/templates/cas_server/form.html
View File

@ -14,7 +14,7 @@
{% endif %}"
{% endspaceless %}>{% spaceless %}
{% if field|is_checkbox %}
<div class="checkbox"><label for="{{field.auto_id}}">{{field}}{{field.label}}</label>
<div class="checkbox"><label for="{{field.auto_id}}">{{field}}{{field.label}}</label></div>
{% else %}
<label class="control-label" for="{{field.auto_id}}">{{field.label}}</label>
{{field}}

17
cas_server/templates/cas_server/login.html
View File

@ -14,10 +14,17 @@
<button class="btn btn-primary btn-block btn-lg" type="submit">{% trans "Login" %}</button>
{% if auto_submit %}</noscript>{% endif %}
</form>
{% if auto_submit %}
<script type="text/javascript">
document.getElementById('login_form').submit(); // SUBMIT FORM
</script>
{% endif %}
{% endblock %}
{% block javascript %}<script type="text/javascript">
jQuery(function( $ ){
$("#id_warn").click(function(e){
if($("#id_warn").is(':checked')){
createCookie("warn", "on", 10 * 365);
} else {
eraseCookie("warn");
}
});
});{% if auto_submit %}
document.getElementById('login_form').submit(); // SUBMIT FORM{% endif %}
</script>{% endblock %}

4
cas_server/tests/test_federate.py
View File

@ -261,7 +261,7 @@ class FederateAuthLoginLogoutTestCase(
# SLO for an unkown ticket should do nothing
response = client.post(
"/federate/%s" % provider.suffix,
{'logoutRequest': tests_utils.logout_request(utils.gen_st())}
{'logoutRequest': utils.logout_request(utils.gen_st())}
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content, b"ok")
@ -288,7 +288,7 @@ class FederateAuthLoginLogoutTestCase(
# 3 or 'CAS_2_SAML_1_0'
response = client.post(
"/federate/%s" % provider.suffix,
{'logoutRequest': tests_utils.logout_request(ticket)}
{'logoutRequest': utils.logout_request(ticket)}
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content, b"ok")

32
cas_server/tests/test_templatetags.py Normal file
View File

@ -0,0 +1,32 @@
# -*- 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) 2016 Valentin Samir
"""tests for the customs template tags"""
from django.test import TestCase
from cas_server import forms
from cas_server.templatetags import cas_server
class TemplateTagsTestCase(TestCase):
"""tests for the customs template tags"""
def test_is_checkbox(self):
"""test for the template filter is_checkbox"""
form = forms.UserCredential()
self.assertFalse(cas_server.is_checkbox(form["username"]))
self.assertTrue(cas_server.is_checkbox(form["warn"]))
def test_is_hidden(self):
"""test for the template filter is_hidden"""
form = forms.UserCredential()
self.assertFalse(cas_server.is_hidden(form["username"]))
self.assertTrue(cas_server.is_hidden(form["lt"]))

22
cas_server/tests/utils.py
View File

@ -115,8 +115,8 @@ def get_validated_ticket(service):
client = Client()
response = client.get('/validate', {'ticket': ticket.value, 'service': service})
assert (response.status_code == 200)
assert (response.content == b'yes\ntest\n')
assert response.status_code == 200
assert response.content == b'yes\ntest\n'
ticket = models.ServiceTicket.objects.get(value=ticket.value)
return (auth_client, ticket)
@ -222,6 +222,10 @@ class Http404Handler(HttpParamsHandler):
class DummyCAS(BaseHTTPServer.BaseHTTPRequestHandler):
"""A dummy CAS that validate for only one (service, ticket) used in federated mode tests"""
#: dict of the last receive GET parameters
params = None
def test_params(self):
"""check that internal and provided (service, ticket) matches"""
if (
@ -340,17 +344,3 @@ class DummyCAS(BaseHTTPServer.BaseHTTPRequestHandler):
httpd_thread.daemon = True
httpd_thread.start()
return (httpd, host, port)
def logout_request(ticket):
"""build a SLO request XML, ready to be send"""
return u"""<samlp:LogoutRequest xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol"
ID="%(id)s" Version="2.0" IssueInstant="%(datetime)s">
<saml:NameID xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"></saml:NameID>
<samlp:SessionIndex>%(ticket)s</samlp:SessionIndex>
</samlp:LogoutRequest>""" % \
{
'id': utils.gen_saml_id(),
'datetime': timezone.now().isoformat(),
'ticket': ticket
}

20
cas_server/utils.py
View File

@ -17,6 +17,7 @@ from django.http import HttpResponseRedirect, HttpResponse
from django.contrib import messages
from django.contrib.messages import constants as DEFAULT_MESSAGE_LEVELS
from django.core.serializers.json import DjangoJSONEncoder
from django.utils import timezone
import random
import string
@ -680,3 +681,22 @@ def dictfetchall(cursor):
dict(zip(columns, row))
for row in cursor.fetchall()
]
def logout_request(ticket):
"""
Forge a SLO logout request
:param unicode ticket: A ticket value
:return: A SLO XML body request
:rtype: unicode
"""
return u"""<samlp:LogoutRequest xmlns:samlp="urn:oasis:names:tc:SAML:2.0:protocol"
ID="%(id)s" Version="2.0" IssueInstant="%(datetime)s">
<saml:NameID xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"></saml:NameID>
<samlp:SessionIndex>%(ticket)s</samlp:SessionIndex>
</samlp:LogoutRequest>""" % {
'id': gen_saml_id(),
'datetime': timezone.now().isoformat(),
'ticket': ticket
}

179
cas_server/views.py
View File

@ -45,6 +45,7 @@ logger = logging.getLogger(__name__)
class LogoutMixin(object):
"""destroy CAS session utils"""
def logout(self, all_session=False):
"""
effectively destroy a CAS session
@ -63,43 +64,59 @@ class LogoutMixin(object):
logger.info("Logging out user %s from all of they sessions." % username)
else:
logger.info("Logging out user %s." % username)
# logout the user from the current session
users = []
# try to get the user from the current session
try:
user = models.User.objects.get(
username=username,
session_key=self.request.session.session_key
users.append(
models.User.objects.get(
username=username,
session_key=self.request.session.session_key
)
)
# flush the session
except models.User.DoesNotExist:
# if user not found in database, flush the session anyway
self.request.session.flush()
# If all_session is set, search all of the user sessions
if all_session:
users.extend(
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
for user in users:
# get the user session
session = SessionStore(session_key=user.session_key)
# flush the session
session.flush()
# send SLO requests
user.logout(self.request)
# delete the user
user.delete()
# increment the destroyed session counter
session_nb += 1
except models.User.DoesNotExist:
# if user not found in database, flush the session anyway
self.request.session.flush()
# If all_session is set logout user from alternative sessions
if all_session:
# Iterate over all user sessions
for user in models.User.objects.filter(username=username):
# get the user session
session = SessionStore(session_key=user.session_key)
# flush the session
session.flush()
# send SLO requests
user.logout(self.request)
# delete the user
user.delete()
# increment the destroyed session counter
session_nb += 1
if username:
logger.info("User %s logged out" % username)
return session_nb
class CsrfExemptView(View):
"""base class for csrf exempt class views"""
@method_decorator(csrf_exempt) # csrf is disabled for allowing SLO requests reception
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
"""
return super(CsrfExemptView, self).dispatch(request, *args, **kwargs)
class LogoutView(View, LogoutMixin):
"""destroy CAS session (logout) view"""
@ -210,17 +227,15 @@ class LogoutView(View, LogoutMixin):
)
class FederateAuth(View):
"""view to authenticated user agains a backend CAS then CAS_FEDERATE is True"""
class FederateAuth(CsrfExemptView):
"""
view to authenticated user agains a backend CAS then CAS_FEDERATE is True
@method_decorator(csrf_exempt) # csrf is disabled for allowing SLO requests reception
def dispatch(self, request, *args, **kwargs):
"""
dispatch different http request to the methods of the same name
csrf is disabled for allowing SLO requests reception.
"""
:param django.http.HttpRequest request: The current request object
"""
return super(FederateAuth, self).dispatch(request, *args, **kwargs)
#: current URL used as service URL by the CAS client
service_url = None
def get_cas_client(self, request, provider, renew=False):
"""
@ -285,7 +300,7 @@ class FederateAuth(View):
"""
method called on GET request
:param django.http.HttpRequest request: The current request object
: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
@ -923,18 +938,13 @@ class LoginView(View, LogoutMixin):
return self.not_authenticated()
class Auth(View):
"""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.
@method_decorator(csrf_exempt)
def dispatch(self, request, *args, **kwargs):
"""
dispatch requests based on method GET, POST, ...
class Auth(CsrfExemptView):
"""
A simple view to validate username/password/service tuple
:param django.http.HttpRequest request: The current request object
"""
return super(Auth, self).dispatch(request, *args, **kwargs)
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.
"""
@staticmethod
def post(request):
@ -1041,8 +1051,9 @@ class Validate(View):
@python_2_unicode_compatible
class ValidateError(Exception):
"""handle service validation error"""
class ValidationBaseError(Exception):
"""Base class for both saml and cas validation error"""
#: The error code
code = None
#: The error message
@ -1051,7 +1062,7 @@ class ValidateError(Exception):
def __init__(self, code, msg=""):
self.code = code
self.msg = msg
super(ValidateError, self).__init__(code)
super(ValidationBaseError, self).__init__(code)
def __str__(self):
return u"%s" % self.msg
@ -1066,12 +1077,27 @@ class ValidateError(Exception):
"""
return render(
request,
"cas_server/serviceValidateError.xml",
{'code': self.code, 'msg': self.msg},
content_type="text/xml; charset=utf-8"
self.template,
self.context(), content_type="text/xml; charset=utf-8"
)
class ValidateError(ValidationBaseError):
"""handle service validation error"""
#: template to be render for the error
template = "cas_server/serviceValidateError.xml"
def context(self):
"""
content to use to render :attr:`template`
:return: A dictionary to contextualize :attr:`template`
:rtype: dict
"""
return {'code': self.code, 'msg': self.msg}
class ValidateService(View):
"""service ticket validation [CAS 2.0] and [CAS 3.0]"""
#: Current :class:`django.http.HttpRequest` object
@ -1333,59 +1359,32 @@ class Proxy(View):
)
@python_2_unicode_compatible
class SamlValidateError(Exception):
class SamlValidateError(ValidationBaseError):
"""handle saml validation error"""
#: The error code
code = None
#: The error message
msg = None
def __init__(self, code, msg=""):
self.code = code
self.msg = msg
super(SamlValidateError, self).__init__(code)
#: template to be render for the error
template = "cas_server/samlValidateError.xml"
def __str__(self):
return u"%s" % self.msg
def render(self, request):
def context(self):
"""
render the error template for the exception
:param django.http.HttpRequest request: The current request object:
:return: the rendered ``cas_server/samlValidateError.xml`` template
:rtype: django.http.HttpResponse
:return: A dictionary to contextualize :attr:`template`
:rtype: dict
"""
return render(
request,
"cas_server/samlValidateError.xml",
{
'code': self.code,
'msg': self.msg,
'IssueInstant': timezone.now().isoformat(),
'ResponseID': utils.gen_saml_id()
},
content_type="text/xml; charset=utf-8"
)
return {
'code': self.code,
'msg': self.msg,
'IssueInstant': timezone.now().isoformat(),
'ResponseID': utils.gen_saml_id()
}
class SamlValidate(View):
class SamlValidate(CsrfExemptView):
"""SAML ticket validation"""
request = None
target = None
ticket = None
root = None
@method_decorator(csrf_exempt)
def dispatch(self, request, *args, **kwargs):
"""
dispatch requests based on method GET, POST, ...
:param django.http.HttpRequest request: The current request object
"""
return super(SamlValidate, self).dispatch(request, *args, **kwargs)
def post(self, request):
"""
methode called on POST request on this view

2
setup.py
View File

@ -62,7 +62,7 @@ if __name__ == '__main__':
'lxml >= 3.4', 'six >= 1'
],
url="https://github.com/nitmir/django-cas-server",
download_url="https://github.com/nitmir/django-cas-server/releases",
download_url="https://github.com/nitmir/django-cas-server/releases/latest",
zip_safe=False,
setup_requires=['pytest-runner'],
tests_require=['pytest', 'pytest-django', 'pytest-pythonpath', 'pytest-warnings', 'mock>=1'],