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
|
# -*- coding: utf-8 -*-
'''
Test the ssh module
'''
# Import python libs
import os
import shutil
# Import Salt Testing libs
from salttesting.helpers import ensure_in_syspath
ensure_in_syspath('../../')
try:
from salttesting.helpers import skip_if_binaries_missing
except ImportError:
from integration import skip_if_binaries_missing
# Import salt libs
import integration
import salt.utils
from salt.exceptions import CommandExecutionError
SUBSALT_DIR = os.path.join(integration.TMP, 'subsalt')
AUTHORIZED_KEYS = os.path.join(SUBSALT_DIR, 'authorized_keys')
KNOWN_HOSTS = os.path.join(SUBSALT_DIR, 'known_hosts')
GITHUB_FINGERPRINT = '16:27:ac:a5:76:28:2d:36:63:1b:56:4d:eb:df:a6:48'
#@skip_if_binaries_missing(['ssh', 'ssh-rsa', 'ssh-keygen'], check_all=True)
class SSHModuleTest(integration.ModuleCase):
'''
Test the ssh module
'''
def setUp(self):
'''
Set up the ssh module tests
'''
super(SSHModuleTest, self).setUp()
if not os.path.isdir(SUBSALT_DIR):
os.makedirs(SUBSALT_DIR)
ssh_raw_path = os.path.join(integration.FILES, 'ssh', 'raw')
with salt.utils.fopen(ssh_raw_path) as fd:
self.key = fd.read().strip()
def tearDown(self):
'''
Tear down the ssh module tests
'''
if os.path.isdir(SUBSALT_DIR):
shutil.rmtree(SUBSALT_DIR)
super(SSHModuleTest, self).tearDown()
def test_auth_keys(self):
'''
test ssh.auth_keys
'''
shutil.copyfile(
os.path.join(integration.FILES, 'ssh', 'authorized_keys'),
AUTHORIZED_KEYS)
ret = self.run_function('ssh.auth_keys', ['root', AUTHORIZED_KEYS])
self.assertEqual(len(list(ret.items())), 1) # exactly one key is found
key_data = list(ret.items())[0][1]
try:
self.assertEqual(key_data['comment'], 'github.com')
self.assertEqual(key_data['enc'], 'ssh-rsa')
self.assertEqual(
key_data['options'], ['command="/usr/local/lib/ssh-helper"']
)
self.assertEqual(key_data['fingerprint'], GITHUB_FINGERPRINT)
except AssertionError as exc:
raise AssertionError(
'AssertionError: {0}. Function returned: {1}'.format(
exc, ret
)
)
def test_bad_enctype(self):
'''
test to make sure that bad key encoding types don't generate an
invalid key entry in authorized_keys
'''
shutil.copyfile(
os.path.join(integration.FILES, 'ssh', 'authorized_badkeys'),
AUTHORIZED_KEYS)
ret = self.run_function('ssh.auth_keys', ['root', AUTHORIZED_KEYS])
# The authorized_badkeys file contains a key with an invalid ssh key
# encoding (dsa-sha2-nistp256 instead of ecdsa-sha2-nistp256)
# auth_keys should skip any keys with invalid encodings. Internally
# the minion will throw a CommandExecutionError so the
# user will get an indicator of what went wrong.
self.assertEqual(len(list(ret.items())), 0) # Zero keys found
def test_get_known_host(self):
'''
Check that known host information is returned from ~/.ssh/config
'''
shutil.copyfile(
os.path.join(integration.FILES, 'ssh', 'known_hosts'),
KNOWN_HOSTS)
arg = ['root', 'github.com']
kwargs = {'config': KNOWN_HOSTS}
ret = self.run_function('ssh.get_known_host', arg, **kwargs)
try:
self.assertEqual(ret['enc'], 'ssh-rsa')
self.assertEqual(ret['key'], self.key)
self.assertEqual(ret['fingerprint'], GITHUB_FINGERPRINT)
except AssertionError as exc:
raise AssertionError(
'AssertionError: {0}. Function returned: {1}'.format(
exc, ret
)
)
def test_recv_known_host(self):
'''
Check that known host information is returned from remote host
'''
ret = self.run_function('ssh.recv_known_host', ['github.com'])
try:
self.assertNotEqual(ret, None)
self.assertEqual(ret['enc'], 'ssh-rsa')
self.assertEqual(ret['key'], self.key)
self.assertEqual(ret['fingerprint'], GITHUB_FINGERPRINT)
except AssertionError as exc:
raise AssertionError(
'AssertionError: {0}. Function returned: {1}'.format(
exc, ret
)
)
def test_check_known_host_add(self):
'''
Check known hosts by its fingerprint. File needs to be updated
'''
arg = ['root', 'github.com']
kwargs = {'fingerprint': GITHUB_FINGERPRINT, 'config': KNOWN_HOSTS}
ret = self.run_function('ssh.check_known_host', arg, **kwargs)
self.assertEqual(ret, 'add')
def test_check_known_host_update(self):
'''
ssh.check_known_host update verification
'''
shutil.copyfile(
os.path.join(integration.FILES, 'ssh', 'known_hosts'),
KNOWN_HOSTS)
arg = ['root', 'github.com']
kwargs = {'config': KNOWN_HOSTS}
# wrong fingerprint
ret = self.run_function('ssh.check_known_host', arg,
**dict(kwargs, fingerprint='aa:bb:cc:dd'))
self.assertEqual(ret, 'update')
# wrong keyfile
ret = self.run_function('ssh.check_known_host', arg,
**dict(kwargs, key='YQ=='))
self.assertEqual(ret, 'update')
def test_check_known_host_exists(self):
'''
Verify check_known_host_exists
'''
shutil.copyfile(
os.path.join(integration.FILES, 'ssh', 'known_hosts'),
KNOWN_HOSTS)
arg = ['root', 'github.com']
kwargs = {'config': KNOWN_HOSTS}
# wrong fingerprint
ret = self.run_function('ssh.check_known_host', arg,
**dict(kwargs, fingerprint=GITHUB_FINGERPRINT))
self.assertEqual(ret, 'exists')
# wrong keyfile
ret = self.run_function('ssh.check_known_host', arg,
**dict(kwargs, key=self.key))
self.assertEqual(ret, 'exists')
def test_rm_known_host(self):
'''
ssh.rm_known_host
'''
shutil.copyfile(
os.path.join(integration.FILES, 'ssh', 'known_hosts'),
KNOWN_HOSTS)
arg = ['root', 'github.com']
kwargs = {'config': KNOWN_HOSTS, 'key': self.key}
# before removal
ret = self.run_function('ssh.check_known_host', arg, **kwargs)
self.assertEqual(ret, 'exists')
# remove
self.run_function('ssh.rm_known_host', arg, config=KNOWN_HOSTS)
# after removal
ret = self.run_function('ssh.check_known_host', arg, **kwargs)
self.assertEqual(ret, 'add')
def test_set_known_host(self):
'''
ssh.set_known_host
'''
# add item
ret = self.run_function('ssh.set_known_host', ['root', 'github.com'],
config=KNOWN_HOSTS)
try:
self.assertEqual(ret['status'], 'updated')
self.assertEqual(ret['old'], None)
self.assertEqual(ret['new']['fingerprint'], GITHUB_FINGERPRINT)
except AssertionError as exc:
raise AssertionError(
'AssertionError: {0}. Function returned: {1}'.format(
exc, ret
)
)
# check that item does exist
ret = self.run_function('ssh.get_known_host', ['root', 'github.com'],
config=KNOWN_HOSTS)
try:
self.assertEqual(ret['fingerprint'], GITHUB_FINGERPRINT)
except AssertionError as exc:
raise AssertionError(
'AssertionError: {0}. Function returned: {1}'.format(
exc, ret
)
)
# add the same item once again
ret = self.run_function('ssh.set_known_host', ['root', 'github.com'],
config=KNOWN_HOSTS)
try:
self.assertEqual(ret['status'], 'exists')
except AssertionError as exc:
raise AssertionError(
'AssertionError: {0}. Function returned: {1}'.format(
exc, ret
)
)
if __name__ == '__main__':
from integration import run_tests
run_tests(SSHModuleTest)
|
