1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
|
from k5test import *
from base64 import b64encode
import shutil
realm = K5Realm(create_host=False, get_creds=False)
usercache = 'FILE:' + os.path.join(realm.testdir, 'usercache')
storagecache = 'FILE:' + os.path.join(realm.testdir, 'save')
# Create two service principals with keys in the default keytab.
service1 = 'service/1@%s' % realm.realm
realm.addprinc(service1)
realm.extract_keytab(service1, realm.keytab)
service2 = 'service/2@%s' % realm.realm
realm.addprinc(service2)
realm.extract_keytab(service2, realm.keytab)
puser = 'p:' + realm.user_princ
pservice1 = 'p:' + service1
pservice2 = 'p:' + service2
# Get forwardable creds for service1 in the default cache.
realm.kinit(service1, None, ['-f', '-k'])
# Try S4U2Self for user with a restricted password.
realm.run([kadminl, 'modprinc', '+needchange', realm.user_princ])
realm.run(['./t_s4u', 'e:user', '-'])
realm.run([kadminl, 'modprinc', '-needchange',
'-pwexpire', '1/1/2000', realm.user_princ])
realm.run(['./t_s4u', 'e:user', '-'])
realm.run([kadminl, 'modprinc', '-pwexpire', 'never', realm.user_princ])
# Try krb5 -> S4U2Proxy with forwardable user creds. This should fail
# at the S4U2Proxy step since the DB2 back end currently has no
# support for allowing it.
realm.kinit(realm.user_princ, password('user'), ['-f', '-c', usercache])
output = realm.run(['./t_s4u2proxy_krb5', usercache, storagecache, '-',
pservice1, pservice2], expected_code=1)
if ('auth1: ' + realm.user_princ not in output or
'NOT_ALLOWED_TO_DELEGATE' not in output):
fail('krb5 -> s4u2proxy')
# Again with SPNEGO.
output = realm.run(['./t_s4u2proxy_krb5', '--spnego', usercache, storagecache,
'-', pservice1, pservice2],
expected_code=1)
if ('auth1: ' + realm.user_princ not in output or
'NOT_ALLOWED_TO_DELEGATE' not in output):
fail('krb5 -> s4u2proxy (SPNEGO)')
# Try krb5 -> S4U2Proxy without forwardable user creds.
realm.kinit(realm.user_princ, password('user'), ['-c', usercache])
output = realm.run(['./t_s4u2proxy_krb5', usercache, storagecache, pservice1,
pservice1, pservice2], expected_code=1)
if ('auth1: ' + realm.user_princ not in output or
'EVIDENCE_TKT_NOT_FORWARDABLE' not in output):
fail('krb5 -> s4u2proxy not-forwardable')
# Try S4U2Self. Ask for an S4U2Proxy step; this won't succeed because
# service/1 isn't allowed to get a forwardable S4U2Self ticket.
realm.run(['./t_s4u', puser, pservice2], expected_code=1,
expected_msg='EVIDENCE_TKT_NOT_FORWARDABLE')
realm.run(['./t_s4u', '--spnego', puser, pservice2], expected_code=1,
expected_msg='EVIDENCE_TKT_NOT_FORWARDABLE')
# Correct that problem and try again. As above, the S4U2Proxy step
# won't actually succeed since we don't support that in DB2.
realm.run([kadminl, 'modprinc', '+ok_to_auth_as_delegate', service1])
realm.run(['./t_s4u', puser, pservice2], expected_code=1,
expected_msg='NOT_ALLOWED_TO_DELEGATE')
# Again with SPNEGO. This uses SPNEGO for the initial authentication,
# but still uses krb5 for S4U2Proxy--the delegated cred is returned as
# a krb5 cred, not a SPNEGO cred, and t_s4u uses the delegated cred
# directly rather than saving and reacquiring it.
realm.run(['./t_s4u', '--spnego', puser, pservice2], expected_code=1,
expected_msg='NOT_ALLOWED_TO_DELEGATE')
realm.stop()
# Set up a realm using the test KDB module so that we can do
# successful S4U2Proxy delegations.
testprincs = {'krbtgt/KRBTEST.COM': {'keys': 'aes128-cts'},
'user': {'keys': 'aes128-cts'},
'service/1': {'flags': '+ok-to-auth-as-delegate',
'keys': 'aes128-cts'},
'service/2': {'keys': 'aes128-cts'}}
conf = {'realms': {'$realm': {'database_module': 'test'}},
'dbmodules': {'test': {'db_library': 'test',
'princs': testprincs,
'delegation': {'service/1': 'service/2'}}}}
realm = K5Realm(create_kdb=False, kdc_conf=conf)
userkeytab = 'FILE:' + os.path.join(realm.testdir, 'userkeytab')
realm.extract_keytab(realm.user_princ, userkeytab)
realm.extract_keytab(service1, realm.keytab)
realm.extract_keytab(service2, realm.keytab)
realm.start_kdc()
# Get forwardable creds for service1 in the default cache.
realm.kinit(service1, None, ['-f', '-k'])
# Successful krb5 -> S4U2Proxy, with krb5 and SPNEGO mechs.
realm.kinit(realm.user_princ, None, ['-f', '-k', '-c', usercache,
'-t', userkeytab])
out = realm.run(['./t_s4u2proxy_krb5', usercache, storagecache, '-',
pservice1, pservice2])
if 'auth1: user@' not in out or 'auth2: user@' not in out:
fail('krb5 -> s4u2proxy')
out = realm.run(['./t_s4u2proxy_krb5', '--spnego', usercache, storagecache,
'-', pservice1, pservice2])
if 'auth1: user@' not in out or 'auth2: user@' not in out:
fail('krb5 -> s4u2proxy')
# Successful S4U2Self -> S4U2Proxy.
out = realm.run(['./t_s4u', puser, pservice2])
# Regression test for #8139: get a user ticket directly for service1 and
# try krb5 -> S4U2Proxy.
realm.kinit(realm.user_princ, None, ['-f', '-k', '-c', usercache,
'-t', userkeytab, '-S', service1])
out = realm.run(['./t_s4u2proxy_krb5', usercache, storagecache, '-',
pservice1, pservice2])
if 'auth1: user@' not in out or 'auth2: user@' not in out:
fail('krb5 -> s4u2proxy')
# Simulate a krbtgt rollover and verify that the user ticket can still
# be validated.
realm.stop_kdc()
newtgt_keys = ['2 aes128-cts', '1 aes128-cts']
newtgt_princs = {'krbtgt/KRBTEST.COM': {'keys': newtgt_keys}}
newtgt_conf = {'dbmodules': {'test': {'princs': newtgt_princs}}}
newtgt_env = realm.special_env('newtgt', True, kdc_conf=newtgt_conf)
realm.start_kdc(env=newtgt_env)
out = realm.run(['./t_s4u2proxy_krb5', usercache, storagecache, '-',
pservice1, pservice2])
if 'auth1: user@' not in out or 'auth2: user@' not in out:
fail('krb5 -> s4u2proxy')
# Get a user ticket after the krbtgt rollover and verify that
# S4U2Proxy delegation works (also a #8139 regression test).
realm.kinit(realm.user_princ, None, ['-f', '-k', '-c', usercache,
'-t', userkeytab])
out = realm.run(['./t_s4u2proxy_krb5', usercache, storagecache, '-',
pservice1, pservice2])
if 'auth1: user@' not in out or 'auth2: user@' not in out:
fail('krb5 -> s4u2proxy')
realm.stop()
mark('S4U2Self with various enctypes')
for realm in multipass_realms(create_host=False, get_creds=False):
service1 = 'service/1@%s' % realm.realm
realm.addprinc(service1)
realm.extract_keytab(service1, realm.keytab)
realm.kinit(service1, None, ['-k'])
realm.run(['./t_s4u', 'e:user', '-'])
# Test cross realm S4U2Self using server referrals.
mark('cross-realm S4U2Self')
testprincs = {'krbtgt/SREALM': {'keys': 'aes128-cts'},
'krbtgt/UREALM': {'keys': 'aes128-cts'},
'user': {'keys': 'aes128-cts', 'flags': '+preauth'},
'other': {'keys': 'aes128-cts'}}
kdcconf1 = {'realms': {'$realm': {'database_module': 'test'}},
'dbmodules': {'test': {'db_library': 'test',
'princs': testprincs,
'alias': {'enterprise@abc': '@UREALM'}}}}
kdcconf2 = {'realms': {'$realm': {'database_module': 'test'}},
'dbmodules': {'test': {'db_library': 'test',
'princs': testprincs,
'alias': {'user@SREALM': '@SREALM',
'enterprise@abc': 'user'}}}}
r1, r2 = cross_realms(2, xtgts=(),
args=({'realm': 'SREALM', 'kdc_conf': kdcconf1},
{'realm': 'UREALM', 'kdc_conf': kdcconf2}),
create_kdb=False)
r1.start_kdc()
r2.start_kdc()
r1.extract_keytab(r1.user_princ, r1.keytab)
r1.kinit(r1.user_princ, None, ['-k', '-t', r1.keytab])
savefile = r1.ccache + '.save'
shutil.copyfile(r1.ccache, savefile)
# Include a regression test for #8741 by unsetting the default realm.
remove_default = {'libdefaults': {'default_realm': None}}
no_default = r1.special_env('no_default', False, krb5_conf=remove_default)
msgs = ('Getting credentials user@UREALM -> user@SREALM',
'/Matching credential not found',
'Getting credentials user@SREALM -> krbtgt/UREALM@SREALM',
'Received creds for desired service krbtgt/UREALM@SREALM',
'via TGT krbtgt/UREALM@SREALM after requesting user\\@SREALM@UREALM',
'krbtgt/SREALM@UREALM differs from requested user\\@SREALM@UREALM',
'via TGT krbtgt/SREALM@UREALM after requesting user@SREALM',
'TGS reply is for user@UREALM -> user@SREALM')
r1.run(['./t_s4u', 'p:' + r2.user_princ, '-', r1.keytab], env=no_default,
expected_trace=msgs)
# Test realm identification of enterprise principal names ([MS-SFU]
# 3.1.5.1.1.1). Attach a bogus realm to the enterprise name to verify
# that we start at the server realm.
mark('cross-realm S4U2Self with enterprise name')
msgs = ('Getting initial credentials for enterprise\\@abc@SREALM',
'Sending unauthenticated request',
'/Realm not local to KDC',
'Following referral to realm UREALM',
'Sending unauthenticated request',
'/Additional pre-authentication required',
'Identified realm of client principal as UREALM',
'Getting credentials enterprise\\@abc@UREALM -> user@SREALM',
'TGS reply is for enterprise\@abc@UREALM -> user@SREALM')
r1.run(['./t_s4u', 'e:enterprise@abc@NOREALM', '-', r1.keytab],
expected_trace=msgs)
mark('S4U2Self using X509 certificate')
# Encode name as a PEM certificate file (sort of) for use by kvno.
def princ_cert(name):
enc = b64encode(name.encode('ascii')).decode('ascii')
return '-----BEGIN CERTIFICATE-----\n%s\n-----END y\n' % enc
cert_path = os.path.join(r1.testdir, 'fake_cert')
with open(cert_path, "w") as cert_file:
cert_file.write(princ_cert('other'))
shutil.copyfile(savefile, r1.ccache)
msgs = ('Getting initial credentials for @SREALM',
'Identified realm of client principal as SREALM',
'TGS reply is for other@SREALM',
'Getting credentials other@SREALM',
'Storing other@SREALM')
r1.run([kvno, '-F', cert_path, r1.user_princ], expected_trace=msgs)
shutil.copyfile(savefile, r1.ccache)
msgs = ('Getting credentials other@SREALM',
'TGS reply is for other@SREALM',
'Storing other@SREALM')
r1.run([kvno, '-I', 'other', '-F', cert_path, r1.user_princ],
expected_trace=msgs)
shutil.copyfile(savefile, r1.ccache)
msgs = ('Getting initial credentials for other@SREALM',
'Identified realm of client principal as SREALM',
'Getting credentials other@SREALM',
'TGS reply is for other@SREALM',
'Storing other@SREALM')
r1.run([kvno, '-U', 'other', '-F', cert_path, r1.user_princ],
expected_trace=msgs)
mark('cross-realm S4U2Self using X509 certificate')
with open(cert_path, "w") as cert_file:
cert_file.write(princ_cert('user@UREALM'))
shutil.copyfile(savefile, r1.ccache)
msgs = ('Getting initial credentials for @SREALM',
'Identified realm of client principal as UREALM',
'TGS reply is for user@UREALM',
'Getting credentials user@UREALM',
'Storing user@UREALM')
r1.run([kvno, '-F', cert_path, r1.user_princ], expected_trace=msgs)
shutil.copyfile(savefile, r1.ccache)
msgs = ('Getting credentials user@UREALM',
'TGS reply is for user@UREALM',
'Storing user@UREALM')
r1.run([kvno, '-I', 'user@UREALM', '-F', cert_path, r1.user_princ],
expected_trace=msgs)
shutil.copyfile(savefile, r1.ccache)
msgs = ('Getting initial credentials for enterprise\\@abc@SREALM',
'Identified realm of client principal as UREALM',
'Getting credentials enterprise\\@abc@UREALM',
'TGS reply is for enterprise\\@abc@UREALM',
'Storing enterprise\\@abc@UREALM')
r1.run([kvno, '-U', 'enterprise@abc', '-F', cert_path, r1.user_princ],
expected_trace=msgs)
r1.stop()
r2.stop()
mark('Resource-based constrained delegation')
a_princs = {'krbtgt/A': {'keys': 'aes128-cts'},
'krbtgt/B': {'keys': 'aes128-cts'},
'user': {'keys': 'aes128-cts', 'flags': '+preauth'},
'sensitive': {'keys': 'aes128-cts',
'flags': '+disallow_forwardable'},
'impersonator': {'keys': 'aes128-cts'},
'rb': {'keys': 'aes128-cts'}}
a_kconf = {'realms': {'$realm': {'database_module': 'test'}},
'dbmodules': {'test': {'db_library': 'test',
'princs': a_princs,
'rbcd': {'rb@A': 'impersonator@A'},
'alias': {'rb@A': 'rb',
'rb@B': '@B',
'rb@C': '@B',
'service/rb.a': 'rb',
'service/rb.b': '@B',
'service/rb.c': '@B' }}}}
b_princs = {'krbtgt/B': {'keys': 'aes128-cts'},
'krbtgt/A': {'keys': 'aes128-cts'},
'krbtgt/C': {'keys': 'aes128-cts'},
'user': {'keys': 'aes128-cts', 'flags': '+preauth'},
'rb': {'keys': 'aes128-cts'}}
b_kconf = {'realms': {'$realm': {'database_module': 'test'}},
'dbmodules': {'test': {'db_library': 'test',
'princs': b_princs,
'rbcd': {'rb@B': 'impersonator@A'},
'alias': {'rb@B': 'rb',
'service/rb.b': 'rb',
'rb@C': '@C',
'impersonator@A': '@A',
'service/rb.c': '@C'}}}}
c_princs = {'krbtgt/C': {'keys': 'aes128-cts'},
'krbtgt/B': {'keys': 'aes128-cts'},
'rb': {'keys': 'aes128-cts'}}
c_kconf = {'realms': {'$realm': {'database_module': 'test'}},
'capaths': { 'A' : { 'C' : 'B' }},
'dbmodules': {'test': {'db_library': 'test',
'princs': c_princs,
'rbcd': {'rb@C': 'impersonator@A'},
'alias': {'rb@C': 'rb',
'service/rb.c': 'rb' }}}}
ra, rb, rc = cross_realms(3, xtgts=(),
args=({'realm': 'A', 'kdc_conf': a_kconf},
{'realm': 'B', 'kdc_conf': b_kconf},
{'realm': 'C', 'kdc_conf': c_kconf}),
create_kdb=False)
ra.start_kdc()
rb.start_kdc()
rc.start_kdc()
domain_realm = {'domain_realm': {'.a':'A', '.b':'B', '.c':'C'}}
domain_conf = ra.special_env('domain_conf', False, krb5_conf=domain_realm)
ra.extract_keytab('impersonator@A', ra.keytab)
ra.kinit('impersonator@A', None, ['-k', '-t', ra.keytab])
mark('Local-realm RBCD')
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'p:rb'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'p:rb@A'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@A'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@A@'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@A@A'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'h:service@rb.a'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'h:service@rb.a'], env=domain_conf)
ra.run(['./t_s4u', 'p:' + 'sensitive@A', 'h:service@rb.a'], expected_code=1)
ra.run(['./t_s4u', 'p:' + rb.user_princ, 'h:service@rb.a'])
mark('Cross-realm RBCD')
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@B'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@B@'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@B@A'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'h:service@rb.b'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'h:service@rb.b'], env=domain_conf)
ra.run(['./t_s4u', 'p:' + 'sensitive@A', 'h:service@rb.b'], expected_code=1)
ra.run(['./t_s4u', 'p:' + rb.user_princ, 'h:service@rb.b'])
mark('RBCD transitive trust')
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@C'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@C@'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'e:rb@C@A'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'h:service@rb.c'])
ra.run(['./t_s4u', 'p:' + ra.user_princ, 'h:service@rb.c'], env=domain_conf)
ra.run(['./t_s4u', 'p:' + 'sensitive@A', 'h:service@rb.c'], expected_code=1)
ra.run(['./t_s4u', 'p:' + rb.user_princ, 'h:service@rb.c'])
ra.stop()
rb.stop()
rc.stop()
success('S4U test cases')
|
