From 3b8342acfba6a6fa1f89b66a58d147b9f2e47b69 Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 29 Jul 2026 22:27:15 +0800 Subject: [PATCH] fix(webcrt): keep device echo intact and stop dual-WS steal. Use raw Telnet reads for ANSI backspace, drop local erase so Tab/completion stays in sync, and make WebSocket attach exclusive so StrictMode remounts cannot drop the first echo. Co-authored-by: Cursor --- netx_api/webcrt_router.py | 16 +++- netx_api/webcrt_service.py | 125 +++++++++++++++++++++------ tests/test_webcrt.py | 44 +++++++++- web/src/components/WebTerminal.tsx | 132 ++++++++++++++++++++++++++--- web/src/pages/WebcrtPage.tsx | 22 ++++- 5 files changed, 298 insertions(+), 41 deletions(-) diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index 09e8623..f00a940 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -75,8 +75,9 @@ def api_close_session(session_id: str, request: Request) -> dict[str, Any]: @router.websocket("/sessions/{session_id}/ws") async def websocket_session(websocket: WebSocket, session_id: str) -> None: await websocket.accept() + attach_gen = 0 try: - sess = mark_attached(session_id) + sess, attach_gen = mark_attached(session_id) except HTTPException as exc: await websocket.send_json({"type": "status", "state": "error", "message": str(exc.detail)}) await websocket.close(code=4404 if exc.status_code == 404 else 4409) @@ -113,7 +114,14 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: async def pump_stdout() -> None: loop = asyncio.get_running_loop() while not stop.is_set(): - chunk = await loop.run_in_executor(None, sess.out_queue.get) + chunk = await loop.run_in_executor( + None, lambda: sess.take_stdout(attach_gen, timeout=0.25) + ) + if chunk == "stale": + # Newer WS owns the session (StrictMode remount); exit without stealing bytes. + break + if chunk == "empty": + continue if chunk is None: stop.set() try: @@ -179,7 +187,7 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: ) break except WebSocketDisconnect: - _log.info("webcrt ws disconnected session=%s", session_id) + _log.info("webcrt ws disconnected session=%s gen=%s", session_id, attach_gen) except Exception: _log.exception("webcrt ws error session=%s", session_id) finally: @@ -190,9 +198,11 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: except Exception: pass # Keep device session briefly so React remount / blip can re-attach. + # Only the current attach_gen may detach — older StrictMode sockets must not. if get_session(session_id) is not None: detach_session( session_id, grace_sec=8.0, client=_client_label(websocket=websocket), + attach_gen=attach_gen, ) diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index 10e57c9..b226416 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -28,12 +28,14 @@ _log = logging.getLogger("netx.webcrt") _sessions_lock = threading.Lock() _sessions: dict[str, "WebcrtSession"] = {} _reaper_started = False +# Sentinel for take_stdout timeout (distinct from device EOF None). +_STDOUT_MISSING = object() -# 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 device CLIs often reject xterm CSI arrows over Telnet/SSH. +# Backspace: map DEL(0x7f) -> BS(0x08), matching common SecureCRT/VT default. _NETWORK_CLI_KEY_SEQS: tuple[tuple[str, str], ...] = ( ("\x1b[1~", "\x01"), # Home -> Ctrl-A - ("\x1b[3~", "\x04"), # Delete -> Ctrl-D + ("\x1b[3~", "\x04"), # Delete key -> Ctrl-D ("\x1b[4~", "\x05"), # End -> Ctrl-E ("\x1b[H", "\x01"), ("\x1b[F", "\x05"), @@ -57,8 +59,9 @@ def uses_network_cli_keymap(device_type: str = "", vendor: str = "") -> bool: return True -def map_network_cli_keys(data: str) -> str: +def map_network_cli_keys(data: str, *, device_type: str = "", vendor: str = "") -> str: """Rewrite xterm key sequences for network-device CLIs.""" + del device_type, vendor # kept for call-site compatibility; keymap is vendor-agnostic text = str(data or "") if not text: return text @@ -233,13 +236,45 @@ class WebcrtSession: close_reason: str = "" bootstrap_output: bytes = b"" needs_live_prompt: bool = True + # React StrictMode remounts open a second WS before the first fully tears down. + # Only the newest attach_gen may consume out_queue / mark detach. + attach_gen: int = 0 out_queue: queue.Queue[bytes | None] = field(default_factory=queue.Queue) _reader: threading.Thread | None = field(default=None, repr=False) _write_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + _stdout_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) def touch(self) -> None: self.last_activity = time.time() + def take_stdout(self, attach_gen: int, *, timeout: float = 0.25) -> bytes | None | str: + """Exclusive stdout take for one WS attach generation. + + Returns: + bytes — device output chunk + None — device reader closed (session end) + \"stale\" — a newer WebSocket owns this session; caller must stop + \"empty\" — no data within timeout (keep polling) + """ + deadline = time.time() + max(0.05, float(timeout)) + while True: + with self._stdout_lock: + if attach_gen != self.attach_gen: + return "stale" + try: + chunk = self.out_queue.get_nowait() + except queue.Empty: + chunk = _STDOUT_MISSING + if chunk is not _STDOUT_MISSING: + if attach_gen != self.attach_gen: + # Put back including EOF sentinel so the new owner still sees close. + self.out_queue.put(chunk) + return "stale" + return chunk # bytes | None + if time.time() >= deadline: + return "empty" + time.sleep(0.02) + def write_stdin(self, data: str) -> None: if self.closed or self.conn is None: raise RuntimeError("session_closed") @@ -247,27 +282,28 @@ class WebcrtSession: if not text: return if self.cli_keymap: - text = map_network_cli_keys(text) + text = map_network_cli_keys( + text, device_type=self.device_type, vendor=self.vendor + ) text = map_network_cli_enter(text, self.conn) if not text: return with self._write_lock: - # Network CLIs: use Netmiko write_channel so RETURN/encoding match the driver. - if self.cli_keymap: - self.conn.write_channel(text) - else: - 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" - channel.write(text.encode(encoding, errors="replace") if isinstance(text, str) else text) - else: - self.conn.write_channel(text) - except Exception: + # Prefer raw channel I/O for interactive typing (char echo / backspace). + # Netmiko write_channel is fine for automation but can feel "half-duplex" + # on some Telnet/VRP sessions when used keystroke-by-keystroke. + 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" + channel.write(text.encode(encoding, errors="replace") if isinstance(text, str) else text) + else: self.conn.write_channel(text) + except Exception: + self.conn.write_channel(text) self.touch() def resize(self, cols: int, rows: int) -> None: if self.closed or self.conn is None: @@ -305,6 +341,7 @@ class WebcrtSession: chunk = b"" try: if channel is not None and hasattr(channel, "recv_ready") and hasattr(channel, "recv"): + # Paramiko SSH: raw bytes keep ANSI / backspace echo intact. if channel.recv_ready(): chunk = channel.recv(4096) if not chunk: @@ -314,6 +351,22 @@ class WebcrtSession: else: time.sleep(0.04) continue + elif channel is not None and hasattr(channel, "read_very_eager"): + # Telnet: do NOT use conn.read_channel() — Netmiko strips ANSI + # escape codes, which removes Huawei backspace echo (\x1b[1D \x1b[1D). + data = channel.read_very_eager() + if data: + chunk = ( + data + if isinstance(data, (bytes, bytearray)) + else str(data).encode( + getattr(conn, "encoding", None) or "utf-8", + errors="replace", + ) + ) + else: + time.sleep(0.04) + continue else: text = conn.read_channel() if text: @@ -544,25 +597,48 @@ def create_session( } -def mark_attached(session_id: str) -> WebcrtSession: +def mark_attached(session_id: str) -> tuple[WebcrtSession, int]: sess = get_session(session_id) if sess is None: raise HTTPException(status_code=404, detail="webcrt_session_not_found") # Allow re-attach after brief WS drop (React StrictMode remount / network blip). + # Bump generation so the previous WS pump stops and does not steal echo bytes. + sess.attach_gen += 1 + attach_gen = sess.attach_gen sess.attached = True sess.detach_deadline = None sess.touch() - _audit("session_attached", session_id=session_id, ne_id=sess.ne_id, ne_ip=sess.ne_ip) - return sess + _audit( + "session_attached", + session_id=session_id, + ne_id=sess.ne_id, + ne_ip=sess.ne_ip, + attach_gen=attach_gen, + ) + return sess, attach_gen -def detach_session(session_id: str, *, grace_sec: float = 8.0, client: str = "") -> dict[str, Any]: +def detach_session( + session_id: str, + *, + grace_sec: float = 8.0, + client: str = "", + attach_gen: int | None = None, +) -> dict[str, Any]: """Mark session unattached but keep device channel open briefly for reconnect.""" sess = get_session(session_id) if sess is None: return {"ok": True, "session_id": session_id, "detached": False} if sess.closed: return {"ok": True, "session_id": session_id, "detached": False} + # Ignore detach from an older StrictMode WS once a newer attach owns the session. + if attach_gen is not None and attach_gen != sess.attach_gen: + return { + "ok": True, + "session_id": session_id, + "detached": False, + "ignored_stale_attach": True, + } sess.attached = False sess.detach_deadline = time.time() + max(1.0, float(grace_sec)) sess.touch() @@ -573,6 +649,7 @@ def detach_session(session_id: str, *, grace_sec: float = 8.0, client: str = "") ne_ip=sess.ne_ip, grace_sec=grace_sec, client=client or "", + attach_gen=attach_gen, ) return {"ok": True, "session_id": session_id, "detached": True} diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index fe4ce07..6dde5f7 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -58,7 +58,7 @@ class WebcrtServiceTests(unittest.TestCase): # xterm Enter (\r) -> Netmiko RETURN (\n for SSH) sess.write_stdin("\r") self.assertEqual(conn.written[-1], "\n") - # xterm DEL / arrows -> network CLI controls + # Backspace DEL -> BS sess.write_stdin("\x7f") self.assertEqual(conn.written[-1], "\x08") sess.write_stdin("\x1b[D\x1b[C\x1b[A\x1b[B") @@ -69,7 +69,10 @@ class WebcrtServiceTests(unittest.TestCase): self.assertTrue(sess.closed) def test_map_network_cli_keys_helpers(self) -> None: + # Backspace: DEL -> BS (SecureCRT/VT default), vendor-agnostic. 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") self.assertTrue(svc.uses_network_cli_keymap("huawei", "Huawei")) self.assertFalse(svc.uses_network_cli_keymap("linux", "bastion")) @@ -178,6 +181,45 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(fake.written[len(before) :], ["\n"]) svc.close_session(out["session_id"], reason="test") + @patch.object(svc, "_audit") + def test_attach_gen_exclusive_stdout_and_stale_detach(self, _mock_audit: MagicMock) -> None: + conn = _FakeConn() + sess = svc.WebcrtSession( + session_id="race", + ne_id="ne1", + ne_name="lab", + ne_ip="1.2.3.4", + protocol="ssh", + cols=80, + rows=24, + conn=conn, # type: ignore[arg-type] + ) + with svc._sessions_lock: + svc._sessions["race"] = sess + + sess1, gen1 = svc.mark_attached("race") + self.assertEqual(gen1, 1) + sess1.out_queue.put(b"a") + sess1.out_queue.put(b"b") + + # Newer StrictMode WS takes ownership before old pump drains. + _sess2, gen2 = svc.mark_attached("race") + self.assertEqual(gen2, 2) + self.assertEqual(sess1.take_stdout(gen1, timeout=0.05), "stale") + self.assertEqual(sess1.take_stdout(gen2, timeout=0.05), b"a") + self.assertEqual(sess1.take_stdout(gen2, timeout=0.05), b"b") + + # Old WS detach must not clear the live attach. + out = svc.detach_session("race", grace_sec=8.0, attach_gen=gen1) + self.assertFalse(out.get("detached")) + self.assertTrue(sess.attached) + self.assertIsNone(sess.detach_deadline) + + out2 = svc.detach_session("race", grace_sec=8.0, attach_gen=gen2) + self.assertTrue(out2.get("detached")) + self.assertFalse(sess.attached) + svc.close_session("race", reason="test") + @patch.object(svc, "_audit") def test_attach_timeout_reaper(self, _mock_audit: MagicMock) -> None: conn = _FakeConn() diff --git a/web/src/components/WebTerminal.tsx b/web/src/components/WebTerminal.tsx index c206c23..c7ac2c4 100644 --- a/web/src/components/WebTerminal.tsx +++ b/web/src/components/WebTerminal.tsx @@ -8,12 +8,14 @@ export type WebTerminalHandle = { copyAll: () => Promise; getText: () => string; fit: () => void; + focus: () => void; }; type Props = { wsUrl: string; title?: string; recording?: boolean; + autoFocus?: boolean; onStatus?: (state: string, message?: string) => void; onReady?: () => void; onStdout?: (data: string) => void; @@ -30,8 +32,40 @@ function serializeTerminal(term: Terminal): string { return lines.join("\n").replace(/\s+$/g, ""); } +function isSidebarSearchTarget(target: EventTarget | null): boolean { + if (!(target instanceof HTMLElement)) return false; + if (target.tagName === "INPUT" || target.tagName === "TEXTAREA" || target.tagName === "SELECT") { + // xterm's hidden textarea must still receive keys normally. + if (target.classList.contains("xterm-helper-textarea")) return false; + return true; + } + return Boolean(target.closest(".webcrt-sidebar__search")); +} + +function isXtermTextarea(target: EventTarget | null): boolean { + return target instanceof HTMLElement && target.classList.contains("xterm-helper-textarea"); +} + +/** Map a browser key event to the bytes xterm/onData would normally emit. */ +function keyEventToStdin(e: KeyboardEvent): string | null { + if (e.ctrlKey || e.altKey || e.metaKey) return null; + if (e.key === "Backspace") return "\x08"; // BS — common SecureCRT/VT default + if (e.key === "Enter") return "\r"; + if (e.key === "Tab") return "\t"; + if (e.key === "Escape") return "\x1b"; + if (e.key === "ArrowUp") return "\x1b[A"; + if (e.key === "ArrowDown") return "\x1b[B"; + if (e.key === "ArrowRight") return "\x1b[C"; + if (e.key === "ArrowLeft") return "\x1b[D"; + if (e.key === "Home") return "\x1b[H"; + if (e.key === "End") return "\x1b[F"; + if (e.key === "Delete") return "\x1b[3~"; + if (e.key.length === 1) return e.key; + return null; +} + export const WebTerminal = forwardRef(function WebTerminal( - { wsUrl, title, recording, onStatus, onReady, onStdout }, + { wsUrl, title, recording, autoFocus = true, onStatus, onReady, onStdout }, ref, ) { const hostRef = useRef(null); @@ -42,6 +76,7 @@ export const WebTerminal = forwardRef(function WebTerm const onReadyRef = useRef(onReady); const onStdoutRef = useRef(onStdout); const recordingRef = useRef(!!recording); + const autoFocusRef = useRef(autoFocus); useEffect(() => { onStatusRef.current = onStatus; @@ -53,6 +88,18 @@ export const WebTerminal = forwardRef(function WebTerm recordingRef.current = !!recording; }, [recording]); + useEffect(() => { + autoFocusRef.current = autoFocus; + }, [autoFocus]); + + const focusTerminal = () => { + try { + termRef.current?.focus(); + } catch { + /* ignore */ + } + }; + useImperativeHandle(ref, () => ({ clear: () => { termRef.current?.clear(); @@ -73,6 +120,7 @@ export const WebTerminal = forwardRef(function WebTerm /* ignore */ } }, + focus: focusTerminal, })); useEffect(() => { @@ -103,9 +151,19 @@ export const WebTerminal = forwardRef(function WebTerm /* ignore */ } }; + const maybeFocus = () => { + if (!autoFocusRef.current) return; + focusTerminal(); + }; + // Focus immediately so the first keystroke is not lost to the sidebar/body. + maybeFocus(); requestAnimationFrame(() => { doFit(); - window.setTimeout(doFit, 50); + maybeFocus(); + window.setTimeout(() => { + doFit(); + maybeFocus(); + }, 50); }); const ws = new WebSocket(wsUrl); @@ -118,6 +176,18 @@ export const WebTerminal = forwardRef(function WebTerm } }; + const sendStdin = (data: string) => { + if (!data) return; + sendJson({ type: "stdin", data }); + }; + + // Display only what the device echoes (chars, Tab completion, BS erase, etc.). + // No local echo / local erase — those desync on Tab complete and prompt redraw. + const writeStdout = (raw: string) => { + if (raw) term.write(raw); + if (recordingRef.current) onStdoutRef.current?.(raw); + }; + const sendResize = () => { doFit(); sendJson({ type: "resize", cols: term.cols, rows: term.rows }); @@ -127,6 +197,7 @@ export const WebTerminal = forwardRef(function WebTerm onStatusRef.current?.("open"); sendResize(); onReadyRef.current?.(); + window.setTimeout(maybeFocus, 30); }; ws.onmessage = (ev) => { @@ -138,13 +209,16 @@ export const WebTerminal = forwardRef(function WebTerm message?: string; }; if (msg.type === "stdout" && typeof msg.data === "string") { - term.write(msg.data); - if (recordingRef.current) onStdoutRef.current?.(msg.data); + writeStdout(msg.data); + maybeFocus(); return; } if (msg.type === "status") { onStatusRef.current?.(String(msg.state || ""), msg.message); - if (msg.state === "connected") return; + if (msg.state === "connected") { + maybeFocus(); + return; + } if (msg.state === "closed" || msg.state === "error") { const detail = msg.message ? `: ${msg.message}` : ""; term.writeln(`\r\n\x1b[33m[session ${msg.state}${detail}]\x1b[0m`); @@ -153,9 +227,7 @@ export const WebTerminal = forwardRef(function WebTerm } if (msg.type === "pong") return; } catch { - const raw = String(ev.data || ""); - term.write(raw); - if (recordingRef.current) onStdoutRef.current?.(raw); + writeStdout(String(ev.data || "")); } }; @@ -171,10 +243,41 @@ export const WebTerminal = forwardRef(function WebTerm } }; + // When focused, xterm onData sends keystrokes. const dataDisposable = term.onData((data) => { - sendJson({ type: "stdin", data }); + // Normalize Backspace: xterm emits DEL(0x7f); devices expect BS(0x08) like default SecureCRT. + const normalized = data.replace(/\x7f/g, "\x08"); + sendStdin(normalized); }); + // When NOT focused (sidebar still focused after click), capture keys and forward + // so the first character / Backspace are not lost to the browser. + const onKeyDownCapture = (e: KeyboardEvent) => { + if (!autoFocusRef.current) return; + if (host.closest("[hidden]")) return; + if (isSidebarSearchTarget(e.target)) return; + + // Always block browser "Backspace = history back" while this session pane is active. + if (e.key === "Backspace") { + e.preventDefault(); + } + + if (isXtermTextarea(e.target)) { + // Let xterm onData handle it; we only prevented browser back above. + maybeFocus(); + return; + } + + const data = keyEventToStdin(e); + if (data == null) return; + + e.preventDefault(); + e.stopPropagation(); + maybeFocus(); + sendStdin(data); + }; + window.addEventListener("keydown", onKeyDownCapture, true); + const pingTimer = window.setInterval(() => { sendJson({ type: "ping" }); }, 25000); @@ -193,6 +296,7 @@ export const WebTerminal = forwardRef(function WebTerm return () => { window.clearInterval(pingTimer); window.removeEventListener("resize", onWinResize); + window.removeEventListener("keydown", onKeyDownCapture, true); ro?.disconnect(); dataDisposable.dispose(); try { @@ -207,5 +311,13 @@ export const WebTerminal = forwardRef(function WebTerm }; }, [wsUrl, title]); - return
; + return ( +
{ + focusTerminal(); + }} + /> + ); }); diff --git a/web/src/pages/WebcrtPage.tsx b/web/src/pages/WebcrtPage.tsx index bcc7099..6345a2b 100644 --- a/web/src/pages/WebcrtPage.tsx +++ b/web/src/pages/WebcrtPage.tsx @@ -336,6 +336,10 @@ export function WebcrtPage() {