Changeset 10416 for main/waeup.cas

Timestamp:

8 Jul 2013, 08:35:22 (11 years ago)

Author:

uli

Message:

Support /validate (CAS 1.0).

Location:

main/waeup.cas/trunk/waeup/cas

Files:

• server.py (modified) (3 diffs)
• tests/test_server.py (modified) (4 diffs)

main/waeup.cas/trunk/waeup/cas/server.py

@@ -59 +59 @@
 
 
+def check_service_ticket(db, ticket, service, renew=False):
+    """Check whether (`ticket`, `service`) represents a valid service
+    ticket in `db`.
+
+    Returns a database set or ``None``.
+    """
+    if None in (ticket, service):
+        return None
+    ticket, service = str(ticket), str(service)
+    q = db.query(ServiceTicket).filter(
+        ServiceTicket.ticket == ticket).filter(
+        ServiceTicket.service == service).first()
+    if renew and q.sso:
+        return None
+    return q
+
+
 def create_login_ticket():
     """Create a unique login ticket.
@@ -74 +91 @@
     if lt_string is None:
         return False
-    q = db.query(LoginTicket).filter(LoginTicket.ticket == lt_string)
+    q = db.query(LoginTicket).filter(LoginTicket.ticket == str(lt_string))
     result = [x for x in q]
     if result:
@@ -283 +300 @@
 
     def validate(self, req):
-        return exc.HTTPNotImplemented()
+        service = req.POST.get('service', req.GET.get('service', None))
+        ticket = req.POST.get('ticket', req.GET.get('ticket', None))
+        renew = req.POST.get('renew', req.GET.get('renew', None))
+        renew = renew is not None
+        st = check_service_ticket(self.db, ticket, service, renew)
+        if st is not None:
+            return Response('yes' + chr(0x0a) + st.user + chr(0x0a))
+        return Response('no' + chr(0x0a) + chr(0x0a))
 
     def logout(self, req):

main/waeup.cas/trunk/waeup/cas/tests/test_server.py

@@ -13 +13 @@
     create_tgc_value, check_login_ticket, set_session_cookie,
     check_session_cookie, get_template, delete_session_cookie,
+    check_service_ticket,
     )
 
@@ -95 +96 @@
         # we can access a validation page
         app = CASServer()
-        req = Request.blank('http://localhost/validate')
-        resp = app(req)
-        assert resp.status == '501 Not Implemented'
+        req = Request.blank('http://localhost/validate?service=foo&ticket=bar')
+        resp = app(req)
+        assert resp.status == '200 OK'
 
     def test_logout(self):
@@ -254 +255 @@
         assert b'like you to' in resp.body
         assert b'http://www.logout.com' in resp.body
+
+    def test_validate_invalid(self):
+        # 2.4.2 validation failures is indicated by a given format
+        app = CASServer()
+        params = 'ticket=foo&service=bar'
+        req = Request.blank('https://localhost/validate?%s' % params)
+        resp = app(req)
+        assert resp.body == b'no\n\n'
+
+    def test_validate_valid(self):
+        # 2.4 validation success is indicated by a given format
+        app = CASServer()
+        sticket = create_service_ticket(
+            'someuser', 'http://service.com/', sso=False)
+        app.db.add(sticket)
+        params = 'ticket=%s&service=%s' % (
+            sticket.ticket, sticket.service)
+        req = Request.blank('https://localhost/validate?%s' % params)
+        resp = app(req)
+        assert resp.body == b'yes\nsomeuser\n'
+
+    def test_validate_renew_invalid(self):
+        # 2.4.1 with `renew` we accept only non-sso issued tickets
+        app = CASServer()
+        sticket = create_service_ticket(
+            'someuser', 'http://service.com/', sso=True)
+        app.db.add(sticket)
+        params = 'ticket=%s&service=%s&renew=true' % (
+            sticket.ticket, sticket.service)
+        req = Request.blank('https://localhost/validate?%s' % params)
+        resp = app(req)
+        assert resp.body == b'no\n\n'
+
+    def test_validate_renew_valid(self):
+        # 2.4.1 with `renew` we accept only non-sso issued tickets
+        app = CASServer()
+        sticket = create_service_ticket(
+            'someuser', 'http://service.com/', sso=False)
+        app.db.add(sticket)
+        params = 'ticket=%s&service=%s&renew=true' % (
+            sticket.ticket, sticket.service)
+        req = Request.blank('https://localhost/validate?%s' % params)
+        resp = app(req)
+        assert resp.body == b'yes\nsomeuser\n'
 
 
@@ -480 +525 @@
             TicketGrantingCookie.value == value)
         assert len(list(q)) == 0
+
+    def test_check_service_ticket(self):
+        db = DB('sqlite:///')
+        st = ServiceTicket(
+            'ST-123456', 'someuser', 'http://myservice.com', True)
+        db.add(st)
+        assert check_service_ticket(db, None, 'foo') is None
+        assert check_service_ticket(db, 'foo', None) is None
+        assert check_service_ticket(db, 'ST-123456', 'foo') is None
+        assert check_service_ticket(db, 'foo', 'http://myservice.com') is None
+        result = check_service_ticket(db, 'ST-123456', 'http://myservice.com')
+        assert isinstance(result, ServiceTicket)
+        assert result.user == 'someuser'
+        assert check_service_ticket(
+            db,  'ST-123456', 'http://myservice.com', True) is None
+        assert check_service_ticket(
+            db,  'ST-123456', 'http://myservice.com', False) is not None