source: TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test/unit/wsgi/authz/test_authz.py @ 6264

Subversion URL: http://proj.badc.rl.ac.uk/svn/ndg/TI12-security/trunk/NDGSecurity/python/ndg_security_test/ndg/security/test/unit/wsgi/authz/test_authz.py@6264
Revision 6264, 14.9 KB checked in by pjkersha, 10 years ago (diff)
  • Refactored PEP result handler code from authz into separate ndg.security.server.wsgi.authz.result_handler package
  • Refactored session handling classes from ndg.security.server.wsgi.authn to new ndg.security.server.wsgi.session module
Line 
1#!/usr/bin/env python
2"""Unit tests for WSGI Authorization handler
3
4NERC DataGrid Project
5"""
6__author__ = "P J Kershaw"
7__date__ = "21/05/09"
8__copyright__ = "(C) 2009 Science and Technology Facilities Council"
9__license__ = "BSD - see LICENSE file in top-level directory"
10__contact__ = "Philip.Kershaw@stfc.ac.uk"
11__revision__ = '$Id: $'
12import logging
13
14
15import unittest
16import os
17from urlparse import urlunsplit
18
19from os.path import expandvars as xpdVars
20from os.path import join as jnPath
21mkPath = lambda file: jnPath(os.environ['NDGSEC_COMBINED_SRVS_UNITTEST_DIR'], 
22                             file)
23
24import paste.fixture
25from paste.deploy import loadapp
26
27from ndg.security.test.unit import BaseTestCase
28from ndg.security.server.wsgi import NDGSecurityMiddlewareBase
29from ndg.security.server.wsgi.authz import (NdgPIPMiddlewareConfigError,
30                                            SamlPIPMiddlewareConfigError, 
31                                            PEPResultHandlerMiddleware)
32from ndg.security.common.authz.msi import Response
33
34
35class RedirectFollowingAccessDenied(PEPResultHandlerMiddleware):
36   
37    @NDGSecurityMiddlewareBase.initCall
38    def __call__(self, environ, start_response):
39       
40        queryString = environ.get('QUERY_STRING', '')
41        if 'admin=1' in queryString:
42            # User has been rejected access to a URI requiring admin rights,
43            # try redirect to the same URI minus the admin query arg, this
44            # request will pass because admin rights aren't needed
45            queryArgs = queryString.split('&')
46            queryList = [arg for arg in queryArgs if arg != 'admin=1']
47            editedQuery = '&'.join(queryList)
48            redirectURI = urlunsplit(('', '', self.pathInfo, editedQuery, ''))
49            return self.redirect(redirectURI)
50        else:
51            return super(RedirectFollowingAccessDenied, self).__call__(
52                                                            environ,
53                                                            start_response)
54
55       
56class TestAuthZMiddleware(object):
57    '''Test Application for the Authentication handler to protect'''
58    response = "Test Authorization application"
59       
60    def __init__(self, app_conf, **local_conf):
61        pass
62   
63    def __call__(self, environ, start_response):
64       
65        if environ['PATH_INFO'] == '/test_401':
66            status = "401 Unauthorized"
67           
68        elif environ['PATH_INFO'] == '/test_403':
69            status = "403 Forbidden"
70           
71        elif environ['PATH_INFO'] == '/test_200':
72            status = "200 OK"
73           
74        elif environ['PATH_INFO'] == '/test_accessDeniedToSecuredURI':
75            # Nb. AuthZ middleware should intercept the request and bypass this
76            # response
77            status = "200 OK"
78           
79        elif environ['PATH_INFO'] == '/test_accessGrantedToSecuredURI':
80            status = "200 OK"
81        else:
82            status = "404 Not found"
83               
84        start_response(status,
85                       [('Content-length', 
86                         str(len(TestAuthZMiddleware.response))),
87                        ('Content-type', 'text/plain')])
88        return [TestAuthZMiddleware.response]
89
90
91class BeakerSessionStub(dict):
92    """Emulate beaker.session session object for purposes of the unit tests
93    """
94    def save(self):
95        pass
96 
97   
98class NdgWSGIAuthZTestCase(BaseTestCase):
99
100    def __init__(self, *args, **kwargs):
101        BaseTestCase.__init__(self, *args, **kwargs)
102       
103        here_dir = os.path.dirname(os.path.abspath(__file__))
104        wsgiapp = loadapp('config:ndg-test.ini', relative_to=here_dir)
105        self.app = paste.fixture.TestApp(wsgiapp)
106       
107        self.startSiteAAttributeAuthority()
108       
109    def test01CatchNoBeakerSessionFound(self):
110       
111        # PEPFilterConfigError is raised if no beaker.session is set in
112        # environ
113        try:
114            response = self.app.get('/test_200')
115        except NdgPIPMiddlewareConfigError, e:
116            print("ok - expected: %s exception: %s" % (e.__class__, e))
117       
118    def test02Ensure200WithNotLoggedInAndUnsecuredURI(self):
119       
120        # Check the authZ middleware leaves the response alone if the URI
121        # is not matched in the policy
122       
123        # Simulate a beaker.session in the environ
124        extra_environ={'beaker.session.ndg.security':BeakerSessionStub()}
125        response = self.app.get('/test_200',
126                                extra_environ=extra_environ)
127
128    def test03Catch401WithLoggedIn(self):
129       
130        # Check that the application being secured can raise a HTTP 401
131        # response and that this respected by the Authorization middleware
132        # even though a user is set in the session
133       
134        extra_environ={'beaker.session.ndg.security':
135                       BeakerSessionStub(username='testuser')}
136        response = self.app.get('/test_401', 
137                                extra_environ=extra_environ,
138                                status=401)
139
140    def test04Catch403WithLoggedIn(self):
141       
142        # Check that the application being secured can raise a HTTP 403
143        # response and that this respected by the Authorization middleware
144        # even though a user is set in the session
145       
146        extra_environ={'beaker.session.ndg.security':
147                       BeakerSessionStub(username='testuser')}
148        response = self.app.get('/test_403', 
149                                extra_environ=extra_environ,
150                                status=403)
151
152    def test05Catch401WithNotLoggedInAndSecuredURI(self):
153       
154        # AuthZ middleware grants access because the URI requested is not
155        # targeted in the policy
156       
157        # AuthZ middleware checks for username key in session set by AuthN
158        # handler
159        extra_environ={'beaker.session.ndg.security':BeakerSessionStub()}       
160        response = self.app.get('/test_accessDeniedToSecuredURI',
161                                extra_environ=extra_environ,
162                                status=401)
163       
164    def test06AccessDeniedForSecuredURI(self):
165       
166        # User is logged in but doesn't have the required credentials for
167        # access
168        extra_environ={'beaker.session.ndg.security':
169                       BeakerSessionStub(username='testuser')}
170       
171        response = self.app.get('/test_accessDeniedToSecuredURI',
172                                extra_environ=extra_environ,
173                                status=403)
174        self.assert_("Insufficient privileges to access the "
175                     "resource" in response)
176        print response
177
178    def test07AccessGrantedForSecuredURI(self):
179       
180        # User is logged in and has credentials for access to a URI secured
181        # by the policy file
182        extra_environ={'beaker.session.ndg.security':
183                       BeakerSessionStub(username='testuser')}
184       
185        response = self.app.get('/test_accessGrantedToSecuredURI',
186                                extra_environ=extra_environ,
187                                status=200)
188        self.assert_(TestAuthZMiddleware.response in response)
189        print response
190
191    def test08AccessDeniedForAdminQueryArg(self):
192       
193        # User is logged in but doesn't have the required credentials for
194        # access
195        extra_environ={'beaker.session.ndg.security':
196                       BeakerSessionStub(username='testuser')}
197       
198        # Try this URI with the query arg admin=1.  This will be picked up
199        # by the policy as a request requiring admin rights.  The request is
200        # denied as the user doesn't have these rights but this then calls
201        # into play the PEP result handler defined in this module,
202        # RedirectFollowingAccessDenied.  This class reinvokes the request
203        # but without the admin query argument.  Access is then granted for
204        # the redirected request
205        response = self.app.get('/test_accessGrantedToSecuredURI',
206                                params={'admin': 1},
207                                extra_environ=extra_environ,
208                                status=302)
209        try:
210            redirectResponse = response.follow(extra_environ=extra_environ)
211        except paste.fixture.AppError, e:
212            self.failIf(TestAuthZMiddleware.response not in response)
213        print response
214
215       
216class TestAuthZMiddleware(object):
217    '''Test Application for the Authentication handler to protect'''
218    response = "Test Authorization application"
219       
220    def __init__(self, app_conf, **local_conf):
221        pass
222   
223    def __call__(self, environ, start_response):
224       
225        if environ['PATH_INFO'] == '/test_401':
226            status = "401 Unauthorized"
227           
228        elif environ['PATH_INFO'] == '/test_403':
229            status = "403 Forbidden"
230           
231        elif environ['PATH_INFO'] == '/test_200':
232            status = "200 OK"
233           
234        elif environ['PATH_INFO'] == '/test_accessDeniedToSecuredURI':
235            # Nb. AuthZ middleware should intercept the request and bypass this
236            # response
237            status = "200 OK"
238           
239        elif environ['PATH_INFO'] == '/test_accessGrantedToSecuredURI':
240            status = "200 OK"
241        else:
242            status = "404 Not found"
243               
244        start_response(status,
245                       [('Content-length', 
246                         str(len(TestAuthZMiddleware.response))),
247                        ('Content-type', 'text/plain')])
248        return [TestAuthZMiddleware.response]
249
250
251class BeakerSessionStub(dict):
252    """Emulate beaker.session session object for purposes of the unit tests
253    """
254    def save(self):
255        pass
256 
257   
258class SamlWSGIAuthZTestCase(BaseTestCase):
259
260    def __init__(self, *args, **kwargs):       
261        BaseTestCase.__init__(self, *args, **kwargs)
262
263        here_dir = os.path.dirname(os.path.abspath(__file__))
264        wsgiapp = loadapp('config:saml-test.ini', relative_to=here_dir)
265        self.app = paste.fixture.TestApp(wsgiapp)
266       
267        self.startSiteAAttributeAuthority(withSSL=True,
268            port=SamlWSGIAuthZTestCase.SITEA_SSL_ATTRIBUTEAUTHORITY_PORTNUM)
269       
270
271    def test01CatchNoBeakerSessionFound(self):
272       
273        # PEPFilterConfigError is raised if no beaker.session is set in
274        # environ
275        try:
276            response = self.app.get('/test_200')
277        except SamlPIPMiddlewareConfigError, e:
278            print("ok - expected: %s exception: %s" % (e.__class__, e))
279       
280    def test02Ensure200WithNotLoggedInAndUnsecuredURI(self):
281       
282        # Check the authZ middleware leaves the response alone if the URI
283        # is not matched in the policy
284       
285        # Simulate a beaker.session in the environ
286        extra_environ={'beaker.session.ndg.security':BeakerSessionStub()}
287        response = self.app.get('/test_200',
288                                extra_environ=extra_environ)
289
290    def test03Catch401WithLoggedIn(self):
291       
292        # Check that the application being secured can raise a HTTP 401
293        # response and that this respected by the Authorization middleware
294        # even though a user is set in the session
295       
296        extra_environ = {
297            'beaker.session.ndg.security':
298                BeakerSessionStub(username=SamlWSGIAuthZTestCase.OPENID_URI)
299        }
300        response = self.app.get('/test_401', 
301                                extra_environ=extra_environ,
302                                status=401)
303
304    def test04Catch403WithLoggedIn(self):
305       
306        # Check that the application being secured can raise a HTTP 403
307        # response and that this respected by the Authorization middleware
308        # even though a user is set in the session
309       
310        extra_environ = {
311            'beaker.session.ndg.security':
312                BeakerSessionStub(username=SamlWSGIAuthZTestCase.OPENID_URI)
313        }
314        response = self.app.get('/test_403', 
315                                extra_environ=extra_environ,
316                                status=403)
317
318    def test05Catch401WithNotLoggedInAndSecuredURI(self):
319       
320        # AuthZ middleware grants access because the URI requested is not
321        # targeted in the policy
322       
323        # AuthZ middleware checks for username key in session set by AuthN
324        # handler
325        extra_environ={'beaker.session.ndg.security':BeakerSessionStub()}       
326        response = self.app.get('/test_accessDeniedToSecuredURI',
327                                extra_environ=extra_environ,
328                                status=401)
329       
330    def test06AccessDeniedForSecuredURI(self):
331       
332        # User is logged in but doesn't have the required credentials for
333        # access
334        extra_environ = {
335            'beaker.session.ndg.security':
336                BeakerSessionStub(username=SamlWSGIAuthZTestCase.OPENID_URI)
337        }
338       
339        response = self.app.get('/test_accessDeniedToSecuredURI',
340                                extra_environ=extra_environ,
341                                status=403)
342        self.assert_("Insufficient privileges to access the "
343                     "resource" in response)
344        print response
345
346    def test07AccessGrantedForSecuredURI(self):
347       
348        # User is logged in and has credentials for access to a URI secured
349        # by the policy file
350        extra_environ = {
351            'beaker.session.ndg.security':
352                BeakerSessionStub(username=SamlWSGIAuthZTestCase.OPENID_URI)
353        }
354       
355        response = self.app.get('/test_accessGrantedToSecuredURI',
356                                extra_environ=extra_environ,
357                                status=200)
358        self.assert_(TestAuthZMiddleware.response in response)
359        print response
360
361    def test08AccessDeniedForAdminQueryArg(self):
362       
363        # User is logged in but doesn't have the required credentials for
364        # access
365        extra_environ = {
366            'beaker.session.ndg.security':
367                BeakerSessionStub(username=SamlWSGIAuthZTestCase.OPENID_URI)
368        }
369       
370        # Try this URI with the query arg admin=1.  This will be picked up
371        # by the policy as a request requiring admin rights.  The request is
372        # denied as the user doesn't have these rights but this then calls
373        # into play the PEP result handler defined in this module,
374        # RedirectFollowingAccessDenied.  This class reinvokes the request
375        # but without the admin query argument.  Access is then granted for
376        # the redirected request
377        response = self.app.get('/test_accessGrantedToSecuredURI',
378                                params={'admin': 1},
379                                extra_environ=extra_environ,
380                                status=302)
381        try:
382            redirectResponse = response.follow(extra_environ=extra_environ)
383        except paste.fixture.AppError, e:
384            self.failIf(TestAuthZMiddleware.response not in response)
385        print response
386
387
388if __name__ == "__main__":
389    unittest.main()       
Note: See TracBrowser for help on using the repository browser.