You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
mitogen/tests/ssh_test.py

237 lines
7.6 KiB
Python

import os
import sys
import tempfile
7 years ago
import mitogen
import mitogen.ssh
import mitogen.utils
import unittest2
7 years ago
import testlib
import plain_old_module
7 years ago
class StubSshMixin(testlib.RouterMixin):
"""
Mix-in that provides :meth:`stub_ssh` executing the stub 'ssh.py'.
"""
def stub_ssh(self, STUBSSH_MODE=None, **kwargs):
os.environ['STUBSSH_MODE'] = str(STUBSSH_MODE)
try:
return self.router.ssh(
hostname='hostname',
username='mitogen__has_sudo',
ssh_path=testlib.data_path('stubs/stub-ssh.py'),
**kwargs
)
finally:
del os.environ['STUBSSH_MODE']
class ConstructorTest(testlib.RouterMixin, testlib.TestCase):
def test_okay(self):
context = self.router.ssh(
hostname='hostname',
username='mitogen__has_sudo',
ssh_path=testlib.data_path('stubs/stub-ssh.py'),
)
#context.call(mitogen.utils.log_to_file, '/tmp/log')
#context.call(mitogen.utils.disable_site_packages)
self.assertEquals(3, context.call(plain_old_module.add, 1, 2))
class SshTest(testlib.DockerMixin, testlib.TestCase):
stream_class = mitogen.ssh.Stream
def test_debug_decoding(self):
# ensure filter_debug_logs() decodes the logged string.
capture = testlib.LogCapturer()
capture.start()
try:
context = self.docker_ssh(
username='mitogen__has_sudo',
password='has_sudo_password',
ssh_debug_level=3,
)
finally:
s = capture.stop()
self.assertTrue("'): debug1: Reading configuration data" in s)
def test_stream_name(self):
context = self.docker_ssh(
username='mitogen__has_sudo',
password='has_sudo_password',
)
name = 'ssh.%s:%s' % (
self.dockerized_ssh.get_host(),
self.dockerized_ssh.port,
)
self.assertEquals(name, context.name)
def test_via_stream_name(self):
context = self.docker_ssh(
username='mitogen__has_sudo_nopw',
password='has_sudo_nopw_password',
)
sudo = self.router.sudo(via=context)
name = 'ssh.%s:%s.sudo.root' % (
self.dockerized_ssh.host,
self.dockerized_ssh.port,
)
self.assertEquals(name, sudo.name)
def test_password_required(self):
try:
context = self.docker_ssh(
username='mitogen__has_sudo',
)
assert 0, 'exception not thrown'
except mitogen.ssh.PasswordError:
e = sys.exc_info()[1]
self.assertEqual(e.args[0], self.stream_class.password_required_msg)
def test_password_incorrect(self):
try:
context = self.docker_ssh(
username='mitogen__has_sudo',
password='badpw',
)
assert 0, 'exception not thrown'
except mitogen.ssh.PasswordError:
e = sys.exc_info()[1]
self.assertEqual(e.args[0], self.stream_class.password_incorrect_msg)
def test_password_specified(self):
context = self.docker_ssh(
username='mitogen__has_sudo',
password='has_sudo_password',
)
self.assertEqual(
'i-am-mitogen-test-docker-image\n',
context.call(plain_old_module.get_sentinel_value),
)
def test_pubkey_required(self):
try:
context = self.docker_ssh(
username='mitogen__has_sudo_pubkey',
)
assert 0, 'exception not thrown'
except mitogen.ssh.PasswordError:
e = sys.exc_info()[1]
self.assertEqual(e.args[0], self.stream_class.password_required_msg)
def test_pubkey_specified(self):
context = self.docker_ssh(
username='mitogen__has_sudo_pubkey',
identity_file=testlib.data_path('docker/mitogen__has_sudo_pubkey.key'),
)
self.assertEqual(
'i-am-mitogen-test-docker-image\n',
context.call(plain_old_module.get_sentinel_value),
)
def test_enforce_unknown_host_key(self):
fp = tempfile.NamedTemporaryFile()
try:
e = self.assertRaises(mitogen.ssh.HostKeyError,
lambda: self.docker_ssh(
username='mitogen__has_sudo_pubkey',
password='has_sudo_password',
ssh_args=['-o', 'UserKnownHostsFile ' + fp.name],
check_host_keys='enforce',
)
)
self.assertEquals(e.args[0], mitogen.ssh.Stream.hostkey_failed_msg)
finally:
fp.close()
def test_accept_enforce_host_keys(self):
fp = tempfile.NamedTemporaryFile()
try:
context = self.docker_ssh(
username='mitogen__has_sudo',
password='has_sudo_password',
ssh_args=['-o', 'UserKnownHostsFile ' + fp.name],
check_host_keys='accept',
)
context.shutdown(wait=True)
fp.seek(0)
# Lame test, but we're about to use enforce mode anyway, which
# verifies the file contents.
self.assertTrue(len(fp.read()) > 0)
context = self.docker_ssh(
username='mitogen__has_sudo',
password='has_sudo_password',
ssh_args=['-o', 'UserKnownHostsFile ' + fp.name],
check_host_keys='enforce',
)
context.shutdown(wait=True)
finally:
fp.close()
class BannerTest(testlib.DockerMixin, testlib.TestCase):
# Verify the ability to disambiguate random spam appearing in the SSHd's
# login banner from a legitimate password prompt.
stream_class = mitogen.ssh.Stream
def test_verbose_enabled(self):
context = self.docker_ssh(
username='mitogen__has_sudo',
password='has_sudo_password',
ssh_debug_level=3,
)
name = 'ssh.%s:%s' % (
self.dockerized_ssh.get_host(),
self.dockerized_ssh.port,
)
self.assertEquals(name, context.name)
class StubPermissionDeniedTest(StubSshMixin, testlib.TestCase):
def test_classic_prompt(self):
self.assertRaises(mitogen.ssh.PasswordError,
lambda: self.stub_ssh(STUBSSH_MODE='permdenied_classic'))
def test_openssh_75_prompt(self):
self.assertRaises(mitogen.ssh.PasswordError,
lambda: self.stub_ssh(STUBSSH_MODE='permdenied_75'))
class StubCheckHostKeysTest(StubSshMixin, testlib.TestCase):
stream_class = mitogen.ssh.Stream
def test_check_host_keys_accept(self):
# required=true, host_key_checking=accept
context = self.stub_ssh(STUBSSH_MODE='ask', check_host_keys='accept')
self.assertEquals('1', context.call(os.getenv, 'STDERR_WAS_TTY'))
def test_check_host_keys_enforce(self):
# required=false, host_key_checking=enforce
context = self.stub_ssh(check_host_keys='enforce')
self.assertEquals(None, context.call(os.getenv, 'STDERR_WAS_TTY'))
def test_check_host_keys_ignore(self):
# required=false, host_key_checking=ignore
context = self.stub_ssh(check_host_keys='ignore')
self.assertEquals(None, context.call(os.getenv, 'STDERR_WAS_TTY'))
def test_password_present(self):
# required=true, password is not None
context = self.stub_ssh(check_host_keys='ignore', password='willick')
self.assertEquals('1', context.call(os.getenv, 'STDERR_WAS_TTY'))
if __name__ == '__main__':
unittest2.main()