Add an option handling UTF-8 decode errors in certain SSH packets

The SSH RFCs define that strings in some SSH packets must be encoded
as UTF-8, and AsyncSSH enforces this by raising a ProtocolError when
UTF-8 decoding fails. This commit adds a new config option to specify
an error handling strategy for the following cases:

  * Disconnect reason
  * Debug message
  * Userauth banner
  * Channel open failure reason
  * Channel exit signal reason
  * SFTP error reason

The default is 'strict', whch preserves the existing behavior of
raising a ProtocolError, but selecting other options allows the
invalid bytes to be removed or replaced, avoiding the exception.

Thanks go to GitHub user Le-Syl21 for suggesting this!

Author: ronf

asyncssh/channel.py

@@ -1125,10 +1125,12 @@ class SSHClientChannel(SSHChannel, Generic[AnyStr]):
     _read_datatypes = {EXTENDED_DATA_STDERR}
 
     def __init__(self, conn: 'SSHClientConnection',
-                 loop: asyncio.AbstractEventLoop, encoding: Optional[str],
-                 errors: str, window: int, max_pktsize: int):
+                 loop: asyncio.AbstractEventLoop, utf8_decode_errors: str,
+                 encoding: Optional[str], errors: str, window: int,
+                 max_pktsize: int):
         super().__init__(conn, loop, encoding, errors, window, max_pktsize)
 
+        self._utf8_decode_errors = utf8_decode_errors
         self._exit_status: Optional[int] = None
         self._exit_signal: Optional[_ExitSignal] = None
 
@@ -1299,7 +1301,7 @@ def _process_exit_signal_request(self, packet: SSHPacket) -> bool:
 
         try:
             signal = signal_bytes.decode('ascii')
-            msg = msg_bytes.decode('utf-8')
+            msg = msg_bytes.decode('utf-8', self._utf8_decode_errors)
             lang = lang_bytes.decode('ascii')
         except UnicodeDecodeError:
             raise ProtocolError('Invalid exit signal request') from None

asyncssh/connection.py

@@ -877,6 +877,7 @@ def __init__(self, loop: asyncio.AbstractEventLoop,
         self._peer_addr = ''
         self._peer_port = 0
         self._tcp_keepalive = options.tcp_keepalive
+        self._utf8_decode_errors = options.utf8_decode_errors
         self._owner: Optional[Union[SSHClient, SSHServer]] = None
         self._extra: Dict[str, object] = {}
 
@@ -1042,6 +1043,11 @@ def logger(self) -> SSHLogger:
 
         return self._logger
 
+    def _decode_utf8(self, msg_bytes) -> str:
+        """Decode UTF-8 bytes, honoring utf8_decode_errors setting"""
+
+        return msg_bytes.decode('utf-8', self._utf8_decode_errors)
+
     def _cleanup(self, exc: Optional[Exception]) -> None:
         """Clean up this connection"""
 
@@ -2193,7 +2199,7 @@ def _process_disconnect(self, _pkttype: int, _pktid: int,
         packet.check_end()
 
         try:
-            reason = reason_bytes.decode('utf-8')
+            reason = self._decode_utf8(reason_bytes)
             lang = lang_bytes.decode('ascii')
         except UnicodeDecodeError:
             raise ProtocolError('Invalid disconnect message') from None
@@ -2236,7 +2242,7 @@ def _process_debug(self, _pkttype: int, _pktid: int,
         packet.check_end()
 
         try:
-            msg = msg_bytes.decode('utf-8')
+            msg = self._decode_utf8(msg_bytes)
             lang = lang_bytes.decode('ascii')
         except UnicodeDecodeError:
             raise ProtocolError('Invalid debug message') from None
@@ -2638,7 +2644,7 @@ def _process_userauth_banner(self, _pkttype: int, _pktid: int,
         packet.check_end()
 
         try:
-            msg = msg_bytes.decode('utf-8')
+            msg = self._decode_utf8(msg_bytes)
             lang = lang_bytes.decode('ascii')
         except UnicodeDecodeError:
             raise ProtocolError('Invalid userauth banner') from None
@@ -2755,7 +2761,7 @@ def _process_channel_open_failure(self, _pkttype: int, _pktid: int,
         packet.check_end()
 
         try:
-            reason = reason_bytes.decode('utf-8')
+            reason = self._decode_utf8(reason_bytes)
             lang = lang_bytes.decode('ascii')
         except UnicodeDecodeError:
             raise ProtocolError('Invalid channel open failure') from None
@@ -4373,8 +4379,8 @@ async def create_session(self, session_factory: SSHClientSessionFactory,
         window: int
         max_pktsize: int
 
-        chan = SSHClientChannel(self, self._loop, encoding, errors,
-                                window, max_pktsize)
+        chan = SSHClientChannel(self, self._loop, self._utf8_decode_errors,
+                                encoding, errors, window, max_pktsize)
 
         session = await chan.create(session_factory, command, subsystem,
                                     new_env, request_pty, term_type, term_size,
@@ -5745,9 +5751,9 @@ async def start_sftp_client(self, env: DefTuple[Optional[Env]] = (),
                                                     env=env, send_env=send_env,
                                                     encoding=None)
 
-        return await start_sftp_client(self, self._loop, reader, writer,
-                                       path_encoding, path_errors,
-                                       sftp_version)
+        return await start_sftp_client(self, self._loop,
+                                       self._utf8_decode_errors, reader, writer,
+                                       path_encoding, path_errors, sftp_version)
 
 
 class SSHServerConnection(SSHConnection):
@@ -7278,6 +7284,7 @@ class SSHConnectionOptions(Options, Generic[_Options]):
     family: int
     local_addr: HostPort
     tcp_keepalive: bool
+    utf8_decode_errors: str
     canonicalize_hostname: Union[bool, str]
     canonical_domains: Sequence[str]
     canonicalize_fallback_local: bool
@@ -7323,6 +7330,7 @@ def prepare(self, config: SSHConfig, # type: ignore
                 passphrase: Optional[BytesOrStr],
                 proxy_command: DefTuple[_ProxyCommand], family: DefTuple[int],
                 local_addr: DefTuple[HostPort], tcp_keepalive: DefTuple[bool],
+                utf8_decode_errors: str,
                 canonicalize_hostname: DefTuple[Union[bool, str]],
                 canonical_domains: DefTuple[Sequence[str]],
                 canonicalize_fallback_local: DefTuple[bool],
@@ -7387,6 +7395,8 @@ def _split_cname_patterns(
         self.tcp_keepalive = cast(bool, tcp_keepalive if tcp_keepalive != ()
             else config.get('TCPKeepAlive', True))
 
+        self.utf8_decode_errors = utf8_decode_errors
+
         self.canonicalize_hostname = \
             cast(Union[bool, str], canonicalize_hostname
                  if canonicalize_hostname != ()
@@ -7812,6 +7822,13 @@ class SSHClientConnectionOptions(SSHConnectionOptions):
        :param tcp_keepalive: (optional)
            Whether or not to enable keepalive probes at the TCP level to
            detect broken connections, defaulting to `True`.
+       :param utf8_decode_errors: (optional)
+           Error handling strategy to apply when UTF-8 decode errors
+           occur in SSH protocol messages, defaulting to 'strict'
+           which shuts down the connection with a ProtocolError.
+           Choosing other strategies can allow the message parsing
+           to proceed with invalid bytes in the message being removed
+           or replaced.
        :param canonicalize_hostname: (optional)
            Whether or not to enable hostname canonicalization, defaulting
            to `False`, in which case hostnames are passed as-is to the
@@ -7984,6 +8001,7 @@ class SSHClientConnectionOptions(SSHConnectionOptions):
        :type keepalive_interval: *see* :ref:`SpecifyingTimeIntervals`
        :type keepalive_count_max: `int`
        :type tcp_keepalive: `bool`
+       :type utf8_decode_errors: `str`
        :type canonicalize_hostname: `bool` or `'always'`
        :type canonical_domains: `list` of `str`
        :type canonicalize_fallback_local: `bool`
@@ -8069,6 +8087,7 @@ def prepare(self, # type: ignore
                 family: DefTuple[int] = (),
                 local_addr: DefTuple[HostPort] = (),
                 tcp_keepalive: DefTuple[bool] = (),
+                utf8_decode_errors: str = 'strict',
                 canonicalize_hostname: DefTuple[Union[bool, str]] = (),
                 canonical_domains: DefTuple[Sequence[str]] = (),
                 canonicalize_fallback_local: DefTuple[bool] = (),
@@ -8180,10 +8199,11 @@ def prepare(self, # type: ignore
 
         super().prepare(config, client_factory or SSHClient, client_version,
                         host, port, tunnel, passphrase, proxy_command, family,
-                        local_addr, tcp_keepalive, canonicalize_hostname,
-                        canonical_domains, canonicalize_fallback_local,
-                        canonicalize_max_dots, canonicalize_permitted_cnames,
-                        kex_algs, encryption_algs, mac_algs, compression_algs,
+                        local_addr, tcp_keepalive, utf8_decode_errors,
+                        canonicalize_hostname, canonical_domains,
+                        canonicalize_fallback_local, canonicalize_max_dots,
+                        canonicalize_permitted_cnames, kex_algs,
+                        encryption_algs, mac_algs, compression_algs,
                         signature_algs, host_based_auth, public_key_auth,
                         kbdint_auth, password_auth, x509_trusted_certs,
                         x509_trusted_cert_paths, x509_purposes, rekey_bytes,
@@ -8636,6 +8656,13 @@ class SSHServerConnectionOptions(SSHConnectionOptions):
        :param tcp_keepalive: (optional)
            Whether or not to enable keepalive probes at the TCP level to
            detect broken connections, defaulting to `True`.
+       :param utf8_decode_errors: (optional)
+           Error handling strategy to apply when UTF-8 decode errors
+           occur in SSH protocol messages, defaulting to 'strict'
+           which shuts down the connection with a ProtocolError.
+           Choosing other strategies can allow the message parsing
+           to proceed with invalid bytes in the message being removed
+           or replaced.
        :param canonicalize_hostname: (optional)
            Whether or not to enable hostname canonicalization, defaulting
            to `False`, in which case hostnames are passed as-is to the
@@ -8732,6 +8759,7 @@ class SSHServerConnectionOptions(SSHConnectionOptions):
        :type keepalive_interval: *see* :ref:`SpecifyingTimeIntervals`
        :type keepalive_count_max: `int`
        :type tcp_keepalive: `bool`
+       :type utf8_decode_errors: `str`
        :type canonicalize_hostname: `bool` or `'always'`
        :type canonical_domains: `list` of `str`
        :type canonicalize_fallback_local: `bool`
@@ -8790,6 +8818,7 @@ def prepare(self, # type: ignore
                 family: DefTuple[int] = (),
                 local_addr: DefTuple[HostPort] = (),
                 tcp_keepalive: DefTuple[bool] = (),
+                utf8_decode_errors: str = 'strict',
                 canonicalize_hostname: DefTuple[Union[bool, str]] = (),
                 canonical_domains: DefTuple[Sequence[str]] = (),
                 canonicalize_fallback_local: DefTuple[bool] = (),
@@ -8865,10 +8894,11 @@ def prepare(self, # type: ignore
 
         super().prepare(config, server_factory or SSHServer, server_version,
                         host, port, tunnel, passphrase, proxy_command, family,
-                        local_addr, tcp_keepalive, canonicalize_hostname,
-                        canonical_domains, canonicalize_fallback_local,
-                        canonicalize_max_dots, canonicalize_permitted_cnames,
-                        kex_algs, encryption_algs, mac_algs, compression_algs,
+                        local_addr, tcp_keepalive, utf8_decode_errors,
+                        canonicalize_hostname, canonical_domains,
+                        canonicalize_fallback_local, canonicalize_max_dots,
+                        canonicalize_permitted_cnames, kex_algs,
+                        encryption_algs, mac_algs, compression_algs,
                         signature_algs, host_based_auth, public_key_auth,
                         kbdint_auth, password_auth, x509_trusted_certs,
                         x509_trusted_cert_paths, x509_purposes,

asyncssh/sftp.py

@@ -959,14 +959,16 @@ class SFTPError(Error):
     """
 
     @staticmethod
-    def construct(packet: SSHPacket) -> Optional['SFTPError']:
+    def construct(packet: SSHPacket, utf8_decode_errors: str) -> \
+            Optional['SFTPError']:
         """Construct an SFTPError from an FXP_STATUS response"""
 
         code = packet.get_uint32()
 
         if packet:
             try:
-                reason = packet.get_string().decode('utf-8')
+                reason = packet.get_string().decode('utf-8',
+                                                    utf8_decode_errors)
                 lang = packet.get_string().decode('ascii')
             except UnicodeDecodeError:
                 raise SFTPBadMessage('Invalid status message') from None
@@ -2596,11 +2598,12 @@ class SFTPClientHandler(SFTPHandler):
     """An SFTP client session handler"""
 
     def __init__(self, loop: asyncio.AbstractEventLoop,
-                 reader: 'SSHReader[bytes]', writer: 'SSHWriter[bytes]',
-                 sftp_version: int):
+                 utf8_decode_errors: str, reader: 'SSHReader[bytes]',
+                 writer: 'SSHWriter[bytes]', sftp_version: int):
         super().__init__(reader, writer)
 
         self._loop = loop
+        self._utf8_decode_errors = utf8_decode_errors
         self._version = sftp_version
         self._next_pktid = 0
         self._requests: Dict[int, _RequestWaiter] = {}
@@ -2694,7 +2697,7 @@ async def _make_request(self, pkttype: Union[int, bytes],
     def _process_status(self, packet: SSHPacket) -> None:
         """Process an incoming SFTP status response"""
 
-        exc = SFTPError.construct(packet)
+        exc = SFTPError.construct(packet, self._utf8_decode_errors)
 
         if self._version < 6:
             packet.check_end()
@@ -8209,13 +8212,15 @@ async def open(self, path: bytes, mode: str) -> SFTPServerFile:
 
 async def start_sftp_client(conn: 'SSHClientConnection',
                             loop: asyncio.AbstractEventLoop,
+                            utf8_decode_errors: str,
                             reader: 'SSHReader[bytes]',
                             writer: 'SSHWriter[bytes]',
                             path_encoding: Optional[str],
                             path_errors: str, sftp_version: int) -> SFTPClient:
     """Start an SFTP client"""
 
-    handler = SFTPClientHandler(loop, reader, writer, sftp_version)
+    handler = SFTPClientHandler(loop, utf8_decode_errors,
+                                reader, writer, sftp_version)
 
     handler.logger.info('Starting SFTP client')
 

tests/test_channel.py

@@ -145,6 +145,9 @@ def _send_request(self, request, *args, want_reply=False):
             if args[0] == String('invalid'):
                 args = (String(b'\xff'),) + args[1:]
 
+            if args[2] == String('invalid'):
+                args = args[:2] + (String(b'\xff'),) + args[3:]
+
             if args[3] == String('invalid'):
                 args = args[:3] + (String(b'\xff'),)
 
@@ -429,6 +432,8 @@ async def _begin_session(self, stdin, stdout, stderr):
             stdin.channel.exit_with_signal('INT', False, 'closed_signal')
         elif action == 'invalid_exit_signal':
             stdin.channel.exit_with_signal('invalid')
+        elif action == 'invalid_exit_msg':
+            stdin.channel.exit_with_signal('INT', False, 'invalid', '')
         elif action == 'invalid_exit_lang':
             stdin.channel.exit_with_signal('INT', False, '', 'invalid')
         elif action == 'window_after_close':
@@ -1593,6 +1598,27 @@ async def test_invalid_exit_signal(self):
             chan, _ = await _create_session(conn, 'invalid_exit_signal')
 
             await chan.wait_closed()
+            self.assertIsNone(chan.get_exit_signal())
+
+    @asynctest
+    async def test_invalid_exit_msg(self):
+        """Test delivery of invalid exit signal message"""
+
+        async with self.connect() as conn:
+            chan, _ = await _create_session(conn, 'invalid_exit_msg')
+
+            await chan.wait_closed()
+            self.assertIsNone(chan.get_exit_signal())
+
+    @asynctest
+    async def test_invalid_exit_msg_error_handler(self):
+        """Test delivery of invalid exit signal message"""
+
+        async with self.connect(utf8_decode_errors='ignore') as conn:
+            chan, _ = await _create_session(conn, 'invalid_exit_msg')
+
+            await chan.wait_closed()
+            self.assertIsNotNone(chan.get_exit_signal())
 
     @asynctest
     async def test_invalid_exit_lang(self):

tests/test_connection.py

@@ -350,16 +350,16 @@ def begin_auth(self, username):
         return False
 
 
-class _VersionRecordingClient(asyncssh.SSHClient):
-    """Client for testing custom client version"""
+class _AuthBannerRecordingClient(asyncssh.SSHClient):
+    """Client which records auth banner"""
 
     def __init__(self):
-        self.reported_version = None
+        self.auth_banner = None
 
     def auth_banner_received(self, msg, lang):
         """Record the client version reported in the auth banner"""
 
-        self.reported_version = msg
+        self.auth_banner = msg
 
 
 class _VersionReportingServer(Server):
@@ -2542,11 +2542,26 @@ async def start_server(cls):
 
     @asynctest
     async def test_invalid_auth_banner(self):
-        """Test server sending invalid auth banner"""
+        """Test server sending invalid UTF-8 in auth banner"""
 
         with self.assertRaises(asyncssh.ProtocolError):
             await self.connect()
 
+    @asynctest
+    async def test_invalid_auth_banner_error_handler(self):
+        """Test error handler for invalid UTF-8 in auth banner"""
+
+        for errors in ('ignore', 'replace', 'backslashreplace',
+                       'surrogateescape'):
+            conn, client = \
+                await self.create_connection(_AuthBannerRecordingClient,
+                                             utf8_decode_errors=errors)
+
+            async with conn:
+                self.assertEqual(client.auth_banner,
+                                 b'\xff'.decode('utf-8', errors))
+
+
 
 class _TestExpiredServerHostCertificate(ServerTestCase):
     """Unit tests for expired server host certificate"""
@@ -2585,11 +2600,11 @@ async def _check_client_version(self, version):
         """Check custom client version"""
 
         conn, client = \
-            await self.create_connection(_VersionRecordingClient,
+            await self.create_connection(_AuthBannerRecordingClient,
                                          client_version=version)
 
         async with conn:
-            self.assertEqual(client.reported_version, 'SSH-2.0-custom')
+            self.assertEqual(client.auth_banner, 'SSH-2.0-custom')
 
     @asynctest
     async def test_custom_client_version(self):