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
|
#
# Copyright (C) 2015 FreeIPA Contributors see COPYING for license
#
import six
from ipalib import api, errors
from ipatests.test_xmlrpc.tracker.base import Tracker
from ipatests.test_xmlrpc import objectclasses
from ipatests.test_xmlrpc.xmlrpc_test import (
fuzzy_string, fuzzy_dergeneralizedtime, raises_exact)
from ipatests.util import assert_deepequal
from ipapython.dn import DN
if six.PY3:
unicode = str
sshpubkey = (u'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQDGAX3xAeLeaJggwTqMjxNwa6X'
'HBUAikXPGMzEpVrlLDCZtv00djsFTBi38PkgxBJVkgRWMrcBsr/35lq7P6w8KGI'
'wA8GI48Z0qBS2NBMJ2u9WQ2hjLN6GdMlo77O0uJY3251p12pCVIS/bHRSq8kHO2'
'No8g7KA9fGGcagPfQH+ee3t7HUkpbQkFTmbPPN++r3V8oVUk5LxbryB3UIIVzNm'
'cSIn3JrXynlvui4MixvrtX6zx+O/bBo68o8/eZD26QrahVbA09fivrn/4h3TM01'
'9Eu/c2jOdckfU3cHUV/3Tno5d6JicibyaoDDK7S/yjdn5jhaz8MSEayQvFkZkiF'
'0L public key test')
sshpubkeyfp = (u'13:67:6B:BF:4E:A2:05:8E:AE:25:8B:A1:31:DE:6F:1B '
'public key test (ssh-rsa)')
class StageUserTracker(Tracker):
""" Tracker class for staged user LDAP object
Implements helper functions for host plugin.
StageUserTracker object stores information about the user.
"""
retrieve_keys = {
u'uid', u'givenname', u'sn', u'homedirectory', u'loginshell',
u'uidnumber', u'gidnumber', u'mail', u'ou', u'telephonenumber',
u'title', u'memberof', u'nsaccountlock', u'memberofindirect',
u'ipauserauthtype', u'userclass', u'ipatokenradiusconfiglink',
u'ipatokenradiususername', u'krbprincipalexpiration',
u'usercertificate', u'dn', u'has_keytab', u'has_password',
u'street', u'postalcode', u'facsimiletelephonenumber',
u'carlicense', u'ipasshpubkey', u'sshpubkeyfp', u'l',
u'st', u'mobile', u'pager', }
retrieve_all_keys = retrieve_keys | {
u'cn', u'ipauniqueid', u'objectclass', u'description',
u'displayname', u'gecos', u'initials', u'krbprincipalname', u'manager'}
create_keys = retrieve_all_keys | {
u'objectclass', u'ipauniqueid', u'randompassword',
u'userpassword', u'krbextradata', u'krblastpwdchange',
u'krbpasswordexpiration', u'krbprincipalkey'}
update_keys = retrieve_keys - {u'dn', u'nsaccountlock'}
activate_keys = retrieve_keys | {
u'has_keytab', u'has_password', u'nsaccountlock'}
def __init__(self, name, givenname, sn, **kwargs):
super(StageUserTracker, self).__init__(default_version=None)
self.uid = name
self.givenname = givenname
self.sn = sn
self.dn = DN(
('uid', self.uid), api.env.container_stageuser, api.env.basedn)
self.kwargs = kwargs
def make_create_command(self, options=None, force=None):
""" Make function that creates a staged user using stageuser-add """
if options is not None:
self.kwargs = options
return self.make_command('stageuser_add', self.uid,
givenname=self.givenname,
sn=self.sn, **self.kwargs)
def make_delete_command(self):
""" Make function that deletes a staged user using stageuser-del """
return self.make_command('stageuser_del', self.uid)
def make_retrieve_command(self, all=False, raw=False):
""" Make function that retrieves a staged user using stageuser-show """
return self.make_command('stageuser_show', self.uid, all=all)
def make_find_command(self, *args, **kwargs):
""" Make function that finds staged user using stageuser-find """
return self.make_command('stageuser_find', *args, **kwargs)
def make_update_command(self, updates):
""" Make function that updates staged user using stageuser-mod """
return self.make_command('stageuser_mod', self.uid, **updates)
def make_activate_command(self):
""" Make function that activates staged user
using stageuser-activate """
return self.make_command('stageuser_activate', self.uid)
def track_create(self):
""" Update expected state for staged user creation """
self.attrs = dict(
dn=self.dn,
uid=[self.uid],
givenname=[self.givenname],
sn=[self.sn],
homedirectory=[u'/home/%s' % self.uid],
displayname=[u'%s %s' % (self.givenname, self.sn)],
cn=[u'%s %s' % (self.givenname, self.sn)],
initials=[u'%s%s' % (self.givenname[0], self.sn[0])],
objectclass=objectclasses.user_base,
description=[u'__no_upg__'],
ipauniqueid=[u'autogenerate'],
uidnumber=[u'-1'],
gidnumber=[u'-1'],
krbprincipalname=[u'%s@%s' % (self.uid, self.api.env.realm)],
mail=[u'%s@%s' % (self.uid, self.api.env.domain)],
gecos=[u'%s %s' % (self.givenname, self.sn)],
loginshell=[u'/bin/sh'],
has_keytab=False,
has_password=False,
nsaccountlock=[u'true'],
)
for key in self.kwargs:
if key == u'krbprincipalname':
self.attrs[key] = [u'%s@%s' % (
(self.kwargs[key].split('@'))[0].lower(),
(self.kwargs[key].split('@'))[1])]
elif key == u'manager':
self.attrs[key] = [self.kwargs[key]]
elif key == u'ipasshpubkey':
self.attrs[u'sshpubkeyfp'] = [sshpubkeyfp]
self.attrs[key] = [self.kwargs[key]]
elif key == u'random' or key == u'userpassword':
self.attrs[u'krbextradata'] = [fuzzy_string]
self.attrs[u'krbpasswordexpiration'] = [
fuzzy_dergeneralizedtime]
self.attrs[u'krblastpwdchange'] = [fuzzy_dergeneralizedtime]
self.attrs[u'krbprincipalkey'] = [fuzzy_string]
self.attrs[u'userpassword'] = [fuzzy_string]
self.attrs[u'has_keytab'] = True
self.attrs[u'has_password'] = True
if key == u'random':
self.attrs[u'randompassword'] = fuzzy_string
else:
self.attrs[key] = [self.kwargs[key]]
self.exists = True
def check_create(self, result):
""" Check 'stageuser-add' command result """
assert_deepequal(dict(
value=self.uid,
summary=u'Added stage user "%s"' % self.uid,
result=self.filter_attrs(self.create_keys),
), result)
def check_delete(self, result):
""" Check 'stageuser-del' command result """
assert_deepequal(dict(
value=[self.uid],
summary=u'Deleted stage user "%s"' % self.uid,
result=dict(failed=[]),
), result)
def check_retrieve(self, result, all=False, raw=False):
""" Check 'stageuser-show' command result """
if all:
expected = self.filter_attrs(self.retrieve_all_keys)
else:
expected = self.filter_attrs(self.retrieve_keys)
# small override because stageuser-find returns different
# type of nsaccountlock value than DS, but overall the value
# fits expected result
if expected[u'nsaccountlock'] == [u'true']:
expected[u'nsaccountlock'] = True
elif expected[u'nsaccountlock'] == [u'false']:
expected[u'nsaccountlock'] = False
assert_deepequal(dict(
value=self.uid,
summary=None,
result=expected,
), result)
def check_find(self, result, all=False, raw=False):
""" Check 'stageuser-find' command result """
if all:
expected = self.filter_attrs(self.retrieve_all_keys)
else:
expected = self.filter_attrs(self.retrieve_keys)
# small override because stageuser-find returns different
# type of nsaccountlock value than DS, but overall the value
# fits expected result
if expected[u'nsaccountlock'] == [u'true']:
expected[u'nsaccountlock'] = True
elif expected[u'nsaccountlock'] == [u'false']:
expected[u'nsaccountlock'] = False
assert_deepequal(dict(
count=1,
truncated=False,
summary=u'1 user matched',
result=[expected],
), result)
def check_find_nomatch(self, result):
""" Check 'stageuser-find' command result when no match is expected """
assert_deepequal(dict(
count=0,
truncated=False,
summary=u'0 users matched',
result=[],
), result)
def check_update(self, result, extra_keys=()):
""" Check 'stageuser-mod' command result """
assert_deepequal(dict(
value=self.uid,
summary=u'Modified stage user "%s"' % self.uid,
result=self.filter_attrs(self.update_keys | set(extra_keys))
), result)
def check_restore_preserved(self, result):
assert_deepequal(dict(
value=[self.uid],
summary=u'Staged user account "%s"' % self.uid,
result=dict(failed=[]),
), result)
def make_fixture_activate(self, request):
"""Make a pytest fixture for a staged user that is to be activated
The fixture ensures the plugin entry does not exist before
and after the tests that use it. It takes into account
that the staged user no longer exists after activation,
therefore the fixture verifies after the tests
that the staged user doesn't exist instead of deleting it.
"""
del_command = self.make_delete_command()
try:
del_command()
except errors.NotFound:
pass
def finish():
with raises_exact(errors.NotFound(
reason=u'%s: stage user not found' % self.uid)):
del_command()
request.addfinalizer(finish)
return self
def create_from_preserved(self, user):
""" Copies values from preserved user - helper function for
restoration tests """
self.attrs = user.attrs
self.uid = user.uid
self.givenname = user.givenname
self.sn = user.sn
self.dn = DN(
('uid', self.uid), api.env.container_stageuser, api.env.basedn)
self.attrs[u'dn'] = self.dn
|