fix(input): discard unrecognised CSI sequences instead of typing them

src/sshdesk/input/terminal.py

@@ -23,6 +23,11 @@
 CSI_TILDE_RE = re.compile(rb"^\x1b\[(\d+)(?:;(\d+))?~")
 LEGACY_MOUSE_PREFIX = b"\x1b[M"
 CURSOR_REPORT_RE = re.compile(rb"^\x1b\[(\d{1,4});(\d{1,4})R")
+# Any complete CSI sequence: ESC [ parameter bytes, intermediate bytes, final byte.
+# Used only after every specific handler has declined, so a sequence this parser
+# does not act on is discarded rather than accumulating in the buffer or being
+# delivered to the remote host one character at a time.
+CSI_ANY_RE = re.compile(rb"^\x1b\[[\x30-\x3f]{0,32}[\x20-\x2f]{0,8}[\x40-\x7e]")
 CSI_KEYS = {
     b"A": KeyCode.UP,
     b"B": KeyCode.DOWN,
@@ -269,6 +274,15 @@ def feed(self, data: bytes, now: float | None = None) -> list[TerminalInputEvent
                         self.escape_since = now
                     if now - self.escape_since < self.ESCAPE_DELAY:
                         break
+                unknown_csi = CSI_ANY_RE.match(bytes(self.buffer))
+                if unknown_csi is not None:
+                    # A complete CSI sequence no handler above claimed. Discard it:
+                    # leaving it buffered stalls every later byte until the buffer
+                    # trips the 8192 guard, and falling through to the ESC handler
+                    # below would type its body at the remote host as text.
+                    del self.buffer[: unknown_csi.end()]
+                    self.escape_since = None
+                    continue
                 if len(self.buffer) >= 2 and self.buffer[1] not in (ord("["), ord("O")):
                     del self.buffer[0]
                     length = self._utf8_length(self.buffer[0])

tests/test_input.py

@@ -139,6 +139,46 @@ def test_oversized_terminal_sequence_is_rejected(self) -> None:
         with self.assertRaisesRegex(ValueError, "8192"):
             TerminalEventParser().feed(b"\x1b[<" + b"1" * 9000)
 
+    def test_unrecognised_csi_is_discarded_not_typed(self) -> None:
+        """An SGR colour sequence must not reach the remote host as keystrokes.
+
+        The ESC fallback used to drop only the ESC byte, leaving '[31m' to be
+        parsed as ordinary characters, so pasting coloured text typed an Escape
+        press followed by the sequence body.
+        """
+        parser = TerminalEventParser()
+        self.assertEqual(parser.feed(b"\x1b[31m", now=0.0), [])
+        self.assertEqual(parser.flush(now=1.0), [])
+
+    def test_coloured_text_yields_only_its_visible_characters(self) -> None:
+        parser = TerminalEventParser()
+        events = parser.feed(b"\x1b[31mhi\x1b[0m", now=0.0)
+        typed = "".join(chr(event.unicode) for event in events if getattr(event, "unicode", 0))
+        self.assertEqual(typed, "hi")
+        self.assertEqual(bytes(parser.buffer), b"")
+
+    def test_bulk_coloured_input_does_not_exhaust_the_buffer(self) -> None:
+        """Unrecognised sequences must not accumulate until the 8192 guard fires.
+
+        The session read loop only calls flush() when select() times out, so a
+        paste arrives as back-to-back feed() calls. Residue that is never
+        consumed crossed the guard and the ValueError ended the session.
+        """
+        parser = TerminalEventParser()
+        chunk = b"\x1b[32mx\x1b[0m" * 256
+        for index in range(10):
+            parser.feed(chunk, now=index * 0.001)
+        self.assertEqual(bytes(parser.buffer), b"")
+
+    def test_known_sequences_still_parse_after_the_csi_fallback(self) -> None:
+        parser = TerminalEventParser()
+        self.assertEqual(parser.feed(b"\x1b[A", now=0.0), [KeyEvent(2, Modifiers.NONE, KeyCode.UP, 0)])
+        parser = TerminalEventParser()
+        self.assertEqual(parser.feed(b"\x1b[12;40R", now=0.0), [TerminalReportEvent(39, 11)])
+        parser = TerminalEventParser()
+        self.assertEqual(parser.feed(b"\x1b[", now=0.0), [])
+        self.assertEqual(bytes(parser.buffer), b"\x1b[")
+
     def test_coordinate_translation_and_letterbox(self) -> None:
         viewport = Viewport(10, 5, 100, 50, 1920, 1080)
         self.assertIsNone(translate_coordinates(9, 5, viewport))