1 | #!/usr/bin/env python |
---|
2 | """Unit tests for WSGI HTTP Basic Auth handler |
---|
3 | |
---|
4 | NERC DataGrid Project |
---|
5 | """ |
---|
6 | __author__ = "P J Kershaw" |
---|
7 | __date__ = "13/10/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$' |
---|
12 | import logging |
---|
13 | logging.basicConfig(level=logging.DEBUG) |
---|
14 | |
---|
15 | import urllib2 |
---|
16 | import base64 |
---|
17 | import paste.fixture |
---|
18 | from ndg.security.test.unit import BaseTestCase |
---|
19 | from ndg.security.server.wsgi.authn import HTTPBasicAuthMiddleware, \ |
---|
20 | HTTPBasicAuthUnauthorized |
---|
21 | |
---|
22 | |
---|
23 | class TestAuthnApp(object): |
---|
24 | '''Test Application for the Authentication handler to protect''' |
---|
25 | response = "Test HTTP Basic Authentication application" |
---|
26 | |
---|
27 | def __init__(self, app_conf, **local_conf): |
---|
28 | pass |
---|
29 | |
---|
30 | def __call__(self, environ, start_response): |
---|
31 | |
---|
32 | if environ['PATH_INFO'] == '/test_200': |
---|
33 | status = "200 OK" |
---|
34 | else: |
---|
35 | status = "404 Not found" |
---|
36 | |
---|
37 | start_response(status, |
---|
38 | [('Content-length', |
---|
39 | str(len(TestAuthnApp.response))), |
---|
40 | ('Content-type', 'text/plain')]) |
---|
41 | return [TestAuthnApp.response] |
---|
42 | |
---|
43 | |
---|
44 | class HTTPBasicAuthPluginMiddleware(object): |
---|
45 | USERNAME = 'testuser' |
---|
46 | PASSWORD = 'password' |
---|
47 | |
---|
48 | def __init__(self, app): |
---|
49 | self._app = app |
---|
50 | |
---|
51 | def __call__(self, environ, start_response): |
---|
52 | def authenticate(environ, username, password): |
---|
53 | if username == HTTPBasicAuthPluginMiddleware.USERNAME and \ |
---|
54 | password == HTTPBasicAuthPluginMiddleware.PASSWORD: |
---|
55 | return |
---|
56 | else: |
---|
57 | raise HTTPBasicAuthUnauthorized("Invalid credentials") |
---|
58 | |
---|
59 | environ['authenticate'] = authenticate |
---|
60 | return self._app(environ, start_response) |
---|
61 | |
---|
62 | |
---|
63 | class HTTPBasicAuthMiddlewareTestCase(BaseTestCase): |
---|
64 | SERVICE_PORTNUM = 10443 |
---|
65 | WGET_CMD = 'wget' |
---|
66 | WGET_USER_OPTNAME = '--http-user' |
---|
67 | WGET_PASSWD_OPTNAME = '--http-password' |
---|
68 | WGET_OUTPUT_OPTNAME = '--output-document' |
---|
69 | WGET_STDOUT = '-' |
---|
70 | |
---|
71 | def __init__(self, *args, **kwargs): |
---|
72 | app = TestAuthnApp({}) |
---|
73 | app = HTTPBasicAuthMiddleware(app, {}, prefix='', |
---|
74 | authnFunc='authenticate') |
---|
75 | self.wsgiapp = HTTPBasicAuthPluginMiddleware(app) |
---|
76 | |
---|
77 | self.app = paste.fixture.TestApp(self.wsgiapp) |
---|
78 | |
---|
79 | BaseTestCase.__init__(self, *args, **kwargs) |
---|
80 | |
---|
81 | def test01PasteFixture(self): |
---|
82 | username = HTTPBasicAuthPluginMiddleware.USERNAME |
---|
83 | password = HTTPBasicAuthPluginMiddleware.PASSWORD |
---|
84 | |
---|
85 | base64String = base64.encodestring('%s:%s' % (username, password))[:-1] |
---|
86 | authHeader = "Basic %s" % base64String |
---|
87 | headers = {'Authorization': authHeader} |
---|
88 | |
---|
89 | url = '/test_200' |
---|
90 | |
---|
91 | response = self.app.get(url, headers=headers, status=200) |
---|
92 | print response |
---|
93 | |
---|
94 | def test02Urllib2Client(self): |
---|
95 | # Thread separate Paster based service |
---|
96 | self.addService(app=self.wsgiapp, |
---|
97 | port=HTTPBasicAuthMiddlewareTestCase.SERVICE_PORTNUM) |
---|
98 | |
---|
99 | username = HTTPBasicAuthPluginMiddleware.USERNAME |
---|
100 | password = HTTPBasicAuthPluginMiddleware.PASSWORD |
---|
101 | url = 'http://localhost:%d/test_200' % \ |
---|
102 | HTTPBasicAuthMiddlewareTestCase.SERVICE_PORTNUM |
---|
103 | |
---|
104 | req = urllib2.Request(url) |
---|
105 | base64String = base64.encodestring('%s:%s' % (username, password))[:-1] |
---|
106 | authHeader = "Basic %s" % base64String |
---|
107 | req.add_header("Authorization", authHeader) |
---|
108 | |
---|
109 | handle = urllib2.urlopen(req) |
---|
110 | |
---|
111 | response = handle.read() |
---|
112 | print (response) |
---|
113 | |
---|
114 | def test03WGetClient(self): |
---|
115 | uri = ('http://localhost:%d/test_200' % |
---|
116 | HTTPBasicAuthMiddlewareTestCase.SERVICE_PORTNUM) |
---|
117 | |
---|
118 | username = HTTPBasicAuthPluginMiddleware.USERNAME |
---|
119 | password = HTTPBasicAuthPluginMiddleware.PASSWORD |
---|
120 | |
---|
121 | import os |
---|
122 | import subprocess |
---|
123 | cmd = "%s %s %s=%s %s=%s %s=%s" % ( |
---|
124 | HTTPBasicAuthMiddlewareTestCase.WGET_CMD, |
---|
125 | uri, |
---|
126 | HTTPBasicAuthMiddlewareTestCase.WGET_USER_OPTNAME, |
---|
127 | username, |
---|
128 | HTTPBasicAuthMiddlewareTestCase.WGET_PASSWD_OPTNAME, |
---|
129 | password, |
---|
130 | HTTPBasicAuthMiddlewareTestCase.WGET_OUTPUT_OPTNAME, |
---|
131 | HTTPBasicAuthMiddlewareTestCase.WGET_STDOUT) |
---|
132 | |
---|
133 | p = subprocess.Popen(cmd, shell=True) |
---|
134 | status = os.waitpid(p.pid, 0) |
---|
135 | self.failIf(status[-1] != 0, "Expecting 0 exit status for %r" % cmd) |
---|
136 | |
---|
137 | |
---|
138 | if __name__ == "__main__": |
---|
139 | unittest.main() |
---|