hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 04:20:45 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2

fix(webcrt): map backspace and arrows for network device CLIs

Browse source 
Huawei/ZTE/Cisco Telnet often rejects xterm DEL and CSI arrows; rewrite them to BS and emacs Ctrl-B/F/P/N so editing and history work.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-07-29 18:44:13 +08:00
parent 1f2e028337
commit 8e3bed562f
2 changed files with 91 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

80
netx_api/webcrt_service.py
View file

@ -27,6 +27,54 @@ _sessions_lock = threading.Lock()
_sessions: dict[str, "WebcrtSession"] = {}
_reaper_started = False
# Network device CLIs (Huawei/ZTE/Cisco) often reject xterm DEL/CSI arrows over Telnet.
# Map to classic emacs-style control keys that VRP/IOS/ZXROS accept.
_NETWORK_CLI_KEY_SEQS: tuple[tuple[str, str], ...] = (
("\x1b[1~", "\x01"), # Home -> Ctrl-A
("\x1b[3~", "\x04"), # Delete -> Ctrl-D
("\x1b[4~", "\x05"), # End -> Ctrl-E
("\x1b[H", "\x01"),
("\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
("\x7f", "\x08"), # DEL -> BS
)
def uses_network_cli_keymap(device_type: str = "", vendor: str = "") -> bool:
blob = f"{device_type} {vendor}".strip().lower()
if not blob:
return True
for token in ("linux", "ubuntu", "centos", "debian", "redhat", "unix"):
if token in blob:
return False
return True
def map_network_cli_keys(data: str) -> str:
"""Rewrite xterm key sequences for network-device CLIs."""
text = str(data or "")
if not text:
return text
out: list[str] = []
i = 0
n = len(text)
while i < n:
matched = False
for seq, repl in _NETWORK_CLI_KEY_SEQS:
if text.startswith(seq, i):
out.append(repl)
i += len(seq)
matched = True
break
if not matched:
out.append(text[i])
i += 1
return "".join(out)
def _utc_now() -> datetime:
return datetime.now(timezone.utc)
@ -62,6 +110,9 @@ class WebcrtSession:
protocol: str
cols: int
rows: int
device_type: str = ""
vendor: str = ""
cli_keymap: bool = True
conn: ConnectHandler | None = None
created_at: float = field(default_factory=time.time)
last_activity: float = field(default_factory=time.time)
@ -80,12 +131,31 @@ class WebcrtSession:
if self.closed or self.conn is None:
raise RuntimeError("session_closed")
text = str(data or "")
if not text:
return
if self.cli_keymap:
text = map_network_cli_keys(text)
if not text:
return
with self._write_lock:
self.conn.write_channel(text)
# Prefer raw channel write so control bytes are not altered.
channel = getattr(self.conn, "remote_conn", None)
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)
elif channel is not None and hasattr(channel, "write") and callable(channel.write):
encoding = getattr(self.conn, "encoding", None) or "utf-8"
if isinstance(text, str):
channel.write(text.encode(encoding, errors="replace"))
else:
channel.write(text)
else:
self.conn.write_channel(text)
except Exception:
# Fallback to netmiko helper.
self.conn.write_channel(text)
self.touch()
def resize(self, cols: int, rows: int) -> None:
if self.closed or self.conn is None:
return
@ -262,6 +332,9 @@ def create_session(
target_ip = str(device.get("ip_address") or "")
target_name = str(device.get("name") or target_ip)
protocol = str(device.get("protocol") or creds.get("protocol") or "ssh")
device_type = str(device.get("device_type") or creds.get("device_type") or "")
vendor = str(device.get("vendor") or creds.get("vendor") or "")
cli_keymap = uses_network_cli_keymap(device_type, vendor)
try:
conn = open_netmiko_connection(creds, session_timeout=connect_timeout)
@ -297,6 +370,9 @@ def create_session(
protocol=protocol,
cols=c,
rows=r,
device_type=device_type,
vendor=vendor,
cli_keymap=cli_keymap,
conn=conn,
)
if leftover:

14
tests/test_webcrt.py
View file

@ -14,7 +14,7 @@ from netx_api import webcrt_service as svc
class _FakeConn:
def __init__(self) -> None:
self.written: list[str] = []
self.remote_conn = MagicMock()
self.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"])
self.remote_conn.recv_ready.return_value = False
self.remote_conn.exit_status_ready.return_value = False
self.remote_conn.resize_pty = MagicMock()
@ -49,15 +49,27 @@ class WebcrtServiceTests(unittest.TestCase):
protocol="ssh",
cols=80,
rows=24,
cli_keymap=True,
conn=conn, # type: ignore[arg-type]
)
sess.write_stdin("show ver\n")
self.assertEqual(conn.written, ["show ver\n"])
# xterm DEL / arrows -> network CLI controls
sess.write_stdin("\x7f")
self.assertEqual(conn.written[-1], "\x08")
sess.write_stdin("\x1b[D\x1b[C\x1b[A\x1b[B")
self.assertEqual(conn.written[-1], "\x02\x06\x10\x0e")
sess.resize(120, 40)
conn.remote_conn.resize_pty.assert_called_with(width=120, height=40)
sess.close("test")
self.assertTrue(sess.closed)
def test_map_network_cli_keys_helpers(self) -> None:
self.assertEqual(svc.map_network_cli_keys("\x7fab"), "\x08ab")
self.assertEqual(svc.map_network_cli_keys("\x1b[D"), "\x02")
self.assertTrue(svc.uses_network_cli_keymap("huawei", "Huawei"))
self.assertFalse(svc.uses_network_cli_keymap("linux", "bastion"))
@patch.object(svc, "_audit")
@patch.object(svc, "open_netmiko_connection")
@patch("netx_api.cli_resolve.resolve_cli_target")