Revision 405bffe2 test/ganeti.backend_unittest.py
b/test/ganeti.backend_unittest.py | ||
---|---|---|
96 | 96 |
"Result from netutils.TcpPing corrupted") |
97 | 97 |
|
98 | 98 |
|
99 |
def _DefRemoteCommandOwner():
|
|
99 |
def _DefRestrictedCmdOwner():
|
|
100 | 100 |
return (os.getuid(), os.getgid()) |
101 | 101 |
|
102 | 102 |
|
103 |
class TestVerifyRemoteCommandName(unittest.TestCase):
|
|
103 |
class TestVerifyRestrictedCmdName(unittest.TestCase):
|
|
104 | 104 |
def testAcceptableName(self): |
105 | 105 |
for i in ["foo", "bar", "z1", "000first", "hello-world"]: |
106 | 106 |
for fn in [lambda s: s, lambda s: s.upper(), lambda s: s.title()]: |
107 |
(status, msg) = backend._VerifyRemoteCommandName(fn(i))
|
|
107 |
(status, msg) = backend._VerifyRestrictedCmdName(fn(i))
|
|
108 | 108 |
self.assertTrue(status) |
109 | 109 |
self.assertTrue(msg is None) |
110 | 110 |
|
111 | 111 |
def testEmptyAndSpace(self): |
112 | 112 |
for i in ["", " ", "\t", "\n"]: |
113 |
(status, msg) = backend._VerifyRemoteCommandName(i)
|
|
113 |
(status, msg) = backend._VerifyRestrictedCmdName(i)
|
|
114 | 114 |
self.assertFalse(status) |
115 | 115 |
self.assertEqual(msg, "Missing command name") |
116 | 116 |
|
117 | 117 |
def testNameWithSlashes(self): |
118 | 118 |
for i in ["/", "./foo", "../moo", "some/name"]: |
119 |
(status, msg) = backend._VerifyRemoteCommandName(i)
|
|
119 |
(status, msg) = backend._VerifyRestrictedCmdName(i)
|
|
120 | 120 |
self.assertFalse(status) |
121 | 121 |
self.assertEqual(msg, "Invalid command name") |
122 | 122 |
|
123 | 123 |
def testForbiddenCharacters(self): |
124 | 124 |
for i in ["#", ".", "..", "bash -c ls", "'"]: |
125 |
(status, msg) = backend._VerifyRemoteCommandName(i)
|
|
125 |
(status, msg) = backend._VerifyRestrictedCmdName(i)
|
|
126 | 126 |
self.assertFalse(status) |
127 | 127 |
self.assertEqual(msg, "Command name contains forbidden characters") |
128 | 128 |
|
129 | 129 |
|
130 |
class TestVerifyRemoteCommandDirectory(unittest.TestCase):
|
|
130 |
class TestVerifyRestrictedCmdDirectory(unittest.TestCase):
|
|
131 | 131 |
def setUp(self): |
132 | 132 |
self.tmpdir = tempfile.mkdtemp() |
133 | 133 |
|
... | ... | |
138 | 138 |
tmpname = utils.PathJoin(self.tmpdir, "foobar") |
139 | 139 |
self.assertFalse(os.path.exists(tmpname)) |
140 | 140 |
(status, msg) = \ |
141 |
backend._VerifyRemoteCommandDirectory(tmpname, _owner=NotImplemented)
|
|
141 |
backend._VerifyRestrictedCmdDirectory(tmpname, _owner=NotImplemented)
|
|
142 | 142 |
self.assertFalse(status) |
143 | 143 |
self.assertTrue(msg.startswith("Can't stat(2) '")) |
144 | 144 |
|
... | ... | |
150 | 150 |
os.chmod(tmpname, mode) |
151 | 151 |
self.assertTrue(os.path.isdir(tmpname)) |
152 | 152 |
(status, msg) = \ |
153 |
backend._VerifyRemoteCommandDirectory(tmpname, _owner=NotImplemented)
|
|
153 |
backend._VerifyRestrictedCmdDirectory(tmpname, _owner=NotImplemented)
|
|
154 | 154 |
self.assertFalse(status) |
155 | 155 |
self.assertTrue(msg.startswith("Permissions on '")) |
156 | 156 |
|
... | ... | |
159 | 159 |
utils.WriteFile(tmpname, data="empty\n") |
160 | 160 |
self.assertTrue(os.path.isfile(tmpname)) |
161 | 161 |
(status, msg) = \ |
162 |
backend._VerifyRemoteCommandDirectory(tmpname,
|
|
163 |
_owner=_DefRemoteCommandOwner())
|
|
162 |
backend._VerifyRestrictedCmdDirectory(tmpname,
|
|
163 |
_owner=_DefRestrictedCmdOwner())
|
|
164 | 164 |
self.assertFalse(status) |
165 | 165 |
self.assertTrue(msg.endswith("is not a directory")) |
166 | 166 |
|
... | ... | |
169 | 169 |
os.mkdir(tmpname) |
170 | 170 |
self.assertTrue(os.path.isdir(tmpname)) |
171 | 171 |
(status, msg) = \ |
172 |
backend._VerifyRemoteCommandDirectory(tmpname,
|
|
173 |
_owner=_DefRemoteCommandOwner())
|
|
172 |
backend._VerifyRestrictedCmdDirectory(tmpname,
|
|
173 |
_owner=_DefRestrictedCmdOwner())
|
|
174 | 174 |
self.assertTrue(status) |
175 | 175 |
self.assertTrue(msg is None) |
176 | 176 |
|
177 | 177 |
|
178 |
class TestVerifyRemoteCommand(unittest.TestCase):
|
|
178 |
class TestVerifyRestrictedCmd(unittest.TestCase):
|
|
179 | 179 |
def setUp(self): |
180 | 180 |
self.tmpdir = tempfile.mkdtemp() |
181 | 181 |
|
... | ... | |
186 | 186 |
tmpname = utils.PathJoin(self.tmpdir, "helloworld") |
187 | 187 |
self.assertFalse(os.path.exists(tmpname)) |
188 | 188 |
(status, msg) = \ |
189 |
backend._VerifyRemoteCommand(self.tmpdir, "helloworld",
|
|
189 |
backend._VerifyRestrictedCmd(self.tmpdir, "helloworld",
|
|
190 | 190 |
_owner=NotImplemented) |
191 | 191 |
self.assertFalse(status) |
192 | 192 |
self.assertTrue(msg.startswith("Can't stat(2) '")) |
... | ... | |
195 | 195 |
tmpname = utils.PathJoin(self.tmpdir, "cmdname") |
196 | 196 |
utils.WriteFile(tmpname, data="empty\n") |
197 | 197 |
(status, msg) = \ |
198 |
backend._VerifyRemoteCommand(self.tmpdir, "cmdname",
|
|
199 |
_owner=_DefRemoteCommandOwner())
|
|
198 |
backend._VerifyRestrictedCmd(self.tmpdir, "cmdname",
|
|
199 |
_owner=_DefRestrictedCmdOwner())
|
|
200 | 200 |
self.assertFalse(status) |
201 | 201 |
self.assertTrue(msg.startswith("access(2) thinks '")) |
202 | 202 |
|
... | ... | |
204 | 204 |
tmpname = utils.PathJoin(self.tmpdir, "cmdname") |
205 | 205 |
utils.WriteFile(tmpname, data="empty\n", mode=0700) |
206 | 206 |
(status, executable) = \ |
207 |
backend._VerifyRemoteCommand(self.tmpdir, "cmdname",
|
|
208 |
_owner=_DefRemoteCommandOwner())
|
|
207 |
backend._VerifyRestrictedCmd(self.tmpdir, "cmdname",
|
|
208 |
_owner=_DefRestrictedCmdOwner())
|
|
209 | 209 |
self.assertTrue(status) |
210 | 210 |
self.assertEqual(executable, tmpname) |
211 | 211 |
|
212 | 212 |
|
213 |
class TestPrepareRemoteCommand(unittest.TestCase):
|
|
213 |
class TestPrepareRestrictedCmd(unittest.TestCase):
|
|
214 | 214 |
_TEST_PATH = "/tmp/some/test/path" |
215 | 215 |
|
216 | 216 |
def testDirFails(self): |
... | ... | |
219 | 219 |
return (False, "test error 31420") |
220 | 220 |
|
221 | 221 |
(status, msg) = \ |
222 |
backend._PrepareRemoteCommand(self._TEST_PATH, "cmd21152",
|
|
222 |
backend._PrepareRestrictedCmd(self._TEST_PATH, "cmd21152",
|
|
223 | 223 |
_verify_dir=fn, |
224 | 224 |
_verify_name=NotImplemented, |
225 | 225 |
_verify_cmd=NotImplemented) |
... | ... | |
232 | 232 |
return (False, "test error 591") |
233 | 233 |
|
234 | 234 |
(status, msg) = \ |
235 |
backend._PrepareRemoteCommand(self._TEST_PATH, "cmd4617",
|
|
235 |
backend._PrepareRestrictedCmd(self._TEST_PATH, "cmd4617",
|
|
236 | 236 |
_verify_dir=lambda _: (True, None), |
237 | 237 |
_verify_name=fn, |
238 | 238 |
_verify_cmd=NotImplemented) |
... | ... | |
246 | 246 |
return (False, "test error 25524") |
247 | 247 |
|
248 | 248 |
(status, msg) = \ |
249 |
backend._PrepareRemoteCommand(self._TEST_PATH, "cmd17577",
|
|
249 |
backend._PrepareRestrictedCmd(self._TEST_PATH, "cmd17577",
|
|
250 | 250 |
_verify_dir=lambda _: (True, None), |
251 | 251 |
_verify_name=lambda _: (True, None), |
252 | 252 |
_verify_cmd=fn) |
... | ... | |
258 | 258 |
return (True, utils.PathJoin(path, cmd)) |
259 | 259 |
|
260 | 260 |
(status, executable) = \ |
261 |
backend._PrepareRemoteCommand(self._TEST_PATH, "cmd22633",
|
|
261 |
backend._PrepareRestrictedCmd(self._TEST_PATH, "cmd22633",
|
|
262 | 262 |
_verify_dir=lambda _: (True, None), |
263 | 263 |
_verify_name=lambda _: (True, None), |
264 | 264 |
_verify_cmd=fn) |
... | ... | |
266 | 266 |
self.assertEqual(executable, utils.PathJoin(self._TEST_PATH, "cmd22633")) |
267 | 267 |
|
268 | 268 |
|
269 |
def _SleepForRemoteCommand(duration):
|
|
269 |
def _SleepForRestrictedCmd(duration):
|
|
270 | 270 |
assert duration > 5 |
271 | 271 |
|
272 | 272 |
|
273 |
def _GenericRemoteCommandError(cmd):
|
|
273 |
def _GenericRestrictedCmdError(cmd):
|
|
274 | 274 |
return "Executing command '%s' failed" % cmd |
275 | 275 |
|
276 | 276 |
|
... | ... | |
283 | 283 |
|
284 | 284 |
def testNonExistantLockDirectory(self): |
285 | 285 |
lockfile = utils.PathJoin(self.tmpdir, "does", "not", "exist") |
286 |
sleep_fn = testutils.CallCounter(_SleepForRemoteCommand)
|
|
286 |
sleep_fn = testutils.CallCounter(_SleepForRestrictedCmd)
|
|
287 | 287 |
self.assertFalse(os.path.exists(lockfile)) |
288 | 288 |
self.assertRaises(backend.RPCFail, |
289 | 289 |
backend.RunRestrictedCmd, "test", |
... | ... | |
298 | 298 |
|
299 | 299 |
@staticmethod |
300 | 300 |
def _TryLock(lockfile): |
301 |
sleep_fn = testutils.CallCounter(_SleepForRemoteCommand)
|
|
301 |
sleep_fn = testutils.CallCounter(_SleepForRestrictedCmd)
|
|
302 | 302 |
|
303 | 303 |
result = False |
304 | 304 |
try: |
... | ... | |
311 | 311 |
_runcmd_fn=NotImplemented, |
312 | 312 |
_enabled=True) |
313 | 313 |
except backend.RPCFail, err: |
314 |
assert str(err) == _GenericRemoteCommandError("test22717"), \
|
|
314 |
assert str(err) == _GenericRestrictedCmdError("test22717"), \
|
|
315 | 315 |
"Did not fail with generic error message" |
316 | 316 |
result = True |
317 | 317 |
|
... | ... | |
337 | 337 |
def testPrepareRaisesException(self): |
338 | 338 |
lockfile = utils.PathJoin(self.tmpdir, "lock") |
339 | 339 |
|
340 |
sleep_fn = testutils.CallCounter(_SleepForRemoteCommand)
|
|
340 |
sleep_fn = testutils.CallCounter(_SleepForRestrictedCmd)
|
|
341 | 341 |
prepare_fn = testutils.CallCounter(self._PrepareRaisingException) |
342 | 342 |
|
343 | 343 |
try: |
... | ... | |
347 | 347 |
_sleep_fn=sleep_fn, _prepare_fn=prepare_fn, |
348 | 348 |
_enabled=True) |
349 | 349 |
except backend.RPCFail, err: |
350 |
self.assertEqual(str(err), _GenericRemoteCommandError("test23122"))
|
|
350 |
self.assertEqual(str(err), _GenericRestrictedCmdError("test23122"))
|
|
351 | 351 |
else: |
352 | 352 |
self.fail("Didn't fail") |
353 | 353 |
|
... | ... | |
362 | 362 |
def testPrepareFails(self): |
363 | 363 |
lockfile = utils.PathJoin(self.tmpdir, "lock") |
364 | 364 |
|
365 |
sleep_fn = testutils.CallCounter(_SleepForRemoteCommand)
|
|
365 |
sleep_fn = testutils.CallCounter(_SleepForRestrictedCmd)
|
|
366 | 366 |
prepare_fn = testutils.CallCounter(self._PrepareFails) |
367 | 367 |
|
368 | 368 |
try: |
... | ... | |
372 | 372 |
_sleep_fn=sleep_fn, _prepare_fn=prepare_fn, |
373 | 373 |
_enabled=True) |
374 | 374 |
except backend.RPCFail, err: |
375 |
self.assertEqual(str(err), _GenericRemoteCommandError("test29327"))
|
|
375 |
self.assertEqual(str(err), _GenericRestrictedCmdError("test29327"))
|
|
376 | 376 |
else: |
377 | 377 |
self.fail("Didn't fail") |
378 | 378 |
|
... | ... | |
412 | 412 |
utils.ShellQuoteArgs(args), |
413 | 413 |
NotImplemented, NotImplemented) |
414 | 414 |
|
415 |
sleep_fn = testutils.CallCounter(_SleepForRemoteCommand)
|
|
415 |
sleep_fn = testutils.CallCounter(_SleepForRestrictedCmd)
|
|
416 | 416 |
prepare_fn = testutils.CallCounter(self._SuccessfulPrepare) |
417 | 417 |
runcmd_fn = testutils.CallCounter(fn) |
418 | 418 |
|
... | ... | |
450 | 450 |
utils.ShellQuoteArgs(args), |
451 | 451 |
NotImplemented, NotImplemented) |
452 | 452 |
|
453 |
sleep_fn = testutils.CallCounter(_SleepForRemoteCommand)
|
|
453 |
sleep_fn = testutils.CallCounter(_SleepForRestrictedCmd)
|
|
454 | 454 |
prepare_fn = testutils.CallCounter(self._SuccessfulPrepare) |
455 | 455 |
runcmd_fn = testutils.CallCounter(fn) |
456 | 456 |
|
Also available in: Unified diff