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
|
# This file is part of cloud-init. See LICENSE file for license information.
import re
from cloudinit import distros
from cloudinit import ssh_util
from cloudinit.tests.helpers import (CiTestCase, mock)
from tests.unittests.util import abstract_to_concrete
@mock.patch("cloudinit.distros.util.system_is_snappy", return_value=False)
@mock.patch("cloudinit.distros.subp.subp")
class TestCreateUser(CiTestCase):
with_logs = True
def setUp(self):
super(TestCreateUser, self).setUp()
self.dist = abstract_to_concrete(distros.Distro)(
name='test', cfg=None, paths=None
)
def _useradd2call(self, args):
# return a mock call for the useradd command in args
# with expected 'logstring'.
args = ['useradd'] + args
logcmd = [a for a in args]
for i in range(len(args)):
if args[i] in ('--password',):
logcmd[i + 1] = 'REDACTED'
return mock.call(args, logstring=logcmd)
def test_basic(self, m_subp, m_is_snappy):
user = 'foouser'
self.dist.create_user(user)
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '-m']),
mock.call(['passwd', '-l', user])])
def test_no_home(self, m_subp, m_is_snappy):
user = 'foouser'
self.dist.create_user(user, no_create_home=True)
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '-M']),
mock.call(['passwd', '-l', user])])
def test_system_user(self, m_subp, m_is_snappy):
# system user should have no home and get --system
user = 'foouser'
self.dist.create_user(user, system=True)
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '--system', '-M']),
mock.call(['passwd', '-l', user])])
def test_explicit_no_home_false(self, m_subp, m_is_snappy):
user = 'foouser'
self.dist.create_user(user, no_create_home=False)
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '-m']),
mock.call(['passwd', '-l', user])])
def test_unlocked(self, m_subp, m_is_snappy):
user = 'foouser'
self.dist.create_user(user, lock_passwd=False)
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '-m'])])
def test_set_password(self, m_subp, m_is_snappy):
user = 'foouser'
password = 'passfoo'
self.dist.create_user(user, passwd=password)
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '--password', password, '-m']),
mock.call(['passwd', '-l', user])])
@mock.patch("cloudinit.distros.util.is_group")
def test_group_added(self, m_is_group, m_subp, m_is_snappy):
m_is_group.return_value = False
user = 'foouser'
self.dist.create_user(user, groups=['group1'])
expected = [
mock.call(['groupadd', 'group1']),
self._useradd2call([user, '--groups', 'group1', '-m']),
mock.call(['passwd', '-l', user])]
self.assertEqual(m_subp.call_args_list, expected)
@mock.patch("cloudinit.distros.util.is_group")
def test_only_new_group_added(self, m_is_group, m_subp, m_is_snappy):
ex_groups = ['existing_group']
groups = ['group1', ex_groups[0]]
m_is_group.side_effect = lambda m: m in ex_groups
user = 'foouser'
self.dist.create_user(user, groups=groups)
expected = [
mock.call(['groupadd', 'group1']),
self._useradd2call([user, '--groups', ','.join(groups), '-m']),
mock.call(['passwd', '-l', user])]
self.assertEqual(m_subp.call_args_list, expected)
@mock.patch("cloudinit.distros.util.is_group")
def test_create_groups_with_whitespace_string(
self, m_is_group, m_subp, m_is_snappy):
# groups supported as a comma delimeted string even with white space
m_is_group.return_value = False
user = 'foouser'
self.dist.create_user(user, groups='group1, group2')
expected = [
mock.call(['groupadd', 'group1']),
mock.call(['groupadd', 'group2']),
self._useradd2call([user, '--groups', 'group1,group2', '-m']),
mock.call(['passwd', '-l', user])]
self.assertEqual(m_subp.call_args_list, expected)
def test_explicit_sudo_false(self, m_subp, m_is_snappy):
user = 'foouser'
self.dist.create_user(user, sudo=False)
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '-m']),
mock.call(['passwd', '-l', user])])
@mock.patch('cloudinit.ssh_util.setup_user_keys')
def test_setup_ssh_authorized_keys_with_string(
self, m_setup_user_keys, m_subp, m_is_snappy):
"""ssh_authorized_keys allows string and calls setup_user_keys."""
user = 'foouser'
self.dist.create_user(user, ssh_authorized_keys='mykey')
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '-m']),
mock.call(['passwd', '-l', user])])
m_setup_user_keys.assert_called_once_with(set(['mykey']), user)
@mock.patch('cloudinit.ssh_util.setup_user_keys')
def test_setup_ssh_authorized_keys_with_list(
self, m_setup_user_keys, m_subp, m_is_snappy):
"""ssh_authorized_keys allows lists and calls setup_user_keys."""
user = 'foouser'
self.dist.create_user(user, ssh_authorized_keys=['key1', 'key2'])
self.assertEqual(
m_subp.call_args_list,
[self._useradd2call([user, '-m']),
mock.call(['passwd', '-l', user])])
m_setup_user_keys.assert_called_once_with(set(['key1', 'key2']), user)
@mock.patch('cloudinit.ssh_util.setup_user_keys')
def test_setup_ssh_authorized_keys_with_integer(
self, m_setup_user_keys, m_subp, m_is_snappy):
"""ssh_authorized_keys warns on non-iterable/string type."""
user = 'foouser'
self.dist.create_user(user, ssh_authorized_keys=-1)
m_setup_user_keys.assert_called_once_with(set([]), user)
match = re.match(
r'.*WARNING: Invalid type \'<(type|class) \'int\'>\' detected for'
' \'ssh_authorized_keys\'.*',
self.logs.getvalue(),
re.DOTALL)
self.assertIsNotNone(
match, 'Missing ssh_authorized_keys invalid type warning')
@mock.patch('cloudinit.ssh_util.setup_user_keys')
def test_create_user_with_ssh_redirect_user_no_cloud_keys(
self, m_setup_user_keys, m_subp, m_is_snappy):
"""Log a warning when trying to redirect a user no cloud ssh keys."""
user = 'foouser'
self.dist.create_user(user, ssh_redirect_user='someuser')
self.assertIn(
'WARNING: Unable to disable SSH logins for foouser given '
'ssh_redirect_user: someuser. No cloud public-keys present.\n',
self.logs.getvalue())
m_setup_user_keys.assert_not_called()
@mock.patch('cloudinit.ssh_util.setup_user_keys')
def test_create_user_with_ssh_redirect_user_with_cloud_keys(
self, m_setup_user_keys, m_subp, m_is_snappy):
"""Disable ssh when ssh_redirect_user and cloud ssh keys are set."""
user = 'foouser'
self.dist.create_user(
user, ssh_redirect_user='someuser', cloud_public_ssh_keys=['key1'])
disable_prefix = ssh_util.DISABLE_USER_OPTS
disable_prefix = disable_prefix.replace('$USER', 'someuser')
disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
m_setup_user_keys.assert_called_once_with(
set(['key1']), 'foouser', options=disable_prefix)
@mock.patch('cloudinit.ssh_util.setup_user_keys')
def test_create_user_with_ssh_redirect_user_does_not_disable_auth_keys(
self, m_setup_user_keys, m_subp, m_is_snappy):
"""Do not disable ssh_authorized_keys when ssh_redirect_user is set."""
user = 'foouser'
self.dist.create_user(
user, ssh_authorized_keys='auth1', ssh_redirect_user='someuser',
cloud_public_ssh_keys=['key1'])
disable_prefix = ssh_util.DISABLE_USER_OPTS
disable_prefix = disable_prefix.replace('$USER', 'someuser')
disable_prefix = disable_prefix.replace('$DISABLE_USER', user)
self.assertEqual(
m_setup_user_keys.call_args_list,
[mock.call(set(['auth1']), user), # not disabled
mock.call(set(['key1']), 'foouser', options=disable_prefix)])
@mock.patch("cloudinit.distros.subp.which")
def test_lock_with_usermod_if_no_passwd(self, m_which, m_subp,
m_is_snappy):
"""Lock uses usermod --lock if no 'passwd' cmd available."""
m_which.side_effect = lambda m: m in ('usermod',)
self.dist.lock_passwd("bob")
self.assertEqual(
[mock.call(['usermod', '--lock', 'bob'])],
m_subp.call_args_list)
@mock.patch("cloudinit.distros.subp.which")
def test_lock_with_passwd_if_available(self, m_which, m_subp,
m_is_snappy):
"""Lock with only passwd will use passwd."""
m_which.side_effect = lambda m: m in ('passwd',)
self.dist.lock_passwd("bob")
self.assertEqual(
[mock.call(['passwd', '-l', 'bob'])],
m_subp.call_args_list)
@mock.patch("cloudinit.distros.subp.which")
def test_lock_raises_runtime_if_no_commands(self, m_which, m_subp,
m_is_snappy):
"""Lock with no commands available raises RuntimeError."""
m_which.return_value = None
with self.assertRaises(RuntimeError):
self.dist.lock_passwd("bob")
# vi: ts=4 expandtab
|
