From e5f31dde76cc7bb25e85139c8b9df50674965128 Mon Sep 17 00:00:00 2001 From: basil-k-aji-dev <70605804+basil-k-aji-dev@users.noreply.github.com> Date: Sun, 13 Sep 2026 23:38:36 +0530 Subject: [PATCH] fix(input): discard unrecognised CSI sequences instead of typing them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TerminalEventParser recognised only a small set of CSI sequences. SGR colour sequences such as \x1b[31m matched none of them, which failed two ways depending on timing. Before ESCAPE_DELAY elapsed the parse loop broke and left the sequence buffered, stalling every byte behind it. The session read loop only calls flush() when select() times out, so during a paste feed() runs back to back and the residue is never cleared; once it crossed the 8192 guard the ValueError propagated out of the input worker and ended the session. Roughly 8KB of coloured text — git diff output, coloured logs — was enough. After the delay the ESC fallback removed only the ESC byte and emitted an ESCAPE key, leaving "[31m" to be parsed as ordinary characters and typed at the remote host. Add CSI_ANY_RE, matching a complete CSI sequence, and consume it once every specific handler has declined. Partial sequences still buffer as before, so split reads keep working. Fixes #5 --- src/sshdesk/input/terminal.py | 14 ++++++++++++ tests/test_input.py | 40 +++++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/src/sshdesk/input/terminal.py b/src/sshdesk/input/terminal.py index 659d84b..88e65a2 100644 --- a/src/sshdesk/input/terminal.py +++ b/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]) diff --git a/tests/test_input.py b/tests/test_input.py index 377f090..2fd422a 100644 --- a/tests/test_input.py +++ b/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))