Enforce zero seqno on kexinit

paramiko/transport.py

@@ -336,6 +336,7 @@ def __init__(
         disabled_algorithms=None,
         server_sig_algs=True,
         strict_kex=True,
+        packetizer_class=None,
     ):
         """
         Create a new SSH session over an existing socket, or socket-like
@@ -406,6 +407,9 @@ def __init__(
             Whether to advertise (and implement, if client also advertises
             support for) a "strict kex" mode for safer handshaking. Default:
             ``True``.
+        :param packetizer_class:
+            Which class to use for instantiating the internal packet handler.
+            Default: ``None`` (i.e.: use `Packetizer` as normal).
 
         .. versionchanged:: 1.15
             Added the ``default_window_size`` and ``default_max_packet_size``
@@ -418,6 +422,8 @@ def __init__(
             Added the ``server_sig_algs`` kwarg.
         .. versionchanged:: 3.4
             Added the ``strict_kex`` kwarg.
+        .. versionchanged:: 3.4
+            Added the ``packetizer_class`` kwarg.
         """
         self.active = False
         self.hostname = None
@@ -467,7 +473,7 @@ def __init__(
         self.sock.settimeout(self._active_check_timeout)
 
         # negotiated crypto parameters
-        self.packetizer = Packetizer(sock)
+        self.packetizer = (packetizer_class or Packetizer)(sock)
         self.local_version = "SSH-" + self._PROTO_ID + "-" + self._CLIENT_ID
         self.remote_version = ""
         self.local_cipher = self.remote_cipher = ""
@@ -2428,7 +2434,8 @@ def _really_parse_kex_init(self, m, ignore_first_byte=False):
 
     def _get_latest_kex_init(self):
         return self._really_parse_kex_init(
-            Message(self._latest_kex_init), ignore_first_byte=True
+            Message(self._latest_kex_init),
+            ignore_first_byte=True,
         )
 
     def _parse_kex_init(self, m):
@@ -2490,6 +2497,13 @@ def _parse_kex_init(self, m):
         for i in to_pop:
             kex_algo_list.pop(i)
 
+        # CVE mitigation: expect zeroed-out seqno anytime we are performing kex
+        # init phase, if strict mode was negotiated.
+        if self.agreed_on_strict_kex and m.seqno != 0:
+            raise MessageOrderError(
+                f"Got nonzero seqno ({m.seqno}) during strict KEXINIT!"
+            )
+
         # as a server, we pick the first item in the client's list that we
         # support.
         # as a client, we pick the first item in our list that the server

sites/www/changelog.rst

@@ -2,6 +2,9 @@
 Changelog
 =========
 
+- :feature:`-` `Transport` grew a new ``packetizer_class`` kwarg for overriding
+  the packet-handler class used internally. Mostly for testing, but advanced
+  users may find this useful when doing deep hacks.
 - :bug:`-` Address `CVE 2023-48795<https://terrapin-attack.com/>`_ (aka the
   "Terrapin Attack", a vulnerability found in the SSH protocol re: treatment of
   packet sequence numbers) as follows:

tests/test_transport.py

@@ -1055,6 +1055,16 @@ def test_server_rejects_client_MSG_USERAUTH_SUCCESS(self):
         # Real fix's behavior
         self._expect_unimplemented()
 
+    def test_can_override_packetizer_used(self):
+        class MyPacketizer(Packetizer):
+            pass
+
+        # control case
+        assert Transport(sock=LoopSocket()).packetizer.__class__ is Packetizer
+        # overridden case
+        tweaked = Transport(sock=LoopSocket(), packetizer_class=MyPacketizer)
+        assert tweaked.packetizer.__class__ is MyPacketizer
+
 
 # TODO: for now this is purely a regression test. It needs actual tests of the
 # intentional new behavior too!
@@ -1243,6 +1253,20 @@ def test_client_uses_server_sig_algs_for_pubkey_auth(self):
             assert tc._agreed_pubkey_algorithm == "rsa-sha2-256"
 
 
+class BadSeqPacketizer(Packetizer):
+    def read_message(self):
+        cmd, msg = super().read_message()
+        # Only mess w/ seqno if kexinit.
+        if cmd is MSG_KEXINIT:
+            # NOTE: this is /only/ the copy of the seqno which gets
+            # transmitted up from Packetizer; it's not modifying
+            # Packetizer's own internal seqno. For these tests,
+            # modifying the latter isn't required, and is also harder
+            # to do w/o triggering MAC mismatches.
+            msg.seqno = 17  # arbitrary nonzero int
+        return cmd, msg
+
+
 class TestStrictKex:
     def test_kex_algos_includes_kex_strict_c(self):
         with server() as (tc, _):
@@ -1277,9 +1301,6 @@ def test_mode_advertised_by_default(self):
                 )
             )
 
-    def test_sequence_numbers_reset_on_newkeys(self):
-        skip()
-
     def test_MessageOrderError_raised_on_out_of_order_messages(self):
         with raises(MessageOrderError):
             with server() as (tc, _):
@@ -1288,12 +1309,41 @@ def test_MessageOrderError_raised_on_out_of_order_messages(self):
                 tc._expect_packet(MSG_KEXINIT)
                 tc.open_session()
 
-    def test_SSHException_raised_on_out_of_order_messages_when_not_strict(self):
+    def test_SSHException_raised_on_out_of_order_messages_when_not_strict(
+        self,
+    ):
         # This is kind of dumb (either situation is still fatal!) but whatever,
         # may as well be strict with our new strict flag...
         with raises(SSHException) as info:  # would be true either way, but
-            with server(client_init=dict(strict_kex=False),
-    ) as (tc, _):
+            with server(
+                client_init=dict(strict_kex=False),
+            ) as (tc, _):
                 tc._expect_packet(MSG_KEXINIT)
                 tc.open_session()
         assert info.type is SSHException  # NOT MessageOrderError!
+
+    def test_error_not_raised_when_kexinit_not_seq_0_but_unstrict(self):
+        with server(
+            client_init=dict(
+                # Disable strict kex
+                strict_kex=False,
+                # Give our clientside a packetizer that sets all kexinit
+                # Message objects to have .seqno==17, which would trigger the
+                # new logic if we'd forgotten to wrap it in strict-kex check
+                packetizer_class=BadSeqPacketizer,
+            ),
+        ):
+            pass  # kexinit happens at connect...
+
+    def test_MessageOrderError_raised_when_kexinit_not_seq_0_and_strict(self):
+        with raises(MessageOrderError):
+            with server(
+                # Give our clientside a packetizer that sets all kexinit
+                # Message objects to have .seqno==17, which should trigger the
+                # new logic (given we are NOT disabling strict-mode)
+                client_init=dict(packetizer_class=BadSeqPacketizer),
+            ):
+                pass  # kexinit happens at connect...
+
+    def test_sequence_numbers_reset_on_newkeys(self):
+        skip()