source: main/waeup.cas/trunk/waeup/cas/tests/test_server.py @ 10461

Last change on this file since 10461 was 10416, checked in by uli, 11 years ago

Support /validate (CAS 1.0).
File size: 21.1 KB
Line 
1import os
2import re
3import shutil
4import tempfile
5import unittest
6from paste.deploy import loadapp
7from webob import Request, Response
8from webtest import TestApp as WebTestApp  # avoid py.test skip message
9from waeup.cas.authenticators import DummyAuthenticator
10from waeup.cas.db import DB, LoginTicket, ServiceTicket, TicketGrantingCookie
11from waeup.cas.server import (
12    CASServer, create_service_ticket, create_login_ticket,
13    create_tgc_value, check_login_ticket, set_session_cookie,
14    check_session_cookie, get_template, delete_session_cookie,
15    check_service_ticket,
16    )
17
18RE_ALPHABET = re.compile('^[a-zA-Z0-9\-]*$')
19RE_COOKIE = re.compile('^cas-tgc=[A-Za-z0-9\-]+; Path=/; secure; HttpOnly$')
20RE_COOKIE_DEL = re.compile(
21    '^cas-tgc=; Max-Age=\-[0-9]+; Path=/; '
22    'expires=Thu, 01-Jan-1970 00:00:00 GMT; secure; HttpOnly$')
23
24
25class CASServerTests(unittest.TestCase):
26
27    def setUp(self):
28        # Create a new location where tempfiles are created.  This way
29        # also temporary dirs of local CASServers can be removed on
30        # tear-down.
31        self._new_tmpdir = tempfile.mkdtemp()
32        self._old_tmpdir = tempfile.tempdir
33        tempfile.tempdir = self._new_tmpdir
34        self.workdir = os.path.join(self._new_tmpdir, 'home')
35        self.db_path = os.path.join(self.workdir, 'mycas.db')
36        os.mkdir(self.workdir)
37        self.paste_conf1 = os.path.join(
38            os.path.dirname(__file__), 'sample1.ini')
39        self.paste_conf2 = os.path.join(
40            os.path.dirname(__file__), 'sample2.ini')
41
42    def tearDown(self):
43        # remove local tempfile and reset old tempdir setting
44        if os.path.isdir(self._new_tmpdir):
45            shutil.rmtree(self._new_tmpdir)
46        tempfile.tempdir = self._old_tmpdir
47
48    def test_paste_deploy_loader(self):
49        # we can load the CAS server via paste.deploy plugin
50        app = loadapp('config:%s' % self.paste_conf1)
51        assert isinstance(app, CASServer)
52        assert hasattr(app, 'db')
53        assert isinstance(app.db, DB)
54        assert hasattr(app, 'auth')
55
56    def test_paste_deploy_options(self):
57        # we can set CAS server-related options via paste.deploy config
58        app = loadapp('config:%s' % self.paste_conf2)
59        assert isinstance(app, CASServer)
60        assert app.db_connection_string == 'sqlite:///:memory:'
61        assert isinstance(app.auth, DummyAuthenticator)
62
63    def test_init(self):
64        # we get a `DB` instance created automatically
65        app = CASServer()
66        assert hasattr(app, 'db')
67        assert app.db is not None
68
69    def test_init_explicit_db_path(self):
70        # we can set a db_path explicitly
71        app = CASServer(db='sqlite:///%s' % self.db_path)
72        assert hasattr(app, 'db')
73        assert isinstance(app.db, DB)
74        assert os.path.isfile(self.db_path)
75
76    def test_get_template(self):
77        app = CASServer()
78        assert app._get_template('login.html') is not None
79        assert app._get_template('not-existent.html') is None
80
81    def test_call_root(self):
82        # the CAS protocol requires no root
83        app = CASServer()
84        req = Request.blank('http://localhost/')
85        resp = app(req)
86        assert resp.status == '404 Not Found'
87
88    def test_first_time_login(self):
89        # we can get a login page
90        app = CASServer()
91        req = Request.blank('http://localhost/login')
92        resp = app(req)
93        assert resp.status == '200 OK'
94
95    def test_validate(self):
96        # we can access a validation page
97        app = CASServer()
98        req = Request.blank('http://localhost/validate?service=foo&ticket=bar')
99        resp = app(req)
100        assert resp.status == '200 OK'
101
102    def test_logout(self):
103        # we can access a logout page
104        app = CASServer()
105        req = Request.blank('http://localhost/logout')
106        resp = app(req)
107        assert resp.status == '200 OK'
108
109    def test_login_simple(self):
110        # a simple login with no service will result in login screen
111        # (2.1.1#service of protocol specs)
112        app = CASServer()
113        req = Request.blank('http://localhost/login')
114        resp = app(req)
115        assert resp.status == '200 OK'
116        assert resp.content_type == 'text/html'
117        assert b'<form ' in resp.body
118
119    def test_login_cred_acceptor_sso_no_service(self):
120        # 2.2.4: successful login via single sign on
121        app = CASServer()
122        tgc = create_tgc_value()
123        app.db.add(tgc)
124        value = str(tgc.value)
125        req = Request.blank('https://localhost/login')
126        req.headers['Cookie'] = 'cas-tgc=%s' % value
127        resp = app(req)
128        assert resp.status == '200 OK'
129        assert b'already' in resp.body
130        assert 'Set-Cookie' not in resp.headers
131        return
132
133    def test_login_renew_with_cookie(self):
134        # 2.1.1: cookie will be ignored when renew is set
135        app = CASServer()
136        tgc = create_tgc_value()
137        app.db.add(tgc)
138        value = str(tgc.value)
139        req = Request.blank('https://localhost/login?renew=true')
140        req.headers['Cookie'] = 'cas-tgc=%s' % value
141        resp = app(req)
142        assert resp.status == '200 OK'
143        assert b'username' in resp.body
144        assert 'Set-Cookie' not in resp.headers
145
146    def test_login_renew_without_cookie(self):
147        # 2.1.1: with renew and no cookie, normal auth will happen
148        app = CASServer()
149        req = Request.blank('https://localhost/login?renew=true')
150        resp = app(req)
151        assert resp.status == '200 OK'
152        assert b'username' in resp.body
153
154    def test_login_renew_as_empty_string(self):
155        # `renew` is handled correctly, even with empty value
156        app = CASServer()
157        tgc = create_tgc_value()
158        app.db.add(tgc)
159        value = str(tgc.value)
160        req = Request.blank('https://localhost/login?renew')
161        req.headers['Cookie'] = 'cas-tgc=%s' % value
162        resp = app(req)
163        assert resp.status == '200 OK'
164        assert b'username' in resp.body
165        assert 'Set-Cookie' not in resp.headers
166
167    def test_login_gateway_no_cookie_with_service(self):
168        # 2.1.1: with gateway but w/o cookie we will be redirected to service
169        # no service ticket will be issued
170        app = CASServer()
171        params = 'gateway=true&service=http%3A%2F%2Fwww.service.com'
172        req = Request.blank('https://localhost/login?%s' % params)
173        resp = app(req)
174        assert resp.status == '303 See Other'
175        assert 'Location' in resp.headers
176        assert resp.headers['Location'] == 'http://www.service.com'
177
178    def test_login_gateway_with_cookie_and_service(self):
179        # 2.1.1: with cookie and gateway we will be redirected to service
180        app = CASServer()
181        tgc = create_tgc_value()
182        app.db.add(tgc)
183        value = str(tgc.value)
184        params = 'gateway=true&service=http%3A%2F%2Fwww.service.com'
185        req = Request.blank('https://localhost/login?%s' % params)
186        req.headers['Cookie'] = 'cas-tgc=%s' % value
187        resp = app(req)
188        assert resp.status == '303 See Other'
189        assert 'Location' in resp.headers
190        assert resp.headers['Location'].startswith(
191            'http://www.service.com?ticket=ST-')
192
193    def test_login_gateway_and_renew(self):
194        # 2.1.1 if both, gateway and renew are specified, only renew is valid
195        app = CASServer()
196        tgc = create_tgc_value()
197        app.db.add(tgc)
198        value = str(tgc.value)
199        req = Request.blank('https://localhost/login?renew=true&gateway=true')
200        req.headers['Cookie'] = 'cas-tgc=%s' % value
201        resp = app(req)
202        # with only gateway true, this would lead to a redirect
203        assert resp.status == '200 OK'
204        assert b'username' in resp.body
205        assert 'Set-Cookie' not in resp.headers
206
207    def test_login_warn(self):
208        # 2.2.1 as a credential acceptor, with `warn` set we require confirm
209        app = CASServer()
210        tgc = create_tgc_value()
211        app.db.add(tgc)
212        value = str(tgc.value)
213        params = 'warn=true&service=http%3A%2F%2Fwww.service.com'
214        req = Request.blank('https://localhost/login?%s' % params)
215        req.headers['Cookie'] = 'cas-tgc=%s' % value
216        resp = app(req)
217        # without warn, we would get a redirect
218        assert resp.status == '200 OK'
219        assert b'CAS login successful' in resp.body
220
221    def test_logout_no_cookie(self):
222        # 2.3 logout displays a logout page.
223        app = CASServer()
224        req = Request.blank('https://localhost/logout')
225        resp = app(req)
226        assert resp.status == '200 OK'
227        assert b'logged out' in resp.body
228
229    def test_logout_with_cookie(self):
230        # 2.3 logout destroys any existing SSO session
231        app = CASServer()
232        tgc = create_tgc_value()
233        app.db.add(tgc)
234        value = str(tgc.value)
235        req = Request.blank('https://localhost/logout')
236        req.headers['Cookie'] = 'cas-tgc=%s' % value
237        resp = app(req)
238        assert resp.status == '200 OK'
239        assert b'logged out' in resp.body
240        assert 'Set-Cookie' in resp.headers
241        cookie = resp.headers['Set-Cookie']
242        assert cookie.startswith('cas-tgc=;')
243        assert 'expires' in cookie
244        assert 'Max-Age' in cookie
245        assert len(list(app.db.query(TicketGrantingCookie))) == 0
246
247    def test_logout_url(self):
248        # 2.3.1 with an `url` given we provide a link on logout
249        app = CASServer()
250        params = 'url=http%3A%2F%2Fwww.logout.com'
251        req = Request.blank('https://localhost/logout?%s' % params)
252        resp = app(req)
253        assert resp.status == '200 OK'
254        assert b'logged out' in resp.body
255        assert b'like you to' in resp.body
256        assert b'http://www.logout.com' in resp.body
257
258    def test_validate_invalid(self):
259        # 2.4.2 validation failures is indicated by a given format
260        app = CASServer()
261        params = 'ticket=foo&service=bar'
262        req = Request.blank('https://localhost/validate?%s' % params)
263        resp = app(req)
264        assert resp.body == b'no\n\n'
265
266    def test_validate_valid(self):
267        # 2.4 validation success is indicated by a given format
268        app = CASServer()
269        sticket = create_service_ticket(
270            'someuser', 'http://service.com/', sso=False)
271        app.db.add(sticket)
272        params = 'ticket=%s&service=%s' % (
273            sticket.ticket, sticket.service)
274        req = Request.blank('https://localhost/validate?%s' % params)
275        resp = app(req)
276        assert resp.body == b'yes\nsomeuser\n'
277
278    def test_validate_renew_invalid(self):
279        # 2.4.1 with `renew` we accept only non-sso issued tickets
280        app = CASServer()
281        sticket = create_service_ticket(
282            'someuser', 'http://service.com/', sso=True)
283        app.db.add(sticket)
284        params = 'ticket=%s&service=%s&renew=true' % (
285            sticket.ticket, sticket.service)
286        req = Request.blank('https://localhost/validate?%s' % params)
287        resp = app(req)
288        assert resp.body == b'no\n\n'
289
290    def test_validate_renew_valid(self):
291        # 2.4.1 with `renew` we accept only non-sso issued tickets
292        app = CASServer()
293        sticket = create_service_ticket(
294            'someuser', 'http://service.com/', sso=False)
295        app.db.add(sticket)
296        params = 'ticket=%s&service=%s&renew=true' % (
297            sticket.ticket, sticket.service)
298        req = Request.blank('https://localhost/validate?%s' % params)
299        resp = app(req)
300        assert resp.body == b'yes\nsomeuser\n'
301
302
303class BrowserTests(unittest.TestCase):
304
305    def setUp(self):
306        self.raw_app = CASServer(auth=DummyAuthenticator())
307        self.app = WebTestApp(self.raw_app)
308
309    def test_login(self):
310        resp = self.app.get('/login')
311        assert resp.status == '200 OK'
312        form = resp.forms[0]
313        # 2.1.3: form must be submitted by POST
314        assert form.method == 'post'
315        fieldnames = form.fields.keys()
316        # 2.1.3: form must contain: username, password, lt
317        assert 'username' in fieldnames
318        assert 'password' in fieldnames
319        assert 'lt' in fieldnames
320        assert RE_ALPHABET.match(form['lt'].value)
321
322    def test_login_no_service(self):
323        # w/o a service passed in, the form should not contain service
324        # (not a strict protocol requirement, but handy)
325        resp = self.app.get('/login')
326        assert 'service' not in resp.forms[0].fields.keys()
327
328    def test_login_service_replayed(self):
329        # 2.1.3: the login form must contain the service param sent
330        resp = self.app.get('/login?service=http%3A%2F%2Fwww.service.com')
331        form = resp.forms[0]
332        assert resp.status == '200 OK'
333        assert 'service' in form.fields.keys()
334        assert form['service'].value == 'http://www.service.com'
335
336    def test_login_cred_acceptor_valid_no_service(self):
337        # 2.2.4: successful login w/o service yields a message
338        lt = create_login_ticket()
339        self.raw_app.db.add(lt)
340        lt_string = lt.ticket
341        resp = self.app.post('/login', dict(
342            username='bird', password='bebop', lt=lt_string))
343        assert resp.status == '200 OK'
344        assert b'successful' in resp.body
345        # single-sign-on session initiated
346        assert 'Set-Cookie' in resp.headers
347        cookie = resp.headers['Set-Cookie']
348        assert cookie.startswith('cas-tgc=')
349
350    def test_login_cred_acceptor_valid_w_service(self):
351        # 2.2.4: successful login with service makes a redirect
352        # Appendix B: safe redirect
353        lt = create_login_ticket()
354        self.raw_app.db.add(lt)
355        lt_string = lt.ticket
356        resp = self.app.post('/login', dict(
357            username='bird', password='bebop', lt=lt_string,
358            service='http://example.com/Login'))
359        assert resp.status == '303 See Other'
360        assert 'Location' in resp.headers
361        assert resp.headers['Location'].startswith(
362            'http://example.com/Login?ticket=ST-')
363        assert 'Pragma' in resp.headers
364        assert resp.headers['Pragma'] == 'no-cache'
365        assert 'Cache-Control' in resp.headers
366        assert resp.headers['Cache-Control'] == 'no-store'
367        assert 'Expires' in resp.headers
368        assert resp.headers['Expires'] == 'Thu, 01 Dec 1994 16:00:00 GMT'
369        assert b'window.location.href' in resp.body
370        assert b'noscript' in resp.body
371        assert b'ticket=ST-' in resp.body
372
373    def test_login_cred_acceptor_failed(self):
374        # 2.2.4: failed login yields a message
375        lt = create_login_ticket()
376        self.raw_app.db.add(lt)
377        lt_string = lt.ticket
378        resp = self.app.post('/login', dict(
379            username='bird', password='cat', lt=lt_string))
380        assert resp.status == '200 OK'
381        assert b'failed' in resp.body
382
383    def test_login_sso_no_service(self):
384        # we can initiate single-sign-on without service
385        resp1 = self.app.get('https://localhost/login')  # HTTPS required!
386        assert resp1.status == '200 OK'
387        assert 'cas-tgc' not in self.app.cookies
388        form = resp1.forms[0]
389        form.set('username', 'bird')
390        form.set('password', 'bebop')
391        resp2 = form.submit('AUTHENTICATE')
392        assert resp2.status == '200 OK'
393        # we got a secure cookie
394        assert 'cas-tgc' in self.app.cookies
395        # when we get the login page again, the cookie will replace creds.
396        resp3 = self.app.get('https://localhost/login')
397        assert b'You logged in already' in resp3.body
398
399    def test_login_sso_with_service(self):
400        resp1 = self.app.get(
401            'https://localhost/login?service=http%3A%2F%2Fservice.com%2F')
402        assert resp1.status == '200 OK'
403        assert 'cas-tgc' not in self.app.cookies
404        form = resp1.forms[0]
405        form.set('username', 'bird')
406        form.set('password', 'bebop')
407        resp2 = form.submit('AUTHENTICATE')
408        assert resp2.status == '303 See Other'
409        assert resp2.headers['Location'].startswith(
410            'http://service.com/?ticket=ST-')
411        # we got a secure cookie
412        assert 'cas-tgc' in self.app.cookies
413        resp3 = self.app.get(
414            'https://localhost/login?service=http%3A%2F%2Fservice.com%2F')
415        assert resp3.status == '303 See Other'
416        assert resp3.headers['Location'].startswith(
417            'http://service.com/?ticket=ST-')
418
419
420class CASServerHelperTests(unittest.TestCase):
421
422    def setUp(self):
423        self.workdir = tempfile.mkdtemp()
424        self.db_file = os.path.join(self.workdir, 'mycas.db')
425        self.conn_string = 'sqlite:///%s' % self.db_file
426        self.db = DB(self.conn_string)
427
428    def tearDown(self):
429        shutil.rmtree(self.workdir)
430
431    def test_create_service_ticket(self):
432        # we can create service tickets
433        st = create_service_ticket(
434            user='bob', service='http://www.example.com')
435        assert isinstance(st, ServiceTicket)
436        # 3.1.1: service not part of ticket
437        assert 'example.com' not in st.ticket
438        # 3.1.1: ticket must start with 'ST-'
439        assert st.ticket.startswith('ST-')
440        # 3.1.1: min. ticket length clients must be able to process is 32
441        assert len(st.ticket) < 33
442        # 3.7: allowed character set == [a-zA-Z0-9\-]
443        assert RE_ALPHABET.match(st.ticket), (
444            'Ticket contains forbidden chars: %s' % st)
445
446    def test_create_login_ticket(self):
447        # we can create login tickets
448        lt = create_login_ticket()
449        # 3.5.1: ticket should start with 'LT-'
450        assert lt.ticket.startswith('LT-')
451        # 3.7: allowed character set == [a-zA-Z0-9\-]
452        assert RE_ALPHABET.match(lt.ticket), (
453            'Ticket contains forbidden chars: %s' % lt)
454
455    def test_create_login_ticket_unique(self):
456        # 3.5.1: login tickets are unique (although not hard to guess)
457        ticket_num = 1000  # increase to test more thoroughly
458        lt_list = [create_login_ticket() for x in range(ticket_num)]
459        assert len(set(lt_list)) == ticket_num
460
461    def test_create_tgc_value(self):
462        # we can create ticket granting cookies
463        tgc = create_tgc_value()
464        assert isinstance(tgc, TicketGrantingCookie)
465        # 3.6.1: cookie value should start with 'TGC-'
466        assert tgc.value.startswith('TGC-')
467        # 3.7: allowed character set == [a-zA-Z0-9\-]
468        assert RE_ALPHABET.match(tgc.value), (
469            'Cookie value contains forbidden chars: %s' % tgc)
470
471    def test_check_login_ticket(self):
472        db = DB('sqlite:///')
473        lt = LoginTicket('LT-123456')
474        db.add(lt)
475        assert check_login_ticket(db, None) is False
476        assert check_login_ticket(db, 'LT-123456') is True
477        # the ticket will be removed after check
478        assert check_login_ticket(db, 'LT-123456') is False
479        assert check_login_ticket(db, 'LT-654321') is False
480
481    def test_set_session_cookie(self):
482        # make sure we can add session cookies to responses
483        db = DB('sqlite:///')
484        resp = set_session_cookie(db, Response())
485        assert 'Set-Cookie' in resp.headers
486        cookie = resp.headers['Set-Cookie']
487        assert RE_COOKIE.match(cookie), (
488            'Cookie in unexpected format: %s' % cookie)
489        # the cookie is stored in database
490        value = cookie.split('=')[1].split(';')[0]
491        q = db.query(TicketGrantingCookie).filter(
492            TicketGrantingCookie.value == value)
493        assert len(list(q)) == 1
494
495    def test_check_session_cookie(self):
496        db = DB('sqlite:///')
497        tgc = create_tgc_value()
498        db.add(tgc)
499        value = tgc.value
500        assert check_session_cookie(db, value) == tgc
501        assert check_session_cookie(db, 'foo') is None
502        assert check_session_cookie(db, b'foo') is None
503        assert check_session_cookie(db, None) is None
504        value2 = value.encode('utf-8')
505        assert check_session_cookie(db, value2) == tgc
506
507    def test_get_template(self):
508        # we can load templates
509        assert get_template('not-existing-template') is None
510        assert get_template('login.html') is not None
511
512    def test_delete_session_cookie(self):
513        # we can unset cookies
514        db = DB('sqlite:///')
515        tgc = create_tgc_value()
516        db.add(tgc)
517        value = tgc.value
518        resp = delete_session_cookie(db, Response(), old_value=value)
519        assert 'Set-Cookie' in resp.headers
520        cookie = resp.headers['Set-Cookie']
521        assert RE_COOKIE_DEL.match(cookie), (
522            'Cookie in unexpected format: %s' % cookie)
523        # the cookie values was deleted from database
524        q = db.query(TicketGrantingCookie).filter(
525            TicketGrantingCookie.value == value)
526        assert len(list(q)) == 0
527
528    def test_check_service_ticket(self):
529        db = DB('sqlite:///')
530        st = ServiceTicket(
531            'ST-123456', 'someuser', 'http://myservice.com', True)
532        db.add(st)
533        assert check_service_ticket(db, None, 'foo') is None
534        assert check_service_ticket(db, 'foo', None) is None
535        assert check_service_ticket(db, 'ST-123456', 'foo') is None
536        assert check_service_ticket(db, 'foo', 'http://myservice.com') is None
537        result = check_service_ticket(db, 'ST-123456', 'http://myservice.com')
538        assert isinstance(result, ServiceTicket)
539        assert result.user == 'someuser'
540        assert check_service_ticket(
541            db,  'ST-123456', 'http://myservice.com', True) is None
542        assert check_service_ticket(
543            db,  'ST-123456', 'http://myservice.com', False) is not None
Note: See TracBrowser for help on using the repository browser.