"""Test specification of encryption algorithms"""

def get_enc_algs(self):

"""Return the selected encryption algorithms"""

return self._enc_algs

def get_server_host_key_algs(self):

"""Return the selected server host key algorithms"""

"""Test SSH messages being split into multiple packets"""

def data_received(self, data, datatype=None):

"""Handle incoming data on the connection"""

super().data_received(data[:3], datatype)

super().data_received(data[3:6], datatype)

super().data_received(data[6:9], datatype)

"""Test starting SSH key exchange while it is in progress"""

def replay_kex(self):

"""Replay last kexinit packet"""

"""Test handling of keepalive requests on client"""

def _process_keepalive_at_openssh_dot_com_global_request(self, packet):

"""Process an incoming OpenSSH keepalive request"""

super()._process_keepalive_at_openssh_dot_com_global_request(packet)

"""Test handling of keepalive failures on client"""

def _process_keepalive_at_openssh_dot_com_global_request(self, packet):

"""Test handling of keepalive requests on server"""

def _process_keepalive_at_openssh_dot_com_global_request(self, packet):

"""Process an incoming OpenSSH keepalive request"""

super()._process_keepalive_at_openssh_dot_com_global_request(packet)

"""Test handling of keepalive failures on server"""

def _process_keepalive_at_openssh_dot_com_global_request(self, packet):

"""Test alternate SSH server version lines"""

def __init__(self, version, leading_text, newline, *args, **kwargs):

super().__init__(*args, **kwargs)

self._version = version

self._leading_text = leading_text

self._newline = newline

@classmethod

def create(cls, version=b'SSH-2.0-AsyncSSH_Test',

leading_text=b'', newline=b'\r\n'):

"""Return a connection factory which sends modified version lines"""

return (lambda *args, **kwargs: cls(version, leading_text,

newline, *args, **kwargs))

def _send_version(self):

"""Start the SSH handshake"""

self._server_version = self._version

self._extra.update(server_version=self._version.decode('ascii'))

"""Test returning invalid server host key"""

def get_server_host_key(self):

"""Return the chosen server host key"""

result = copy(super().get_server_host_key())

result.public_data = b'xxx'

"""Test adding an unrecognized extension in extension info"""

def _send_ext_info(self):

"""Send extension information"""

self._extensions_to_send['xxx'] = b''

"""Test returning a bad signature in host keys prove request"""

def _process_hostkeys_prove_00_at_openssh_dot_com_global_request(

self, packet):

"""Prove the server has private keys for all requested host keys"""

"""Test returning failure in host keys prove request"""

def _process_hostkeys_prove_00_at_openssh_dot_com_global_request(

self, packet):

"""Prove the server has private keys for all requested host keys"""

super()._process_hostkeys_prove_00_at_openssh_dot_com_global_request(

"""Replace HMAC class with FailingMAC"""

class _FailingMAC(_HMAC):

"""Test error in MAC validation"""

def verify(self, seq, packet, sig):

"""Verify the signature of a message"""

return super().verify(seq, packet + b'\xff', sig)

_, hash_size, args = _mac_handler[alg]

"""Test error in GCM tag verification"""

def verify_and_decrypt(self, header, data, mac):

"""Verify the signature of and decrypt a block of data"""

"""Test server host key/CA validation callbacks"""

def __init__(self, host_key=None, ca_key=None):

self._host_key = \

asyncssh.read_public_key(host_key) if host_key else None

self._ca_key = \

asyncssh.read_public_key(ca_key) if ca_key else None

def validate_host_public_key(self, host, addr, port, key):

"""Return whether key is an authorized key for this host"""

# pylint: disable=unused-argument

return key == self._host_key

def validate_host_ca_key(self, host, addr, port, key):

"""Return whether key is an authorized CA key for this host"""

# pylint: disable=unused-argument

"""Test sending a request prior to auth complete"""

def __init__(self):

self._conn = None

def connection_made(self, conn):

"""Save connection for use later"""

self._conn = conn

def password_auth_requested(self):

"""Attempt to execute a command before authentication is complete"""

# pylint: disable=protected-access

self._conn._auth_complete = True

self._conn.send_packet(MSG_GLOBAL_REQUEST, String(b'\xff'),

Boolean(True))

"""Test of internal error exception handler"""

def connection_made(self, conn):

"""Raise an error when a new connection is opened"""

# pylint: disable=unused-argument

"""Test of exception during client cleanup"""

def connection_lost(self, exc):

"""Raise an error when a client is cleaned up"""

# pylint: disable=unused-argument

"""Allow forwarding to test server host key request tunneling"""

def connection_requested(self, dest_host, dest_port, orig_host, orig_port):

"""Handle a request to create a new connection"""

"""Server for testing connection abort during auth"""

def begin_auth(self, username):

"""Abort the connection during auth"""

self._conn.abort()

"""Server for testing connection close during long auth callback"""

def password_auth_supported(self):

"""Return that password auth is supported"""

return True

async def validate_password(self, username, password):

"""Delay validating password"""

# pylint: disable=unused-argument

await asyncio.sleep(1)

"""Server for testing internal error during auth"""

def debug_msg_received(self, msg, lang, always_display):

"""Process a debug message"""

# pylint: disable=unused-argument

"""Server for testing invalid auth banner"""

def begin_auth(self, username):

"""Send an invalid auth banner"""

self._conn.send_auth_banner(b'\xff')

"""Client which records auth banner"""

def __init__(self):

self.auth_banner = None

def auth_banner_received(self, msg, lang):

"""Record the client version reported in the auth banner"""

"""Server for testing custom client version"""

def begin_auth(self, username):

"""Report the client's version in the auth banner"""

version = self._conn.get_extra_info('client_version')

self._conn.send_auth_banner(version)

"""Unit tests for AsyncSSH connection API"""

# pylint: disable=too-many-public-methods

@classmethod

async def start_server(cls):

"""Start an SSH server to connect to"""

def acceptor(conn):

"""Acceptor for SSH connections"""

conn.logger.info('Acceptor called')

return (await cls.create_server(_TunnelServer, gss_host=(),

compression_algs='*',

encryption_algs='*',

kex_algs='*', mac_algs='*',

acceptor=acceptor))

async def get_server_host_key(self, **kwargs):

"""Get host key from the test server"""

return (await asyncssh.get_server_host_key(self._server_addr,

self._server_port,

**kwargs))

async def _check_version(self, *args, **kwargs):

"""Check alternate SSH server version lines"""

with patch('asyncssh.connection.SSHServerConnection',

_VersionedServerConnection.create(*args, **kwargs)):

async with self.connect():

pass

@asynctest

async def test_connect(self):

"""Test connecting with async context manager"""

async with self.connect() as conn:

pass

self.assertTrue(conn.is_closed())

@asynctest

async def test_connect_sock(self):

"""Test connecting using an already-connected socket"""

sock = socket.socket()

await self.loop.sock_connect(sock, (self._server_addr,

self._server_port))

async with asyncssh.connect(sock=sock):

pass

@unittest.skipUnless(nc_available, 'Netcat not available')

@asynctest

async def test_connect_non_tcp_sock(self):

"""Test connecting using an non-TCP socket"""

sock1, sock2 = socket.socketpair()

proc = await asyncio.create_subprocess_exec(

'nc', str(self._server_addr), str(self._server_port),

stdin=sock1, stdout=sock1, stderr=sock1)

async with asyncssh.connect(

self._server_addr, self._server_port, sock=sock2):

pass

await proc.wait()

sock1.close()

@asynctest

async def test_run_client(self):

"""Test running an SSH client on an already-connected socket"""

sock = socket.socket()

await self.loop.sock_connect(sock, (self._server_addr,

self._server_port))

async with self.run_client(sock):

pass

@asynctest

async def test_connect_encrypted_key(self):

"""Test connecting with encrypted client key and no passphrase"""

async with self.connect(client_keys='ckey_encrypted',

ignore_encrypted=True):

pass

with self.assertRaises(asyncssh.KeyImportError):

await self.connect(client_keys='ckey_encrypted')

with open('config', 'w') as f:

f.write('IdentityFile ckey_encrypted')

async with self.connect(config='config'):

pass

with self.assertRaises(asyncssh.KeyImportError):

await self.connect(config='config', ignore_encrypted=False)

@asynctest

async def test_connect_invalid_options_type(self):

"""Test connecting using options using incorrect type of options"""

options = asyncssh.SSHServerConnectionOptions()

with self.assertRaises(TypeError):

await self.connect(options=options)

@asynctest

async def test_connect_invalid_option_name(self):

"""Test connecting using incorrect option name"""

with self.assertRaises(TypeError):

await self.connect(xxx=1)

@asynctest

async def test_connect_failure(self):

"""Test failure connecting"""

with self.assertRaises(OSError):

await asyncssh.connect('\xff')

@asynctest

async def test_connect_failure_without_agent(self):

"""Test failure connecting with SSH agent disabled"""

with self.assertRaises(OSError):

await asyncssh.connect('\xff', agent_path=None)

@asynctest

async def test_connect_timeout_exceeded(self):

"""Test connect timeout exceeded"""

with self.assertRaises(asyncio.TimeoutError):

with patch('asyncio.BaseEventLoop.create_connection',

_slow_connect):

await asyncssh.connect('', connect_timeout=1)

@asynctest

async def test_connect_timeout_exceeded_string(self):

"""Test connect timeout exceeded with string value"""

with self.assertRaises(asyncio.TimeoutError):

with patch('asyncio.BaseEventLoop.create_connection',

_slow_connect):

await asyncssh.connect('', connect_timeout='0m1s')

@asynctest

async def test_connect_timeout_exceeded_tunnel(self):

"""Test connect timeout exceeded"""

with self.assertRaises(asyncio.TimeoutError):

with patch('asyncio.BaseEventLoop.create_connection',

_slow_connect):

await asyncssh.listen(server_host_keys=['skey'],

tunnel='', connect_timeout=1)

@asynctest

async def test_invalid_connect_timeout(self):

"""Test invalid connect timeout"""

with self.assertRaises(ValueError):

await self.connect(connect_timeout=-1)

@asynctest

async def test_connect_tcp_keepalive_off(self):

"""Test connecting with TCP keepalive disabled"""

async with self.connect(tcp_keepalive=False) as conn:

sock = conn.get_extra_info('socket')

self.assertEqual(bool(sock.getsockopt(socket.SOL_SOCKET,

socket.SO_KEEPALIVE)), False)

@asynctest

async def test_split_version(self):

"""Test version split across two packets"""

with patch('asyncssh.connection.SSHClientConnection',

_SplitClientConnection):

async with self.connect():

pass

@asynctest

async def test_version_1_99(self):

"""Test SSH server version 1.99"""

await self._check_version(b'SSH-1.99-Test')

@asynctest

async def test_banner_before_version(self):

"""Test banner lines before SSH server version"""

await self._check_version(leading_text=b'Banner 1\r\nBanner 2\r\n')

@asynctest

async def test_banner_line_too_long(self):

"""Test excessively long banner line"""

with self.assertRaises(asyncssh.ProtocolError):

await self._check_version(leading_text=8192*b'*' + b'\r\n')

@asynctest

async def test_too_many_banner_lines(self):

"""Test too many banner lines"""

with self.assertRaises(asyncssh.ProtocolError):

await self._check_version(leading_text=2048*b'Banner line\r\n')

@asynctest

async def test_version_without_cr(self):

"""Test SSH server version with LF instead of CRLF"""

await self._check_version(newline=b'\n')

@asynctest

async def test_version_line_too_long(self):

"""Test excessively long version line"""

with self.assertRaises(asyncssh.ProtocolError):

await self._check_version(newline=256*b'*' + b'\r\n')

@asynctest

async def test_unknown_version(self):

"""Test unknown SSH server version"""

with self.assertRaises(asyncssh.ProtocolNotSupported):

await self._check_version(b'SSH-1.0-Test')

@asynctest

async def test_no_server_host_keys(self):

"""Test starting a server with no host keys"""

with self.assertRaises(ValueError):

await asyncssh.create_server(Server, server_host_keys=[],

gss_host=None)

@asynctest

async def test_duplicate_type_server_host_keys(self):

"""Test starting a server with duplicate host key types"""

with self.assertRaises(ValueError):

await asyncssh.listen(server_host_keys=['skey', 'skey'])

@asynctest

async def test_reserved_server_host_keys(self):

"""Test reserved host keys with host key sending enabled"""

async with self.listen(server_host_keys=['skey', 'skey'],

send_server_host_keys=True):

pass

@asynctest

async def test_get_server_host_key(self):

"""Test retrieving a server host key"""

keylist = asyncssh.load_public_keys('skey.pub')

key = await self.get_server_host_key()

self.assertEqual(key, keylist[0])

@asynctest

async def test_get_server_host_key_tunnel(self):

"""Test retrieving a server host key while tunneling over SSH"""

keylist = asyncssh.load_public_keys('skey.pub')

async with self.connect() as conn:

key = await self.get_server_host_key(tunnel=conn)

self.assertEqual(key, keylist[0])

@asynctest

async def test_get_server_host_key_connect_failure(self):

"""Test failure connecting when retrieving a server host key"""

with self.assertRaises(OSError):

await asyncssh.get_server_host_key('\xff')

@unittest.skipUnless(nc_available, 'Netcat not available')

@asynctest

async def test_get_server_host_key_proxy(self):

"""Test retrieving a server host key using proxy command"""

keylist = asyncssh.load_public_keys('skey.pub')

proxy_command = ('nc', str(self._server_addr), str(self._server_port))

key = await self.get_server_host_key(proxy_command=proxy_command)

self.assertEqual(key, keylist[0])

@unittest.skipUnless(nc_available, 'Netcat not available')

@asynctest

async def test_get_server_host_key_proxy_failure(self):

"""Test failure retrieving a server host key using proxy command"""

# Leave out arguments to 'nc' to trigger a failure

proxy_command = 'nc'

with self.assertRaises((OSError, asyncssh.ConnectionLost)):

await self.connect(proxy_command=proxy_command)

@asynctest

async def test_known_hosts_not_present(self):

"""Test connecting with default known hosts file not present"""

try:

os.rename(os.path.join('.ssh', 'known_hosts'),

os.path.join('.ssh', 'known_hosts.save'))

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.connect()

finally:

os.rename(os.path.join('.ssh', 'known_hosts.save'),

os.path.join('.ssh', 'known_hosts'))

@unittest.skipIf(sys.platform == 'win32', 'skip chmod tests on Windows')

@asynctest

async def test_known_hosts_not_readable(self):

"""Test connecting with default known hosts file not readable"""

try:

os.chmod(os.path.join('.ssh', 'known_hosts'), 0)

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.connect()

finally:

os.chmod(os.path.join('.ssh', 'known_hosts'), 0o644)

@asynctest

async def test_known_hosts_none(self):

"""Test connecting with known hosts checking disabled"""

default_algs = get_default_x509_certificate_algs() + \

get_default_certificate_algs() + \

get_default_public_key_algs()

async with self.connect(known_hosts=None) as conn:

self.assertEqual(conn.get_server_host_key_algs(), default_algs)

@asynctest

async def test_known_hosts_none_in_config(self):

"""Test connecting with known hosts checking disabled in config file"""

with open('config', 'w') as f:

f.write('UserKnownHostsFile none')

async with self.connect(config='config'):

pass

@asynctest

async def test_known_hosts_none_without_x509(self):

"""Test connecting with known hosts checking and X.509 disabled"""

non_x509_algs = get_default_certificate_algs() + \

get_default_public_key_algs()

async with self.connect(known_hosts=None,

x509_trusted_certs=None) as conn:

self.assertEqual(conn.get_server_host_key_algs(), non_x509_algs)

@asynctest

async def test_known_hosts_multiple_keys(self):

"""Test connecting with multiple trusted known hosts keys"""

rsa_algs = [alg for alg in get_default_public_key_algs()

if b'rsa' in alg]

async with self.connect(x509_trusted_certs=None,

known_hosts=(['skey.pub', 'skey.pub'],

[], [])) as conn:

self.assertEqual(conn.get_server_host_key_algs(), rsa_algs)

@asynctest

async def test_known_hosts_ca(self):

"""Test connecting with a known hosts CA"""

async with self.connect(known_hosts=([], ['skey.pub'], [])) as conn:

self.assertEqual(conn.get_server_host_key_algs(),

get_default_x509_certificate_algs() +

get_default_certificate_algs())

@asynctest

async def test_known_hosts_bytes(self):

"""Test connecting with known hosts passed in as bytes"""

with open('skey.pub', 'rb') as f:

skey = f.read()

async with self.connect(known_hosts=([skey], [], [])):

pass

@asynctest

async def test_known_hosts_keylist_file(self):

"""Test connecting with known hosts passed as a keylist file"""

async with self.connect(known_hosts=('skey.pub', [], [])):

pass

@asynctest

async def test_known_hosts_sshkeys(self):

"""Test connecting with known hosts passed in as SSHKeys"""

keylist = asyncssh.load_public_keys('skey.pub')

async with self.connect(known_hosts=(keylist, [], [])) as conn:

self.assertEqual(conn.get_server_host_key(), keylist[0])

@asynctest

async def test_read_known_hosts(self):

"""Test connecting with known hosts object from read_known_hosts"""

known_hosts = asyncssh.read_known_hosts('~/.ssh/known_hosts')

async with self.connect(known_hosts=known_hosts):

pass

@asynctest

async def test_read_known_hosts_filelist(self):

"""Test connecting with known hosts from read_known_hosts file list"""

known_hosts = asyncssh.read_known_hosts(['~/.ssh/known_hosts'])

async with self.connect(known_hosts=known_hosts):

pass

@asynctest

async def test_import_known_hosts(self):

"""Test connecting with known hosts object from import_known_hosts"""

known_hosts_path = os.path.join('.ssh', 'known_hosts')

with open(known_hosts_path) as f:

known_hosts = asyncssh.import_known_hosts(f.read())

async with self.connect(known_hosts=known_hosts):

pass

@asynctest

async def test_validate_host_ca_callback(self):

"""Test callback to validate server CA key"""

def client_factory():

"""Return an SSHClient which can validate the sevrer CA key"""

return _ValidateHostKeyClient(ca_key='skey.pub')

conn, _ = await self.create_connection(client_factory,

known_hosts=([], [], []))

async with conn:

pass

@asynctest

async def test_untrusted_known_hosts_ca(self):

"""Test untrusted server CA key"""

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.connect(known_hosts=([], ['ckey.pub'], []))

@asynctest

async def test_untrusted_host_key_callback(self):

"""Test callback to validate server host key returning failure"""

def client_factory():

"""Return an SSHClient which can validate the sevrer host key"""

return _ValidateHostKeyClient(host_key='ckey.pub')

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.create_connection(client_factory,

known_hosts=([], [], []))

@asynctest

async def test_untrusted_host_ca_callback(self):

"""Test callback to validate server CA key returning failure"""

def client_factory():

"""Return an SSHClient which can validate the sevrer CA key"""

return _ValidateHostKeyClient(ca_key='ckey.pub')

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.create_connection(client_factory,

known_hosts=([], [], []))

@asynctest

async def test_revoked_known_hosts_key(self):

"""Test revoked server host key"""

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.connect(known_hosts=(['ckey.pub'], [], ['skey.pub']))

@asynctest

async def test_revoked_known_hosts_ca(self):

"""Test revoked server CA key"""

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.connect(known_hosts=([], ['ckey.pub'], ['skey.pub']))

@asynctest

async def test_empty_known_hosts(self):

"""Test empty known hosts list"""

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.connect(known_hosts=([], [], []))

@asynctest

async def test_invalid_server_host_key(self):

"""Test invalid server host key"""

with patch('asyncssh.connection.SSHServerConnection',

_BadHostKeyServerConnection):

with self.assertRaises(asyncssh.HostKeyNotVerifiable):

await self.connect()

@asynctest

async def test_changing_server_host_key(self):

"""Test changing server host key"""

self._server.update(server_host_keys=['skey_ecdsa'])

async with self.connect(known_hosts=None):

pass

self._server.update(server_host_keys=['skey'])

with self.assertRaises(asyncssh.KeyExchangeFailed):

await self.connect(known_hosts=(['skey_ecdsa.pub'], [], []))

@asynctest

async def test_kex_algs(self):

"""Test connecting with different key exchange algorithms"""

for kex in get_kex_algs():

kex = kex.decode('ascii')

if kex.startswith('gss-') and not gss_available: # pragma: no cover

continue

with self.subTest(kex_alg=kex):

async with self.connect(kex_algs=[kex], gss_host='1'):

pass

@asynctest

async def test_duplicate_encryption_algs(self):

"""Test connecting with a duplicated encryption algorithm"""

with patch('asyncssh.connection.SSHClientConnection',

_CheckAlgsClientConnection):

async with self.connect(

encryption_algs=['aes256-ctr', 'aes256-ctr']) as conn:

self.assertEqual(conn.get_enc_algs(), [b'aes256-ctr'])

@asynctest

async def test_leading_encryption_alg(self):

"""Test adding a new first encryption algorithm"""

with patch('asyncssh.connection.SSHClientConnection',

_CheckAlgsClientConnection):

async with self.connect(encryption_algs='^aes256-ctr') as conn:

self.assertEqual(conn.get_enc_algs()[0], b'aes256-ctr')

@asynctest

async def test_trailing_encryption_alg(self):

"""Test adding a new last encryption algorithm"""

with patch('asyncssh.connection.SSHClientConnection',

_CheckAlgsClientConnection):

async with self.connect(encryption_algs='+3des-cbc') as conn:

self.assertEqual(conn.get_enc_algs()[-1], b'3des-cbc')

@asynctest

async def test_removing_encryption_alg(self):

"""Test removing an encryption algorithm"""

with patch('asyncssh.connection.SSHClientConnection',

_CheckAlgsClientConnection):

async with self.connect(encryption_algs='-aes256-ctr') as conn:

self.assertTrue(b'aes256-ctr' not in conn.get_enc_algs())

@asynctest

async def test_empty_kex_algs(self):

"""Test connecting with an empty list of key exchange algorithms"""

with self.assertRaises(ValueError):

await self.connect(kex_algs=[])

@asynctest

async def test_invalid_kex_alg(self):

"""Test connecting with invalid key exchange algorithm"""

with self.assertRaises(ValueError):

await self.connect(kex_algs=['xxx'])

@asynctest

async def test_invalid_kex_alg_str(self):

"""Test connecting with invalid key exchange algorithm pattern"""

with self.assertRaises(ValueError):

await self.connect(kex_algs='diffie-hallman-group14-sha1,xxx')

@asynctest

async def test_invalid_kex_alg_config(self):

"""Test connecting with invalid key exchange algorithm config"""

with open('config', 'w') as f:

f.write('KexAlgorithms diffie-hellman-group14-sha1,xxx')

async with self.connect(config='config'):

pass

@asynctest

async def test_unsupported_kex_alg(self):

"""Test connecting with unsupported key exchange algorithm"""

def unsupported_kex_alg():

"""Patched version of get_kex_algs to test unsupported algorithm"""

return [b'fail'] + get_kex_algs()