# Copyright (c) 2016-2024 by Ron Frederick and others. # # This program and the accompanying materials are made available under # the terms of the Eclipse Public License v2.0 which accompanies this # distribution and is available at: # # http://www.eclipse.org/legal/epl-2.0/ # # This program may also be made available under the following secondary # licenses when the conditions for such availability set forth in the # Eclipse Public License v2.0 are satisfied: # # GNU General Public License, Version 2.0, or any later versions of # that license # # SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later # # Contributors: # Ron Frederick - initial implementation, API, and documentation """Unit tests for AsyncSSH channel API""" import asyncio import os import tempfile import unittest from signal import SIGINT from unittest.mock import patch import asyncssh from asyncssh.constants import DEFAULT_LANG, MSG_USERAUTH_REQUEST from asyncssh.constants import MSG_CHANNEL_OPEN_CONFIRMATION from asyncssh.constants import MSG_CHANNEL_OPEN_FAILURE from asyncssh.constants import MSG_CHANNEL_WINDOW_ADJUST from asyncssh.constants import MSG_CHANNEL_DATA from asyncssh.constants import MSG_CHANNEL_EXTENDED_DATA from asyncssh.constants import MSG_CHANNEL_EOF, MSG_CHANNEL_CLOSE from asyncssh.constants import MSG_CHANNEL_SUCCESS from asyncssh.packet import Byte, String, UInt32 from asyncssh.public_key import CERT_TYPE_USER from asyncssh.stream import SSHTCPStreamSession, SSHUNIXStreamSession from asyncssh.stream import SSHTunTapStreamSession from asyncssh.tuntap import SSH_TUN_MODE_POINTTOPOINT, SSH_TUN_MODE_ETHERNET from .server import Server, ServerTestCase from .util import asynctest, echo, make_certificate PTY_OP_PARTIAL = 158 PTY_OP_NO_END = 159 class _ClientChannel(asyncssh.SSHClientChannel): """Patched SSH client channel for unit testing""" def _send_request(self, request, *args, want_reply=False): """Send a channel request""" if request == b'pty-req': if args[5][-6:-5] == Byte(PTY_OP_PARTIAL): args = args[:5] + (String(args[5][4:-5]),) elif args[5][-6:-5] == Byte(PTY_OP_NO_END): args = args[:5] + (String(args[5][4:-6]),) super()._send_request(request, *args, want_reply=want_reply) def get_send_pktsize(self): """Return the sender's max packet size """ return self._send_pktsize def send_request(self, request, *args): """Send a custom request (for unit testing)""" self._send_request(request, *args) async def make_request(self, request, *args): """Make a custom request (for unit testing)""" return await self._make_request(request, *args) class _ClientSession(asyncssh.SSHClientSession): """Unit test SSH client session""" def __init__(self): self._chan = None self.recv_buf = {None: [], asyncssh.EXTENDED_DATA_STDERR: []} self.xon_xoff = None self.exit_status = None self.exit_signal_msg = None self.exc = None def connection_made(self, chan): """Handle connection open""" self._chan = chan def connection_lost(self, exc): """Handle connection close""" self.exc = exc self._chan = None def data_received(self, data, datatype): """Handle data from the channel""" self.recv_buf[datatype].append(data) def xon_xoff_requested(self, client_can_do): """Handle request to enable/disable XON/XOFF flow control""" self.xon_xoff = client_can_do def exit_status_received(self, status): """Handle remote exit status""" # pylint: disable=unused-argument self.exit_status = status def exit_signal_received(self, signal, core_dumped, msg, lang): """Handle remote exit signal""" # pylint: disable=unused-argument self.exit_signal_msg = msg async def _create_session(conn, command=None, *, subsystem=None, **kwargs): """Create a client session""" return await conn.create_session(_ClientSession, command, subsystem=subsystem, **kwargs) class _ServerChannel(asyncssh.SSHServerChannel): """Patched SSH server channel class for unit testing""" def _send_request(self, request, *args, want_reply=False): """Send a channel request""" if request == b'exit-signal': if args[0] == String('invalid'): args = (String(b'\xff'),) + args[1:] if args[3] == String('invalid'): args = args[:3] + (String(b'\xff'),) super()._send_request(request, *args, want_reply=want_reply) def _process_delayed_request(self, packet): """Process a request that delays before responding""" packet.check_end() asyncio.get_event_loop().call_later(0.1, self._report_response, True) def get_send_pktsize(self): """Return the sender's max packet size """ return self._send_pktsize async def open_session(self): """Attempt to open a session on the client""" return await self._open(b'session') class _EchoServerSession(asyncssh.SSHServerSession): """A shell session which echoes data from stdin to stdout/stderr""" def __init__(self): self._chan = None def connection_made(self, chan): """Handle session open""" self._chan = chan username = self._chan.get_extra_info('username') if username == 'close': self._chan.close() elif username == 'task_error': raise RuntimeError('Exception handler test') def shell_requested(self): """Handle shell request""" return True def data_received(self, data, datatype): """Handle data from the channel""" self._chan.write(data[:1]) self._chan.writelines([data[1:]]) self._chan.write_stderr(data[:1]) self._chan.writelines_stderr([data[1:]]) def eof_received(self): """Handle EOF on the channel""" self._chan.write_eof() self._chan.close() class _PTYServerSession(asyncssh.SSHServerSession): """Server for testing PTY requests""" def __init__(self): self._chan = None self._pty_ok = True def connection_made(self, chan): """Handle session open""" self._chan = chan username = self._chan.get_extra_info('username') if username == 'no_pty': self._pty_ok = False def pty_requested(self, term_type, term_size, term_modes): """Handle pseudo-terminal request""" self._chan.set_extra_info( pty_args=(term_type, term_size, term_modes.get(asyncssh.PTY_OP_OSPEED))) return self._pty_ok def shell_requested(self): """Handle shell request""" return True def session_started(self): """Handle session start""" chan = self._chan chan.write(f'Req: {chan.get_extra_info("pty_args")}\n') chan.close() class _ClientSessionCleanupError(asyncssh.SSHClientSession): """Test of exception during client session cleanup""" def connection_lost(self, exc): """Raise an error when a client session is cleaned up""" raise RuntimeError('Exception in session cleanup test') class _ChannelServer(Server): """Server for testing the AsyncSSH channel API""" async def _begin_session(self, stdin, stdout, stderr): """Begin processing a new session""" # pylint: disable=too-many-statements action = stdin.channel.get_command() or stdin.channel.get_subsystem() if not action: action = 'echo' if action == 'echo': await echo(stdin, stdout, stderr) elif action == 'conn_close': await stdin.read(1) stdout.write('\n') self._conn.close() elif action == 'close': await stdin.read(1) stdout.write('\n') elif action == 'agent': try: async with asyncssh.connect_agent(self._conn) as agent: stdout.write(str(len(await agent.get_keys())) + '\n') except (OSError, asyncssh.ChannelOpenError): stdout.channel.exit(1) elif action == 'agent_sock': agent_path = stdin.channel.get_agent_path() if agent_path: async with asyncssh.connect_agent(agent_path) as agent: await asyncio.sleep(0.1) stdout.write(str(len(await agent.get_keys())) + '\n') else: stdout.channel.exit(1) elif action == 'rejected_agent': agent_path = stdin.channel.get_agent_path() stdout.write(str(bool(agent_path)) + '\n') chan = self._conn.create_agent_channel() try: await chan.open(SSHUNIXStreamSession) except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'rejected_session': chan = _ServerChannel(self._conn, asyncio.get_event_loop(), False, False, False, 0, 1024, None, 'strict', 1, 32768) try: await chan.open_session() except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'rejected_tcpip_direct': chan = self._conn.create_tcp_channel() try: await chan.connect(SSHTCPStreamSession, '', 0, '', 0) except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'unknown_tcpip_listener': chan = self._conn.create_tcp_channel() try: await chan.accept(SSHTCPStreamSession, 'xxx', 0, '', 0) except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'invalid_tcpip_listener': chan = self._conn.create_tcp_channel() try: await chan.accept(SSHTCPStreamSession, b'\xff', 0, '', 0) except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'rejected_unix_direct': chan = self._conn.create_unix_channel() try: await chan.connect(SSHUNIXStreamSession, '') except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'unknown_unix_listener': chan = self._conn.create_unix_channel() try: await chan.accept(SSHUNIXStreamSession, 'xxx') except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'invalid_unix_listener': chan = self._conn.create_unix_channel() try: await chan.accept(SSHUNIXStreamSession, b'\xff') except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'rejected_tun_request': chan = self._conn.create_tuntap_channel() try: await chan.open(SSHTunTapStreamSession, SSH_TUN_MODE_POINTTOPOINT, 0) except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'rejected_tap_request': chan = self._conn.create_tuntap_channel() try: await chan.open(SSHTunTapStreamSession, SSH_TUN_MODE_ETHERNET, 0) except asyncssh.ChannelOpenError: stdout.channel.exit(1) elif action == 'late_auth_banner': try: self._conn.send_auth_banner('auth banner') except OSError: stdin.channel.exit(1) elif action == 'invalid_open_confirm': stdin.channel.send_packet(MSG_CHANNEL_OPEN_CONFIRMATION, UInt32(0), UInt32(0), UInt32(0)) elif action == 'invalid_open_failure': stdin.channel.send_packet(MSG_CHANNEL_OPEN_FAILURE, UInt32(0), String(''), String('')) elif action == 'env': value = stdin.channel.get_environment_bytes().get(b'TEST', b'') stdout.write(value.decode('utf-8', 'backslashreplace') + '\n') elif action == 'env_binary_key': value = stdin.channel.get_environment_bytes().get(b'TEST\xff', b'') stdout.write(value.decode('utf-8', 'backslashreplace') + '\n') elif action == 'env_str': value = stdin.channel.get_environment().get('TEST', '') stdout.write(value + '\n') elif action == 'env_str_cached': value1 = stdin.channel.get_environment().get('TEST', '') value2 = stdin.channel.get_environment().get('TEST', '') stdout.write(value1 + value2 + '\n') elif action == 'env_non_string_key': value = stdin.channel.get_environment().get('1', '') stdout.write(value + '\n') elif action == 'term': chan = stdin.channel info = str((chan.get_terminal_type(), chan.get_terminal_size(), chan.get_terminal_mode(asyncssh.PTY_OP_OSPEED))) stdout.write(info + '\n') elif action == 'xon_xoff': stdin.channel.set_xon_xoff(True) elif action == 'no_xon_xoff': stdin.channel.set_xon_xoff(False) elif action == 'signals': try: await stdin.readline() except asyncssh.BreakReceived as exc: stdin.channel.exit_with_signal('ABRT', False, str(exc.msec)) except asyncssh.SignalReceived as exc: stdin.channel.exit_with_signal('ABRT', False, exc.signal) except asyncssh.TerminalSizeChanged as exc: size = (exc.width, exc.height, exc.pixwidth, exc.pixheight) stdin.channel.exit_with_signal('ABRT', False, str(size)) elif action == 'exit_status': stdin.channel.exit(1) elif action == 'closed_status': stdin.channel.close() stdin.channel.exit(1) elif action == 'exit_signal': stdin.channel.exit_with_signal('INT', False, 'exit_signal') elif action == 'unknown_signal': stdin.channel.exit_with_signal('unknown', False, 'unknown_signal') elif action == 'closed_signal': stdin.channel.close() stdin.channel.exit_with_signal('INT', False, 'closed_signal') elif action == 'invalid_exit_signal': stdin.channel.exit_with_signal('invalid') elif action == 'invalid_exit_lang': stdin.channel.exit_with_signal('INT', False, '', 'invalid') elif action == 'window_after_close': stdin.channel.send_packet(MSG_CHANNEL_CLOSE) stdin.channel.send_packet(MSG_CHANNEL_WINDOW_ADJUST, UInt32(0)) elif action == 'empty_data': stdin.channel.send_packet(MSG_CHANNEL_DATA, String('')) elif action == 'partial_unicode': data = '\xff\xff'.encode() stdin.channel.send_packet(MSG_CHANNEL_DATA, String(data[:3])) stdin.channel.send_packet(MSG_CHANNEL_DATA, String(data[3:])) elif action == 'partial_unicode_at_eof': data = '\xff\xff'.encode() stdin.channel.send_packet(MSG_CHANNEL_DATA, String(data[:3])) elif action == 'unicode_error': stdin.channel.send_packet(MSG_CHANNEL_DATA, String(b'\xff')) elif action == 'data_past_window': stdin.channel.send_packet(MSG_CHANNEL_DATA, String(2*1025*1024*'\0')) elif action == 'ext_data_past_window': stdin.channel.send_packet(MSG_CHANNEL_EXTENDED_DATA, UInt32(asyncssh.EXTENDED_DATA_STDERR), String(2*1025*1024*'\0')) elif action == 'data_after_eof': stdin.channel.send_packet(MSG_CHANNEL_EOF) stdout.write('xxx') elif action == 'data_after_close': await asyncio.sleep(0.1) stdout.write('xxx') elif action == 'ext_data_after_eof': stdin.channel.send_packet(MSG_CHANNEL_EOF) stdin.channel.write_stderr('xxx') elif action == 'invalid_datatype': stdin.channel.send_packet(MSG_CHANNEL_EXTENDED_DATA, UInt32(255), String('')) elif action == 'double_eof': await asyncio.sleep(0.1) stdin.channel.send_packet(MSG_CHANNEL_EOF) stdin.channel.write_eof() elif action == 'double_close': await asyncio.sleep(0.1) stdout.write('xxx') stdin.channel.send_packet(MSG_CHANNEL_CLOSE) elif action == 'request_after_close': stdin.channel.send_packet(MSG_CHANNEL_CLOSE) stdin.channel.exit(1) elif action == 'unexpected_auth': self._conn.send_packet(MSG_USERAUTH_REQUEST, String('guest'), String('ssh-connection'), String('none')) elif action == 'invalid_response': stdin.channel.send_packet(MSG_CHANNEL_SUCCESS) elif action == 'send_pktsize': stdout.write(str(stdout.channel.get_send_pktsize())) stdout.close() else: stdin.channel.exit(255) stdin.channel.close() await stdin.channel.wait_closed() async def _conn_close(self): """Close the connection during a channel open""" self._conn.close() await asyncio.sleep(0.1) return _EchoServerSession() def begin_auth(self, username): """Handle client authentication request""" return username not in {'guest', 'conn_close_startup', 'conn_close_open', 'close', 'echo', 'no_channels', 'no_pty', 'request_pty', 'task_error'} def session_requested(self): """Handle a request to create a new session""" username = self._conn.get_extra_info('username') with patch('asyncssh.connection.SSHServerChannel', _ServerChannel): channel = self._conn.create_server_channel() if username == 'conn_close_startup': self._conn.close() return False elif username == 'conn_close_open': return (channel, self._conn_close()) elif username in {'close', 'echo', 'task_error'}: return (channel, _EchoServerSession()) elif username in {'request_pty', 'no_pty'}: return (channel, _PTYServerSession()) elif username != 'no_channels': return (channel, self._begin_session) else: return False class _TestChannel(ServerTestCase): """Unit tests for AsyncSSH channel API""" # pylint: disable=too-many-public-methods @classmethod async def start_server(cls): """Start an SSH server for the tests to use""" return (await cls.create_server( _ChannelServer, authorized_client_keys='authorized_keys')) async def _check_action(self, command, expected_result): """Run a command on a remote session and check for a specific result""" async with self.connect() as conn: chan, session = await _create_session(conn, command) await chan.wait_closed() self.assertEqual(session.exit_status, expected_result) async def _check_session(self, conn, command=(), *, large_block=False, **kwargs): """Open a session and test if an input line is echoed back""" chan, session = await _create_session(conn, command, **kwargs) if large_block: data = 4 * [1025*1024*'\0'] else: data = [str(id(self))] chan.writelines(data) self.assertTrue(chan.can_write_eof()) self.assertFalse(chan.is_closing()) chan.write_eof() self.assertTrue(chan.is_closing()) await chan.wait_closed() data = ''.join(data) for buf in session.recv_buf.values(): self.assertEqual(data, ''.join(buf)) chan.close() @asynctest async def test_shell(self): """Test starting a shell""" async with self.connect(username='echo') as conn: await self._check_session(conn) @asynctest async def test_shell_failure(self): """Test failure to start a shell""" async with self.connect(username='no_channels') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn) @asynctest async def test_shell_internal_error(self): """Test internal error in callback to start a shell""" async with self.connect(username='task_error') as conn: with self.assertRaises((OSError, asyncssh.ConnectionLost)): await _create_session(conn) @asynctest async def test_shell_large_block(self): """Test starting a shell and sending a large block of data""" async with self.connect(username='echo') as conn: await self._check_session(conn, large_block=True) @asynctest async def test_exec(self): """Test execution of a remote command""" async with self.connect() as conn: await self._check_session(conn, 'echo', window=1024*1024, max_pktsize=16384) @asynctest async def test_exec_from_connect(self): """Test execution of a remote command set on connection""" async with self.connect(command='echo') as conn: await self._check_session(conn) @asynctest async def test_forced_exec(self): """Test execution of a forced remote command""" ckey = asyncssh.read_private_key('ckey') cert = make_certificate('ssh-rsa-cert-v01@openssh.com', CERT_TYPE_USER, ckey, ckey, ['ckey'], options={'force-command': String('echo')}) async with self.connect(username='ckey', client_keys=[(ckey, cert)], agent_path=None) as conn: await self._check_session(conn) @asynctest async def test_invalid_exec(self): """Test execution of an invalid remote command""" async with self.connect() as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn, b'\xff') @asynctest async def test_exec_failure(self): """Test failure to execute a remote command""" async with self.connect(username='no_channels') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn, 'echo') @asynctest async def test_subsystem(self): """Test starting a subsystem""" async with self.connect() as conn: await self._check_session(conn, subsystem='echo') @asynctest async def test_invalid_subsystem(self): """Test starting an invalid subsystem""" async with self.connect() as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn, subsystem=b'\xff') @asynctest async def test_subsystem_failure(self): """Test failure to start a subsystem""" async with self.connect(username='no_channels') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn, subsystem='echo') @asynctest async def test_conn_close_during_startup(self): """Test connection close during channel startup""" async with self.connect(username='conn_close_startup') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn) @asynctest async def test_conn_close_during_open(self): """Test connection close during channel open""" async with self.connect(username='conn_close_open') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn) @asynctest async def test_close_during_startup(self): """Test channel close during startup""" async with self.connect(username='close') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn) @asynctest async def test_inbound_conn_close_while_read_paused(self): """Test inbound connection close while reading is paused""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'conn_close') chan.pause_reading() chan.write('\n') await asyncio.sleep(0.1) conn.close() await chan.wait_closed() @asynctest async def test_outbound_conn_close_while_read_paused(self): """Test outbound connection close while reading is paused""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'close') chan.pause_reading() chan.write('\n') await asyncio.sleep(0.1) conn.close() await chan.wait_closed() @asynctest async def test_close_while_read_paused(self): """Test closing a remotely closed channel while reading is paused""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'close') chan.pause_reading() chan.write('\n') await asyncio.sleep(0.1) chan.close() await chan.wait_closed() @asynctest async def test_keepalive(self): """Test keepalive channel requests""" with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: chan, _ = await _create_session(conn) result = await chan.make_request(b'keepalive@openssh.com') self.assertFalse(result) @asynctest async def test_invalid_open_confirmation(self): """Test receiving an open confirmation on already open channel""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'invalid_open_confirm') await chan.wait_closed() @asynctest async def test_invalid_open_failure(self): """Test receiving an open failure on already open channel""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'invalid_open_failure') await chan.wait_closed() @asynctest async def test_unknown_channel_request(self): """Test sending unknown channel request""" with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: chan, _ = await _create_session(conn) self.assertFalse(await chan.make_request('unknown')) @asynctest async def test_invalid_channel_request(self): """Test sending non-ASCII channel request""" with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: chan, _ = await _create_session(conn) with self.assertRaises(asyncssh.ProtocolError): await chan.make_request('\xff') @asynctest async def test_delayed_channel_request(self): """Test queuing channel requests with delayed response""" with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: chan, _ = await _create_session(conn) chan.send_request(b'delayed') chan.send_request(b'delayed') @asynctest async def test_invalid_channel_response(self): """Test receiving response for non-existent channel request""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'invalid_response') chan.close() @asynctest async def test_already_open(self): """Test connect on an already open channel""" async with self.connect() as conn: chan, _ = await _create_session(conn) with self.assertRaises(OSError): await chan.create(None, None, None, {}, False, None, None, None, False, None, None, False, False) chan.close() @asynctest async def test_write_buffer(self): """Test setting write buffer limits""" async with self.connect() as conn: chan, _ = await _create_session(conn) chan.set_write_buffer_limits() chan.set_write_buffer_limits(low=8192) chan.set_write_buffer_limits(high=32768) chan.set_write_buffer_limits(32768, 8192) with self.assertRaises(ValueError): chan.set_write_buffer_limits(8192, 32768) self.assertEqual(chan.get_write_buffer_size(), 0) chan.close() @asynctest async def test_empty_write(self): """Test writing an empty block of data""" async with self.connect() as conn: chan, _ = await _create_session(conn) chan.write('') chan.close() @asynctest async def test_invalid_write_extended(self): """Test writing using an invalid extended data type""" async with self.connect() as conn: chan, _ = await _create_session(conn) with self.assertRaises(OSError): chan.write('test', -1) @asynctest async def test_unneeded_resume_reading(self): """Test resume reading when not paused""" async with self.connect() as conn: chan, _ = await _create_session(conn) await asyncio.sleep(0.1) chan.resume_reading() chan.close() @asynctest async def test_agent_forwarding_explicit(self): """Test SSH agent forwarding with explicit path""" if not self.agent_available(): # pragma: no cover self.skipTest('ssh-agent not available') async with self.connect(username='ckey', agent_forwarding='agent') as conn: chan, session = await _create_session(conn, 'agent') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, '3\n') chan, session = await _create_session(conn, 'agent') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, '3\n') @asynctest async def test_agent_forwarding_sock(self): """Test SSH agent forwarding via UNIX domain socket""" if not self.agent_available(): # pragma: no cover self.skipTest('ssh-agent not available') async with self.connect(username='ckey', agent_forwarding=True) as conn: chan, session = await _create_session(conn, 'agent_sock') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, '3\n') @asynctest async def test_rejected_session(self): """Test receiving inbound session request""" await self._check_action('rejected_session', 1) @asynctest async def test_rejected_tcpip_direct(self): """Test receiving inbound direct TCP/IP connection""" await self._check_action('rejected_tcpip_direct', 1) @asynctest async def test_unknown_tcpip_listener(self): """Test receiving connection on unknown TCP/IP listener""" await self._check_action('unknown_tcpip_listener', 1) @asynctest async def test_invalid_tcpip_listener(self): """Test receiving connection on invalid TCP/IP listener path""" await self._check_action('invalid_tcpip_listener', None) @asynctest async def test_rejected_unix_direct(self): """Test receiving inbound direct UNIX connection""" await self._check_action('rejected_unix_direct', 1) @asynctest async def test_unknown_unix_listener(self): """Test receiving connection on unknown UNIX listener""" await self._check_action('unknown_unix_listener', 1) @asynctest async def test_invalid_unix_listener(self): """Test receiving connection on invalid UNIX listener path""" await self._check_action('invalid_unix_listener', None) @asynctest async def test_rejected_tun_request(self): """Test receiving inbound TUN request""" await self._check_action('rejected_tun_request', 1) @asynctest async def test_rejected_tap_request(self): """Test receiving inbound TAP request""" await self._check_action('rejected_tap_request', 1) @asynctest async def test_agent_forwarding_failure(self): """Test failure of SSH agent forwarding""" ckey = asyncssh.read_private_key('ckey') cert = make_certificate('ssh-rsa-cert-v01@openssh.com', CERT_TYPE_USER, ckey, ckey, ['ckey'], extensions={'no-agent-forwarding': ''}) async with self.connect(username='ckey', client_keys=[(ckey, cert)], agent_path=None, agent_forwarding=True) as conn: chan, session = await _create_session(conn, 'agent') await chan.wait_closed() self.assertEqual(session.exit_status, 1) @asynctest async def test_agent_forwarding_sock_failure(self): """Test failure to create SSH agent forwarding socket""" old_tempdir = tempfile.tempdir try: tempfile.tempdir = 'xxx' async with self.connect(username='ckey', agent_forwarding=True) as conn: chan, session = await _create_session(conn, 'agent_sock') await chan.wait_closed() self.assertEqual(session.exit_status, 1) finally: tempfile.tempdir = old_tempdir @asynctest async def test_agent_forwarding_not_offered(self): """Test SSH agent forwarding not offered by client""" async with self.connect() as conn: chan, session = await _create_session(conn, 'agent') await chan.wait_closed() self.assertEqual(session.exit_status, 1) @asynctest async def test_agent_forwarding_rejected(self): """Test rejection of SSH agent forwarding by client""" async with self.connect() as conn: chan, session = await _create_session(conn, 'rejected_agent') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'False\n') self.assertEqual(session.exit_status, 1) @asynctest async def test_request_pty(self): """Test reuquesting a PTY with terminal information""" modes = {asyncssh.PTY_OP_OSPEED: 9600} async with self.connect(username='request_pty') as conn: chan, session = await _create_session(conn, term_type='ansi', term_size=(80, 24), term_modes=modes) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, "Req: ('ansi', (80, 24, 0, 0), 9600)\r\n") @asynctest async def test_terminal_full_size(self): """Test sending terminal information with full size""" modes = {asyncssh.PTY_OP_OSPEED: 9600} async with self.connect() as conn: chan, session = await _create_session(conn, 'term', term_type='ansi', term_size=(80, 24, 480, 240), term_modes=modes) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, "('ansi', (80, 24, 480, 240), 9600)\r\n") @asynctest async def test_pty_without_term_type(self): """Test requesting a PTY without setting the terminal type""" async with self.connect() as conn: chan, session = await _create_session(conn, 'term', request_pty='force') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, "('', (0, 0, 0, 0), None)\n") @asynctest async def test_invalid_terminal_size(self): """Test sending invalid terminal size""" async with self.connect() as conn: with self.assertRaises(ValueError): await _create_session(conn, 'term', term_type='ansi', term_size=(0, 0, 0)) @asynctest async def test_invalid_terminal_modes(self): """Test sending invalid terminal modes""" modes = {asyncssh.PTY_OP_RESERVED: 0} async with self.connect() as conn: with self.assertRaises(ValueError): await _create_session(conn, 'term', term_type='ansi', term_modes=modes) @asynctest async def test_pty_disallowed_by_cert(self): """Test rejection of pty request by certificate""" ckey = asyncssh.read_private_key('ckey') cert = make_certificate('ssh-rsa-cert-v01@openssh.com', CERT_TYPE_USER, ckey, ckey, ['ckey'], extensions={'no-pty': ''}) async with self.connect(username='ckey', client_keys=[(ckey, cert)], agent_path=None) as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn, 'term', term_type='ansi') @asynctest async def test_pty_disallowed_by_session(self): """Test rejection of pty request by session""" async with self.connect(username='no_pty') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await _create_session(conn, term_type='ansi') @asynctest async def test_invalid_term_type(self): """Test requesting an invalid terminal type""" with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: with self.assertRaises(asyncssh.ProtocolError): await _create_session(conn, term_type=b'\xff') @asynctest async def test_term_modes_missing_end(self): """Test sending terminal modes without PTY_OP_END""" modes = {asyncssh.PTY_OP_OSPEED: 9600, PTY_OP_NO_END: 0} with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: chan, session = await _create_session(conn, 'term', term_type='ansi', term_modes=modes) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, "('ansi', (0, 0, 0, 0), 9600)\r\n") @asynctest async def test_term_modes_incomplete(self): """Test sending terminal modes with incomplete value""" modes = {asyncssh.PTY_OP_OSPEED: 9600, PTY_OP_PARTIAL: 0} with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: with self.assertRaises(asyncssh.ProtocolError): await _create_session(conn, 'term', term_type='ansi', term_modes=modes) @asynctest async def test_env(self): """Test setting environment with byte strings""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env', env={b'TEST': b'test'}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @asynctest async def test_env_str(self): """Test setting environment using Unicode strings""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env_str', env={'TEST': 'test'}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @asynctest async def test_env_str_cached(self): """Test caching of Unicode string environment dict on server""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env_str_cached', env={'TEST': 'test'}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'testtest\n') @asynctest async def test_env_invalid_str(self): """Test trying to access binary envionment value as a Unicode string""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env_str', env={'TEST': b'test\xff'}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, '\n') @asynctest async def test_env_binary_key(self): """Test setting environment with binary data in key""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env_binary_key', env={b'TEST\xff': 'test'}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @asynctest async def test_env_binary_value(self): """Test setting environment with binary data in value""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env', env={'TEST': b'test\xff'}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\\xff\n') @asynctest async def test_env_non_string_key(self): """Test setting environment with non-string as a key""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env_non_string_key', env={1: 'test'}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @asynctest async def test_env_non_string_value(self): """Test setting environment with non-string as a value""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env', env={'TEST': 1}) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, '1\n') @asynctest async def test_invalid_env(self): """Test sending invalid environment""" async with self.connect() as conn: with self.assertRaises(ValueError): await _create_session(conn, 'env', env=1) @asynctest async def test_env_from_connect(self): """Test setting environment on connection""" async with self.connect(env={'TEST': 'test'}) as conn: chan, session = await _create_session(conn, 'env') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @asynctest async def test_env_list(self): """Test setting environment using a list of name=value strings""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env', env=['TEST=test']) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @asynctest async def test_env_list_binary(self): """Test setting environment using a list of name=value byte strings""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env', env=[b'TEST=test\xff']) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\\xff\n') @asynctest async def test_env_tuple(self): """Test setting environment using a tuple of name=value strings""" async with self.connect() as conn: chan, session = await _create_session(conn, 'env', env=('TEST=test',)) await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @asynctest async def test_invalid_env_list(self): """Test setting environment using an invalid string""" with self.assertRaises(ValueError): async with self.connect() as conn: await _create_session(conn, 'env', env=['XXX']) @asynctest async def test_send_env(self): """Test sending local environment""" async with self.connect() as conn: try: os.environ['TEST'] = 'test' chan, session = await _create_session(conn, 'env', send_env=['TEST']) finally: del os.environ['TEST'] await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') @unittest.skipUnless(os.supports_bytes_environ, 'skip binary send env if not supported by OS') @asynctest async def test_send_env_binary(self): """Test sending local environment using a byte string""" async with self.connect() as conn: try: os.environb[b'TEST'] = b'test\xff' chan, session = await _create_session(conn, 'env', send_env=[b'TEST']) finally: del os.environb[b'TEST'] await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\\xff\n') @asynctest async def test_send_env_from_connect(self): """Test sending local environment on connection""" try: os.environ['TEST'] = 'test' async with self.connect(send_env=['TEST']) as conn: chan, session = await _create_session(conn, 'env') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, 'test\n') finally: del os.environ['TEST'] @asynctest async def test_mixed_env(self): """Test sending a mix of local environment and new values""" async with self.connect() as conn: try: os.environ['TEST'] = '1' chan, session = await _create_session(conn, 'env', env={'TEST': 2}, send_env='TEST') finally: del os.environ['TEST'] await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, '2\n') @asynctest async def test_xon_xoff_enable(self): """Test enabling XON/XOFF flow control""" async with self.connect() as conn: chan, session = await _create_session(conn, 'xon_xoff') await chan.wait_closed() self.assertEqual(session.xon_xoff, True) @asynctest async def test_xon_xoff_disable(self): """Test disabling XON/XOFF flow control""" async with self.connect() as conn: chan, session = await _create_session(conn, 'no_xon_xoff') await chan.wait_closed() self.assertEqual(session.xon_xoff, False) @asynctest async def test_break(self): """Test sending a break""" async with self.connect() as conn: chan, session = await _create_session(conn, 'signals') chan.send_break(1000) await chan.wait_closed() self.assertEqual(session.exit_signal_msg, '1000') @asynctest async def test_signal(self): """Test sending a signal""" async with self.connect() as conn: chan, session = await _create_session(conn, 'signals') chan.send_signal('INT') await chan.wait_closed() self.assertEqual(session.exit_signal_msg, 'INT') @asynctest async def test_numeric_signal(self): """Test sending a signal using a numeric value""" async with self.connect() as conn: chan, session = await _create_session(conn, 'signals') chan.send_signal(SIGINT) await chan.wait_closed() self.assertEqual(session.exit_signal_msg, 'INT') @asynctest async def test_unknown_signal(self): """Test sending a signal with an unknown numeric value""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'signals') with self.assertRaises(ValueError): chan.send_signal(123) chan.close() @asynctest async def test_terminate(self): """Test sending a terminate signal""" async with self.connect() as conn: chan, session = await _create_session(conn, 'signals') chan.terminate() await chan.wait_closed() self.assertEqual(session.exit_signal_msg, 'TERM') @asynctest async def test_kill(self): """Test sending a kill signal""" async with self.connect() as conn: chan, session = await _create_session(conn, 'signals') chan.kill() await chan.wait_closed() self.assertEqual(session.exit_signal_msg, 'KILL') @asynctest async def test_invalid_signal(self): """Test sending an invalid signal""" with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect() as conn: chan, session = await _create_session(conn, 'signals') chan.send_signal(b'\xff') chan.write('\n') await chan.wait_closed() self.assertEqual(session.exit_status, None) @asynctest async def test_terminal_size_change(self): """Test sending terminal size change""" async with self.connect() as conn: chan, session = await _create_session(conn, 'signals', term_type='ansi') chan.change_terminal_size(80, 24) await chan.wait_closed() self.assertEqual(session.exit_signal_msg, '(80, 24, 0, 0)') @asynctest async def test_full_terminal_size_change(self): """Test sending full terminal size change""" async with self.connect() as conn: chan, session = await _create_session(conn, 'signals', term_type='ansi') chan.change_terminal_size(80, 24, 480, 240) await chan.wait_closed() self.assertEqual(session.exit_signal_msg, '(80, 24, 480, 240)') @asynctest async def test_exit_status(self): """Test receiving exit status""" async with self.connect() as conn: chan, session = await _create_session(conn, 'exit_status') await chan.wait_closed() self.assertEqual(session.exit_status, 1) self.assertEqual(chan.get_exit_status(), 1) self.assertIsNone(chan.get_exit_signal()) self.assertEqual(chan.get_returncode(), 1) @asynctest async def test_exit_status_after_close(self): """Test delivery of exit status after remote close""" async with self.connect() as conn: chan, session = await _create_session(conn, 'closed_status') await chan.wait_closed() self.assertIsNone(session.exit_status) self.assertIsNone(chan.get_exit_status()) self.assertIsNone(chan.get_exit_signal()) self.assertIsNone(chan.get_returncode()) @asynctest async def test_exit_signal(self): """Test receiving exit signal""" async with self.connect() as conn: chan, session = await _create_session(conn, 'exit_signal') await chan.wait_closed() self.assertEqual(session.exit_signal_msg, 'exit_signal') self.assertEqual(chan.get_exit_status(), -1) self.assertEqual(chan.get_exit_signal(), ('INT', False, 'exit_signal', DEFAULT_LANG)) self.assertEqual(chan.get_returncode(), -SIGINT) @asynctest async def test_exit_signal_after_close(self): """Test delivery of exit signal after remote close""" async with self.connect() as conn: chan, session = await _create_session(conn, 'closed_signal') await chan.wait_closed() self.assertIsNone(session.exit_signal_msg) self.assertIsNone(chan.get_exit_status()) self.assertIsNone(chan.get_exit_signal()) self.assertIsNone(chan.get_returncode()) @asynctest async def test_unknown_exit_signal(self): """Test receiving unknown exit signal""" async with self.connect() as conn: chan, session = await _create_session(conn, 'unknown_signal') await chan.wait_closed() self.assertEqual(session.exit_signal_msg, 'unknown_signal') self.assertEqual(chan.get_exit_status(), -1) self.assertEqual(chan.get_exit_signal(), ('unknown', False, 'unknown_signal', DEFAULT_LANG)) self.assertEqual(chan.get_returncode(), -99) @asynctest async def test_invalid_exit_signal(self): """Test delivery of invalid exit signal""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'invalid_exit_signal') await chan.wait_closed() @asynctest async def test_invalid_exit_lang(self): """Test delivery of invalid exit signal language""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'invalid_exit_lang') await chan.wait_closed() @asynctest async def test_window_adjust_after_eof(self): """Test receiving window adjust after EOF""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'window_after_close') await chan.wait_closed() @asynctest async def test_empty_data(self): """Test receiving empty data packet""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'empty_data') chan.close() @asynctest async def test_partial_unicode(self): """Test receiving Unicode data spread across two packets""" async with self.connect() as conn: chan, session = await _create_session(conn, 'partial_unicode') await chan.wait_closed() result = ''.join(session.recv_buf[None]) self.assertEqual(result, '\xff\xff') @asynctest async def test_partial_unicode_at_eof(self): """Test receiving partial Unicode data and then EOF""" async with self.connect() as conn: chan, session = await _create_session( conn, 'partial_unicode_at_eof') await chan.wait_closed() self.assertIsInstance(session.exc, asyncssh.ProtocolError) @asynctest async def test_unicode_error(self): """Test receiving bad Unicode data""" async with self.connect() as conn: chan, session = await _create_session(conn, 'unicode_error') await chan.wait_closed() self.assertIsInstance(session.exc, asyncssh.ProtocolError) @asynctest async def test_data_past_window(self): """Test receiving a data packet past the advertised window""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'data_past_window') await chan.wait_closed() @asynctest async def test_ext_data_past_window(self): """Test receiving an extended data packet past the advertised window""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'ext_data_past_window') await chan.wait_closed() @asynctest async def test_data_after_eof(self): """Test receiving data after EOF""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'data_after_eof') await chan.wait_closed() @asynctest async def test_data_after_close(self): """Test receiving data after close""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'data_after_close') chan.write(4*1025*1024*'\0') chan.close() await asyncio.sleep(0.2) await chan.wait_closed() @asynctest async def test_extended_data_after_eof(self): """Test receiving extended data after EOF""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'ext_data_after_eof') await chan.wait_closed() @asynctest async def test_invalid_datatype(self): """Test receiving data with invalid data type""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'invalid_datatype') await chan.wait_closed() @asynctest async def test_double_eof(self): """Test receiving two EOF messages""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'double_eof') await chan.wait_closed() @asynctest async def test_double_close(self): """Test receiving two close messages""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'double_close') chan.pause_reading() await asyncio.sleep(0.2) chan.resume_reading() await chan.wait_closed() @asynctest async def test_request_after_close(self): """Test receiving a channel request after a close""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'request_after_close') await chan.wait_closed() @asynctest async def test_late_auth_banner(self): """Test server sending authentication banner after auth completes""" async with self.connect() as conn: chan, session = await _create_session(conn, 'late_auth_banner') await chan.wait_closed() self.assertEqual(session.exit_status, 1) @asynctest async def test_unexpected_userauth_request(self): """Test userauth request sent to client""" async with self.connect() as conn: chan, _ = await _create_session(conn, 'unexpected_auth') await chan.wait_closed() @asynctest async def test_unknown_action(self): """Test unknown action""" async with self.connect() as conn: chan, session = await _create_session(conn, 'unknown') await chan.wait_closed() self.assertEqual(session.exit_status, 255) @asynctest async def test_client_session_cleanup_error(self): """Test error in client session cleanup""" async with self.connect() as conn: await conn.create_session(_ClientSessionCleanupError) class _TestChannelNoPTY(ServerTestCase): """Unit tests for AsyncSSH channel module with PTYs disallowed""" @classmethod async def start_server(cls): """Start an SSH server for the tests to use""" return (await cls.create_server( _ChannelServer, authorized_client_keys='authorized_keys', allow_pty=False)) @asynctest async def test_shell_pty(self): """Test starting a shell that request a PTY""" async with self.connect() as conn: with self.assertRaises(asyncssh.ChannelOpenError): await conn.run(term_type='ansi') @asynctest async def test_shell_no_pty(self): """Test starting a shell that doesn't request a PTY""" async with self.connect() as conn: await conn.run(request_pty=False, stdin=asyncssh.DEVNULL) @asynctest async def test_exec_pty(self): """Test execution of a remote command that requests a PTY""" async with self.connect() as conn: with self.assertRaises(asyncssh.ChannelOpenError): await conn.run('echo', request_pty='force') @asynctest async def test_exec_pty_from_connect(self): """Test execution of a command that requests a PTY on the connection""" async with self.connect(request_pty='force') as conn: with self.assertRaises(asyncssh.ChannelOpenError): await conn.run('echo') @asynctest async def test_exec_no_pty(self): """Test execution of a remote command that doesn't request a PTY""" async with self.connect() as conn: await conn.run('echo', term_type='ansi', request_pty='auto', stdin=asyncssh.DEVNULL) class _TestChannelNoAgentForwarding(ServerTestCase): """Unit tests for channel module with agent forwarding disallowed""" @classmethod async def start_server(cls): """Start an SSH server with agent forwarding disabled""" return (await cls.create_server( _ChannelServer, authorized_client_keys='authorized_keys', agent_forwarding=False)) @asynctest async def test_agent_forwarding_disallowed(self): """Test starting a shell that request a PTY""" async with self.connect(agent_forwarding=True) as conn: result = await conn.run('agent') self.assertEqual(result.exit_status, 1) class _TestConnectionDropbearClient(ServerTestCase): """Unit tests for testing Dropbear client compatibility fix""" @classmethod async def start_server(cls): """Start an SSH server to connect to""" return await cls.create_server(_ChannelServer) @asynctest async def test_dropbear_client(self): """Test reduced dropbear send packet size""" with patch('asyncssh.connection.SSHServerChannel', _ServerChannel): async with self.connect( client_version='dropbear', max_pktsize=32759, compression_algs=['zlib@openssh.com']) as conn: _, stdout, _ = await conn.open_session('send_pktsize') self.assertEqual((await stdout.read()), '32758') async with self.connect(client_version='dropbear', max_pktsize=32759, compression_algs=None) as conn: _, stdout, _ = await conn.open_session('send_pktsize') self.assertEqual((await stdout.read()), '32759') class _TestConnectionDropbearServer(ServerTestCase): """Unit tests for testing Dropbear server compatibility fix""" @classmethod async def start_server(cls): """Start an SSH server to connect to""" return await cls.create_server( _ChannelServer, server_version='dropbear', max_pktsize=32759) @asynctest async def test_dropbear_server(self): """Test reduced dropbear send packet size""" with patch('asyncssh.connection.SSHClientChannel', _ClientChannel): async with self.connect( compression_algs='zlib@openssh.com') as conn: stdin, _, _ = await conn.open_session() self.assertEqual(stdin.channel.get_send_pktsize(), 32758) async with self.connect(compression_algs=None) as conn: stdin, _, _ = await conn.open_session() self.assertEqual(stdin.channel.get_send_pktsize(), 32759)