diff --git a/netx_api/ne_cli_hop.py b/netx_api/ne_cli_hop.py index 807d9f8..83a45dd 100644 --- a/netx_api/ne_cli_hop.py +++ b/netx_api/ne_cli_hop.py @@ -10,7 +10,9 @@ from netmiko import ConnectHandler _CLI_HOP_NESTED_END_RE = re.compile( r"(?is)" r"(?:^|\n)\s*(?:" - r"connection\s+closed(?:\s+by\s+(?:foreign|remote)\s+host)?" + # Huawei VRP: "Info: The connection was closed by the remote host." + r"(?:info:\s*)?(?:the\s+)?connection\s+was\s+closed\s+by\s+(?:the\s+)?(?:foreign|remote)\s+host" + r"|connection\s+closed(?:\s+by\s+(?:foreign|remote)\s+host)?" r"|closed\s+by\s+foreign\s+host" r"|connection\s+to\s+\S+\s+closed" r"|%\s*connection\s+closed(?:\s+by\s+(?:foreign|remote)\s+host)?" diff --git a/netx_api/ne_session_connect.py b/netx_api/ne_session_connect.py index 3813473..049a850 100644 --- a/netx_api/ne_session_connect.py +++ b/netx_api/ne_session_connect.py @@ -545,18 +545,26 @@ def _connect_direct( def _read_channel(conn: ConnectHandler, wait: float = 0.5, max_loops: int = 40) -> str: - if wait > 0: - time.sleep(wait) + """Collect device output. + + ``wait`` is the longest we park for the *first* bytes; once output has arrived the read + returns as soon as the channel is quiet again. (It used to sleep the whole ``wait`` and + then poll at 50-150ms, which added ~0.1-0.4s of dead time to every login-prompt step.) + """ + quiet = 0.04 if wait <= 0.15 else 0.12 + first_deadline = time.monotonic() + max(0.0, wait) + got_at: float | None = None chunks: list[str] = [] - for _ in range(max_loops): + reads = 0 + while reads < max_loops: try: part = conn.read_channel() except Exception: break if part is None or part is False: - break + part = "" # MagicMock truthy-but-empty: stop when unit tests leave read_channel unconfigured. - if not isinstance(part, (str, bytes, bytearray)): + elif not isinstance(part, (str, bytes, bytearray)): text = str(part) # unittest.mock default str looks like "" if text.startswith("= first_deadline: + break + elif now - got_at >= quiet: break - chunks.append(part) - time.sleep(0.05 if wait <= 0.15 else 0.15) + time.sleep(0.01) return "".join(chunks) @@ -801,11 +817,11 @@ def _interactive_target_auth( return else: empty_after_pass = 0 - time.sleep(0.12) + time.sleep(0.02) continue if not buf.strip(): - time.sleep(0.2) + time.sleep(0.02) continue # Huawei stelnet often reprints the hop ``[sysname]`` after host-key trust and # *before* ``Enter password:``. Do not treat that as target login success. @@ -826,7 +842,7 @@ def _interactive_target_auth( if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") return - time.sleep(0.15) + time.sleep(0.02) denied = _target_auth_failure_snippet(acc, sent_pass=sent_pass) if denied: raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") @@ -879,7 +895,7 @@ def _huawei_enter_system_view( for _ in range(8): part = _read_channel(conn, wait=0.2, max_loops=10) if not part: - time.sleep(0.15) + time.sleep(0.02) continue acc += part if emit_raw: diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index 32645d3..46ae67a 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -8,6 +8,7 @@ import queue import re import threading import time +from collections import deque from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -44,6 +45,18 @@ def uses_network_cli_keymap(device_type: str = "", vendor: str = "") -> bool: return True +_NETWORK_PROMPT_RE = re.compile( + r"^(?:<[^<>\s]{1,63}>" + r"|\[[~*]?[\w.\-/:+]{1,80}\]" + r"|[A-Za-z0-9][\w.\-/:+]{0,62}(?:\([^()\s]{1,60}\))*[#>])$" +) + + +def looks_like_network_prompt(line: str) -> bool: + """```` / ``[~r1]`` / ``R2#`` / ``R2(config)#`` — never a Linux ``user@host:~$`` prompt.""" + return bool(_NETWORK_PROMPT_RE.match(str(line or "").strip())) + + def map_network_cli_keys( data: str, *, @@ -56,6 +69,8 @@ def map_network_cli_keys( text = str(data or "") if not text: return text + if "\x1b" not in text and "\x7f" not in text: + return text # plain printable typing: nothing to rewrite out: list[str] = [] i = 0 n = len(text) @@ -86,7 +101,7 @@ def channel_return(conn: ConnectHandler | None) -> str: def map_network_cli_enter(data: str, conn: ConnectHandler | None) -> str: """Map xterm Enter (\\r) to the device's Netmiko RETURN.""" text = str(data or "") - if not text: + if not text or ("\r" not in text and "\n" not in text): return text ret = channel_return(conn) # Replace once: chained replacements expand the CR in Telnet CRLF again. @@ -367,10 +382,10 @@ def _encode_text(text: str, encoding: str) -> bytes: class _BoundedByteQueue: - """Thread-safe queue that drops oldest chunks when full (backpressure).""" + """Thread-safe chunk queue that drops the oldest chunks when full (backpressure).""" def __init__(self, maxsize: int = 2000) -> None: - self._q: queue.Queue[bytes | None] = queue.Queue() + self._dq: deque[bytes | None] = deque() self._max = max(8, int(maxsize or 2000)) self._cond = threading.Condition() self.dropped = 0 @@ -378,13 +393,12 @@ class _BoundedByteQueue: def put(self, item: bytes | None) -> None: with self._cond: - while self._q.qsize() >= self._max: - try: - self._q.get_nowait() + dq = self._dq + if item is not None: + while len(dq) >= self._max: + dq.popleft() self.dropped += 1 - except queue.Empty: - break - self._q.put(item) + dq.append(item) self._cond.notify() def put_nowait(self, item: bytes | None) -> None: @@ -392,22 +406,42 @@ class _BoundedByteQueue: def get_nowait(self) -> bytes | None: with self._cond: - return self._q.get_nowait() + if not self._dq: + raise queue.Empty + return self._dq.popleft() def get(self, timeout: float = 0.25) -> bytes | None: """Block until a chunk is available or timeout (raises queue.Empty).""" - deadline = time.time() + max(0.0, float(timeout)) + deadline = time.monotonic() + max(0.0, float(timeout)) with self._cond: - while self._q.empty(): - remaining = deadline - time.time() + while not self._dq: + remaining = deadline - time.monotonic() if remaining <= 0: raise queue.Empty self._cond.wait(timeout=remaining) - return self._q.get_nowait() + return self._dq.popleft() + + def drain(self, max_bytes: int = 262144) -> tuple[list[bytes], bool]: + """Pop every queued chunk (up to ~max_bytes). Returns ``(chunks, saw_eof)``.""" + out: list[bytes] = [] + total = 0 + with self._cond: + dq = self._dq + while dq: + item = dq[0] + if item is None: + dq.popleft() + return out, True + if out and total + len(item) > max_bytes: + break + dq.popleft() + out.append(item) + total += len(item) + return out, False def qsize(self) -> int: with self._cond: - return self._q.qsize() + return len(self._dq) def take_drop_delta(self) -> int: """Return newly dropped chunk count since last call (for client notice).""" @@ -1254,6 +1288,9 @@ def feed_command_line_buffer(buf: str, data: str, *, max_line: int = 512) -> tup completed: list[str] = [] limit = max(64, min(int(max_line or 512), 4096)) raw = str(data or "") + if raw.isprintable(): + # Plain typing (no Enter / BS / ESC): just append. + return (cur + raw)[:limit] if len(cur) < limit else cur, completed i = 0 while i < len(raw): ch = raw[i] diff --git a/netx_api/webcrt_cmd_audit.py b/netx_api/webcrt_cmd_audit.py new file mode 100644 index 0000000..3580ca7 --- /dev/null +++ b/netx_api/webcrt_cmd_audit.py @@ -0,0 +1,525 @@ +"""Server-side command audit for WebCRT: track the device's own input row. + +Why: inferring commands from browser keystrokes / xterm snapshots races with the +device echo (fast typing, Tab, history recall, slow VMs) and records fragments such +as ``R2#s`` or ``cddontainer/r/``. The device itself always knows what it executes: +when it echoes the line feed for our Enter, the *current input row* is exactly the +prompt + command it is about to run. + +``LineScreen`` is a tiny single-row VT100 model (CR/BS/CSI cursor + erase). +``CommandTracker`` owns one ``LineScreen`` per session and resolves each Enter we +send against the first line feed the device echoes back. Everything here is pure +(no I/O) and cheap: output before the last line feed is skipped unless an Enter +is waiting for its echo. +""" + +from __future__ import annotations + +import re +import threading +import time +from collections import deque +from dataclasses import dataclass, field + +# --------------------------------------------------------------------------- # +# Single-row terminal model +# --------------------------------------------------------------------------- # + +_TOKEN_RE = re.compile( + r"(\x1b\[[0-?]*[ -/]*[@-~]" # CSI + r"|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)" # OSC + r"|\x1b[()][0-9A-Za-z]" # charset + r"|\x1b[@-Z\\-_0-9=>]" # 2-byte ESC + r"|[\x00-\x1f\x7f])" # C0 + DEL +) +# Incomplete escape at the end of a chunk — carry into the next feed. +_PARTIAL_ESC_RE = re.compile(r"\x1b(?:\[[0-?]*[ -/]*|\][^\x07\x1b]*\x1b?|[()])?\Z") +_MAX_CARRY = 96 + + +class LineScreen: + """Current input row of a terminal (no scrolling / wrapping).""" + + __slots__ = ("cells", "cur", "_carry") + + def __init__(self) -> None: + self.cells: list[str] = [] + self.cur = 0 + self._carry = "" + + def reset_row(self) -> None: + self.cells = [] + self.cur = 0 + + def text(self) -> str: + return "".join(self.cells).rstrip() + + def _put(self, run: str) -> None: + if not run: + return + n = len(run) + cells = self.cells + cur = self.cur + if cur == len(cells): + cells.extend(run) + else: + if cur > len(cells): + cells.extend(" " * (cur - len(cells))) + cells[cur : cur + n] = run + self.cur = cur + n + + def apply_token(self, tok: str) -> None: + """Apply one control token (not LF — the tracker owns line feeds).""" + c0 = tok[0] + if c0 == "\r": + self.cur = 0 + elif c0 == "\b": + if self.cur > 0: + self.cur -= 1 + elif c0 == "\t": + self._put(" " * (8 - (self.cur % 8))) + elif c0 == "\x1b": + if len(tok) >= 3 and tok[1] == "[": + self._csi(tok[2:-1], tok[-1]) + elif tok == "\x1bc": + self.reset_row() + # every other control byte (BEL, NUL, DEL, SO/SI …) is ignored + + def _csi(self, params: str, final: str) -> None: + if final not in "DCKP@XGHfJ": + return + if params and params[0] in "?>=": + return + try: + first = params.split(";", 1)[0] + n = int(first) if first else 0 + except ValueError: + n = 0 + cells = self.cells + if final == "D": + self.cur = max(0, self.cur - (n or 1)) + elif final == "C": + self.cur += n or 1 + elif final == "K": + if n == 0: + del cells[self.cur :] + elif n == 1: + for i in range(min(self.cur + 1, len(cells))): + cells[i] = " " + else: + self.reset_row() + elif final == "P": + del cells[self.cur : self.cur + (n or 1)] + elif final == "@": + if self.cur > len(cells): + cells.extend(" " * (self.cur - len(cells))) + cells[self.cur : self.cur] = [" "] * (n or 1) + elif final == "X": + for i in range(self.cur, min(self.cur + (n or 1), len(cells))): + cells[i] = " " + elif final == "G": + self.cur = max(0, (n or 1) - 1) + elif final in "Hf": + parts = params.split(";") + try: + col = int(parts[1]) if len(parts) > 1 and parts[1] else 1 + except ValueError: + col = 1 + self.cur = max(0, col - 1) + elif final == "J": + if n in (2, 3): + self.reset_row() + elif n == 0: + del cells[self.cur :] + + def feed_row(self, text: str) -> None: + """Apply text that contains no line feed.""" + if not text: + return + if not _has_ctl(text): + self._put(text) + return + parts = _TOKEN_RE.split(text) + for i, part in enumerate(parts): + if not part: + continue + if i & 1: + self.apply_token(part) + else: + self._put(part) + + def take_carry(self, text: str) -> str: + """Prepend a carried partial escape; stash a new trailing partial.""" + if self._carry: + text = self._carry + text + self._carry = "" + if text and "\x1b" in text[-_MAX_CARRY:]: + m = _PARTIAL_ESC_RE.search(text, max(0, len(text) - _MAX_CARRY)) + if m and m.group(0): + self._carry = m.group(0) + text = text[: m.start()] + return text + + +def _has_ctl(text: str) -> bool: + # str.isprintable is a fast C scan; only fall back to token split when needed. + return not text.isprintable() + + +# --------------------------------------------------------------------------- # +# Prompt splitting +# --------------------------------------------------------------------------- # + +_PROMPT_RES: tuple[re.Pattern[str], ...] = ( + # Huawei / H3C / ZTE user & system view: [r1] [~r1] [*r1-GigabitEthernet0/0/1] + re.compile(r"^<[^<>\s]{1,63}>"), + re.compile(r"^\[[~*]?[A-Za-z0-9][\w.\-/:+]{0,79}\]"), + # Cisco / Arista / ZTE / Ruijie: R2# R2> R2(config-if)# SW-1(config)# + re.compile(r"^[A-Za-z0-9][\w.\-/:+]{0,62}(?:\([^()\s]{1,60}\))*[#>]"), + # MikroTik: [admin@MikroTik] > [admin@MikroTik] /ip> + re.compile(r"^\[[\w.\-]+@[\w.\-]+\][^\n]{0,40}?>\s?"), +) +_LINUX_PROMPT_RES: tuple[re.Pattern[str], ...] = ( + # [user@host dir]$ user@host:~/dir$ (venv) user@host:~# + re.compile(r"^(?:\([^()]{1,40}\)\s*)?\[[^\]]{1,80}\]\s?[$#%>]\s?"), + re.compile(r"^(?:\([^()]{1,40}\)\s*)?[\w.\-]+@[\w.\-]+(?::|\s)[^\s$#%>]{0,120}\s?[$#%>]\s?"), + re.compile(r"^[^\s$#%>]{1,80}[$#%]\s"), +) +_NOT_PROMPT_RE = re.compile(r"^\[[YyNn]/[YyNn]\]") +_PASSWORD_ROW_RE = re.compile(r"(?:password|passwd|passphrase|密码)\s*[:：>]\s*\**\s*$", re.IGNORECASE) +_PAGER_ROW_RE = re.compile(r"-{2,}\s*more\s*-{2,}|--\s*more\s*--|\(END\)|lines?\s+\d+-\d+", re.IGNORECASE) + + +def split_prompt(row: str, hint: str = "", *, linux_like: bool = False) -> tuple[str, str] | None: + """Split ``row`` into ``(prompt, command)``; None when no CLI prompt is recognised. + + ``prompt`` keeps the whitespace before the command, so ``prompt + command`` is the + row exactly as the device displayed it. + """ + s = row.rstrip() + if not s or _NOT_PROMPT_RE.match(s) or _PAGER_ROW_RE.search(s[:80]): + return None + h = hint.rstrip() + if h and s.startswith(h): + return _cut(s, len(h)) + pats = _LINUX_PROMPT_RES + _PROMPT_RES if linux_like else _PROMPT_RES + _LINUX_PROMPT_RES + for pat in pats: + m = pat.match(s) + if m: + return _cut(s, m.end()) + return None + + +def _cut(s: str, end: int) -> tuple[str, str]: + rest = s[end:] + cmd = rest.lstrip() + return s[: len(s) - len(cmd)], cmd.rstrip() + + +def looks_like_password_row(row: str) -> bool: + return bool(_PASSWORD_ROW_RE.search(row.rstrip())) + + +# --------------------------------------------------------------------------- # +# Command tracker +# --------------------------------------------------------------------------- # + + +@dataclass +class AuditCmd: + command: str # prompt + command as the device displayed it + prompt: str = "" + input: str = "" # command only + redacted: bool = False + source: str = "stdin" + via: str = "device_echo" + + +@dataclass +class _Pending: + t: float + typed: str + snapshot: str # input row at the time we sent Enter + prompt_hint: str + client_hints: list[str] + source: str + # "bare": Enter with nothing typed in this payload (history recall / empty Enter) — + # consumes the first line feed whatever it shows. + bare: bool = False + # Tab can make the device redraw the line after a CRLF before the Enter echo; keep the + # latest prompt+command row seen and settle once the real output starts. + has_tab: bool = False + cand: tuple[str, str] | None = None + cand_t: float = 0.0 + + +_TAB_SETTLE_SEC = 1.2 + + +@dataclass +class CommandTracker: + """Per-session: device input row + FIFO of Enters waiting for their echo.""" + + linux_like: bool = False + expire_sec: float = 3.0 + max_command: int = 512 + screen: LineScreen = field(default_factory=LineScreen) + lock: threading.Lock = field(default_factory=threading.Lock) + pending: deque[_Pending] = field(default_factory=deque) + prompt_hint: str = "" + line_started: bool = False + password_mode: bool = False + # Rows abandoned with Ctrl-C: some devices (IOS, bash) echo that row + CRLF, and that + # line feed must not be mistaken for the echo of the next Enter. + abandoned: list[tuple[str, float]] = field(default_factory=list) + _typed: str = "" + + # ----- device output -------------------------------------------------- # + def feed(self, text: str) -> list[AuditCmd]: + """Feed decoded device output; returns commands resolved by line-feed echoes.""" + out: list[AuditCmd] = [] + with self.lock: + scr = self.screen + text = scr.take_carry(text) + if not text: + return out + while text: + nl = text.find("\n") + if nl < 0: + scr.feed_row(text) + break + self.line_started = False + if self.pending: + scr.feed_row(text[:nl]) + self._resolve_one(out) + scr.reset_row() + text = text[nl + 1 :] + continue + # Nothing waiting: only the last row matters — skip the bulk. + last = text.rfind("\n") + scr.reset_row() + text = text[last + 1 :] + scr.feed_row(text) + break + self.password_mode = looks_like_password_row(scr.text()) + if self.line_started and not scr.text(): + # Row wiped by the device (screen clear / new prompt): start over. + self.line_started = False + return out + + def _resolve_one(self, out: list[AuditCmd]) -> None: + row = self.screen.text() + p = self.pending[0] + if self.abandoned and not p.bare: + now = time.monotonic() + self.abandoned = [(r, t) for r, t in self.abandoned if now - t < 5.0] + typed = p.typed.replace("\t", "").strip() + for i, (ab, _t) in enumerate(self.abandoned): + # The device reprints the aborted row; ``startswith`` covers a slow echo + # (row snapshot was taken before every typed char came back). A VRP that + # keeps the text yields ``garbage`` + our command, which contains ``typed``. + if row == ab or (row.startswith(ab) and typed and typed not in row): + del self.abandoned[i] + return # echo of the aborted line, not our Enter + parsed = split_prompt(row, p.prompt_hint, linux_like=self.linux_like) + has_cmd = bool(parsed and parsed[1]) + if p.bare: + # A bare Enter (empty line, or executing a recalled history line) is echoed by + # the first *prompt* row's line feed; output rows of a command that is still + # printing must not consume it (Enter typed ahead during output). + if parsed is None: + return + self.pending.popleft() + if has_cmd: + out.append(self._make(parsed[0], parsed[1], p.source, "device_echo")) # type: ignore[index] + # Enter pressed at a pager left an orphan bare Enter in front of the next + # typed line: that line feed belongs to both — do not record it twice. + nxt = self.pending[0] if self.pending else None + if nxt is not None and not nxt.bare: + typed = nxt.typed.replace("\t", "").replace(" ", "").lower() + if typed and parsed[1].replace(" ", "").lower().startswith(typed[:3]): + self.pending.popleft() + return + if p.has_tab: + if has_cmd: + p.cand = (parsed[0], parsed[1]) # type: ignore[index] + p.cand_t = time.monotonic() + elif p.cand is not None: + # Prompt / output row after the echo: the last candidate is the executed line. + self._finish_candidate(p, out) + return + if not has_cmd: + # Prompt-only / non-CLI row: earlier output, a prompt reprint — not our echo. + return + self.pending.popleft() + out.append(self._make(parsed[0], parsed[1], p.source, "device_echo")) # type: ignore[index] + + def _finish_candidate(self, p: _Pending, out: list[AuditCmd]) -> None: + if self.pending and self.pending[0] is p: + self.pending.popleft() + if p.cand is not None: + out.append(self._make(p.cand[0], p.cand[1], p.source, "device_echo")) + p.cand = None + + def _make(self, prompt: str, cmd: str, source: str, via: str) -> AuditCmd: + cmd = cmd[: self.max_command] + base = prompt.rstrip() + # Keep one space after user@host style prompts; network CLIs print none. + sep = " " if (len(prompt) > len(base) and (self.linux_like or "@" in base)) else "" + return AuditCmd( + command=(base + sep + cmd)[: self.max_command], + prompt=base, + input=cmd, + source=source, + via=via, + ) + + def expire(self, now: float | None = None) -> list[AuditCmd]: + """Settle Tab candidates and resolve Enters whose echo never arrived.""" + out: list[AuditCmd] = [] + if not self.pending: + return out + t = time.monotonic() if now is None else now + with self.lock: + while self.pending: + p = self.pending[0] + if p.cand is not None and (t - p.cand_t) >= _TAB_SETTLE_SEC: + self._finish_candidate(p, out) + continue + if (t - p.t) >= self.expire_sec: + self.pending.popleft() + cmd = self._fallback(p) + if cmd is not None: + out.append(cmd) + continue + break + return out + + def _fallback(self, p: _Pending) -> AuditCmd | None: + parsed = split_prompt(p.snapshot, p.prompt_hint, linux_like=self.linux_like) + if parsed and parsed[1]: + return self._make(parsed[0], parsed[1], p.source, "snapshot") + for hint in reversed(p.client_hints): + parsed = split_prompt(hint, p.prompt_hint, linux_like=self.linux_like) + if parsed and parsed[1]: + return self._make(parsed[0], parsed[1], p.source, "client_hint") + # Last resort: what the operator typed, under the prompt that was on screen. + typed = p.typed.replace("\t", "").strip() + if typed and p.prompt_hint: + return self._make(p.prompt_hint, typed, p.source, "typed") + return None + + def flush_all(self) -> list[AuditCmd]: + """Session closing: resolve every still-pending Enter with the best fallback.""" + out: list[AuditCmd] = [] + with self.lock: + while self.pending: + p = self.pending.popleft() + if p.cand is not None: + out.append(self._make(p.cand[0], p.cand[1], p.source, "device_echo")) + continue + cmd = self._fallback(p) + if cmd is not None: + out.append(cmd) + return out + + # ----- operator input -------------------------------------------------- # + def on_stdin( + self, + lines: list[str], + *, + saw_input: bool, + had_enter: bool, + client_hints: list[str] | None = None, + source: str = "stdin", + now: float | None = None, + abort: bool = False, + abort_text: str = "", + ) -> list[AuditCmd]: + """Register keystrokes already split into completed ``lines`` by the caller. + + ``saw_input``: payload had anything besides bare Enter (captures the prompt). + ``had_enter``: payload contained at least one line terminator. + Returns immediately-final records (password redaction); echo-resolved ones + come from ``feed``. + """ + out: list[AuditCmd] = [] + t = time.monotonic() if now is None else now + with self.lock: + row = self.screen.text() + if abort: + parsed_row = split_prompt(row, self.prompt_hint, linux_like=self.linux_like) + if parsed_row and parsed_row[1]: + self.abandoned.append((row, t)) + # What we sent before the Ctrl-C may not have been echoed yet. + if abort_text.strip() and self.prompt_hint: + self.abandoned.append((self.prompt_hint + abort_text.strip(), t)) + del self.abandoned[:-6] + if saw_input and not self.line_started: + # First keystroke of a line: the row is still just the prompt. Prefer the + # regex-detected prompt so typed echo never leaks into the hint. + if row and not looks_like_password_row(row): + known = split_prompt(row, "", linux_like=self.linux_like) + if known: + self.prompt_hint = known[0] + elif row[-1] in "#>$%]": + self.prompt_hint = row # exotic CLI prompt + else: + # "Username:" / "Continue? [y/n]:" … a dialog, not a CLI prompt. + self.prompt_hint = "" + self.line_started = True + if not had_enter: + return out + hints = [h for h in (client_hints or []) if h] + if self.password_mode or looks_like_password_row(row): + # Typed secret: never keep it, record a redacted marker. + n = max(1, len(lines) or 1) + out.extend(AuditCmd(command="***", redacted=True, source=source, via="password") for _ in range(n)) + self.password_mode = False + self.line_started = False + return out + # No completed text: a bare Enter, possibly executing a recalled history line + # whose redraw has not reached us yet. + queue_lines = list(lines) or [""] + for i, typed in enumerate(queue_lines): + self.pending.append( + _Pending( + t=t, + typed=typed, + snapshot=row if i == 0 else "", + prompt_hint=self.prompt_hint, + client_hints=hints[-1:] if i == len(queue_lines) - 1 else [], + source=source, + bare=not typed, + has_tab="\t" in typed, + ) + ) + self.line_started = False + # Bound runaway growth (device never echoing). + while len(self.pending) > 64: + self.pending.popleft() + return out + + def immediate(self, lines: list[str], source: str) -> list[AuditCmd]: + """Trusted non-interactive input (post-login commands): record as typed.""" + return [ + AuditCmd(command=ln[: self.max_command], input=ln[: self.max_command], source=source, via="trusted") + for ln in lines + if ln.strip() + ] + + def current_row(self) -> str: + with self.lock: + return self.screen.text() + + def seed(self, text: str) -> None: + """Prime the row from the login transcript (prompt printed before the reader).""" + self.feed(text) + + def set_row(self, text: str) -> None: + """Force the current input row (e.g. after a duplicate prompt was dropped).""" + with self.lock: + self.screen.reset_row() + self.screen._carry = "" + self.screen.feed_row(text) + self.line_started = False diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index a7e3493..54e6dd4 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import codecs import json import logging import time @@ -686,6 +687,7 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: replay_text = _decode_bytes(raw_boot, "utf-8") sess.bootstrap_replayed = True else: + sess.flush_session_log() replay_text = read_session_log_tail(session_id, max_bytes=49152) if replay_text: replay_bytes = _encode_text(replay_text, "utf-8") @@ -699,59 +701,72 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: _log.debug("webcrt bootstrap/replay send failed session=%s", session_id, exc_info=True) stop = asyncio.Event() - stdin_buf: list[str] = [] - stdin_audit_lines: list[str] = [] - stdin_flush_task: asyncio.Task[None] | None = None + loop = asyncio.get_running_loop() + io_pool = webcrt_io_executor() + wake = asyncio.Event() + # Keystrokes are written as they arrive (no timer delay); whatever piled up while a + # write was in flight is coalesced into the next one. + stdin_q: asyncio.Queue[tuple[str, str | None]] = asyncio.Queue() - async def flush_stdin() -> None: - nonlocal stdin_buf, stdin_audit_lines - if not stdin_buf: - return - data = "".join(stdin_buf) - audit_lines = list(stdin_audit_lines) - stdin_buf = [] - stdin_audit_lines = [] + async def stdin_writer() -> None: + while True: + data, audit = await stdin_q.get() + parts = [data] + audits = [audit] if audit else [] + taken = 1 + while True: + try: + d2, a2 = stdin_q.get_nowait() + except asyncio.QueueEmpty: + break + parts.append(d2) + taken += 1 + if a2: + audits.append(a2) + blob = "".join(parts) + try: + await loop.run_in_executor( + io_pool, + lambda b=blob, al=audits: sess.write_stdin( + b, + audit_lines=al or None, + audit_line=al[-1] if al else None, + ), + ) + except Exception as exc: + try: + await websocket.send_json( + {"type": "status", "state": "error", "message": f"write_failed:{exc}"} + ) + except Exception: + pass + stop.set() + wake.set() + try: + await websocket.close(code=4500) + except Exception: + pass + finally: + for _ in range(taken): + stdin_q.task_done() + + async def drain_stdin() -> None: try: - await asyncio.get_running_loop().run_in_executor( - webcrt_io_executor(), - lambda: sess.write_stdin( - data, - audit_lines=audit_lines or None, - audit_line=audit_lines[-1] if audit_lines else None, - ), - ) - except Exception as exc: - await websocket.send_json( - {"type": "status", "state": "error", "message": f"write_failed:{exc}"} - ) - stop.set() - - async def schedule_stdin_flush() -> None: - await asyncio.sleep(0.008) - await flush_stdin() + await asyncio.wait_for(stdin_q.join(), timeout=5.0) + except Exception: + pass async def pump_stdout() -> None: - loop = asyncio.get_running_loop() - pending: list[bytes] = [] - last_flush = loop.time() + """Event-driven stdout pump: the reader thread wakes us, we drain and send at once.""" + sess.set_notifier(loop, wake) + dec = None + if _normalize_encoding(sess.encoding) != "utf-8": + dec = codecs.getincrementaldecoder("gbk")(errors="replace") - def _to_browser_bytes(raw: bytes) -> bytes: - if _normalize_encoding(sess.encoding) == "utf-8": + def _to_browser(raw: bytes) -> bytes: + if dec is None: return raw - return _decode_bytes(raw, sess.encoding).encode("utf-8", errors="replace") - - async def _flush_pending() -> bool: - nonlocal pending, last_flush - if not pending: - return True - blob = _to_browser_bytes(b"".join(pending)) - pending = [] - last_flush = loop.time() - try: - await websocket.send_bytes(blob) - return True - except Exception: - return False + return dec.decode(raw).encode("utf-8", errors="replace") async def _notify_queue_drops() -> None: try: @@ -772,48 +787,48 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: except Exception: pass - while not stop.is_set(): - # Longer block is cheap now (Condition wait); cuts executor churn when idle. - chunk = await loop.run_in_executor( - webcrt_io_executor(), lambda: sess.take_stdout(attach_gen, timeout=0.2) - ) - if chunk == "stale": - break - if chunk == "empty": - if pending and (loop.time() - last_flush) >= 0.016: - if not await _flush_pending(): + wake.set() # output queued before the pump started + try: + while not stop.is_set(): + await wake.wait() + wake.clear() + while not stop.is_set(): + status, data = sess.drain_stdout(attach_gen) + if status == "stale": + return + if data: + try: + await websocket.send_bytes(_to_browser(data)) + except Exception: + stop.set() + return + await _notify_queue_drops() + if status == "eof": + await _notify_queue_drops() stop.set() + try: + await websocket.send_json( + { + "type": "status", + "state": "closed", + "message": sess.close_reason or "device_closed", + } + ) + except Exception: + pass + return + if status != "data": break - await _notify_queue_drops() - continue - if chunk is None: - await _flush_pending() - await _notify_queue_drops() - stop.set() - try: - await websocket.send_json( - { - "type": "status", - "state": "closed", - "message": sess.close_reason or "device_closed", - } - ) - except Exception: - pass - break - pending.append(chunk) - if sum(len(p) for p in pending) >= 8192 or (loop.time() - last_flush) >= 0.016: - if not await _flush_pending(): - stop.set() - break - await _notify_queue_drops() + finally: + sess.clear_notifier(wake) reader_task = asyncio.create_task(pump_stdout()) + writer_task = asyncio.create_task(stdin_writer()) if sess.needs_live_prompt: sess.needs_live_prompt = False try: - await asyncio.get_running_loop().run_in_executor( - webcrt_io_executor(), + await loop.run_in_executor( + io_pool, lambda: sess.write_stdin("\r", audit_source="prompt_sync"), ) except Exception: @@ -829,9 +844,7 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: text = _decode_bytes(bytes(msg_raw["bytes"]), sess.encoding) except Exception: continue - stdin_buf.append(text) - if stdin_flush_task is None or stdin_flush_task.done(): - stdin_flush_task = asyncio.create_task(schedule_stdin_flush()) + stdin_q.put_nowait((text, None)) continue raw = msg_raw.get("text") if raw is None: @@ -846,23 +859,16 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: if data is None: continue audit_raw = msg.get("audit_line") - if audit_raw is not None and str(audit_raw).strip(): - stdin_audit_lines.append(str(audit_raw).strip()[:512]) - stdin_buf.append(str(data)) - # Coalesce high-frequency keystrokes briefly. - if len(stdin_buf) >= 8: - if stdin_flush_task and not stdin_flush_task.done(): - stdin_flush_task.cancel() - await flush_stdin() - elif stdin_flush_task is None or stdin_flush_task.done(): - stdin_flush_task = asyncio.create_task(schedule_stdin_flush()) + audit = str(audit_raw).strip()[:512] if audit_raw is not None and str(audit_raw).strip() else None + stdin_q.put_nowait((str(data), audit)) elif mtype == "resize": cols = int(msg.get("cols") or sess.cols) rows = int(msg.get("rows") or sess.rows) - await asyncio.get_running_loop().run_in_executor(webcrt_io_executor(), sess.resize, cols, rows) + await loop.run_in_executor(io_pool, sess.resize, cols, rows) elif mtype == "break": + await drain_stdin() try: - await asyncio.get_running_loop().run_in_executor(webcrt_io_executor(), sess.send_break) + await loop.run_in_executor(io_pool, sess.send_break) except Exception as exc: await websocket.send_json( {"type": "status", "state": "error", "message": f"break_failed:{exc}"} @@ -872,7 +878,7 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: await websocket.send_json({"type": "pong"}) elif mtype == "close": stop.set() - await flush_stdin() + await drain_stdin() close_session( session_id, reason="client_close", @@ -885,17 +891,18 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: _log.exception("webcrt ws error session=%s", session_id) finally: stop.set() - if stdin_flush_task and not stdin_flush_task.done(): - stdin_flush_task.cancel() + wake.set() try: - await flush_stdin() + await drain_stdin() except Exception: pass + writer_task.cancel() reader_task.cancel() - try: - await reader_task - except Exception: - pass + for t in (writer_task, reader_task): + try: + await t + except BaseException: + pass # Keep device session so UI reconnect / remount can re-attach. if get_session(session_id) is not None: grace = float(getattr(settings, "webcrt_detach_grace_sec", 120) or 120) diff --git a/netx_api/webcrt_session_model.py b/netx_api/webcrt_session_model.py index 0879395..17b9c37 100644 --- a/netx_api/webcrt_session_model.py +++ b/netx_api/webcrt_session_model.py @@ -1,8 +1,12 @@ """WebCRT interactive session object (I/O, SFTP, close).""" from __future__ import annotations +import codecs import logging import queue +import re +import select +import socket import threading import time from dataclasses import dataclass, field @@ -16,8 +20,10 @@ from .ne_session_factory import ( extract_cli_prompt_marker, should_close_cli_hop_session, ) +from .webcrt_cmd_audit import AuditCmd, CommandTracker from .webcrt_channel import ( _BoundedByteQueue, + _normalize_encoding, _audit, _decode_bytes, _encode_text, @@ -31,6 +37,9 @@ from .webcrt_channel import ( _log = logging.getLogger("netx.webcrt") +# Operator typed an exit-style command (arms the hop prompt-return heuristic). +_HOP_EXIT_RE = re.compile(r"(?:^|[\r\n])\s*(?:quit|exit|logout|logoff|q|qu|qui|ex|exi)\s*[\r\n]", re.IGNORECASE) + @dataclass class WebcrtSession: session_id: str @@ -98,6 +107,17 @@ class WebcrtSession: _last_prompt_line: str = field(default="", repr=False) _last_audited_command: str = field(default="", repr=False) _last_audited_at: float = field(default=0.0, repr=False) + # Live audit: device-echo based command tracker (active while the reader runs). + _tracker: CommandTracker = field(default_factory=CommandTracker, repr=False) + # asyncio wake-up for the WS stdout pump: (loop, event) + "already scheduled" flag. + _notify: Any = field(default=None, repr=False) + _wake_flag: bool = field(default=False, repr=False) + _log_last_flush: float = field(default=0.0, repr=False) + _log_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + _hop_exit_until: float = field(default=0.0, repr=False) + + def __post_init__(self) -> None: + self._tracker.linux_like = not bool(self.cli_keymap) def touch(self) -> None: self.last_activity = time.time() @@ -256,39 +276,55 @@ class WebcrtSession: self._log_fh = None def append_session_log(self, text: str) -> None: - if not text or self._log_fh is None: + """Buffered transcript write; flushed by the reader every ~250ms (and on attach).""" + fh = self._log_fh + if not text or fh is None: return try: - self._log_fh.write(text) - self._log_fh.flush() - max_bytes = int(getattr(settings, "webcrt_session_log_max_bytes", 0) or 0) - if max_bytes > 0: - try: - pos = int(self._log_fh.tell()) - except Exception: # noqa: BLE001 - pos = 0 - if pos >= max_bytes: - self._log_fh.write( - f"\n# rotated at {pos} bytes (cap={max_bytes}) ts={_utc_iso()}\n" - ) - self._log_fh.flush() - self._log_fh.close() - self._log_fh = None - # Re-open fresh file (append mode continues same path; rotate by rename). - try: - path = _session_log_path(self.session_id) - rotated = path.with_suffix(path.suffix + f".{int(pos)}.old") - if path.exists(): - path.replace(rotated) - except Exception: # noqa: BLE001 - _log.debug("webcrt session log rotate rename failed", exc_info=True) - self.open_session_log() + with self._log_lock: + fh.write(text) + if time.monotonic() - self._log_last_flush >= 0.25: + self.flush_session_log() except Exception: _log.warning("webcrt session log append failed session=%s", self.session_id, exc_info=True) - def close_session_log(self) -> None: + def flush_session_log(self) -> None: fh = self._log_fh - self._log_fh = None + if fh is None: + return + try: + with self._log_lock: + fh.flush() + self._log_last_flush = time.monotonic() + max_bytes = int(getattr(settings, "webcrt_session_log_max_bytes", 0) or 0) + if max_bytes <= 0: + return + try: + pos = int(fh.tell()) + except Exception: # noqa: BLE001 + pos = 0 + if pos < max_bytes: + return + fh.write(f"\n# rotated at {pos} bytes (cap={max_bytes}) ts={_utc_iso()}\n") + fh.flush() + fh.close() + self._log_fh = None + # Re-open fresh file (rotate by rename). + try: + path = _session_log_path(self.session_id) + rotated = path.with_suffix(path.suffix + f".{int(pos)}.old") + if path.exists(): + path.replace(rotated) + except Exception: # noqa: BLE001 + _log.debug("webcrt session log rotate rename failed", exc_info=True) + self.open_session_log() + except Exception: + _log.warning("webcrt session log flush failed session=%s", self.session_id, exc_info=True) + + def close_session_log(self) -> None: + with self._log_lock: + fh = self._log_fh + self._log_fh = None if fh is None: return try: @@ -326,6 +362,57 @@ class WebcrtSession: return "stale" return chunk # bytes | None + # ---- asyncio hand-off (no polling thread per WebSocket) -------------------- # + def set_notifier(self, loop: Any, event: Any) -> None: + with self._stdout_lock: + self._notify = (loop, event) + self._wake_flag = False + + def clear_notifier(self, event: Any) -> None: + with self._stdout_lock: + n = self._notify + if n is not None and n[1] is event: + self._notify = None + + def bump_attach(self) -> int: + """New WebSocket owns stdout: invalidate the previous pump and wake it.""" + with self._stdout_lock: + self.attach_gen += 1 + gen = self.attach_gen + self._wake(force=True) + return gen + + def _wake(self, *, force: bool = False) -> None: + """Called from any thread after queue changes; schedules at most one loop wake-up.""" + n = self._notify + if n is None: + return + if self._wake_flag and not force: + return + self._wake_flag = True + try: + n[0].call_soon_threadsafe(n[1].set) + except Exception: + self._wake_flag = False + self._notify = None + + def drain_stdout(self, attach_gen: int, max_bytes: int = 262144) -> tuple[str, bytes]: + """Non-blocking take for the WS pump. + + Returns ``(status, data)``: ``stale`` (newer WS owns stdout), ``eof`` (reader + ended; ``data`` may still hold the final bytes), ``data`` or ``empty``. + """ + self._wake_flag = False + with self._stdout_lock: + if attach_gen != self.attach_gen: + return "stale", b"" + chunks, eof = self.out_queue.drain(max_bytes) + data = b"".join(chunks) + if eof: + return "eof", data + return ("data" if data else "empty"), data + + # ---- stdin ----------------------------------------------------------------- # def write_stdin( self, data: str, @@ -351,24 +438,51 @@ class WebcrtSession: text = map_network_cli_enter(text, self.conn) if not text: return - self._note_stdin_for_audit( - text, - source=audit_source, - audit_line=audit_override, - audit_lines=audit_list, - ) + if self.cli_hop_guard and (_HOP_EXIT_RE.search(text) or "\x04" in text): + self._hop_exit_until = time.monotonic() + 8.0 + live_cmds: list[AuditCmd] = [] + if self._audit_live(): + # Register Enter first (cheap), send to the device, then publish: the audit + # row is resolved from the device echo, so nothing here delays the keystroke. + hints = list(audit_list or []) + if audit_override and (not hints or hints[-1] != audit_override): + hints.append(audit_override) + live_cmds = self._audit_stdin_live(text, source=str(audit_source or "stdin"), hints=hints) + else: + self._note_stdin_for_audit( + text, + source=audit_source, + audit_line=audit_override, + audit_lines=audit_list, + ) + self._send_raw(text) + if live_cmds: + self._emit_audit_cmds(live_cmds) + self.touch() + + def _send_raw(self, text: str) -> None: with self._write_lock: # Prefer raw channel I/O for interactive typing (char echo / backspace). - channel = getattr(self.conn, "remote_conn", None) + conn = self.conn + if conn is None: + raise RuntimeError("session_closed") + channel = getattr(conn, "remote_conn", None) try: if channel is not None and hasattr(channel, "send") and callable(channel.send): payload = _encode_text(text, self.encoding) - # Paramiko may write partially when the window is full. + # Paramiko may write partially when the window is full; the reader + # keeps a short channel timeout, so a timed-out send is simply retried. view = memoryview(payload) + deadline = time.monotonic() + 30.0 while len(view): - n = int(channel.send(view) or 0) + try: + n = int(channel.send(view) or 0) + except socket.timeout: + if time.monotonic() > deadline or self.closed: + raise + continue if n <= 0: - time.sleep(0.01) + time.sleep(0.005) continue view = view[n:] self.bytes_in += len(payload) @@ -377,12 +491,79 @@ class WebcrtSession: channel.write(payload) self.bytes_in += len(payload) else: - self.conn.write_channel(text) + conn.write_channel(text) self.bytes_in += len(text) except Exception: + if self.closed or self.conn is None: + raise self.conn.write_channel(text) self.bytes_in += len(text) - self.touch() + + # ---- command audit ------------------------------------------------------- # + def _audit_live(self) -> bool: + """Device-echo audit is active only while the reader thread feeds the tracker.""" + t = self._reader + return t is not None and t.is_alive() + + def _audit_stdin_live(self, text: str, *, source: str, hints: list[str]) -> list[AuditCmd]: + if source == "prompt_sync": + return [] + had_enter = "\r" in text or "\n" in text + abort_text = "" + with self._cmd_buf_lock: + cut = text.find("\x03") + if cut >= 0: + abort_text, _ = feed_command_line_buffer(self._cmd_buf, text[:cut]) + self._cmd_buf, buf_lines = feed_command_line_buffer(self._cmd_buf, text) + if source == "post_login": + return self._tracker.immediate(buf_lines, source) if had_enter else [] + saw_input = bool(text.strip("\r\n")) + return self._tracker.on_stdin( + buf_lines, + saw_input=saw_input, + had_enter=had_enter, + client_hints=hints, + source=source, + abort="\x03" in text, + abort_text=abort_text, + ) + + def _emit_audit_cmds(self, cmds: list[AuditCmd]) -> None: + for c in cmds: + if not str(c.command).strip(): + continue + try: + _audit( + "command", + session_id=self.session_id, + ne_id=self.ne_id, + ne_name=self.ne_name, + ne_ip=self.ne_ip, + protocol=self.protocol, + owner_user_id=self.owner_user_id, + owner_username=self.owner_username, + command="***" if c.redacted else str(c.command)[:512], + redacted=bool(c.redacted), + source=str(c.source or "stdin")[:32], + via=c.via, + prompt=str(c.prompt or "")[:128], + ) + except Exception: + _log.debug("webcrt command audit failed session=%s", self.session_id, exc_info=True) + + def seed_audit_row(self, transcript: str) -> None: + """Prime the input row from the login transcript (prompt printed before the reader).""" + try: + self._tracker.set_row("") + self._tracker.seed(str(transcript or "")[-4000:]) + except Exception: + _log.debug("webcrt audit seed failed session=%s", self.session_id, exc_info=True) + + def reset_audit_row(self, row: str) -> None: + try: + self._tracker.set_row(row) + except Exception: + pass def _note_stdin_for_audit( self, @@ -461,6 +642,26 @@ class WebcrtSession: _log.debug("webcrt command audit failed session=%s", self.session_id, exc_info=True) def _note_stdout_for_audit(self, text: str) -> None: + """Feed device output to the audit tracker (live) or the legacy tail buffer.""" + chunk = str(text or "") + if not chunk: + return + if self._audit_live(): + cmds = self._tracker.feed(chunk) + self._password_mode = self._tracker.password_mode + if cmds: + self._emit_audit_cmds(cmds) + return + self._note_stdout_legacy(chunk) + + def _audit_housekeeping(self) -> None: + if self._tracker.pending: + cmds = self._tracker.expire() + if cmds: + self._emit_audit_cmds(cmds) + self.flush_session_log() + + def _note_stdout_legacy(self, text: str) -> None: """Track device prompts so the next typed line can be redacted if it is a password.""" chunk = str(text or "") if not chunk: @@ -534,50 +735,74 @@ class WebcrtSession: ) self._reader.start() + @staticmethod + def _wait_readable(channel: Any, timeout: float) -> None: + """Block until a telnet-style socket is readable (or timeout).""" + sock = getattr(channel, "sock", None) + try: + if sock is not None and sock.fileno() >= 0: + select.select([sock], [], [], timeout) + return + except Exception: + pass + time.sleep(min(timeout, 0.01)) + def _reader_loop(self) -> None: conn = self.conn if conn is None: self.out_queue.put(None) + self._wake() return channel = getattr(conn, "remote_conn", None) hop_return = False poll = max(0.002, float(getattr(settings, "webcrt_reader_poll_sec", 0.01) or 0.01)) + # Incremental decode: a multi-byte char split across recv() chunks stays intact. + dec = codecs.getincrementaldecoder("gbk" if _normalize_encoding(self.encoding) == "gbk" else "utf-8")( + errors="replace" + ) + # Paramiko: park in recv() with a short timeout (wakes the moment data arrives) + # instead of spin-polling recv_ready(). + blocking = False + if ( + channel is not None + and hasattr(channel, "recv") + and hasattr(channel, "recv_ready") + and hasattr(channel, "settimeout") + ): + try: + channel.settimeout(0.2) + blocking = True + except Exception: + blocking = False + last_house = time.monotonic() try: while not self.closed: chunk = b"" try: - if channel is not None and hasattr(channel, "recv_ready") and hasattr(channel, "recv"): - # Paramiko SSH: prefer short blocking recv over fixed spin-sleep. - ready = False + if blocking: try: - ready = bool(channel.recv_ready()) - except Exception: - ready = False - if ready: - chunk = channel.recv(16384) + chunk = channel.recv(65536) + except socket.timeout: + chunk = b"" + else: + if not chunk: + break # EOF + if not chunk: + if channel.recv_ready(): + continue + if getattr(channel, "closed", False) or ( + hasattr(channel, "exit_status_ready") and channel.exit_status_ready() + ): + break + elif channel is not None and hasattr(channel, "recv_ready") and hasattr(channel, "recv"): + if channel.recv_ready(): + chunk = channel.recv(65536) if not chunk: break elif hasattr(channel, "exit_status_ready") and channel.exit_status_ready(): break else: - # Brief block: settimeout + recv wakes sooner than sleep(0.04). - prev_timeout = None - try: - prev_timeout = channel.gettimeout() - except Exception: - prev_timeout = None - try: - channel.settimeout(poll) - chunk = channel.recv(16384) - except Exception: - chunk = b"" - finally: - try: - channel.settimeout(prev_timeout) - except Exception: - pass - if not chunk: - continue + time.sleep(poll) elif channel is not None and hasattr(channel, "read_very_eager"): # Telnet: do NOT use conn.read_channel() — Netmiko strips ANSI. data = channel.read_very_eager() @@ -588,39 +813,52 @@ class WebcrtSession: else _encode_text(str(data), self.encoding) ) else: - time.sleep(poll) - continue + self._wait_readable(channel, 0.2) else: text = conn.read_channel() if text: chunk = _encode_text(str(text), self.encoding) else: time.sleep(poll) - continue + except EOFError: + break except Exception as exc: if self.closed: break _log.debug("webcrt reader error session=%s: %s", self.session_id, exc) + if channel is not None and getattr(channel, "closed", False): + break time.sleep(0.05) continue + now = time.monotonic() if chunk: - self.touch() + self.last_activity = time.time() self.bytes_out += len(chunk) self.out_queue.put(chunk) + self._wake() + decoded = "" try: - decoded = _decode_bytes(chunk, self.encoding) - self.append_session_log(decoded) - self._note_stdout_for_audit(decoded) + decoded = dec.decode(bytes(chunk)) + if decoded: + self.append_session_log(decoded) + self._note_stdout_for_audit(decoded) except Exception: - pass - if self.cli_hop_guard and self._note_cli_hop_output(chunk): + _log.debug("webcrt reader side-effects failed session=%s", self.session_id, exc_info=True) + if self.cli_hop_guard and self._note_cli_hop_output(chunk, decoded): hop_return = True notice = ( "\r\n*** WebCRT: 目标会话已结束，已断开代理连接 " "(target session ended; closing hop proxy) ***\r\n" ) self.out_queue.put(notice.encode("utf-8", errors="replace")) + self._wake() break + if now - last_house >= 0.25: + last_house = now + try: + self._audit_housekeeping() + except Exception: + _log.debug("webcrt housekeeping failed session=%s", self.session_id, exc_info=True) finally: if hop_return and not self.closed: try: @@ -630,22 +868,33 @@ class WebcrtSession: except Exception: self.close("cli_hop_return") self.out_queue.put(None) + self._wake() + try: + self.flush_session_log() + except Exception: + pass - def _note_cli_hop_output(self, chunk: bytes) -> bool: + def _note_cli_hop_output(self, chunk: bytes, text: str | None = None) -> bool: """Accumulate stdout and return True when nested CLI hop has returned to proxy.""" - try: - text = _decode_bytes(chunk, self.encoding) - except Exception: - text = str(chunk) + if text is None: + try: + text = _decode_bytes(chunk, self.encoding) + except Exception: + text = str(chunk) self._hop_scan_buf = (self._hop_scan_buf + text)[-12000:] marker = str(self.cli_hop_prompt or "").strip() last = extract_cli_prompt_marker(self._hop_scan_buf) if last and (not marker or last != marker): self._cli_hop_seen_other_prompt = True + # Prompt-only "back on the hop" is ambiguous when hop and target share a hostname + # (target ``system-view`` prints the same ``[~r1]`` as the hop). Only trust it + # shortly after the operator typed an exit command; the nested "connection closed" + # message needs no such gate. + armed = self._cli_hop_seen_other_prompt and time.monotonic() <= self._hop_exit_until return should_close_cli_hop_session( self._hop_scan_buf, self.cli_hop_prompt, - seen_other_prompt=self._cli_hop_seen_other_prompt, + seen_other_prompt=armed, ) def run_post_login_commands(self) -> None: @@ -660,6 +909,19 @@ class WebcrtSession: _log.debug("post_login command failed session=%s", self.session_id, exc_info=True) break + def _teardown_conn(self, conn: Any) -> None: + self.close_sftp() + if conn is None: + return + for attr in ("remote_conn", "remote_conn_pre", "_netx_jump_client"): + obj = getattr(conn, attr, None) + if obj is None: + continue + try: + obj.close() + except Exception: + pass + def close(self, reason: str = "closed") -> None: if self.closed: return @@ -667,14 +929,27 @@ class WebcrtSession: self.state = "closed" self.close_reason = reason or "closed" self._ready_event.set() - self.close_sftp() + # Enters still waiting for a device echo would be lost with the session. try: - close_netmiko_connection(self.conn) + pending_cmds = self._tracker.flush_all() + if pending_cmds: + self._emit_audit_cmds(pending_cmds) except Exception: - pass + _log.debug("webcrt audit flush on close failed session=%s", self.session_id, exc_info=True) + conn = self.conn self.conn = None + # Tear the transport down off-thread: Netmiko's disconnect() runs vendor cleanup + # (prompt probing, several seconds on Huawei) which would stall the event loop / + # reaper, and keeps the device VTY occupied longer than needed. + threading.Thread( + target=self._teardown_conn, + args=(conn,), + name=f"webcrt-close-{self.session_id[:8]}", + daemon=True, + ).start() try: self.out_queue.put_nowait(None) except Exception: pass + self._wake(force=True) self.close_session_log() diff --git a/netx_api/webcrt_session_registry.py b/netx_api/webcrt_session_registry.py index 2bdf91b..9794649 100644 --- a/netx_api/webcrt_session_registry.py +++ b/netx_api/webcrt_session_registry.py @@ -34,6 +34,7 @@ from .webcrt_channel import ( _normalize_encoding, _prime_interactive_channel, _session_log_text, + looks_like_network_prompt, prepare_bootstrap_output, uses_network_cli_keymap, ) @@ -284,14 +285,16 @@ def _finish_connect( # Pull post-auth banner/MOTD from the PTY. With interactive no-op session_preparation # (generic_termserver), Netmiko session_log is often empty — do not discard these bytes. # Keep this short: live echo already streamed login; long settle blocks WS stdin. + # Login already ended on a CLI prompt: only a short look for trailing MOTD bytes. + prompted_early = _looks_like_cli_prompt(pre_log) try: - early = _capture_raw_channel(conn, duration=0.2) + early = _capture_raw_channel(conn, duration=0.06 if prompted_early else 0.2) except Exception: early = "" if early: sess.push_connect_echo(early) seed = f"{pre_log}{early}" - already_prompted = _looks_like_cli_prompt(seed) or _looks_like_cli_prompt(pre_log) + already_prompted = _looks_like_cli_prompt(seed) or prompted_early saw_password_change = bool( re.search( r"(?i)(change\s*now|please\s*choose|password\s+needs\s+to\s+be\s+changed).{0,80}:", @@ -356,7 +359,7 @@ def _finish_connect( primed = "" combined = f"{seed}{primed}" # Brief settle only — do not burn seconds here while the terminal shows a prompt. - settle_sec = 0.12 if already_prompted or _looks_like_cli_prompt(combined) else 0.3 + settle_sec = 0.05 if already_prompted or _looks_like_cli_prompt(combined) else 0.3 try: more = _capture_raw_channel(conn, duration=settle_sec) if more: @@ -395,7 +398,7 @@ def _finish_connect( if not echoed.strip(): bootstrap = prepare_bootstrap_output(combined) try: - leftover = _capture_raw_channel(conn, duration=0.1) + leftover = _capture_raw_channel(conn, duration=0.05) except Exception: leftover = "" if ( @@ -407,7 +410,7 @@ def _finish_connect( bootstrap = prepare_bootstrap_output(f"{bootstrap}{leftover}") else: try: - leftover = _capture_raw_channel(conn, duration=0.08) + leftover = _capture_raw_channel(conn, duration=0.04) except Exception: leftover = "" if ( @@ -421,6 +424,19 @@ def _finish_connect( sess.conn = conn sess.cli_hop_guard = bool(hop_guard) sess.cli_hop_prompt = str((hop_guard or {}).get("hop_prompt") or "") + # Inventory type "generic" (e.g. a quick-connect to a Cisco box) disables the network + # key map, so Home/Delete reach IOS/VRP as raw escape sequences. The login prompt tells + # us what it really is. + if not sess.cli_keymap and str(sess.device_type or "").strip().lower() in ( + "", + "generic", + "generic_termserver", + "generic_telnet", + ): + last_line = str(echoed or combined or "").replace("\r\n", "\n").replace("\r", "\n").rstrip(" \n").split("\n")[-1] + if looks_like_network_prompt(prompt_hint or last_line) or looks_like_network_prompt(last_line): + sess.cli_keymap = True + sess._tracker.linux_like = False sess.bootstrap_output = _encode_text(str(bootstrap or ""), sess.encoding) # Nudge Enter on WS attach only when we still need a shell prompt. # Never when already at CLI prompt or Username:/Password: (would empty-submit login). @@ -432,6 +448,8 @@ def _finish_connect( and not _looks_like_login_prompt(final_view) ) sess.open_session_log() + # Prime the audit tracker with the prompt printed during login (reader starts later). + sess.seed_audit_row(echoed or bootstrap or combined) log_seed = echoed or bootstrap if log_seed: sess.append_session_log(log_seed if str(log_seed).endswith("\n") else str(log_seed) + "\n") @@ -439,7 +457,7 @@ def _finish_connect( # Drop late prompt echoes that race into the queue right after reader start. if not prompt_hint and final_view: prompt_hint = str(final_view).replace("\r\n", "\n").replace("\r", "\n").strip().split("\n")[-1].strip() - settle_deadline = time.time() + (0.2 if already_prompted or saw_password_change else 0.35) + settle_deadline = time.time() + (0.08 if already_prompted or saw_password_change else 0.35) while time.time() < settle_deadline: try: chunk = sess.out_queue.get_nowait() @@ -456,6 +474,8 @@ def _finish_connect( # A slow VM may emit its FIRST prompt after the reader starts. # Only drop duplicates when a prompt was already delivered during login. if _looks_like_cli_prompt(final_view) and _is_prompt_only_echo(text, prompt_hint): + # Dropped from the browser stream: keep the audit row in step with it. + sess.reset_audit_row(prompt_hint) continue if _looks_like_cli_prompt(text): sess.needs_live_prompt = False @@ -696,8 +716,7 @@ def mark_attached(session_id: str) -> tuple[WebcrtSession, int]: 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 + attach_gen = sess.bump_attach() sess.attached = True sess.detach_deadline = None sess.touch() diff --git a/scripts/bench_webcrt.py b/scripts/bench_webcrt.py new file mode 100644 index 0000000..06f7c32 --- /dev/null +++ b/scripts/bench_webcrt.py @@ -0,0 +1,239 @@ +#!/usr/bin/env python3 +"""WebCRT latency / throughput / audit benchmark against a live netx + lab device. + + python scripts/bench_webcrt.py --ne-id [--prompt-re "|\\[~?r1\\]"] [--audit] + +Measures (client-side, real WebSocket): + * connect_ms create session -> first prompt visible + * echo latency single keystroke -> echoed byte (p50/p95/max) + * burst echo latency 20 keys sent 3ms apart -> last echo byte + * command RTT Enter -> next prompt for a short command + * bulk throughput long output command, bytes / seconds +With --audit it also types a set of tricky command lines (backspace, tab, history +recall, bare Enter) and prints what landed in audit.jsonl for the session. +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import re +import statistics +import sys +import time + +import httpx +import websockets + +BASE = os.environ.get("NETX_BASE", "http://127.0.0.1:8890").rstrip("/") +USER = os.environ.get("NETX_USER", "admin") +PWD = os.environ.get("NETX_PASSWORD", "Admin@1234") + + +def pct(vals: list[float], p: float) -> float: + if not vals: + return float("nan") + s = sorted(vals) + return s[min(len(s) - 1, int(len(s) * p))] + + +async def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--ne-id", default="") + ap.add_argument("--quick", default="", help="ip:port:user:pass -> quick-connect instead of --ne-id") + ap.add_argument("--prompt-re", default=r"(?:<[^>\r\n]+>|\[[^\]\r\n]+\]|[\w.\-]+(?:\([^)]+\))*[#>])\s*$") + ap.add_argument("--bulk-cmd", default="display current-configuration") + ap.add_argument("--probe-cmd", default="display clock") + ap.add_argument("--audit", action="store_true") + ap.add_argument("--echo-only", action="store_true", help="only connect + echo latency") + ap.add_argument("--audit-cmds", default="") + ap.add_argument("--audit-file", default=os.path.join("data", "webcrt", "audit.jsonl")) + args = ap.parse_args() + prompt_re = re.compile(args.prompt_re) + + c = httpx.Client(base_url=BASE, timeout=60) + tok = c.post("/v1/auth/login", json={"username": USER, "password": PWD}).json()["access_token"] + h = {"Authorization": f"Bearer {tok}"} + t_create = time.perf_counter() + if args.quick: + ip, port, user, pw = args.quick.split(":", 3) + r = c.post("/v1/webcrt/sessions/quick-connect", headers=h, json={ + "name": "bench", "ip_address": ip, "port": int(port), "protocol": "ssh", "username": user, + "password": pw, "save_password": False, "cols": 160, "rows": 50}) + else: + r = c.post("/v1/webcrt/sessions", headers=h, json={"ne_id": args.ne_id, "cols": 160, "rows": 50}) + r.raise_for_status() + sid = r.json()["session_id"] + ticket = c.post("/v1/webcrt/ws-ticket", headers=h).json()["ticket"] + url = f"{BASE.replace('http', 'ws', 1)}/v1/webcrt/sessions/{sid}/ws?ws_ticket={ticket}" + print(f"session={sid}") + + async with websockets.connect(url, max_size=64 * 1024 * 1024) as ws: + buf = "" + ev = asyncio.Event() + last_rx = 0.0 + rx_bytes = 0 + + async def reader() -> None: + nonlocal buf, last_rx, rx_bytes + dec = __import__("codecs").getincrementaldecoder("utf-8")(errors="replace") + async for raw in ws: + if isinstance(raw, bytes): + buf += dec.decode(raw) + rx_bytes += len(raw) + last_rx = time.perf_counter() + ev.set() + else: + m = json.loads(raw) + if m.get("type") == "status" and m.get("state") in ("error", "closed", "warning"): + print("STATUS", m) + if m.get("type") == "stdout": + buf += m.get("data", "") + last_rx = time.perf_counter() + ev.set() + + rt = asyncio.create_task(reader()) + + def at_prompt() -> bool: + tail = re.sub(r"\x1b\[[0-9;?]*[A-Za-z]", "", buf).replace("\r", "\n").rstrip(" ").split("\n")[-1] + return bool(prompt_re.search(tail)) and len(tail) < 80 + + async def wait_prompt(timeout: float = 60.0, quiet: float = 0.05) -> bool: + end = time.perf_counter() + timeout + while time.perf_counter() < end: + if at_prompt() and time.perf_counter() - last_rx >= quiet: + return True + await asyncio.sleep(0.005) + return False + + async def send(data: str, audit_line: str | None = None) -> None: + p = {"type": "stdin", "data": data} + if audit_line: + p["audit_line"] = audit_line + await ws.send(json.dumps(p)) + + ok = await wait_prompt(90, 0.3) + connect_ms = (time.perf_counter() - t_create) * 1000 + if not ok: + await send("\r") + ok = await wait_prompt(15, 0.3) + print(f"connect_ms={connect_ms:.0f} prompt={'yes' if ok else 'NO'} tail={buf[-80:]!r}") + if not ok: + return 2 + + # --- single-key echo latency --- + lat: list[float] = [] + for ch in "abcdefghijklmnopqrst": + n0 = rx_bytes + t0 = time.perf_counter() + await send(ch) + while rx_bytes == n0 and time.perf_counter() - t0 < 2: + await asyncio.sleep(0.0005) + lat.append((time.perf_counter() - t0) * 1000) + await asyncio.sleep(0.05) + await send("\x08" * 20) + await asyncio.sleep(0.4) + print(f"echo_ms p50={statistics.median(lat):.1f} p95={pct(lat, .95):.1f} max={max(lat):.1f}") + + if args.echo_only: + await ws.send(json.dumps({"type": "close"})) + await asyncio.sleep(0.3) + rt.cancel() + return 0 + + # --- burst typing: last echo arrival --- + await wait_prompt(5, 0.1) + n0 = len(buf) + t0 = time.perf_counter() + for ch in "abcdefghijklmnopqrst": + await send(ch) + await asyncio.sleep(0.003) + t_sent = time.perf_counter() + while time.perf_counter() - t_sent < 3 and not buf.endswith("abcdefghijklmnopqrst"): + await asyncio.sleep(0.002) + print(f"burst_echo_ms={(time.perf_counter() - t_sent) * 1000:.1f} (after last key sent)") + await send("\x08" * 20) + await asyncio.sleep(0.4) + + # --- command RTT --- + rtts: list[float] = [] + for _ in range(5): + await wait_prompt(10, 0.05) + n0 = len(buf) + t0 = time.perf_counter() + await send(args.probe_cmd + "\r") + await asyncio.sleep(0.02) + await wait_prompt(15, 0.0) + # prompt re-appeared after output + while len(buf) == n0 and time.perf_counter() - t0 < 10: + await asyncio.sleep(0.005) + rtts.append((time.perf_counter() - t0) * 1000) + print(f"cmd_rtt_ms p50={statistics.median(rtts):.0f} max={max(rtts):.0f} ({args.probe_cmd})") + + # --- bulk --- + await wait_prompt(10, 0.2) + rx0 = rx_bytes + t0 = time.perf_counter() + await send(args.bulk_cmd + "\r") + await asyncio.sleep(0.2) + # page through "More" prompts if any + deadline = time.perf_counter() + 90 + while time.perf_counter() < deadline: + tail = buf[-200:] + if "More" in tail[-40:]: + await send(" ") + await asyncio.sleep(0.01) + continue + if at_prompt() and time.perf_counter() - last_rx > 0.4 and rx_bytes - rx0 > 100: + break + await asyncio.sleep(0.01) + dt = time.perf_counter() - t0 - 0.4 + print(f"bulk bytes={rx_bytes - rx0} secs={dt:.2f} KB/s={(rx_bytes - rx0) / 1024 / max(dt, 1e-3):.0f}") + + if args.audit: + await wait_prompt(10, 0.2) + cmds = [__import__("codecs").decode(x, "unicode_escape") for x in args.audit_cmds.split("|") if x] or [ + "display clock\r", + "disX\x08play ver\t\r", + "dis ip int br\t\r", + "\r", + "display clock\r", + "\x1b[A\r", + "\x1b[A\x1b[A\x1b[B\r", + "system-view\r", + "sysname r1\r", + "quit\r", + ] + for cmd in cmds: + await send(cmd) + await asyncio.sleep(0.2) + await wait_prompt(10, 0.25) + if "More" in buf[-40:]: + await send(" ") + await asyncio.sleep(0.5) + + await ws.send(json.dumps({"type": "close"})) + await asyncio.sleep(0.3) + rt.cancel() + + time.sleep(0.8) + if args.audit: + print("--- audit.jsonl commands for session ---") + try: + with open(args.audit_file, encoding="utf-8") as fh: + for line in fh: + try: + o = json.loads(line) + except Exception: + continue + if o.get("event") == "command" and o.get("session_id") == sid: + print(repr(o.get("command")), o.get("source")) + except OSError as exc: + print("cannot read audit:", exc) + return 0 + + +if __name__ == "__main__": + raise SystemExit(asyncio.run(main())) diff --git a/scripts/e2e_webcrt_input_scenarios.py b/scripts/e2e_webcrt_input_scenarios.py new file mode 100644 index 0000000..157e198 --- /dev/null +++ b/scripts/e2e_webcrt_input_scenarios.py @@ -0,0 +1,189 @@ +#!/usr/bin/env python3 +"""Keystroke-level WebCRT audit scenarios against a live device. + +Every key is sent as its own stdin frame (~40ms apart, like a person typing) and the +command that landed in audit.jsonl for the session is compared with what the device +actually executed. + + python scripts/e2e_webcrt_input_scenarios.py --ne-id --vendor huawei + python scripts/e2e_webcrt_input_scenarios.py --ne-id --vendor cisco +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import os +import re +import sys +import time + +import httpx +import websockets + +BASE = os.environ.get("NETX_BASE", "http://127.0.0.1:8890").rstrip("/") +USER = os.environ.get("NETX_USER", "admin") +PWD = os.environ.get("NETX_PASSWORD", "Admin@1234") + +LEFT, RIGHT, UP, DOWN = "\x1b[D", "\x1b[C", "\x1b[A", "\x1b[B" +HOME, END, DEL, BS, TAB, CR, CTRL_C = "\x1b[H", "\x1b[F", "\x1b[3~", "\x08", "\t", "\r", "\x03" + + +def keys(*parts: str | list[str]) -> list[str]: + """Flatten: a plain str is typed char by char, escape sequences stay whole.""" + out: list[str] = [] + for p in parts: + if isinstance(p, list): + out.extend(p) + elif p.startswith("\x1b") or p in (BS, TAB, CR, CTRL_C): + out.append(p) + else: + out.extend(list(p)) + return out + + +def scenarios(vendor: str) -> list[tuple[str, list[str], "str | list[str]"]]: + """(name, keystrokes, expected command text after the prompt).""" + if vendor == "cisco": + full, partial, tabbed, other = "show clock", "show cl", "sh clo", "show version" + wrong = "show clxock" + else: + full, partial, tabbed, other = "display clock", "display cl", "display clo", "display version" + wrong = "display clxock" + word = full.split()[-1] # clock + return [ + ("plain", keys(full, CR), full), + ("backspace fix", keys(full[:-3] + "x", BS, full[-3:], CR), full), + ("tab completion", keys(tabbed, TAB, CR), "sh clock" if vendor == "cisco" else full), + ("left-arrow insert mid-line", keys(full.replace("clock", "cock"), [LEFT] * 3, "l", CR), full), + ("left-arrow + backspace delete", keys(wrong, [LEFT] * 3, BS, CR), full), + ("left-arrow + Delete key", keys(wrong, [LEFT] * 4, DEL, CR), full), + ("left-arrow replace char", keys(full.replace("clock", "clxck"), [LEFT] * 2, BS, "o", CR), full), + ("home + insert first char", keys(full[1:], HOME, full[0], END, CR), full), + ("right-arrow after left", keys(full, LEFT, LEFT, RIGHT, RIGHT, CR), full), + ("up-arrow recall", keys(UP, CR), full), + ("up-arrow recall then edit tail", keys(UP, [BS] * len(word), other.split()[-1], CR), other), + ("up up down recall", keys(UP, UP, DOWN, CR), other), + ("recall + left-arrow delete+insert", keys(UP, [LEFT] * 2, BS, "c", CR), other[:-3] + "con"), + ("ctrl-c (device decides if line clears)", keys("garbage", CTRL_C, full, CR), ("garbage" + full) if vendor == "huawei" else full), + ("empty enter x2 (not audited)", keys(CR, CR), ""), + ("run then recall and run again", keys(full, CR, UP, CR), [full, full]), + ] + + +async def run(args: argparse.Namespace) -> int: + c = httpx.Client(base_url=BASE, timeout=60) + tok = c.post("/v1/auth/login", json={"username": USER, "password": PWD}).json()["access_token"] + h = {"Authorization": f"Bearer {tok}"} + sid = c.post("/v1/webcrt/sessions", headers=h, json={"ne_id": args.ne_id, "cols": 160, "rows": 50}).json()["session_id"] + ticket = c.post("/v1/webcrt/ws-ticket", headers=h).json()["ticket"] + url = f"{BASE.replace('http', 'ws', 1)}/v1/webcrt/sessions/{sid}/ws?ws_ticket={ticket}" + prompt_re = re.compile(r"(?:<[^>\r\n]+>|\[[^\]\r\n]+\]|[\w.\-]+(?:\([^)]+\))*[#>])\s*$") + results: list[tuple[str, str | None, list[str]]] = [] + + async with websockets.connect(url, max_size=None) as ws: + buf = "" + last_rx = 0.0 + + async def reader() -> None: + nonlocal buf, last_rx + try: + async for raw in ws: + if isinstance(raw, bytes): + buf += raw.decode("utf-8", "replace") + last_rx = time.perf_counter() + except Exception: + pass + + rt = asyncio.create_task(reader()) + + def at_prompt() -> bool: + plain = re.sub(r"\x1b\[[0-9;?]*[A-Za-z]", "", buf).replace("\r", "\n") + tail = plain.rstrip(" ").split("\n")[-1] + return bool(prompt_re.search(tail)) and len(tail) < 80 + + async def settle(timeout: float = 20.0, quiet: float = 0.6) -> None: + end = time.perf_counter() + timeout + while time.perf_counter() < end: + if at_prompt() and time.perf_counter() - last_rx >= quiet: + return + await asyncio.sleep(0.02) + + async def send(data: str) -> None: + await ws.send(json.dumps({"type": "stdin", "data": data})) + + await settle(120, 1.0) + if not at_prompt(): + await send("\r") + await settle(15, 1.0) + sel = [x.strip() for x in args.only.split(",") if x.strip()] + for name, ks, expected in scenarios(args.vendor): + if sel and not name.startswith(tuple(sel)): + continue + before = len(read_cmds(sid, args.audit_file)) + for k in ks: + await send(k) + await asyncio.sleep(args.key_gap) + # Wait for the device (slow VMs echo seconds late), not a fixed delay: first for + # output after the last key, then until the expected audit rows exist. + sent_at = time.perf_counter() + while last_rx < sent_at and time.perf_counter() - sent_at < 15: + await asyncio.sleep(0.02) + await settle() + want = len(expected) if isinstance(expected, list) else (0 if expected == "" else 1) + waited = time.perf_counter() + while time.perf_counter() - waited < (2.0 if want == 0 else 15.0): + if len(read_cmds(sid, args.audit_file)) - before >= want: + break + await asyncio.sleep(0.1) + await asyncio.sleep(1.5) # Tab lines settle ~1.2s; also catches stray extra rows + got = read_cmds(sid, args.audit_file)[before:] + results.append((name, expected, got)) + await ws.send(json.dumps({"type": "close"})) + await asyncio.sleep(0.5) + rt.cancel() + + bad = 0 + print(f"{'scenario':38} {'result':6} recorded") + for name, expected, got in results: + tails = [re.sub(r"^(?:<[^>]*>|\[[^\]]*\]|[\w.\-]+(?:\([^)]*\))*[#>])\s*", "", g) for g in got] + if isinstance(expected, list): + ok = tails == expected + elif expected == "": + ok = not got + else: + ok = tails == [expected] + bad += 0 if ok else 1 + print(f"{name:38} {'OK' if ok else 'FAIL':6} {got!r}" + ("" if ok else f" expected {expected!r}")) + print(f"\n{len(results) - bad}/{len(results)} scenarios correct") + return 1 if bad else 0 + + +def read_cmds(sid: str, path: str) -> list[str]: + out: list[str] = [] + try: + with open(path, encoding="utf-8") as fh: + for line in fh: + if '"event": "command"' not in line: + continue + o = json.loads(line) + if o.get("session_id") == sid: + out.append(str(o.get("command") or "")) + except OSError: + pass + return out + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--ne-id", required=True) + ap.add_argument("--vendor", choices=["huawei", "cisco"], required=True) + ap.add_argument("--only", default="", help="comma-separated scenario name prefixes") + ap.add_argument("--key-gap", type=float, default=0.04, help="seconds between keystrokes") + ap.add_argument("--audit-file", default=os.path.join("data", "webcrt", "audit.jsonl")) + return asyncio.run(run(ap.parse_args())) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/fake_ssh_device.py b/scripts/fake_ssh_device.py new file mode 100644 index 0000000..6b1cdbc --- /dev/null +++ b/scripts/fake_ssh_device.py @@ -0,0 +1,167 @@ +#!/usr/bin/env python3 +"""Local fake network-device SSH server for WebCRT pipeline benchmarks. + +Zero device latency, so a benchmark against it measures netx overhead only. + + python scripts/fake_ssh_device.py [--port 2222] [--user lab] [--password lab] + +Behaviour: Cisco-like ``test#`` prompt, server-side echo with Tab completion +(``sh`` -> ``show ``), history on Up arrow, ``show clock`` (short output), +``bulk `` (large output) and ``quit``. +""" + +from __future__ import annotations + +import argparse +import socket +import threading +import time + +import paramiko + +HOST_KEY = paramiko.RSAKey.generate(2048) +PROMPT = b"test#" + + +class _Server(paramiko.ServerInterface): + def __init__(self, user: str, password: str) -> None: + self.user, self.password = user, password + self.event = threading.Event() + + def check_auth_password(self, username, password): # noqa: ANN001 + ok = username == self.user and password == self.password + return paramiko.AUTH_SUCCESSFUL if ok else paramiko.AUTH_FAILED + + def get_allowed_auths(self, username): # noqa: ANN001 + return "password" + + def check_channel_request(self, kind, chanid): # noqa: ANN001 + return paramiko.OPEN_SUCCEEDED if kind == "session" else paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED + + def check_channel_pty_request(self, *a): # noqa: ANN002 + return True + + def check_channel_shell_request(self, channel): # noqa: ANN001 + self.event.set() + return True + + def check_channel_window_change_request(self, *a): # noqa: ANN002 + return True + + +def _shell(chan: paramiko.Channel) -> None: + chan.send(b"\r\nFake device banner\r\n" + PROMPT) + line = b"" + history: list[bytes] = [] + hidx = 0 + try: + while True: + data = chan.recv(1024) + if not data: + return + i = 0 + while i < len(data): + b = data[i : i + 1] + if data[i : i + 3] == b"\x1b[A": # Up + if history: + hidx = max(0, hidx - 1) if hidx else len(history) - 1 + chan.send(b"\x08" * len(line) + b" " * len(line) + b"\x08" * len(line)) + line = history[hidx] + chan.send(line) + i += 3 + continue + if b == b"\t": + if line.startswith(b"sh") and b" " not in line: + chan.send(b"ow "[len(line) - 2 :] if len(line) <= 4 else b"") + line = b"show " + i += 1 + continue + if b in (b"\r", b"\n"): + chan.send(b"\r\n") + cmd = line.strip() + line = b"" + if cmd: + history.append(cmd) + hidx = len(history) + if cmd == b"quit": + chan.close() + return + if cmd == b"show clock": + chan.send(b"2026-10-12 12:00:00\r\nMonday\r\n") + elif cmd.startswith(b"bulk"): + mb = float(cmd.split()[1]) if len(cmd.split()) > 1 else 1.0 + row = b"Ethernet1/0/%d unassigned up down --\r\n" + sent = 0 + n = 0 + buf = [] + while sent < mb * 1024 * 1024: + r = row % (n % 999) + buf.append(r) + sent += len(r) + n += 1 + if len(buf) >= 400: + chan.sendall(b"".join(buf)) + buf = [] + if buf: + chan.sendall(b"".join(buf)) + elif cmd: + chan.send(b"%% Unknown command\r\n") + chan.send(PROMPT) + i += 1 + continue + if b in (b"\x08", b"\x7f"): + if line: + line = line[:-1] + chan.send(b"\x08 \x08") + i += 1 + continue + line += b + chan.send(b) + i += 1 + except Exception: # noqa: BLE001 + pass + finally: + try: + chan.close() + except Exception: # noqa: BLE001 + pass + + +def _serve(sock: socket.socket, user: str, password: str) -> None: + t = paramiko.Transport(sock) + t.add_server_key(HOST_KEY) + srv = _Server(user, password) + try: + t.start_server(server=srv) + chan = t.accept(30) + if chan is None: + return + srv.event.wait(10) + _shell(chan) + except Exception: # noqa: BLE001 + pass + finally: + try: + t.close() + except Exception: # noqa: BLE001 + pass + + +def main() -> None: + ap = argparse.ArgumentParser() + ap.add_argument("--port", type=int, default=2222) + ap.add_argument("--user", default="lab") + ap.add_argument("--password", default="lab") + a = ap.parse_args() + s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + s.bind(("127.0.0.1", a.port)) + s.listen(20) + print(f"fake device on 127.0.0.1:{a.port} user={a.user}", flush=True) + while True: + c, _ = s.accept() + threading.Thread(target=_serve, args=(c, a.user, a.password), daemon=True).start() + + +if __name__ == "__main__": + main() diff --git a/tests/test_webcrt_cmd_audit.py b/tests/test_webcrt_cmd_audit.py new file mode 100644 index 0000000..2e2a4e0 --- /dev/null +++ b/tests/test_webcrt_cmd_audit.py @@ -0,0 +1,295 @@ +"""Device-echo command audit: LineScreen / CommandTracker and the live session wiring.""" + +from __future__ import annotations + +import time +import unittest +from unittest.mock import MagicMock, patch + +from netx_api.webcrt_cmd_audit import CommandTracker, LineScreen, split_prompt +from netx_api.webcrt_session_model import WebcrtSession + + +def _run(steps, *, linux=False, settle=False): + t = CommandTracker(linux_like=linux) + res = [] + for kind, val in steps: + if kind == "out": + res += t.feed(val) + else: + lines, saw, enter = val + res += t.on_stdin(lines, saw_input=saw, had_enter=enter) + if settle: + res += t.expire(time.monotonic() + 5.0) + return [r.command for r in res] + + +class LineScreenTests(unittest.TestCase): + def _row(self, text: str) -> str: + s = LineScreen() + s.feed_row(text) + return s.text() + + def test_plain_and_cr(self) -> None: + self.assertEqual(self._row("abc\rX"), "Xbc") + + def test_backspace_overwrite(self) -> None: + self.assertEqual(self._row("disX\x08 \x08play"), "display") + + def test_csi_left_erase_and_delete(self) -> None: + self.assertEqual(self._row("display clock\x1b[13D\x1b[K"), "") + self.assertEqual(self._row("abcdef\x1b[3D\x1b[2P"), "abcf") + + def test_huawei_history_redraw(self) -> None: + raw = "display clock\x1b[13D \x1b[13Ddis ip int brief \x1b[17D \x1b[17Ddisplay clock" + self.assertEqual(self._row(raw), "display clock") + + def test_more_pager_wipe_leaves_blank_row(self) -> None: + self.assertEqual(self._row(" ---- More ----\x1b[16D \x1b[16D"), "") + + def test_partial_escape_is_carried_between_chunks(self) -> None: + t = CommandTracker() + t.feed("a\x1b[") + t.feed("1Db") + self.assertEqual(t.current_row(), "b") + + +class SplitPromptTests(unittest.TestCase): + def test_vendors(self) -> None: + self.assertEqual(split_prompt("display clock"), ("", "display clock")) + self.assertEqual(split_prompt("[~r1-GigabitEthernet0/0/0]ip address 1.1.1.1 24")[1], "ip address 1.1.1.1 24") + self.assertEqual(split_prompt("R2(config-if)#no shut"), ("R2(config-if)#", "no shut")) + self.assertEqual(split_prompt("[admin@MikroTik] /ip> print")[1], "print") + self.assertEqual(split_prompt("[user@host ~]$ ls -l")[1], "ls -l") + + def test_not_a_prompt(self) -> None: + self.assertIsNone(split_prompt("[Y/N]:")) + self.assertIsNone(split_prompt(" ---- More ----")) + self.assertIsNone(split_prompt("Username:admin")) + + +class TrackerScenarioTests(unittest.TestCase): + def test_echo_row_wins_over_lagging_typed_text(self) -> None: + got = _run([("out", ""), ("in", (["show clock"], True, True)), ("out", "sh"), ("out", "ow clock\r\nT\r\n")]) + self.assertEqual(got, ["show clock"]) + + def test_backspace_edit(self) -> None: + got = _run([("out", "R2#"), ("in", (["shoX"], True, False)), ("out", "shoX\b \bw"), + ("in", (["show clock"], True, True)), ("out", " clock\r\nT\r\nR2#")]) + self.assertEqual(got, ["R2#show clock"]) + + def test_bare_enter_not_audited(self) -> None: + self.assertEqual(_run([("out", ""), ("in", ([], False, True)), ("out", "\r\n")]), []) + + def test_history_recall_same_frame_and_separate(self) -> None: + a = _run([("out", ""), ("in", ([], True, True)), ("out", "display clock\r\nx\r\n")]) + b = _run([("out", ""), ("in", ([], True, False)), ("out", "display clock"), + ("in", ([], False, True)), ("out", "\r\nx\r\n")]) + self.assertEqual(a, ["display clock"]) + self.assertEqual(b, ["display clock"]) + + def test_tab_redraw_before_enter_echo(self) -> None: + # Huawei redraws the line after CRLF on Tab; the executed line is the later row. + got = _run([("out", ""), ("in", (["display ver\t"], True, True)), + ("out", "\tdisX\x1b[1D \x1b[1Dplay ver\r\ndisplay version \r\nVRP\r\n")]) + self.assertEqual(got, ["display version"]) + + def test_password_redacted(self) -> None: + self.assertEqual(_run([("out", "Password:"), ("in", (["secret"], True, True))]), ["***"]) + + def test_login_dialog_not_audited(self) -> None: + got = _run([("out", "Username:"), ("in", (["admin"], True, True)), ("out", "admin\r\nPassword:")], settle=True) + self.assertEqual(got, []) + + def test_multiline_paste_keeps_order_and_skips_empty_echo(self) -> None: + got = _run([("out", "R2#"), ("in", (["a", "b"], True, True)), ("out", "a\r\nR2#\r\nR2#b\r\nR2#")]) + self.assertEqual(got, ["R2#a", "R2#b"]) + + def test_pager_enter_is_not_a_command(self) -> None: + got = _run([("out", ""), ("in", (["display cur"], True, True)), + ("out", "display cur\r\nl1\r\n ---- More ----"), ("in", ([], False, True)), + ("out", "\x1b[16D \x1b[16D\r\nl2\r\n")]) + self.assertEqual(got, ["display cur"]) + + def test_no_echo_falls_back_to_snapshot(self) -> None: + self.assertEqual(_run([("out", "R2#show clock"), ("in", ([], False, True))], settle=True), ["R2#show clock"]) + + def test_linux_prompt_keeps_single_space(self) -> None: + got = _run([("out", "[u@h ~]$ "), ("in", (["ls"], True, True)), ("out", "ls\r\nf\r\n[u@h ~]$ ")], linux=True) + self.assertEqual(got, ["[u@h ~]$ ls"]) + + def test_enter_typed_ahead_during_output_then_recall(self) -> None: + # Up+Enter pressed while the previous command is still printing: output rows must + # not swallow the bare Enter. + got = _run([("out", ""), ("in", (["display clock"], True, True)), + ("in", ([], True, False)), ("in", ([], False, True)), + ("out", "display clock\r\n2026\r\nSunday\r\n"), + ("out", "display clock\r\n2026\r\nSunday\r\n")]) + self.assertEqual(got, ["display clock", "display clock"]) + + def test_pager_enter_then_typed_line_recorded_once(self) -> None: + got = _run([("out", ""), ("in", (["display cur"], True, True)), + ("out", "display cur\r\nl1\r\n ---- More ----"), ("in", ([], False, True)), + ("out", "\x1b[16D \x1b[16D\r\nl2\r\n"), + ("in", (["display clock"], True, True)), ("out", "display clock\r\nt\r\n")]) + self.assertEqual(got, ["display cur", "display clock"]) + + def test_ctrl_c_echo_does_not_steal_next_enter(self) -> None: + # IOS/bash echo the aborted row + CRLF; keys typed meanwhile must still be audited. + t = CommandTracker() + t.feed("R2#garbage") + res = t.on_stdin([], saw_input=True, had_enter=False, abort=True) + res += t.on_stdin(["show clock"], saw_input=True, had_enter=True) + res += t.feed("\r\nR2#show clock\r\n*22:57:17 UTC\r\nR2#") + self.assertEqual([r.command for r in res], ["R2#show clock"]) + + def test_ctrl_c_with_slow_echo_still_skips_abort_linefeed(self) -> None: + # Only "R2#garb" was echoed when Ctrl-C was typed; the device later prints the + # whole aborted row + CRLF, then echoes the line typed meanwhile. + t = CommandTracker() + t.feed("R2#") + t.on_stdin(["garbage"], saw_input=True, had_enter=False) + t.feed("garb") + res = t.on_stdin([], saw_input=True, had_enter=False, abort=True, abort_text="garbage") + res += t.on_stdin(["show clock"], saw_input=True, had_enter=True) + res += t.feed("age\r\nR2#show clock\r\n*22:57 UTC\r\nR2#") + self.assertEqual([r.command for r in res], ["R2#show clock"]) + + def test_ctrl_c_that_keeps_text_is_audited_as_displayed(self) -> None: + # VRP ignores Ctrl-C at the prompt: the executed line really is garbage+command. + t = CommandTracker() + t.feed("garbage") + res = t.on_stdin([], saw_input=True, had_enter=False, abort=True) + res += t.on_stdin(["display clock"], saw_input=True, had_enter=True) + res += t.feed("display clock\r\nError\r\n") + self.assertEqual([r.command for r in res], ["garbagedisplay clock"]) + + def test_network_prompt_classifier(self) -> None: + from netx_api.webcrt_channel import looks_like_network_prompt as f + + for ok in ("R2#", "", "[~r1]", "R2(config)#"): + self.assertTrue(f(ok), ok) + for bad in ("[u@h ~]$", "root@h:~#", "~ #", "user@host>"): + self.assertFalse(f(bad), bad) + + def test_bulk_output_skips_to_last_row_when_idle(self) -> None: + t = CommandTracker() + t.feed("".join(f"line {i}\r\n" for i in range(5000)) + "") + self.assertEqual(t.current_row(), "") + + +def _live_session(reader_alive: bool = True) -> tuple[WebcrtSession, MagicMock]: + conn = MagicMock() + conn.RETURN = "\n" + conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty", "send"]) + conn.remote_conn.send.side_effect = lambda b: len(b) + sess = WebcrtSession( + session_id="s-live", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", protocol="ssh", + cols=80, rows=24, cli_keymap=False, owner_user_id="u1", owner_username="bob", conn=conn, + ) + alive = MagicMock() + alive.is_alive.return_value = reader_alive + sess._reader = alive + return sess, conn + + +class LiveSessionAuditTests(unittest.TestCase): + @patch("netx_api.webcrt_session_model._audit") + def test_command_resolved_from_device_echo_not_client_hint(self, mock_audit: MagicMock) -> None: + sess, _ = _live_session() + sess._note_stdout_for_audit("") + sess.write_stdin("display clock\r", audit_line="dis") # stale xterm snapshot + mock_audit.assert_not_called() # nothing recorded until the device echoes + sess._note_stdout_for_audit("display clock\r\n2026-10-12\r\n") + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["command"], "display clock") + self.assertEqual(kwargs["via"], "device_echo") + self.assertEqual(kwargs["owner_username"], "bob") + + @patch("netx_api.webcrt_session_model._audit") + def test_keystroke_sent_before_audit_work(self, mock_audit: MagicMock) -> None: + sess, conn = _live_session() + sess._note_stdout_for_audit("") + sess.write_stdin("display clock\r") + self.assertTrue(conn.remote_conn.send.called) + + @patch("netx_api.webcrt_session_model._audit") + def test_password_prompt_redacted_immediately(self, mock_audit: MagicMock) -> None: + sess, _ = _live_session() + sess._note_stdout_for_audit("Password:") + self.assertTrue(sess._password_mode) + sess.write_stdin("hunter2\r") + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["command"], "***") + self.assertTrue(kwargs["redacted"]) + + @patch("netx_api.webcrt_session_model._audit") + def test_post_login_is_trusted(self, mock_audit: MagicMock) -> None: + sess, _ = _live_session() + sess.write_stdin("screen-length 0 temporary\r", audit_source="post_login") + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["command"], "screen-length 0 temporary") + self.assertEqual(kwargs["source"], "post_login") + + @patch("netx_api.webcrt_session_model._audit") + def test_prompt_sync_enter_not_audited(self, mock_audit: MagicMock) -> None: + sess, _ = _live_session() + sess._note_stdout_for_audit("display clock") + sess.write_stdin("\r", audit_source="prompt_sync") + sess._note_stdout_for_audit("\r\n") + mock_audit.assert_not_called() + + +class StdoutHandOffTests(unittest.TestCase): + def test_drain_stdout_batches_and_reports_eof_and_stale(self) -> None: + sess, _ = _live_session() + sess.attach_gen = 3 + sess.out_queue.put(b"a") + sess.out_queue.put(b"b") + self.assertEqual(sess.drain_stdout(2), ("stale", b"")) + self.assertEqual(sess.drain_stdout(3), ("data", b"ab")) + self.assertEqual(sess.drain_stdout(3), ("empty", b"")) + sess.out_queue.put(b"z") + sess.out_queue.put(None) + self.assertEqual(sess.drain_stdout(3), ("eof", b"z")) + + def test_wake_is_coalesced(self) -> None: + sess, _ = _live_session() + loop = MagicMock() + sess.set_notifier(loop, MagicMock()) + sess._wake() + sess._wake() + sess._wake() + self.assertEqual(loop.call_soon_threadsafe.call_count, 1) + sess.drain_stdout(sess.attach_gen) # pump consumed -> next put wakes again + sess._wake() + self.assertEqual(loop.call_soon_threadsafe.call_count, 2) + + def test_reader_keeps_split_utf8_intact(self) -> None: + import threading + + sess, conn = _live_session(reader_alive=False) + chunks = [b"\xe4\xbd", b"\xa0\xe5\xa5\xbd\r\n", None] + ch = conn.remote_conn + ch.recv_ready.return_value = True + ch.exit_status_ready.return_value = False + seen: list[str] = [] + it = iter(chunks) + + def _recv(_n: int) -> bytes: + nxt = next(it) + if nxt is None: + sess.closed = True + return b"" + return nxt + + ch.recv.side_effect = _recv + sess._note_stdout_for_audit = lambda text: seen.append(text) # type: ignore[method-assign] + t = threading.Thread(target=sess._reader_loop, daemon=True) + t.start() + t.join(timeout=3) + self.assertEqual("".join(seen), "你好\r\n") + + +if __name__ == "__main__": + unittest.main() diff --git a/web/package-lock.json b/web/package-lock.json index 8eb6317..a286dc3 100644 --- a/web/package-lock.json +++ b/web/package-lock.json @@ -12,6 +12,7 @@ "@heroui/styles": "^3.2.5", "@tanstack/react-query": "^5.100.8", "@xterm/addon-fit": "^0.11.0", + "@xterm/addon-webgl": "^0.19.0", "@xterm/xterm": "^6.0.0", "@xyflow/react": "^12.11.2", "dagre": "^0.8.5", @@ -2065,6 +2066,12 @@ "integrity": "sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g==", "license": "MIT" }, + "node_modules/@xterm/addon-webgl": { + "version": "0.19.0", + "resolved": "https://registry.npmjs.org/@xterm/addon-webgl/-/addon-webgl-0.19.0.tgz", + "integrity": "sha512-b3fMOsyLVuCeNJWxolACEUED0vm7qC0cy4wRvf3oURSzDTYVQiGPhTnhWZwIHdvC48Y+oLhvYXnY4XDXPoJo6A==", + "license": "MIT" + }, "node_modules/@xterm/xterm": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/@xterm/xterm/-/xterm-6.0.0.tgz", diff --git a/web/package.json b/web/package.json index 9580af5..28bee6a 100644 --- a/web/package.json +++ b/web/package.json @@ -18,6 +18,7 @@ "@heroui/styles": "^3.2.5", "@tanstack/react-query": "^5.100.8", "@xterm/addon-fit": "^0.11.0", + "@xterm/addon-webgl": "^0.19.0", "@xterm/xterm": "^6.0.0", "@xyflow/react": "^12.11.2", "dagre": "^0.8.5", diff --git a/web/src/features/webcrt/components/WebTerminal.tsx b/web/src/features/webcrt/components/WebTerminal.tsx index 3bf6380..efa0525 100644 --- a/web/src/features/webcrt/components/WebTerminal.tsx +++ b/web/src/features/webcrt/components/WebTerminal.tsx @@ -1,6 +1,7 @@ import { Button, Input } from "@heroui/react"; import { forwardRef, useEffect, useImperativeHandle, useRef, useState } from "react"; import { FitAddon } from "@xterm/addon-fit"; +import { WebglAddon } from "@xterm/addon-webgl"; import { Terminal } from "@xterm/xterm"; import "@xterm/xterm/css/xterm.css"; import { useI18n } from "../../../i18n"; @@ -250,13 +251,20 @@ function keyEventToStdin(e: KeyboardEvent): string | null { return null; } -function decodeBytes(buf: ArrayBuffer, _encoding: string): string { - // Server normalizes device encodings (e.g. GBK) to UTF-8 on the wire. - try { - return new TextDecoder("utf-8").decode(buf); - } catch { - return ""; - } +/** + * Server normalizes device encodings (e.g. GBK) to UTF-8 on the wire. Frames are cut at + * arbitrary byte boundaries, so decode as a stream (a split multi-byte char must not + * turn into U+FFFD). + */ +function createStdoutDecoder(): (buf: ArrayBuffer) => string { + const dec = new TextDecoder("utf-8"); + return (buf) => { + try { + return dec.decode(buf, { stream: true }); + } catch { + return ""; + } + }; } const PREF_KEY = "netx.webcrt.termPrefs"; @@ -540,6 +548,28 @@ export const WebTerminal = forwardRef(function WebTerm const fit = new FitAddon(); term.loadAddon(fit); term.open(host); + // GPU renderer: the default DOM renderer is the main cost when a device dumps + // thousands of lines. Fall back silently (headless/old GPU/context loss). + let webgl: WebglAddon | null = null; + try { + webgl = new WebglAddon(); + webgl.onContextLoss(() => { + try { + webgl?.dispose(); + } catch { + /* ignore */ + } + webgl = null; + }); + term.loadAddon(webgl); + } catch { + try { + webgl?.dispose(); + } catch { + /* ignore */ + } + webgl = null; + } termRef.current = term; fitRef.current = fit; @@ -594,27 +624,31 @@ export const WebTerminal = forwardRef(function WebTerm const isActiveSocket = () => !cancelled && !!ws && wsRef.current === ws; - // Coalesce high-rate stdout into one paint frame to cut xterm write churn. + // Echo goes straight to xterm when the terminal is idle (no frame of added latency); + // only bursts are coalesced into one paint frame to cut xterm write churn. let writeBuf = ""; let writeRaf = 0; + let lastWriteAt = 0; + const decodeStdout = createStdoutDecoder(); const flushWriteBuf = () => { - writeRaf = 0; + if (writeRaf) { + cancelAnimationFrame(writeRaf); + writeRaf = 0; + } const chunk = writeBuf; writeBuf = ""; if (!chunk || cancelled) return; + lastWriteAt = performance.now(); term.write(applyKeywordHighlight(chunk, keywordHighlightRef.current)); }; const writeStdout = (raw: string) => { if (!raw || !isActiveSocket()) return; - lastStdoutAtRef.current = performance.now(); + const now = performance.now(); + lastStdoutAtRef.current = now; // Always forward to parent (log buffer / frozen failure transcript); recording flag only gates download. onStdoutRef.current?.(raw); writeBuf += raw; - if (writeBuf.length >= 16384) { - if (writeRaf) { - cancelAnimationFrame(writeRaf); - writeRaf = 0; - } + if (writeBuf.length >= 16384 || (!writeRaf && now - lastWriteAt >= 16)) { flushWriteBuf(); return; } @@ -641,13 +675,13 @@ export const WebTerminal = forwardRef(function WebTerm socket.onmessage = (ev) => { if (!isActiveSocket()) return; if (ev.data instanceof ArrayBuffer) { - writeStdout(decodeBytes(ev.data, encodingRef.current)); + writeStdout(decodeStdout(ev.data)); return; } if (typeof Blob !== "undefined" && ev.data instanceof Blob) { void ev.data.arrayBuffer().then((buf) => { if (!isActiveSocket()) return; - writeStdout(decodeBytes(buf, encodingRef.current)); + writeStdout(decodeStdout(buf)); }); return; } @@ -902,6 +936,12 @@ export const WebTerminal = forwardRef(function WebTerm return () => { cancelled = true; + try { + webgl?.dispose(); + } catch { + /* ignore */ + } + webgl = null; if (writeRaf) { cancelAnimationFrame(writeRaf); writeRaf = 0;