SingleLogOut unit tests

cas_server/models.py

@@ -20,7 +20,6 @@ from django.utils import timezone
 from picklefield.fields import PickledObjectField
 
 import re
-import os
 import sys
 import logging
 from importlib import import_module
@@ -428,46 +427,27 @@ class Ticket(models.Model):
                     self.user.username
                 )
             )
-            try:
-                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': os.urandom(20).encode("hex"),
-                        'datetime': timezone.now().isoformat(),
-                        'ticket':  self.value
-                    }
-                if self.service_pattern.single_log_out_callback:
-                    url = self.service_pattern.single_log_out_callback
-                else:
-                    url = self.service
-                async_list.append(
-                    session.post(
-                        url.encode('utf-8'),
-                        data={'logoutRequest': xml.encode('utf-8')},
-                        timeout=settings.CAS_SLO_TIMEOUT
-                    )
+            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
+                }
+            if self.service_pattern.single_log_out_callback:
+                url = self.service_pattern.single_log_out_callback
+            else:
+                url = self.service
+            async_list.append(
+                session.post(
+                    url.encode('utf-8'),
+                    data={'logoutRequest': xml.encode('utf-8')},
+                    timeout=settings.CAS_SLO_TIMEOUT
                 )
-            except Exception as error:
-                error = utils.unpack_nested_exception(error)
-                logger.warning(
-                    "Error durring SLO for user %s on service %s: %s" % (
-                        self.user.username,
-                        self.service,
-                        error
-                    )
-                )
-                if request is not None:
-                    messages.add_message(
-                        request,
-                        messages.WARNING,
-                        _(u'Error during service logout %(service)s:\n%(error)s') %
-                        {'service':  self.service, 'error': error}
-                    )
-                else:
-                    sys.stderr.write("%r\n" % error)
+            )
 
 
 class ServiceTicket(Ticket):

cas_server/tests/test_utils.py

@@ -1,3 +1,4 @@
+"""Tests module for utils"""
 from django.test import TestCase
 
 import six

cas_server/tests/test_view.py

@@ -20,7 +20,8 @@ from cas_server.tests.utils import (
     get_auth_client,
     get_user_ticket_request,
     get_pgt,
-    get_proxy_ticket
+    get_proxy_ticket,
+    get_validated_ticket
 )
 from cas_server.tests.mixin import BaseServicePattern, XmlContent
 
@@ -386,6 +387,15 @@ class LoginTestCase(TestCase, BaseServicePattern):
 @override_settings(CAS_AUTH_CLASS='cas_server.auth.TestAuthUser')
 class LogoutTestCase(TestCase):
     """test fot the logout view"""
+    def setUp(self):
+        self.service = 'http://127.0.0.1:45678'
+        self.service_pattern = models.ServicePattern.objects.create(
+            name="localhost",
+            pattern="^https?://127\.0\.0\.1(:[0-9]+)?(/.*)?$",
+            single_log_out=True
+        )
+        models.ReplaceAttributName.objects.create(name="*", service_pattern=self.service_pattern)
+
     def test_logout(self):
         """logout is idempotent"""
         client = Client()
@@ -477,6 +487,37 @@ class LogoutTestCase(TestCase):
         response = client.get('/logout?service=https://www.example.com')
         self.assert_redirect_to_service(client, response)
 
+    def test_logout_slo(self):
+        """test logout from a service with SLO support"""
+        (httpd, host, port) = utils.HttpParamsHandler.run()[0:3]
+        service = "http://%s:%s" % (host, port)
+
+        (client, ticket) = get_validated_ticket(service)[:2]
+
+        client.get('/logout')
+
+        params = httpd.PARAMS
+        self.assertTrue(b'logoutRequest' in params and params[b'logoutRequest'])
+
+        root = etree.fromstring(params[b'logoutRequest'][0])
+        self.assertTrue(
+            root.xpath(
+                "//samlp:LogoutRequest",
+                namespaces={"samlp": "urn:oasis:names:tc:SAML:2.0:protocol"}
+            )
+        )
+        session_index = root.xpath(
+            "//samlp:SessionIndex",
+            namespaces={"samlp": "urn:oasis:names:tc:SAML:2.0:protocol"}
+        )
+        self.assertEqual(len(session_index), 1)
+        self.assertEqual(session_index[0].text, ticket.value)
+
+        # SLO error are displayed on logout page
+        (client, ticket) = get_validated_ticket(self.service)[:2]
+        response = client.get('/logout')
+        self.assertTrue(b"Error during service logout" in response.content)
+
     def test_ajax_logout(self):
         """test ajax logout"""
         client = get_auth_client()

cas_server/tests/utils.py

@@ -50,7 +50,19 @@ def get_user_ticket_request(service):
         session_key=client.session.session_key
     )
     ticket = models.ServiceTicket.objects.get(value=ticket_value)
-    return (user, ticket)
+    return (user, ticket, client)
+
+
+def get_validated_ticket(service):
+    (ticket, auth_client) = get_user_ticket_request(service)[1:3]
+
+    client = Client()
+    response = client.get('/validate', {'ticket': ticket.value, 'service': service})
+    assert (response.status_code == 200)
+    assert (response.content == b'yes\ntest\n')
+
+    ticket = models.ServiceTicket.objects.get(value=ticket.value)
+    return (auth_client, ticket)
 
 
 def get_pgt():
@@ -71,6 +83,7 @@ def get_pgt():
 
 
 def get_proxy_ticket(service):
+    """Return a ProxyTicket waiting for validation"""
     params = get_pgt()
 
     # get a proxy ticket