1 | #!/usr/bin/env python |
---|
2 | |
---|
3 | """Test harness for NDG Session Manager client - makes requests for |
---|
4 | authentication and authorisation. An Attribute Authority and Simple CA |
---|
5 | services must be running for the reqAuthorisation and addUser tests |
---|
6 | |
---|
7 | NERC Data Grid Project |
---|
8 | |
---|
9 | @author P J Kershaw |
---|
10 | |
---|
11 | 23/02/06 |
---|
12 | |
---|
13 | Renamed from SessionClientTest.py 27/0/4/06 |
---|
14 | Moved and renamed SessionMgrClientTest.py 23/11/06 |
---|
15 | |
---|
16 | @copyright (C) 2007 CCLRC & NERC |
---|
17 | |
---|
18 | @license This software may be distributed under the terms of the Q Public |
---|
19 | License, version 1.0 or later. |
---|
20 | """ |
---|
21 | reposID = "$Id:$" |
---|
22 | |
---|
23 | import unittest |
---|
24 | import os, sys, getpass |
---|
25 | from ConfigParser import SafeConfigParser |
---|
26 | |
---|
27 | from ndg.security.common.SessionMgr import SessionMgrClient, \ |
---|
28 | AttributeRequestDenied |
---|
29 | |
---|
30 | from ndg.security.common.SessionCookie import SessionCookie |
---|
31 | |
---|
32 | |
---|
33 | class SessionMgrClientTestCase(unittest.TestCase): |
---|
34 | |
---|
35 | def setUp(self): |
---|
36 | |
---|
37 | configParser = SafeConfigParser() |
---|
38 | configParser.read("./sessionMgrClientTest.cfg") |
---|
39 | |
---|
40 | self.cfg = {} |
---|
41 | for section in configParser.sections(): |
---|
42 | self.cfg[section] = dict(configParser.items(section)) |
---|
43 | |
---|
44 | tracefile = sys.stderr |
---|
45 | |
---|
46 | try: |
---|
47 | if self.cfg['setUp'].get('clntprikeypwd') is None: |
---|
48 | clntPriKeyPwd = getpass.getpass(\ |
---|
49 | prompt="\nsetUp - client private key password: ") |
---|
50 | else: |
---|
51 | clntPriKeyPwd = self.cfg['setUp'].get('clntprikeypwd') |
---|
52 | except KeyboardInterrupt: |
---|
53 | sys.exit(0) |
---|
54 | |
---|
55 | # Initialise the Session Manager client connection |
---|
56 | # Omit traceFile keyword to leave out SOAP debug info |
---|
57 | self.clnt = SessionMgrClient(uri=self.cfg['setUp']['smuri'], |
---|
58 | verifyingCertFilePath=self.cfg['setUp']['srvcertfilepath'], |
---|
59 | signingCertFilePath=self.cfg['setUp']['clntcertfilepath'], |
---|
60 | signingPriKeyFilePath=self.cfg['setUp']['clntprikeyfilepath'], |
---|
61 | signingPriKeyPwd=clntPriKeyPwd, |
---|
62 | tracefile=tracefile) |
---|
63 | |
---|
64 | self.sessCookie = None |
---|
65 | self.proxyCert = None |
---|
66 | self.proxyPriKey = None |
---|
67 | self.userCert = None |
---|
68 | |
---|
69 | def test1AddUser(self): |
---|
70 | """Add a new user ID to the MyProxy repository""" |
---|
71 | |
---|
72 | passphrase = self.cfg['test1AddUser'].get('passphrase') or \ |
---|
73 | getpass.getpass(prompt="\ntest1AddUser pass-phrase for new user: ") |
---|
74 | |
---|
75 | # Note the pass-phrase is read from the file tmp. To pass |
---|
76 | # explicitly as a string use the 'passphrase' keyword instead |
---|
77 | self.clnt.addUser(self.cfg['test1AddUser']['username'], |
---|
78 | passphrase=passphrase) |
---|
79 | print "Added user '%s'" % self.cfg['test1AddUser']['username'] |
---|
80 | |
---|
81 | |
---|
82 | def test2CookieConnect(self): |
---|
83 | """test2CookieConnect: Connect as if acting as a browser client - |
---|
84 | a cookie is returned""" |
---|
85 | |
---|
86 | passphrase = self.cfg['test2CookieConnect'].get('passphrase') or \ |
---|
87 | getpass.getpass(prompt="\ntest2CookieConnect pass-phrase for user: ") |
---|
88 | |
---|
89 | self.proxyCert, self.proxyPriKey, self.userCert, cookie = \ |
---|
90 | self.clnt.connect(self.cfg['test2CookieConnect']['username'], |
---|
91 | passphrase=passphrase, |
---|
92 | getCookie=True) |
---|
93 | |
---|
94 | self.sessCookie = SessionCookie(cookie) |
---|
95 | print "User '%s' connected to Session Manager:\n%s" % \ |
---|
96 | (self.cfg['test2CookieConnect']['username'], self.sessCookie) |
---|
97 | |
---|
98 | |
---|
99 | def test3ProxyCertConnect(self): |
---|
100 | """test3ProxyCertConnect: Connect as a command line client - |
---|
101 | a proxyCert is returned""" |
---|
102 | |
---|
103 | passphrase = self.cfg['test3ProxyCertConnect'].get('passphrase') or \ |
---|
104 | getpass.getpass(\ |
---|
105 | prompt="\ntest3ProxyCertConnect pass-phrase for user: ") |
---|
106 | |
---|
107 | self.proxyCert, self.proxyPriKey, self.userCert, null = \ |
---|
108 | self.clnt.connect(self.cfg['test3ProxyCertConnect']['username'], |
---|
109 | passphrase=passphrase, |
---|
110 | getCookie=False) |
---|
111 | print "User '%s' connected to Session Manager:\n%s" % \ |
---|
112 | (self.cfg['test3ProxyCertConnect']['username'], self.proxyCert) |
---|
113 | |
---|
114 | |
---|
115 | def test4CookieDisconnect(self): |
---|
116 | """test4CookieDisconnect: disconnect as if acting as a browser client - |
---|
117 | a cookie is returned""" |
---|
118 | |
---|
119 | print "\n\t" + self.test4CookieDisconnect.__doc__ |
---|
120 | self.test2CookieConnect() |
---|
121 | |
---|
122 | # Use proxy cert / private key just obtained from connect call for |
---|
123 | # signature generation |
---|
124 | self.clnt.signatureHandler.signingCert = self.proxyCert |
---|
125 | self.clnt.signatureHandler.signingCertPriKey = self.proxyPriKey |
---|
126 | |
---|
127 | self.clnt.disconnect(#userCert=self.userCert, |
---|
128 | #sessCookie=str(self.sessCookie) |
---|
129 | #sessID="A", |
---|
130 | #encrSessionMgrURI="B" |
---|
131 | ) |
---|
132 | |
---|
133 | print "User disconnected from Session Manager:\n%s" % self.sessCookie |
---|
134 | |
---|
135 | |
---|
136 | def test5ProxyCertDisconnect(self): |
---|
137 | """test5ProxyCertDisconnect: Connect as a command line client - |
---|
138 | a proxyCert is returned""" |
---|
139 | |
---|
140 | print "\n\t" + self.test5ProxyCertDisconnect.__doc__ |
---|
141 | self.test3ProxyCertConnect() |
---|
142 | |
---|
143 | # Use proxy to sign outbound SOAP message |
---|
144 | self.clnt.signingCert = self.proxyCert |
---|
145 | self.clnt.signingKey = self.proxyPriKey |
---|
146 | self.clnt.signingPriKeyPwd = None |
---|
147 | |
---|
148 | self.clnt.disconnect(proxyCert=self.proxyCert) |
---|
149 | print "User disconnected from Session Manager:\n%s" % self.proxyCert |
---|
150 | |
---|
151 | |
---|
152 | def test6CookieGetAttCert(self): |
---|
153 | """test6CookieGetAttCert: make an attribute request using |
---|
154 | a cookie as authentication credential""" |
---|
155 | |
---|
156 | print "\n\t" + self.test6CookieGetAttCert.__doc__ |
---|
157 | self.test2CookieConnect() |
---|
158 | |
---|
159 | attCert, extAttCertList = self.clnt.getAttCert(\ |
---|
160 | sessID=self.sessCookie.sessionID, |
---|
161 | encrSessionMgrURI=self.sessCookie.encrSessionMgrURI, |
---|
162 | attAuthorityURI=self.cfg['test6CookieGetAttCert']['aauri']) |
---|
163 | |
---|
164 | print "Attribute Certificate:\n%s" % attCert |
---|
165 | print "External Attribute Certificate List:\n%s" % extAttCertList |
---|
166 | |
---|
167 | |
---|
168 | def test6aCookieGetAttCertRefused(self): |
---|
169 | """test6aCookieGetAttCertRefused: make an attribute request using |
---|
170 | a cookie as authentication credential requesting an AC from an |
---|
171 | Attribute Authority where the user is NOT registered""" |
---|
172 | |
---|
173 | print "\n\t" + self.test6aCookieGetAttCertRefused.__doc__ |
---|
174 | self.test2CookieConnect() |
---|
175 | |
---|
176 | aaURI = self.cfg['test6aCookieGetAttCertRefused']['aauri'] |
---|
177 | |
---|
178 | try: |
---|
179 | attCert, extAttCertList = self.clnt.getAttCert(\ |
---|
180 | sessID=self.sessCookie.sessionID, |
---|
181 | encrSessionMgrURI=self.sessCookie.encrSessionMgrURI, |
---|
182 | attAuthorityURI=aaURI, |
---|
183 | mapFromTrustedHosts=False) |
---|
184 | except AttributeRequestDenied, e: |
---|
185 | print "SUCCESS - obtained expected result: %s" % e |
---|
186 | return |
---|
187 | |
---|
188 | self.fail("Request allowed from AA where user is NOT registered!") |
---|
189 | |
---|
190 | |
---|
191 | def test6bCookieGetMappedAttCert(self): |
---|
192 | """test6bCookieGetMappedAttCert: make an attribute request using |
---|
193 | a cookie as authentication credential""" |
---|
194 | |
---|
195 | print "\n\t" + self.test6bCookieGetMappedAttCert.__doc__ |
---|
196 | self.test2CookieConnect() |
---|
197 | |
---|
198 | attCert, extAttCertList = self.clnt.getAttCert(\ |
---|
199 | sessID=self.sessCookie.sessionID, |
---|
200 | encrSessionMgrURI=self.sessCookie.encrSessionMgrURI, |
---|
201 | attAuthorityURI=self.cfg['test6bCookieGetMappedAttCert']['aauri']) |
---|
202 | |
---|
203 | print "Attribute Certificate:\n%s" % attCert |
---|
204 | print "External Attribute Certificate List:\n%s" % extAttCertList |
---|
205 | |
---|
206 | |
---|
207 | def test6bCookieGetMappedAttCert(self): |
---|
208 | """test6CookieGetAttCert: make an attribute request using |
---|
209 | a cookie as authentication credential""" |
---|
210 | |
---|
211 | print "\n\t" + self.test6bCookieGetMappedAttCert.__doc__ |
---|
212 | self.test2CookieConnect() |
---|
213 | |
---|
214 | attCert, statusCode, msg, extAttCertList = self.clnt.getAttCert(\ |
---|
215 | sessID=self.sessCookie.sessionID, |
---|
216 | encrSessionMgrURI=self.sessCookie.encrSessionMgrURI, |
---|
217 | attAuthorityURI=self.cfg['test6bCookieGetMappedAttCert']['aauri']) |
---|
218 | |
---|
219 | print "Attribute Certificate:\n%s" % attCert |
---|
220 | print "Status: %s" % statusCode |
---|
221 | print "Message: %s" % msg |
---|
222 | print "External Attribute Certificate List:\n%s" % extAttCertList |
---|
223 | |
---|
224 | |
---|
225 | def test6cCookieGetAttCertWithExtAttCertList(self): |
---|
226 | """test6CookieGetAttCert: make an attribute request using |
---|
227 | a cookie as authentication credential""" |
---|
228 | |
---|
229 | print "\n\t" + self.test6cCookieGetAttCertWithExtAttCertList.__doc__ |
---|
230 | self.test2CookieConnect() |
---|
231 | |
---|
232 | aaURI = \ |
---|
233 | self.cfg['test6cCookieGetAttCertWithExtAttCertList']['aauri'] |
---|
234 | |
---|
235 | attCert, statusCode, msg, extAttCertList = self.clnt.getAttCert(\ |
---|
236 | sessID=self.sessCookie.sessionID, |
---|
237 | encrSessionMgrURI=self.sessCookie.encrSessionMgrURI, |
---|
238 | attAuthorityURI=aaURI, |
---|
239 | extAttCertList=['AC1', 'AC2', 'AC3']) |
---|
240 | |
---|
241 | print "Attribute Certificate:\n%s" % attCert |
---|
242 | print "Status: %s" % statusCode |
---|
243 | print "Message: %s" % msg |
---|
244 | print "External Attribute Certificate List:\n%s" % extAttCertList |
---|
245 | |
---|
246 | |
---|
247 | def test7ProxyCertGetAttCert(self): |
---|
248 | """test7ProxyCertGetAttCert: make an attribute request using |
---|
249 | a proxy cert as authentication credential""" |
---|
250 | print "\n\t" + self.test7ProxyCertGetAttCert.__doc__ |
---|
251 | self.test3ProxyCertConnect() |
---|
252 | |
---|
253 | # Request an attribute certificate from an Attribute Authority |
---|
254 | # using the proxyCert returned from connect() |
---|
255 | |
---|
256 | aaURI = self.cfg['test7ProxyCertGetAttCert']['aauri'] |
---|
257 | attCert, statusCode, msg, extAttCertList = self.clnt.getAttCert(\ |
---|
258 | proxyCert=self.proxyCert, |
---|
259 | attAuthorityURI=aaURI) |
---|
260 | |
---|
261 | print "Attribute Certificate:\n%s" % attCert |
---|
262 | print "Status: %s" % statusCode |
---|
263 | print "Message: %s" % msg |
---|
264 | print "External Attribute Certificate List:\n%s" % extAttCertList |
---|
265 | |
---|
266 | |
---|
267 | def test8GetX509Cert(self): |
---|
268 | "test8GetX509Cert: return the Session Manager's X.509 Cert." |
---|
269 | cert = self.clnt.getX509Cert() |
---|
270 | |
---|
271 | print "Session Manager X.509 Certificate:\n" + cert |
---|
272 | |
---|
273 | |
---|
274 | #_____________________________________________________________________________ |
---|
275 | class SessionMgrClientTestSuite(unittest.TestSuite): |
---|
276 | |
---|
277 | def __init__(self): |
---|
278 | map = map(SessionMgrClientTestCase, |
---|
279 | ( |
---|
280 | "test1AddUser", |
---|
281 | "test2CookieConnect", |
---|
282 | "test3ProxyCertConnect", |
---|
283 | "test4CookieDisconnect", |
---|
284 | "test5ProxyCertDisconnect", |
---|
285 | "test6CookieGetAttCert", |
---|
286 | "test6bCookieGetMappedAttCert", |
---|
287 | "test6cCookieGetAttCertWithExtAttCertList", |
---|
288 | "test7ProxyCertGetAttCert", |
---|
289 | "test8GetX509Cert", |
---|
290 | )) |
---|
291 | unittest.TestSuite.__init__(self, map) |
---|
292 | |
---|
293 | |
---|
294 | if __name__ == "__main__": |
---|
295 | unittest.main() |
---|