#!/usr/bin/env python import sys import os import shutil import tempfile import unittest if sys.platform != 'win32': from pexpect import pxssh from .PexpectTestCase import PexpectTestCase class SSHTestBase(PexpectTestCase): def setUp(self): super(SSHTestBase, self).setUp() self.tempdir = tempfile.mkdtemp() self.orig_path = os.environ.get('PATH') os.symlink(self.PYTHONBIN, os.path.join(self.tempdir, 'python')) fakessh_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), 'fakessh')) os.environ['PATH'] = self.tempdir + os.pathsep + fakessh_dir + \ ((os.pathsep + self.orig_path) if self.orig_path else '') def tearDown(self): shutil.rmtree(self.tempdir) if self.orig_path: os.environ['PATH'] = self.orig_path else: del os.environ['PATH'] class PxsshTestCase(SSHTestBase): def test_fake_ssh(self): ssh = pxssh.pxssh() #ssh.logfile_read = sys.stdout # DEBUG ssh.login('server', 'me', password='s3cret') ssh.sendline('ping') ssh.expect('pong', timeout=10) assert ssh.prompt(timeout=10) ssh.logout() def test_wrong_pw(self): ssh = pxssh.pxssh() try: ssh.login('server', 'me', password='wr0ng') except pxssh.ExceptionPxssh: pass else: assert False, 'Password should have been refused' def test_failed_set_unique_prompt(self): ssh = pxssh.pxssh() ssh.set_unique_prompt = lambda: False try: ssh.login('server', 'me', password='s3cret', auto_prompt_reset=True) except pxssh.ExceptionPxssh: pass else: assert False, 'should have raised exception, pxssh.ExceptionPxssh' def test_connection_refused(self): ssh = pxssh.pxssh() try: ssh.login('noserver', 'me', password='s3cret') except pxssh.ExceptionPxssh: pass else: assert False, 'should have raised exception, pxssh.ExceptionPxssh' def test_ssh_tunnel_string(self): ssh = pxssh.pxssh(debug_command_string=True) tunnels = { 'local': ['2424:localhost:22'],'remote': ['2525:localhost:22'], 'dynamic': [8888] } confirmation_strings = 0 confirmation_array = ['-R 2525:localhost:22','-L 2424:localhost:22','-D 8888'] string = ssh.login('server', 'me', password='s3cret', ssh_tunnels=tunnels) for confirmation in confirmation_array: if confirmation in string: confirmation_strings+=1 if confirmation_strings!=len(confirmation_array): assert False, 'String generated from tunneling is incorrect.' def test_remote_ssh_tunnel_string(self): ssh = pxssh.pxssh(debug_command_string=True) tunnels = { 'local': ['2424:localhost:22'],'remote': ['2525:localhost:22'], 'dynamic': [8888] } confirmation_strings = 0 confirmation_array = ['-R 2525:localhost:22','-L 2424:localhost:22','-D 8888'] string = ssh.login('server', 'me', password='s3cret', ssh_tunnels=tunnels, spawn_local_ssh=False) for confirmation in confirmation_array: if confirmation in string: confirmation_strings+=1 if confirmation_strings!=len(confirmation_array): assert False, 'String generated from remote tunneling is incorrect.' def test_ssh_config_passing_string(self): ssh = pxssh.pxssh(debug_command_string=True) temp_file = tempfile.NamedTemporaryFile() config_path = temp_file.name string = ssh.login('server', 'me', password='s3cret', spawn_local_ssh=False, ssh_config=config_path) if not '-F '+config_path in string: assert False, 'String generated from SSH config passing is incorrect.' def test_username_or_ssh_config(self): try: ssh = pxssh.pxssh(debug_command_string=True) temp_file = tempfile.NamedTemporaryFile() config_path = temp_file.name string = ssh.login('server') raise AssertionError('Should have failed due to missing username and missing ssh_config.') except TypeError: pass def test_ssh_config_user(self): ssh = pxssh.pxssh(debug_command_string=True) temp_file = tempfile.NamedTemporaryFile() config_path = temp_file.name temp_file.write(b'HosT server\n' b'UsEr me\n' b'hOSt not-server\n') temp_file.seek(0) string = ssh.login('server', ssh_config=config_path) def test_ssh_config_no_username_empty_config(self): ssh = pxssh.pxssh(debug_command_string=True) temp_file = tempfile.NamedTemporaryFile() config_path = temp_file.name try: string = ssh.login('server', ssh_config=config_path) raise AssertionError('Should have failed due to no Host.') except TypeError: pass def test_ssh_config_wrong_Host(self): ssh = pxssh.pxssh(debug_command_string=True) temp_file = tempfile.NamedTemporaryFile() config_path = temp_file.name temp_file.write(b'Host not-server\n' b'Host also-not-server\n') temp_file.seek(0) try: string = ssh.login('server', ssh_config=config_path) raise AssertionError('Should have failed due to no matching Host.') except TypeError: pass def test_ssh_config_no_user(self): ssh = pxssh.pxssh(debug_command_string=True) temp_file = tempfile.NamedTemporaryFile() config_path = temp_file.name temp_file.write(b'Host server\n' b'Host not-server\n') temp_file.seek(0) try: string = ssh.login('server', ssh_config=config_path) raise AssertionError('Should have failed due to no user.') except TypeError: pass def test_ssh_config_empty_user(self): ssh = pxssh.pxssh(debug_command_string=True) temp_file = tempfile.NamedTemporaryFile() config_path = temp_file.name temp_file.write(b'Host server\n' b'user \n' b'Host not-server\n') temp_file.seek(0) try: string = ssh.login('server', ssh_config=config_path) raise AssertionError('Should have failed due to empty user.') except TypeError: pass def test_ssh_key_string(self): ssh = pxssh.pxssh(debug_command_string=True) confirmation_strings = 0 confirmation_array = [' -A'] string = ssh.login('server', 'me', password='s3cret', ssh_key=True) for confirmation in confirmation_array: if confirmation in string: confirmation_strings+=1 if confirmation_strings!=len(confirmation_array): assert False, 'String generated from forcing the SSH agent sock is incorrect.' confirmation_strings = 0 temp_file = tempfile.NamedTemporaryFile() ssh_key = temp_file.name confirmation_array = [' -i '+ssh_key] string = ssh.login('server', 'me', password='s3cret', ssh_key=ssh_key) for confirmation in confirmation_array: if confirmation in string: confirmation_strings+=1 if confirmation_strings!=len(confirmation_array): assert False, 'String generated from adding an SSH key is incorrect.' def test_custom_ssh_cmd_debug(self): ssh = pxssh.pxssh(debug_command_string=True) cipher_string = '-c aes128-ctr,aes192-ctr,aes256-ctr,arcfour256,arcfour128,' \ + 'aes128-cbc,3des-cbc,blowfish-cbc,cast128-cbc,aes192-cbc,' \ + 'aes256-cbc,arcfour' confirmation_strings = 0 confirmation_array = [cipher_string, '-2'] string = ssh.login('server', 'me', password='s3cret', cmd='ssh ' + cipher_string + ' -2') for confirmation in confirmation_array: if confirmation in string: confirmation_strings+=1 if confirmation_strings!=len(confirmation_array): assert False, 'String generated for custom ssh client command is incorrect.' def test_custom_ssh_cmd_debug(self): ssh = pxssh.pxssh(debug_command_string=True) cipher_string = '-c aes128-ctr,aes192-ctr,aes256-ctr,arcfour256,arcfour128,' \ + 'aes128-cbc,3des-cbc,blowfish-cbc,cast128-cbc,aes192-cbc,' \ + 'aes256-cbc,arcfour' confirmation_strings = 0 confirmation_array = [cipher_string, '-2'] string = ssh.login('server', 'me', password='s3cret', cmd='ssh ' + cipher_string + ' -2') for confirmation in confirmation_array: if confirmation in string: confirmation_strings+=1 if confirmation_strings!=len(confirmation_array): assert False, 'String generated for custom ssh client command is incorrect.' def test_failed_custom_ssh_cmd_debug(self): ssh = pxssh.pxssh(debug_command_string=True) cipher_string = '-c invalid_cipher' confirmation_strings = 0 confirmation_array = [cipher_string, '-2'] string = ssh.login('server', 'me', password='s3cret', cmd='ssh ' + cipher_string + ' -2') for confirmation in confirmation_array: if confirmation in string: confirmation_strings+=1 if confirmation_strings!=len(confirmation_array): assert False, 'String generated for custom ssh client command is incorrect.' def test_custom_ssh_cmd(self): try: ssh = pxssh.pxssh() cipher_string = '-c aes128-ctr,aes192-ctr,aes256-ctr,arcfour256,arcfour128,' \ + 'aes128-cbc,3des-cbc,blowfish-cbc,cast128-cbc,aes192-cbc,' \ + 'aes256-cbc,arcfour' result = ssh.login('server', 'me', password='s3cret', cmd='ssh ' + cipher_string + ' -2') ssh.PROMPT = r'Closed connection' ssh.sendline('exit') ssh.prompt(timeout=5) string = str(ssh.before) + str(ssh.after) if 'Closed connection' not in string: assert False, 'should have logged into Mock SSH client and exited' except pxssh.ExceptionPxssh as e: assert False, 'should not have raised exception, pxssh.ExceptionPxssh' else: pass def test_failed_custom_ssh_cmd(self): try: ssh = pxssh.pxssh() cipher_string = '-c invalid_cipher' result = ssh.login('server', 'me', password='s3cret', cmd='ssh ' + cipher_string + ' -2') ssh.PROMPT = r'Closed connection' ssh.sendline('exit') ssh.prompt(timeout=5) string = str(ssh.before) + str(ssh.after) if 'Closed connection' not in string: assert False, 'should not have completed logging into Mock SSH client and exited' except pxssh.ExceptionPxssh as e: pass else: assert False, 'should have raised exception, pxssh.ExceptionPxssh' def test_login_bash(self): ssh = pxssh.pxssh() result = ssh.login('server bash', 'me', password='s3cret') ssh.sendline('ping') ssh.expect('pong', timeout=10) assert ssh.prompt(timeout=10) ssh.logout() def test_login_zsh(self): ssh = pxssh.pxssh() result = ssh.login('server zsh', 'me', password='s3cret') ssh.sendline('ping') ssh.expect('pong', timeout=10) assert ssh.prompt(timeout=10) ssh.logout() def test_login_tcsh(self): ssh = pxssh.pxssh() result = ssh.login('server tcsh', 'me', password='s3cret') ssh.sendline('ping') ssh.expect('pong', timeout=10) assert ssh.prompt(timeout=10) ssh.logout() if __name__ == '__main__': unittest.main()