Łukasz Rekucki <lrekucki@gmail.com>
Marek Stepniowski <marek@stepniowski.com>
Fred Wenzel <fwenzel@mozilla.com>
+Sebastian Annies <Sebastian.Annies@castlabs.com>
recursive-include cas_provider/templates *.html
+recursive-include cas_provider/fixtures *
include AUTHORS.txt
include README.rst
include LICENSE
\ No newline at end of file
return ticket
def clean(self):
- super(LoginForm, self).clean(self)
+ AuthenticationForm.clean(self)
self.cleaned_data.get('lt').delete()
return self.cleaned_data
# Python <2.6 compatibility
from cgi import parse_qs
-
__all__ = ['ServiceTicket', 'LoginTicket']
-
class BaseTicket(models.Model):
ticket = models.CharField(_('ticket'), max_length=32)
created = models.DateTimeField(_('created'), auto_now=True)
parsed = urlparse.urlparse(self.service)
query = parse_qs(parsed.query)
query['ticket'] = [self.ticket]
- query = [ ((k, v) if len(v) > 1 else (k, v[0])) for k, v in query.iteritems()]
+ query = [((k, v) if len(v) > 1 else (k, v[0])) for k, v in query.iteritems()]
parsed = urlparse.ParseResult(parsed.scheme, parsed.netloc,
- parsed.path, parsed.params,
- urllib.urlencode(query), parsed.fragment)
+ parsed.path, parsed.params,
+ urllib.urlencode(query), parsed.fragment)
return parsed.geturl()
class LoginTicket(BaseTicket):
-
prefix = 'LT'
class Meta:
verbose_name = _('Login Ticket')
verbose_name_plural = _('Login Tickets')
+
+
+class ProxyGrantingTicket(BaseTicket):
+ serviceTicket = models.ForeignKey(ServiceTicket, null=True)
+ pgtiou = models.CharField(max_length=256, verbose_name=_('PGTiou'))
+ prefix = 'PGT'
+
+ def __init__(self, *args, **kwargs):
+ if 'pgtiou' not in kwargs:
+ kwargs['pgtiou'] = u"PGTIOU-%s" % (''.join(Random().sample(string.ascii_letters + string.digits, 50)))
+ super(ProxyGrantingTicket, self).__init__(*args, **kwargs)
+
+ class Meta:
+ verbose_name = _('Proxy Granting Ticket')
+ verbose_name_plural = _('Proxy Granting Tickets')
+
+
+class ProxyTicket(ServiceTicket):
+ proxyGrantingTicket = models.ForeignKey(ProxyGrantingTicket, verbose_name=_('Proxy Granting Ticket'))
+
+ prefix = 'PT'
+
+ class Meta:
+ verbose_name = _('Proxy Ticket')
+ verbose_name_plural = _('Proxy Tickets')
+
+
+class ProxyGrantingTicketIOU(BaseTicket):
+ proxyGrantingTicket = models.ForeignKey(ProxyGrantingTicket, verbose_name=_('Proxy Granting Ticket'))
+
+ prefix = 'PGTIOU'
+
+ class Meta:
+ verbose_name = _('Proxy Granting Ticket IOU')
+ verbose_name_plural = _('Proxy Granting Tickets IOU')
+
+
+import StringIO
+import urllib2
+from xml import etree
+from xml.etree import ElementTree
+from cas_provider.attribute_formatters import CAS, NSMAP
from cas_provider.models import ServiceTicket
-from cas_provider.views import _cas2_sucess_response, _cas2_error_response, \
- INVALID_TICKET
+from cas_provider.views import _cas2_sucess_response, INVALID_TICKET, _cas2_error_response
from django.contrib.auth.models import User
from django.core.urlresolvers import reverse
from django.test import TestCase
from urlparse import urlparse
from django.conf import settings
+import cas_provider
+
+def dummy_urlopen(url):
+ pass
class ViewsTest(TestCase):
- fixtures = ['cas_users.json', ]
+ fixtures = ['cas_users', ]
def setUp(self):
self.service = 'http://example.com/'
+ def test_successful_login_with_proxy(self):
+ urllib2.urlopen = dummy_urlopen # monkey patching urllib2.urlopen so that the testcase doesnt really opens a url
+ proxyTarget = "http://my.sweet.service"
+
+ response = self._login_user('root', '123')
+ response = self._validate_cas2(response, True, proxyTarget )
+
+ # Test: I'm acting as the service that will call another service
+ # Step 1: Get the proxy granting ticket
+ responseXml = ElementTree.parse(StringIO.StringIO(response.content))
+ auth_success = responseXml.find(CAS + 'authenticationSuccess', namespaces=NSMAP)
+ pgt = auth_success.find(CAS + "proxyGrantingTicket", namespaces=NSMAP)
+ user = auth_success.find(CAS + "user", namespaces=NSMAP)
+ self.assertEqual('root', user.text)
+ self.assertIsNotNone(pgt.text)
+ self.assertTrue(pgt.text.startswith('PGTIOU'))
+
+ #Step 2: Get the actual proxy ticket
+ proxyTicketResponse = self.client.get(reverse('proxy'), {'targetService': proxyTarget, 'pgt': pgt.text}, follow=False)
+ proxyTicketResponseXml = ElementTree.parse(StringIO.StringIO(proxyTicketResponse.content))
+ self.assertIsNotNone(proxyTicketResponseXml.find(CAS + "proxySuccess", namespaces=NSMAP))
+ self.assertIsNotNone(proxyTicketResponseXml.find(CAS + "proxySuccess/cas:proxyTicket", namespaces=NSMAP))
+ proxyTicket = proxyTicketResponseXml.find(CAS + "proxySuccess/cas:proxyTicket", namespaces=NSMAP);
+
+ #Step 3: I have the proxy ticket I can talk to some other backend service as the currently logged in user!
+ proxyValidateResponse = self.client.get(reverse('cas_proxy_validate'), {'ticket': proxyTicket.text, 'service': proxyTarget, 'pgtUrl': None})
+ proxyValidateResponseXml = ElementTree.parse(StringIO.StringIO(proxyValidateResponse.content))
+
+ auth_success_2 = proxyValidateResponseXml.find(CAS + 'authenticationSuccess', namespaces=NSMAP)
+ user_2 = auth_success.find(CAS + "user", namespaces=NSMAP)
+ proxies = auth_success.find(CAS + "proxies")
+ self.assertIsNotNone(auth_success_2)
+ self.assertEqual(user.text, user_2.text)
+ self.assertIsNotNone(proxies)
+
+
+ def test_successful_proxy_chaining(self):
+ self.assertFalse(True)
+
+ def test_successful_service_not_matching_in_request_to_proxy(self):
+ self.assertFalse(True)
+
+
def test_succeessful_login(self):
response = self._login_user('root', '123')
self._validate_cas1(response, True)
response = self._validate_cas2(response, True)
self.assertEqual(response.content, '''<cas:serviceResponse xmlns:cas="http://www.yale.edu/tp/cas">'''
- '''<cas:authenticationSuccess>'''
- '''<cas:user>editor</cas:user>'''
- '''<cas:attributes>'''
- '''<cas:attraStyle>Jasig</cas:attraStyle>'''
- '''<cas:group>editor</cas:group>'''
- '''<cas:is_staff>True</cas:is_staff>'''
- '''<cas:is_active>True</cas:is_active>'''
- '''<cas:email>editor@exapmle.com</cas:email>'''
- '''</cas:attributes>'''
- '''</cas:authenticationSuccess>'''
- '''</cas:serviceResponse>''')
+ '''<cas:authenticationSuccess>'''
+ '''<cas:user>editor</cas:user>'''
+ '''<cas:attributes>'''
+ '''<cas:attraStyle>Jasig</cas:attraStyle>'''
+ '''<cas:group>editor</cas:group>'''
+ '''<cas:is_staff>True</cas:is_staff>'''
+ '''<cas:is_active>True</cas:is_active>'''
+ '''<cas:email>editor@exapmle.com</cas:email>'''
+ '''</cas:attributes>'''
+ '''</cas:authenticationSuccess>'''
+ '''</cas:serviceResponse>''')
self._cas_logout()
response = self._login_user('editor', '123')
settings.CAS_CUSTOM_ATTRIBUTES_FORMATER = 'cas_provider.attribute_formatters.ruby_cas'
response = self._validate_cas2(response, True)
self.assertEqual(response.content, '''<cas:serviceResponse xmlns:cas="http://www.yale.edu/tp/cas">'''
- '''<cas:authenticationSuccess>'''
- '''<cas:user>editor</cas:user>'''
- '''<cas:attraStyle>RubyCAS</cas:attraStyle>'''
- '''<cas:group>editor</cas:group>'''
- '''<cas:is_staff>True</cas:is_staff>'''
- '''<cas:is_active>True</cas:is_active>'''
- '''<cas:email>editor@exapmle.com</cas:email>'''
- '''</cas:authenticationSuccess>'''
- '''</cas:serviceResponse>''')
+ '''<cas:authenticationSuccess>'''
+ '''<cas:user>editor</cas:user>'''
+ '''<cas:attraStyle>RubyCAS</cas:attraStyle>'''
+ '''<cas:group>editor</cas:group>'''
+ '''<cas:is_staff>True</cas:is_staff>'''
+ '''<cas:is_active>True</cas:is_active>'''
+ '''<cas:email>editor@exapmle.com</cas:email>'''
+ '''</cas:authenticationSuccess>'''
+ '''</cas:serviceResponse>''')
self._cas_logout()
response = self._login_user('editor', '123')
settings.CAS_CUSTOM_ATTRIBUTES_FORMATER = 'cas_provider.attribute_formatters.name_value'
response = self._validate_cas2(response, True)
self.assertEqual(response.content, '''<cas:serviceResponse xmlns:cas="http://www.yale.edu/tp/cas">'''
- '''<cas:authenticationSuccess>'''
- '''<cas:user>editor</cas:user>'''
- '''<cas:attribute name="attraStyle" value="Name-Value"/>'''
- '''<cas:attribute name="group" value="editor"/>'''
- '''<cas:attribute name="is_staff" value="True"/>'''
- '''<cas:attribute name="is_active" value="True"/>'''
- '''<cas:attribute name="email" value="editor@exapmle.com"/>'''
- '''</cas:authenticationSuccess>'''
- '''</cas:serviceResponse>''')
+ '''<cas:authenticationSuccess>'''
+ '''<cas:user>editor</cas:user>'''
+ '''<cas:attribute name="attraStyle" value="Name-Value"/>'''
+ '''<cas:attribute name="group" value="editor"/>'''
+ '''<cas:attribute name="is_staff" value="True"/>'''
+ '''<cas:attribute name="is_active" value="True"/>'''
+ '''<cas:attribute name="email" value="editor@exapmle.com"/>'''
+ '''</cas:authenticationSuccess>'''
+ '''</cas:serviceResponse>''')
def test_cas2_fail_validate(self):
self.assertEqual(response.content, u'no\n\n')
- def _validate_cas2(self, response, is_correct=True):
+ def _validate_cas2(self, response, is_correct=True, pgtUrl = None):
if is_correct:
self.assertEqual(response.status_code, 302)
self.assertTrue(response.has_header('location'))
location = urlparse(response['location'])
ticket = location.query.split('=')[1]
-
- response = self.client.get(reverse('cas_service_validate'), {'ticket': ticket, 'service': self.service}, follow=False)
+ if pgtUrl:
+ response = self.client.get(reverse('cas_service_validate'), {'ticket': ticket, 'service': self.service, 'pgtUrl': pgtUrl}, follow=False)
+ else:
+ response = self.client.get(reverse('cas_service_validate'), {'ticket': ticket, 'service': self.service}, follow=False)
self.assertEqual(response.status_code, 200)
else:
self.assertEqual(response.status_code, 200)
self.assertEqual(response.content, _cas2_error_response(INVALID_TICKET).content)
return response
-
class ModelsTestCase(TestCase):
fixtures = ['cas_users.json', ]
urlpatterns = patterns('cas_provider.views',
url(r'^login/?$', 'login', name='cas_login'),
url(r'^validate/?$', 'validate', name='cas_validate'),
+ url(r'^proxy/?$', 'proxy', name='proxy'),
url(r'^serviceValidate/?$', 'service_validate', name='cas_service_validate'),
+ url(r'^proxyValidate/?$', 'proxy_validate', name='cas_proxy_validate'),
url(r'^logout/?$', 'logout', name='cas_logout'),
)
+from lxml import etree
+from urllib import urlencode
+import urllib2
+import urlparse
from django.conf import settings
from django.contrib.auth import login as auth_login, logout as auth_logout
from django.core.urlresolvers import get_callable
from django.http import HttpResponse, HttpResponseRedirect
from django.shortcuts import render_to_response
from django.template import RequestContext
+from cas_provider.attribute_formatters import NSMAP, CAS
+from cas_provider.models import ProxyGrantingTicket, ProxyTicket
from forms import LoginForm
from models import ServiceTicket, LoginTicket
__all__ = ['login', 'validate', 'logout', 'service_validate']
-
INVALID_TICKET = 'INVALID_TICKET'
INVALID_SERVICE = 'INVALID_SERVICE'
INVALID_REQUEST = 'INVALID_REQUEST'
(INVALID_SERVICE, u'Service is invalid'),
(INVALID_REQUEST, u'Not all required parameters were sent.'),
(INTERNAL_ERROR, u'An internal error occurred during ticket validation'),
-)
+ )
-def login(request, template_name='cas/login.html', \
- success_redirect=settings.LOGIN_REDIRECT_URL,
- warn_template_name='cas/warn.html',
- form_class=LoginForm):
+def login(request, template_name='cas/login.html',\
+ success_redirect=settings.LOGIN_REDIRECT_URL,
+ warn_template_name='cas/warn.html',
+ form_class=LoginForm):
service = request.GET.get('service', None)
if request.user.is_authenticated():
if service is not None:
return render_to_response(template_name, {
'form': form,
'errors': form.get_errors() if hasattr(form, 'get_errors') else None,
- }, context_instance=RequestContext(request))
+ }, context_instance=RequestContext(request))
def validate(request):
ticket = ServiceTicket.objects.get(ticket=ticket_string)
assert ticket.service == service
username = ticket.user.username
- ticket.delete()
return HttpResponse("yes\n%s\n" % username)
except:
pass
return HttpResponse("no\n\n")
-def logout(request, template_name='cas/logout.html', \
- auto_redirect=settings.CAS_AUTO_REDIRECT_AFTER_LOGOUT):
+def logout(request, template_name='cas/logout.html',
+ auto_redirect=settings.CAS_AUTO_REDIRECT_AFTER_LOGOUT):
url = request.GET.get('url', None)
if request.user.is_authenticated():
+ for ticket in ServiceTicket.objects.filter(user = request.user):
+ ticket.delete()
auth_logout(request)
if url and auto_redirect:
return HttpResponseRedirect(url)
- return render_to_response(template_name, {'url': url}, \
+ return render_to_response(template_name, {'url': url},\
context_instance=RequestContext(request))
+def proxy(request):
+ targetService = request.GET['targetService']
+ pgtiou = request.GET['pgt']
-def service_validate(request):
- """Validate ticket via CAS v.2 protocol"""
- service = request.GET.get('service', None)
- ticket_string = request.GET.get('ticket', None)
+ try:
+ proxyGrantingTicket = ProxyGrantingTicket.objects.get(pgtiou=pgtiou)
+ except ProxyGrantingTicket.DoesNotExist:
+ return _cas2_error_response(INVALID_TICKET)
+
+ pt = ProxyTicket.objects.create(proxyGrantingTicket = proxyGrantingTicket,
+ user=proxyGrantingTicket.serviceTicket.user,
+ service=targetService)
+ return _cas2_proxy_success(pt.ticket)
+
+
+def ticket_validate(service, ticket_string, pgtUrl):
if service is None or ticket_string is None:
return _cas2_error_response(INVALID_REQUEST)
return _cas2_error_response(INVALID_TICKET)
if ticket.service != service:
- ticket.delete()
return _cas2_error_response(INVALID_SERVICE)
+
+ pgtIouId = None
+ proxies = ()
+ if pgtUrl is not None:
+ pgt = generate_proxy_granting_ticket(pgtUrl, ticket)
+ if pgt:
+ pgtIouId = pgt.pgtiou
+
+ while pgt:
+ proxies += (pgt.serviceTicket.service,)
+ pgt = pgt.serviceTicket.proxyGrantingTicket if hasattr(pgt.serviceTicket, 'proxyGrantingTicket') else None
+
+
user = ticket.user
- ticket.delete()
- return _cas2_sucess_response(user)
+ return _cas2_sucess_response(user, pgtIouId, proxies)
+
+
+def service_validate(request):
+ """Validate ticket via CAS v.2 protocol"""
+ service = request.GET.get('service', None)
+ ticket_string = request.GET.get('ticket', None)
+ pgtUrl = request.GET.get('pgtUrl', None)
+ if ticket_string.startswith('PT-'):
+ return _cas2_error_response(INVALID_TICKET, "serviceValidate cannot verify proxy tickets")
+ else:
+ return ticket_validate(service, ticket_string, pgtUrl)
-def _cas2_error_response(code):
+def proxy_validate(request):
+ """Validate ticket via CAS v.2 protocol"""
+ service = request.GET.get('service', None)
+ ticket_string = request.GET.get('ticket', None)
+ pgtUrl = request.GET.get('pgtUrl', None)
+ return ticket_validate(service, ticket_string, pgtUrl)
+
+def generate_proxy_granting_ticket(pgt_url, ticket):
+ proxy_callback_good_status = (200, 202, 301, 302, 304)
+ uri = list(urlparse.urlsplit(pgt_url))
+
+ pgt = ProxyGrantingTicket()
+ pgt.serviceTicket = ticket
+
+ if hasattr(ticket, 'proxyGrantingTicket'):
+ # here we got a proxy ticket! tata!
+ pgt.pgt = ticket.proxyGrantingTicket
+
+ params = {'pgtId': pgt.ticket, 'pgtIou': pgt.pgtiou}
+
+ query = dict(urlparse.parse_qsl(uri[4]))
+ query.update(params)
+
+ uri[4] = urlencode(query)
+
+
+ try:
+ response = urllib2.urlopen(urlparse.urlunsplit(uri))
+ except urllib2.HTTPError, e:
+ if not e.code in proxy_callback_good_status:
+ return
+ except urllib2.URLError, e:
+ return
+
+ pgt.save()
+ return pgt
+
+
+def _cas2_proxy_success(pt):
+ return HttpResponse(proxy_success(pt))
+
+def _cas2_sucess_response(user, pgt = None, proxies = None):
+ return HttpResponse(auth_success_response(user, pgt, proxies), mimetype='text/xml')
+
+def _cas2_error_response(code, message = None):
return HttpResponse(u''''<cas:serviceResponse xmlns:cas="http://www.yale.edu/tp/cas">
<cas:authenticationFailure code="%(code)s">
%(message)s
</cas:authenticationFailure>
</cas:serviceResponse>''' % {
'code': code,
- 'message': dict(ERROR_MESSAGES).get(code)
+ 'message': message if message else dict(ERROR_MESSAGES).get(code)
}, mimetype='text/xml')
+def proxy_success(pt):
+ response = etree.Element(CAS + 'serviceResponse', nsmap=NSMAP)
+ proxySuccess = etree.SubElement(response, CAS + 'proxySuccess')
+ proxyTicket = etree.SubElement(proxySuccess, CAS + 'proxyTicket')
+ proxyTicket.text = pt
+ return unicode(etree.tostring(response, encoding='utf-8'), 'utf-8')
-def _cas2_sucess_response(user):
- return HttpResponse(auth_success_response(user), mimetype='text/xml')
-
+def auth_success_response(user, pgt, proxies):
-def auth_success_response(user):
- from attribute_formatters import CAS, NSMAP, etree
response = etree.Element(CAS + 'serviceResponse', nsmap=NSMAP)
auth_success = etree.SubElement(response, CAS + 'authenticationSuccess')
if len(attrs) > 0:
formater = get_callable(settings.CAS_CUSTOM_ATTRIBUTES_FORMATER)
formater(auth_success, attrs)
+
+
+ if pgt:
+ pgtElement = etree.SubElement(auth_success, CAS + 'proxyGrantingTicket')
+ pgtElement.text = pgt
+
+ if proxies:
+ proxiesElement = etree.SubElement(auth_success , CAS + "proxies")
+ for proxy in proxies:
+ proxyElement = etree.SubElement(proxiesElement, CAS + "proxy")
+ proxyElement.text = proxy
+
+
+
return unicode(etree.tostring(response, encoding='utf-8'), 'utf-8')
include_package_data=True,
zip_safe=False,
install_requires=['setuptools'],
+ classifiers = [
+ "Development Status :: 3 - Alpha",
+ "Framework :: Django",
+ "License :: OSI Approved :: MIT License",
+ ]
)
fnp / django-cas-provider.git / commitdiff