Fix standard urls. Test for Django up to 5.0.
[django-cas-provider.git] / cas_provider_examples / simple / tests.py
1 from io import StringIO
2 from urllib.parse import urlparse, parse_qsl, parse_qs
3 from urllib.request import install_opener
4 from xml import etree
5 from xml.etree import ElementTree
6 import cas_provider
7 from cas_provider.attribute_formatters import CAS
8 from cas_provider.models import ServiceTicket
9 from cas_provider.signals import cas_collect_custom_attributes
10 from cas_provider.views import _cas2_sucess_response, INVALID_TICKET, _cas2_error_response, generate_proxy_granting_ticket
11 from django.contrib.auth.models import User, UserManager
12 from django.test import TestCase
13 from django.conf import settings
14 from django.urls import reverse
15
16
17 class DummyOpener(object):
18     url = None
19
20     @staticmethod
21     def open(url, *args, **kwargs):
22         DummyOpener.url = url
23
24
25 class ViewsTest(TestCase):
26
27     fixtures = ['cas_users', ]
28
29
30
31     def setUp(self):
32         self.service = 'http://example.com/'
33
34
35     def test_successful_login_with_proxy(self):
36         install_opener(DummyOpener) # Don't really open any URLs
37         proxyTarget = "http://my.sweet.service"
38
39         response = self._login_user('root', '123')
40         response = self._validate_cas2(response, True, proxyTarget )
41
42         # Test: I'm acting as the service that will call another service
43         # Step 1: Get the proxy granting ticket
44         responseXml = ElementTree.parse(StringIO(response.content.decode('utf-8')))
45         auth_success = responseXml.find(CAS + 'authenticationSuccess')
46         pgt = auth_success.find(CAS + "proxyGrantingTicket")
47         user = auth_success.find(CAS + "user")
48         self.assertEqual('root', user.text)
49         self.assertIsNotNone(pgt.text)
50         self.assertTrue(pgt.text.startswith('PGTIOU'))
51
52         pgtId = parse_qs(urlparse(DummyOpener.url).query)['pgtId']
53
54         #Step 2: Get the actual proxy ticket
55         proxyTicketResponse = self.client.get(reverse('proxy'), {'targetService': proxyTarget, 'pgt': pgtId}, follow=False)
56         proxyTicketResponseXml = ElementTree.parse(StringIO(proxyTicketResponse.content.decode('utf-8')))
57         self.assertIsNotNone(proxyTicketResponseXml.find(CAS + "proxySuccess"))
58         self.assertIsNotNone(proxyTicketResponseXml.find(CAS + "proxySuccess/" + CAS + "proxyTicket"))
59         proxyTicket = proxyTicketResponseXml.find(CAS + "proxySuccess/" + CAS + "proxyTicket");
60
61         #Step 3: I have the proxy ticket I can talk to some other backend service as the currently logged in user!
62         proxyValidateResponse = self.client.get(reverse('cas_proxy_validate'), {'ticket': proxyTicket.text, 'service': proxyTarget})
63         proxyValidateResponseXml = ElementTree.parse(StringIO(proxyValidateResponse.content.decode('utf-8')))
64
65         auth_success_2 = proxyValidateResponseXml.find(CAS + 'authenticationSuccess')
66         user_2 = auth_success.find(CAS + "user")
67         proxies_1  = auth_success_2.find(CAS + "proxies")
68         self.assertIsNone(proxies_1) # there are no proxies. I am issued by a Service Ticket
69         
70         self.assertEqual(user.text, user_2.text)
71
72
73     def test_successful_proxy_chaining(self):
74         install_opener(DummyOpener) # Don't really open any URLs
75         proxyTarget_1 = "http://my.sweet.service_1"
76         proxyTarget_2 = "http://my.sweet.service_2"
77
78         response = self._login_user('root', '123')
79         response = self._validate_cas2(response, True, proxyTarget_1 )
80
81         # Test: I'm acting as the service that will call another service
82         # Step 1: Get the proxy granting ticket
83         responseXml = ElementTree.parse(StringIO(response.content.decode('utf-8')))
84         auth_success_1 = responseXml.find(CAS + 'authenticationSuccess')
85         pgt_1 = auth_success_1.find(CAS + "proxyGrantingTicket")
86         user_1 = auth_success_1.find(CAS + "user")
87         self.assertEqual('root', user_1.text)
88         self.assertIsNotNone(pgt_1.text)
89         self.assertTrue(pgt_1.text.startswith('PGTIOU'))
90
91         pgtId_1 = parse_qs(urlparse(DummyOpener.url).query)['pgtId']
92
93         #Step 2: Get the actual proxy ticket
94         proxyTicketResponse_1 = self.client.get(reverse('proxy'), {'targetService': proxyTarget_1, 'pgt': pgtId_1}, follow=False)
95         proxyTicketResponseXml_1 = ElementTree.parse(StringIO(proxyTicketResponse_1.content.decode('utf-8')))
96         self.assertIsNotNone(proxyTicketResponseXml_1.find(CAS + "proxySuccess"))
97         self.assertIsNotNone(proxyTicketResponseXml_1.find(CAS + "proxySuccess/" + CAS + "proxyTicket"))
98         proxyTicket_1 = proxyTicketResponseXml_1.find(CAS + "proxySuccess/" + CAS +"proxyTicket");
99
100         #Step 3: I'm backend service 1 - I have the proxy ticket - I want to talk to back service 2
101         #
102         proxyValidateResponse_1 = self.client.get(reverse('cas_proxy_validate'), {'ticket': proxyTicket_1.text, 'service': proxyTarget_1, 'pgtUrl': proxyTarget_2})
103         proxyValidateResponseXml_1 = ElementTree.parse(StringIO(proxyValidateResponse_1.content.decode('utf-8')))
104
105         auth_success_2 = proxyValidateResponseXml_1.find(CAS + 'authenticationSuccess')
106         user_2 = auth_success_2.find(CAS + "user")
107
108         proxies_1  = auth_success_2.find(CAS + "proxies")
109         self.assertIsNone(proxies_1) # there are no proxies. I am issued by a Service Ticket
110         self.assertIsNotNone(auth_success_2)
111         self.assertEqual('root', user_2.text)
112
113         pgt_2 = auth_success_2.find(CAS + "proxyGrantingTicket")
114         user = auth_success_2.find(CAS + "user")
115         self.assertEqual('root', user.text)
116         self.assertIsNotNone(pgt_2.text)
117         self.assertTrue(pgt_2.text.startswith('PGTIOU'))
118
119         pgtId_2 = parse_qs(urlparse(DummyOpener.url).query)['pgtId']
120
121         #Step 4: Get the second proxy ticket
122         proxyTicketResponse_2 = self.client.get(reverse('proxy'), {'targetService': proxyTarget_2, 'pgt': pgtId_2})
123         proxyTicketResponseXml_2 = ElementTree.parse(StringIO(proxyTicketResponse_2.content.decode('utf-8')))
124         self.assertIsNotNone(proxyTicketResponseXml_2.find(CAS + "proxySuccess"))
125         self.assertIsNotNone(proxyTicketResponseXml_2.find(CAS + "proxySuccess/" + CAS + "proxyTicket"))
126         proxyTicket_2 = proxyTicketResponseXml_2.find(CAS + "proxySuccess/" + CAS + "proxyTicket")
127
128         proxyValidateResponse_3 = self.client.get(reverse('cas_proxy_validate'), {'ticket': proxyTicket_2.text, 'service': proxyTarget_2, 'pgtUrl': ''})
129         proxyValidateResponseXml_3 = ElementTree.parse(StringIO(proxyValidateResponse_3.content.decode('utf-8')))
130
131         auth_success_3 = proxyValidateResponseXml_3.find(CAS + 'authenticationSuccess')
132         user_3 = auth_success_3.find(CAS + "user")
133
134         proxies_3  = auth_success_3.find(CAS + "proxies")
135         self.assertIsNotNone(proxies_3) # there should be a proxy. I am issued by a Proxy Ticket
136         proxy_3 = proxies_3.find(CAS + "proxy")
137         self.assertEqual(proxyTarget_1, proxy_3.text )
138
139         self.assertIsNotNone(auth_success_2)
140         self.assertEqual('root', user_3.text)
141
142
143     def test_succeessful_login(self):
144         response = self._login_user('root', '123')
145         self._validate_cas1(response, True)
146
147         response = self.client.get(reverse('cas_login'), {'service': self.service}, follow=False)
148         self.assertEqual(response.status_code, 302)
149         self.assertTrue(response['location'].startswith('%s?ticket=' % self.service))
150
151         response = self.client.get(reverse('cas_login'), follow=False)
152         self.assertEqual(response.status_code, 302)
153         self.assertTrue(response['location'].startswith('/'))
154
155         response = self.client.get(response['location'], follow=False)
156         self.assertIn(response.status_code, [302, 200])
157
158         response = self.client.get(reverse('cas_login'), {'service': self.service, 'warn': True}, follow=False)
159         self.assertEqual(response.status_code, 200)
160         self.assertTemplateUsed(response, 'cas/warn.html')
161
162
163     def _cas_logout(self):
164         response = self.client.get(reverse('cas_logout'), follow=False)
165         self.assertEqual(response.status_code, 200)
166
167
168     def test_logout(self):
169         response = self._login_user('root', '123')
170         self._validate_cas1(response, True)
171
172         self._cas_logout()
173
174         response = self.client.get(reverse('cas_login'), follow=False)
175         self.assertEqual(response.status_code, 200)
176         self.assertTrue(response.context['user'].is_anonymous)
177
178
179     def test_broken_pwd(self):
180         self._fail_login('root', '321')
181
182     def test_broken_username(self):
183         self._fail_login('notroot', '123')
184
185     def test_nonactive_user_login(self):
186         self._fail_login('nonactive', '123')
187
188     def test_cas2_success_validate(self):
189         response = self._login_user('root', '123')
190         response = self._validate_cas2(response, True)
191         user = User.objects.get(username=self.username)
192         self.assertEqual(response.content, _cas2_sucess_response(user).content)
193
194     def test_cas2_validate_twice(self):
195         response = self._login_user('root', '123')
196         response2 = self._validate_cas2(response, True)
197         user = User.objects.get(username=self.username)
198         self.assertEqual(response2.content, _cas2_sucess_response(user).content)
199
200         response2 = self._validate_cas2(response, True)
201         self.assertEqual(response2.content, _cas2_error_response(INVALID_TICKET).content)
202
203     def test_cas2_validate_twice_with_pgt(self):
204         proxyTarget = "http://my.sweet.service_1"
205         response = self._login_user('root', '123')
206         response2 = self._validate_cas2(response, True, proxyTarget)
207         user = User.objects.get(username=self.username)
208         self.assertEqual(response2.content, _cas2_sucess_response(user).content)
209
210         response2 = self._validate_cas2(response, True, proxyTarget)
211         self.assertEqual(response2.content, _cas2_error_response(INVALID_TICKET).content)
212
213     def test_cas2_custom_attrs(self):
214         cas_collect_custom_attributes.connect(cas_mapping)
215         response = self._login_user('editor', '123')
216         self.maxDiff=None
217
218         response = self._validate_cas2(response, True)
219         self.assertEqual(response.content.decode('utf-8'),
220                                            '''<cas:serviceResponse xmlns:cas="http://www.yale.edu/tp/cas">'''
221                                            '''<cas:authenticationSuccess>'''
222                                            '''<cas:user>editor</cas:user>'''
223                                            '''<cas:attributes>'''
224                                            '''<cas:attraStyle>Jasig</cas:attraStyle>'''
225                                            '''<cas:email>editor@exapmle.com</cas:email>'''
226                                            '''<cas:group>editor</cas:group>'''
227                                            '''<cas:is_active>True</cas:is_active>'''
228                                            '''<cas:is_staff>True</cas:is_staff>'''
229                                            '''</cas:attributes>'''
230                                            '''</cas:authenticationSuccess>'''
231                                            '''</cas:serviceResponse>''')
232
233         self._cas_logout()
234         response = self._login_user('editor', '123')
235         settings.CAS_CUSTOM_ATTRIBUTES_FORMATER = 'cas_provider.attribute_formatters.ruby_cas'
236         response = self._validate_cas2(response, True)
237         self.assertEqual(response.content.decode('utf-8'),
238                                            '''<cas:serviceResponse xmlns:cas="http://www.yale.edu/tp/cas">'''
239                                            '''<cas:authenticationSuccess>'''
240                                            '''<cas:user>editor</cas:user>'''
241                                            '''<cas:attraStyle>RubyCAS</cas:attraStyle>'''
242                                            '''<cas:email>editor@exapmle.com</cas:email>'''
243                                            '''<cas:group>editor</cas:group>'''
244                                            '''<cas:is_active>True</cas:is_active>'''
245                                            '''<cas:is_staff>True</cas:is_staff>'''
246                                            '''</cas:authenticationSuccess>'''
247                                            '''</cas:serviceResponse>''')
248
249         self._cas_logout()
250         response = self._login_user('editor', '123')
251         settings.CAS_CUSTOM_ATTRIBUTES_FORMATER = 'cas_provider.attribute_formatters.name_value'
252         response = self._validate_cas2(response, True)
253         self.assertEqual(response.content.decode('utf-8'),
254                                            '''<cas:serviceResponse xmlns:cas="http://www.yale.edu/tp/cas">'''
255                                            '''<cas:authenticationSuccess>'''
256                                            '''<cas:user>editor</cas:user>'''
257                                            '''<cas:attribute name="attraStyle" value="Name-Value"/>'''
258                                            '''<cas:attribute name="email" value="editor@exapmle.com"/>'''
259                                            '''<cas:attribute name="group" value="editor"/>'''
260                                            '''<cas:attribute name="is_active" value="True"/>'''
261                                            '''<cas:attribute name="is_staff" value="True"/>'''
262                                            '''</cas:authenticationSuccess>'''
263                                            '''</cas:serviceResponse>''')
264
265
266     def test_cas2_fail_validate(self):
267         for user, pwd in (('root', '321'), ('notroot', '123'), ('nonactive', '123')):
268             response = self._login_user(user, pwd)
269             self._validate_cas2(response, False)
270
271
272     def test_generate_proxy_granting_ticket(self):
273         install_opener(DummyOpener) # Don't really open any URLs
274         url = 'http://my.call.back/callhere'
275
276         user = User.objects.get(username = 'root')
277         st = ServiceTicket.objects.create(user = user )
278         pgt = generate_proxy_granting_ticket(url, st)
279         self.assertIsNotNone(pgt)
280
281         calledUrl = DummyOpener.url
282         parsedUrl = urlparse(calledUrl)
283         params = parse_qs(parsedUrl.query)
284         self.assertIsNotNone(params['pgtId'])
285         self.assertIsNotNone(params['pgtIou'])
286
287
288
289     def _fail_login(self, username, password):
290         response = self._login_user(username, password)
291         self._validate_cas1(response, False)
292
293         response = self.client.get(reverse('cas_login'), {'service': self.service}, follow=False)
294         self.assertEqual(response.status_code, 200)
295         response = self.client.get(reverse('cas_login'), follow=False)
296         self.assertEqual(response.status_code, 200)
297
298
299
300     def _login_user(self, username, password):
301         self.username = username
302         response = self.client.get(reverse('cas_login'), {'service': self.service})
303         self.assertEqual(response.status_code, 200)
304         self.assertTemplateUsed(response, 'cas/login.html')
305         form = response.context['form']
306         service = form['service'].value()
307         return self.client.post(reverse('cas_login'), {
308             'username': username,
309             'password': password,
310             'service': service
311         }, follow=False)
312
313
314     def _validate_cas1(self, response, is_correct=True):
315         if is_correct:
316             self.assertEqual(response.status_code, 302)
317             self.assertTrue(response.has_header('location'))
318             location = urlparse(response['location'])
319             ticket = location.query.split('=')[1]
320
321             response = self.client.get(reverse('cas_validate'), {'ticket': ticket, 'service': self.service}, follow=False)
322             self.assertEqual(response.status_code, 200)
323             self.assertEqual(response.content.decode('utf-8'), 'yes\n%s\n' % self.username)
324         else:
325             self.assertEqual(response.status_code, 200)
326             self.assertEqual(len(response.context['errors']), 1)
327
328             response = self.client.get(reverse('cas_validate'), {'ticket': 'ST-12312312312312312312312', 'service': self.service}, follow=False)
329             self.assertEqual(response.status_code, 200)
330             self.assertEqual(response.content.decode('utf-8'), 'no\n\n')
331
332
333     def _validate_cas2(self, response, is_correct=True, pgtUrl = None):
334         if is_correct:
335             self.assertEqual(response.status_code, 302)
336             self.assertTrue(response.has_header('location'))
337             location = urlparse(response['location'])
338             ticket = location.query.split('=')[1]
339             if pgtUrl:
340                 response = self.client.get(reverse('cas_service_validate'), {'ticket': ticket, 'service': self.service, 'pgtUrl': pgtUrl}, follow=False)
341             else:
342                 response = self.client.get(reverse('cas_service_validate'), {'ticket': ticket, 'service': self.service}, follow=False)
343             self.assertEqual(response.status_code, 200)
344         else:
345             self.assertEqual(response.status_code, 200)
346             self.assertEqual(len(response.context['errors']), 1)
347
348             response = self.client.get(reverse('cas_service_validate'), {'ticket': 'ST-12312312312312312312312', 'service': self.service}, follow=False)
349             self.assertEqual(response.status_code, 200)
350             self.assertEqual(response.content, _cas2_error_response(INVALID_TICKET).content)
351         return response
352
353 class ModelsTestCase(TestCase):
354
355     fixtures = ['cas_users.json', ]
356
357     def setUp(self):
358         self.user = User.objects.get(username='root')
359
360     def test_redirects(self):
361         ticket = ServiceTicket.objects.create(service='http://example.com', user=self.user)
362         self.assertEqual(ticket.get_redirect_url(), '%(service)s?ticket=%(ticket)s' % ticket.__dict__)
363
364
365 def cas_mapping(sender, user, **kwargs):
366     return {
367         'is_staff': str(user.is_staff),
368         'is_active': str(user.is_active),
369         'email': user.email,
370         'group': [g.name for g in user.groups.all()]
371     }