From b690914f18e4a3f480032df19b05ab2a224766d1 Mon Sep 17 00:00:00 2001 From: hansjone Date: Thu, 30 Jul 2026 02:34:25 +0000 Subject: [PATCH] fix(webcrt): restore arrow keys under application cursor mode Normalize SS3 arrows to CSI and stop remapping to Ctrl-B so left/right actually move the device cursor after login. Co-authored-by: Cursor --- netx_api/webcrt_service.py | 41 ++++++++++++++++++++++++++++---------- tests/test_webcrt.py | 10 ++++++++-- 2 files changed, 39 insertions(+), 12 deletions(-) diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index 357d025..fbc53d6 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -37,8 +37,13 @@ _reaper_started = False # Sentinel for take_stdout timeout (distinct from device EOF None). _STDOUT_MISSING = object() -# Network device CLIs often reject xterm CSI arrows over Telnet/SSH. -# Backspace: map DEL(0x7f) -> BS(0x08), matching common SecureCRT/VT default. +# Network device CLI key rewrites (SecureCRT-like). +# - Backspace: DEL(0x7f) -> BS(0x08) +# - Home/End/Delete: emacs controls (widely accepted) +# - Arrows: after login many boxes enable DECCKM (application cursor), so xterm +# sends SS3 forms (\x1bOD) which VRP/IOS ignore; normalize SS3 -> CSI and +# pass CSI through. Do not rewrite arrows to Ctrl-B/F — that leaves the +# device cursor stuck at EOL when SS3 was what actually arrived. _NETWORK_CLI_KEY_SEQS: tuple[tuple[str, str], ...] = ( ("\x1b[1~", "\x01"), # Home -> Ctrl-A ("\x1b[3~", "\x04"), # Delete key -> Ctrl-D @@ -47,10 +52,10 @@ _NETWORK_CLI_KEY_SEQS: tuple[tuple[str, str], ...] = ( ("\x1b[F", "\x05"), ("\x1bOH", "\x01"), ("\x1bOF", "\x05"), - ("\x1b[A", "\x10"), # Up -> Ctrl-P (history) - ("\x1b[B", "\x0e"), # Down -> Ctrl-N - ("\x1b[C", "\x06"), # Right -> Ctrl-F - ("\x1b[D", "\x02"), # Left -> Ctrl-B + ("\x1bOA", "\x1b[A"), # App Up -> CSI Up + ("\x1bOB", "\x1b[B"), + ("\x1bOC", "\x1b[C"), + ("\x1bOD", "\x1b[D"), # App Left -> CSI Left ("\x7f", "\x08"), # DEL -> BS ) @@ -65,9 +70,15 @@ def uses_network_cli_keymap(device_type: str = "", vendor: str = "") -> bool: return True -def map_network_cli_keys(data: str, *, device_type: str = "", vendor: str = "") -> str: +def map_network_cli_keys( + data: str, + *, + device_type: str = "", + vendor: str = "", + protocol: str = "", +) -> str: """Rewrite xterm key sequences for network-device CLIs.""" - del device_type, vendor # kept for call-site compatibility; keymap is vendor-agnostic + del device_type, vendor, protocol # protocol kept for call-site compatibility text = str(data or "") if not text: return text @@ -294,7 +305,10 @@ class WebcrtSession: return if self.cli_keymap: text = map_network_cli_keys( - text, device_type=self.device_type, vendor=self.vendor + text, + device_type=self.device_type, + vendor=self.vendor, + protocol=self.protocol, ) text = map_network_cli_enter(text, self.conn) if not text: @@ -307,7 +321,14 @@ class WebcrtSession: try: if channel is not None and hasattr(channel, "send") and callable(channel.send): payload = text.encode(getattr(self.conn, "encoding", None) or "utf-8", errors="replace") - channel.send(payload) + # Paramiko may write partially when the window is full. + view = memoryview(payload) + while len(view): + n = int(channel.send(view) or 0) + if n <= 0: + time.sleep(0.01) + continue + view = view[n:] elif channel is not None and hasattr(channel, "write") and callable(channel.write): encoding = getattr(self.conn, "encoding", None) or "utf-8" channel.write(text.encode(encoding, errors="replace") if isinstance(text, str) else text) diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index 9ceb980..59bd8eb 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -61,8 +61,11 @@ class WebcrtServiceTests(unittest.TestCase): # Backspace DEL -> BS sess.write_stdin("\x7f") self.assertEqual(conn.written[-1], "\x08") + # CSI arrows pass through; application-cursor SS3 -> CSI. sess.write_stdin("\x1b[D\x1b[C\x1b[A\x1b[B") - self.assertEqual(conn.written[-1], "\x02\x06\x10\x0e") + self.assertEqual(conn.written[-1], "\x1b[D\x1b[C\x1b[A\x1b[B") + sess.write_stdin("\x1bOD") + self.assertEqual(conn.written[-1], "\x1b[D") sess.resize(120, 40) conn.remote_conn.resize_pty.assert_called_with(width=120, height=40) sess.close("test") @@ -73,7 +76,10 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(svc.map_network_cli_keys("\x7fab"), "\x08ab") self.assertEqual(svc.map_network_cli_keys("\x7fab", device_type="cisco_ios", vendor="Cisco"), "\x08ab") self.assertEqual(svc.map_network_cli_keys("\x7fab", device_type="huawei", vendor="Huawei"), "\x08ab") - self.assertEqual(svc.map_network_cli_keys("\x1b[D"), "\x02") + # Keep CSI left; normalize SS3 application-cursor left. + self.assertEqual(svc.map_network_cli_keys("\x1b[D"), "\x1b[D") + self.assertEqual(svc.map_network_cli_keys("\x1bOD"), "\x1b[D") + self.assertEqual(svc.map_network_cli_keys("\x1bOA\x1bOB\x1bOC\x1bOD"), "\x1b[A\x1b[B\x1b[C\x1b[D") self.assertTrue(svc.uses_network_cli_keymap("huawei", "Huawei")) self.assertFalse(svc.uses_network_cli_keymap("linux", "bastion")) self.assertEqual(svc.map_network_cli_enter("\r", _FakeConn()), "\n") # type: ignore[arg-type]