hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-14 23:40:54 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 3

Speed up WebCRT and record audited commands from the device echo.

Browse source 
Performance: event-driven stdout pump (no polling thread per socket, no 200ms
coalescing stall), keystrokes written immediately, blocking SSH reads, bounded
deque queue with batch drain, incremental UTF-8 decode, buffered session log,
non-blocking session teardown, shorter login/settle waits, xterm WebGL renderer
and idle-time direct writes. Zero-latency fake device: echo p50 11ms -> 2.3ms,
command RTT 219ms -> 20ms, bulk output 2.1 -> 8.3 MB/s.

Audit: track the device's own input row server-side (webcrt_cmd_audit) and
resolve each Enter from the echoed line feed, so Tab, backspace, cursor-key
edits, history recall, paste, Ctrl-C and pagers record what actually ran.
Password lines are redacted; pending Enters are flushed on close.

Hop: stop closing the hop session when the target prints the same prompt as the
hop (system-view); recognise the Huawei "closed by the remote host" message.
Generic NEs that show a network prompt now get the network key map.

Adds bench_webcrt.py, fake_ssh_device.py and e2e_webcrt_input_scenarios.py.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
oliver 2026-10-12 07:34:30 +08:00
parent 0d894efe5e
commit 45de8144a3
14 changed files with 2060 additions and 241 deletions
Download patch file Download diff file Expand all files Collapse all files

4
netx_api/ne_cli_hop.py
View file

@ -10,7 +10,9 @@ from netmiko import ConnectHandler
_CLI_HOP_NESTED_END_RE = re.compile( _CLI_HOP_NESTED_END_RE = re.compile(
r"(?is)" r"(?is)"
r"(?:^|\n)\s*(?:" 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"|closed\s+by\s+foreign\s+host"
r"|connection\s+to\s+\S+\s+closed" r"|connection\s+to\s+\S+\s+closed"
r"|%\s*connection\s+closed(?:\s+by\s+(?:foreign|remote)\s+host)?" r"|%\s*connection\s+closed(?:\s+by\s+(?:foreign|remote)\s+host)?"

40
netx_api/ne_session_connect.py
View file

@ -545,18 +545,26 @@ def _connect_direct(
def _read_channel(conn: ConnectHandler, wait: float = 0.5, max_loops: int = 40) -> str: def _read_channel(conn: ConnectHandler, wait: float = 0.5, max_loops: int = 40) -> str:
if wait > 0: """Collect device output.
time.sleep(wait)
``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] = [] chunks: list[str] = []
for _ in range(max_loops): reads = 0
while reads < max_loops:
try: try:
part = conn.read_channel() part = conn.read_channel()
except Exception: except Exception:
break break
if part is None or part is False: if part is None or part is False:
break part = ""
# MagicMock truthy-but-empty: stop when unit tests leave read_channel unconfigured. # 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) text = str(part)
# unittest.mock default str looks like "<MagicMock name=...>" # unittest.mock default str looks like "<MagicMock name=...>"
if text.startswith("<MagicMock") or text.startswith("<Mock"): if text.startswith("<MagicMock") or text.startswith("<Mock"):
@ -564,10 +572,18 @@ def _read_channel(conn: ConnectHandler, wait: float = 0.5, max_loops: int = 40)
part = text part = text
elif isinstance(part, (bytes, bytearray)): elif isinstance(part, (bytes, bytearray)):
part = bytes(part).decode("utf-8", errors="replace") part = bytes(part).decode("utf-8", errors="replace")
if not part: now = time.monotonic()
if part:
chunks.append(part)
reads += 1
got_at = now
continue
if got_at is None:
if now >= first_deadline:
break
elif now - got_at >= quiet:
break break
chunks.append(part) time.sleep(0.01)
time.sleep(0.05 if wait <= 0.15 else 0.15)
return "".join(chunks) return "".join(chunks)
@ -801,11 +817,11 @@ def _interactive_target_auth(
return return
else: else:
empty_after_pass = 0 empty_after_pass = 0
time.sleep(0.12) time.sleep(0.02)
continue continue
if not buf.strip(): if not buf.strip():
time.sleep(0.2) time.sleep(0.02)
continue continue
# Huawei stelnet often reprints the hop ``[sysname]`` after host-key trust and # Huawei stelnet often reprints the hop ``[sysname]`` after host-key trust and
# *before* ``Enter password:``. Do not treat that as target login success. # *before* ``Enter password:``. Do not treat that as target login success.
@ -826,7 +842,7 @@ def _interactive_target_auth(
if denied: if denied:
raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}")
return return
time.sleep(0.15) time.sleep(0.02)
denied = _target_auth_failure_snippet(acc, sent_pass=sent_pass) denied = _target_auth_failure_snippet(acc, sent_pass=sent_pass)
if denied: if denied:
raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}") raise paramiko.AuthenticationException(f"target_auth_rejected: {denied}")
@ -879,7 +895,7 @@ def _huawei_enter_system_view(
for _ in range(8): for _ in range(8):
part = _read_channel(conn, wait=0.2, max_loops=10) part = _read_channel(conn, wait=0.2, max_loops=10)
if not part: if not part:
time.sleep(0.15) time.sleep(0.02)
continue continue
acc += part acc += part
if emit_raw: if emit_raw:

67
netx_api/webcrt_channel.py
View file

@ -8,6 +8,7 @@ import queue
import re import re
import threading import threading
import time import time
from collections import deque
from datetime import datetime, timezone from datetime import datetime, timezone
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
@ -44,6 +45,18 @@ def uses_network_cli_keymap(device_type: str = "", vendor: str = "") -> bool:
return True 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>`` / ``[~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( def map_network_cli_keys(
data: str, data: str,
*, *,
@ -56,6 +69,8 @@ def map_network_cli_keys(
text = str(data or "") text = str(data or "")
if not text: if not text:
return text return text
if "\x1b" not in text and "\x7f" not in text:
return text # plain printable typing: nothing to rewrite
out: list[str] = [] out: list[str] = []
i = 0 i = 0
n = len(text) 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: def map_network_cli_enter(data: str, conn: ConnectHandler | None) -> str:
"""Map xterm Enter (\\r) to the device's Netmiko RETURN.""" """Map xterm Enter (\\r) to the device's Netmiko RETURN."""
text = str(data or "") text = str(data or "")
if not text: if not text or ("\r" not in text and "\n" not in text):
return text return text
ret = channel_return(conn) ret = channel_return(conn)
# Replace once: chained replacements expand the CR in Telnet CRLF again. # 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: 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: 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._max = max(8, int(maxsize or 2000))
self._cond = threading.Condition() self._cond = threading.Condition()
self.dropped = 0 self.dropped = 0
@ -378,13 +393,12 @@ class _BoundedByteQueue:
def put(self, item: bytes | None) -> None: def put(self, item: bytes | None) -> None:
with self._cond: with self._cond:
while self._q.qsize() >= self._max: dq = self._dq
try: if item is not None:
self._q.get_nowait() while len(dq) >= self._max:
dq.popleft()
self.dropped += 1 self.dropped += 1
except queue.Empty: dq.append(item)
break
self._q.put(item)
self._cond.notify() self._cond.notify()
def put_nowait(self, item: bytes | None) -> None: def put_nowait(self, item: bytes | None) -> None:
@ -392,22 +406,42 @@ class _BoundedByteQueue:
def get_nowait(self) -> bytes | None: def get_nowait(self) -> bytes | None:
with self._cond: 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: def get(self, timeout: float = 0.25) -> bytes | None:
"""Block until a chunk is available or timeout (raises queue.Empty).""" """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: with self._cond:
while self._q.empty(): while not self._dq:
remaining = deadline - time.time() remaining = deadline - time.monotonic()
if remaining <= 0: if remaining <= 0:
raise queue.Empty raise queue.Empty
self._cond.wait(timeout=remaining) 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: def qsize(self) -> int:
with self._cond: with self._cond:
return self._q.qsize() return len(self._dq)
def take_drop_delta(self) -> int: def take_drop_delta(self) -> int:
"""Return newly dropped chunk count since last call (for client notice).""" """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] = [] completed: list[str] = []
limit = max(64, min(int(max_line or 512), 4096)) limit = max(64, min(int(max_line or 512), 4096))
raw = str(data or "") 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 i = 0
while i < len(raw): while i < len(raw):
ch = raw[i] ch = raw[i]

525
netx_api/webcrt_cmd_audit.py Normal file
View file

@ -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] [*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

217
netx_api/webcrt_router.py
View file

@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import codecs
import json import json
import logging import logging
import time import time
@ -686,6 +687,7 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None:
replay_text = _decode_bytes(raw_boot, "utf-8") replay_text = _decode_bytes(raw_boot, "utf-8")
sess.bootstrap_replayed = True sess.bootstrap_replayed = True
else: else:
sess.flush_session_log()
replay_text = read_session_log_tail(session_id, max_bytes=49152) replay_text = read_session_log_tail(session_id, max_bytes=49152)
if replay_text: if replay_text:
replay_bytes = _encode_text(replay_text, "utf-8") 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) _log.debug("webcrt bootstrap/replay send failed session=%s", session_id, exc_info=True)
stop = asyncio.Event() stop = asyncio.Event()
stdin_buf: list[str] = [] loop = asyncio.get_running_loop()
stdin_audit_lines: list[str] = [] io_pool = webcrt_io_executor()
stdin_flush_task: asyncio.Task[None] | None = None 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: async def stdin_writer() -> None:
nonlocal stdin_buf, stdin_audit_lines while True:
if not stdin_buf: data, audit = await stdin_q.get()
return parts = [data]
data = "".join(stdin_buf) audits = [audit] if audit else []
audit_lines = list(stdin_audit_lines) taken = 1
stdin_buf = [] while True:
stdin_audit_lines = [] 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: try:
await asyncio.get_running_loop().run_in_executor( await asyncio.wait_for(stdin_q.join(), timeout=5.0)
webcrt_io_executor(), except Exception:
lambda: sess.write_stdin( pass
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()
async def pump_stdout() -> None: async def pump_stdout() -> None:
loop = asyncio.get_running_loop() """Event-driven stdout pump: the reader thread wakes us, we drain and send at once."""
pending: list[bytes] = [] sess.set_notifier(loop, wake)
last_flush = loop.time() dec = None
if _normalize_encoding(sess.encoding) != "utf-8":
dec = codecs.getincrementaldecoder("gbk")(errors="replace")
def _to_browser_bytes(raw: bytes) -> bytes: def _to_browser(raw: bytes) -> bytes:
if _normalize_encoding(sess.encoding) == "utf-8": if dec is None:
return raw return raw
return _decode_bytes(raw, sess.encoding).encode("utf-8", errors="replace") return dec.decode(raw).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
async def _notify_queue_drops() -> None: async def _notify_queue_drops() -> None:
try: try:
@ -772,48 +787,48 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None:
except Exception: except Exception:
pass pass
while not stop.is_set(): wake.set() # output queued before the pump started
# Longer block is cheap now (Condition wait); cuts executor churn when idle. try:
chunk = await loop.run_in_executor( while not stop.is_set():
webcrt_io_executor(), lambda: sess.take_stdout(attach_gen, timeout=0.2) await wake.wait()
) wake.clear()
if chunk == "stale": while not stop.is_set():
break status, data = sess.drain_stdout(attach_gen)
if chunk == "empty": if status == "stale":
if pending and (loop.time() - last_flush) >= 0.016: return
if not await _flush_pending(): 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() 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 break
await _notify_queue_drops() finally:
continue sess.clear_notifier(wake)
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()
reader_task = asyncio.create_task(pump_stdout()) reader_task = asyncio.create_task(pump_stdout())
writer_task = asyncio.create_task(stdin_writer())
if sess.needs_live_prompt: if sess.needs_live_prompt:
sess.needs_live_prompt = False sess.needs_live_prompt = False
try: try:
await asyncio.get_running_loop().run_in_executor( await loop.run_in_executor(
webcrt_io_executor(), io_pool,
lambda: sess.write_stdin("\r", audit_source="prompt_sync"), lambda: sess.write_stdin("\r", audit_source="prompt_sync"),
) )
except Exception: 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) text = _decode_bytes(bytes(msg_raw["bytes"]), sess.encoding)
except Exception: except Exception:
continue continue
stdin_buf.append(text) stdin_q.put_nowait((text, None))
if stdin_flush_task is None or stdin_flush_task.done():
stdin_flush_task = asyncio.create_task(schedule_stdin_flush())
continue continue
raw = msg_raw.get("text") raw = msg_raw.get("text")
if raw is None: if raw is None:
@ -846,23 +859,16 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None:
if data is None: if data is None:
continue continue
audit_raw = msg.get("audit_line") audit_raw = msg.get("audit_line")
if audit_raw is not None and str(audit_raw).strip(): audit = str(audit_raw).strip()[:512] if audit_raw is not None and str(audit_raw).strip() else None
stdin_audit_lines.append(str(audit_raw).strip()[:512]) stdin_q.put_nowait((str(data), audit))
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())
elif mtype == "resize": elif mtype == "resize":
cols = int(msg.get("cols") or sess.cols) cols = int(msg.get("cols") or sess.cols)
rows = int(msg.get("rows") or sess.rows) 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": elif mtype == "break":
await drain_stdin()
try: 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: except Exception as exc:
await websocket.send_json( await websocket.send_json(
{"type": "status", "state": "error", "message": f"break_failed:{exc}"} {"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"}) await websocket.send_json({"type": "pong"})
elif mtype == "close": elif mtype == "close":
stop.set() stop.set()
await flush_stdin() await drain_stdin()
close_session( close_session(
session_id, session_id,
reason="client_close", 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) _log.exception("webcrt ws error session=%s", session_id)
finally: finally:
stop.set() stop.set()
if stdin_flush_task and not stdin_flush_task.done(): wake.set()
stdin_flush_task.cancel()
try: try:
await flush_stdin() await drain_stdin()
except Exception: except Exception:
pass pass
writer_task.cancel()
reader_task.cancel() reader_task.cancel()
try: for t in (writer_task, reader_task):
await reader_task try:
except Exception: await t
pass except BaseException:
pass
# Keep device session so UI reconnect / remount can re-attach. # Keep device session so UI reconnect / remount can re-attach.
if get_session(session_id) is not None: if get_session(session_id) is not None:
grace = float(getattr(settings, "webcrt_detach_grace_sec", 120) or 120) grace = float(getattr(settings, "webcrt_detach_grace_sec", 120) or 120)

441
netx_api/webcrt_session_model.py
View file

@ -1,8 +1,12 @@
"""WebCRT interactive session object (I/O, SFTP, close).""" """WebCRT interactive session object (I/O, SFTP, close)."""
from __future__ import annotations from __future__ import annotations
import codecs
import logging import logging
import queue import queue
import re
import select
import socket
import threading import threading
import time import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
@ -16,8 +20,10 @@ from .ne_session_factory import (
extract_cli_prompt_marker, extract_cli_prompt_marker,
should_close_cli_hop_session, should_close_cli_hop_session,
) )
from .webcrt_cmd_audit import AuditCmd, CommandTracker
from .webcrt_channel import ( from .webcrt_channel import (
_BoundedByteQueue, _BoundedByteQueue,
_normalize_encoding,
_audit, _audit,
_decode_bytes, _decode_bytes,
_encode_text, _encode_text,
@ -31,6 +37,9 @@ from .webcrt_channel import (
_log = logging.getLogger("netx.webcrt") _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 @dataclass
class WebcrtSession: class WebcrtSession:
session_id: str session_id: str
@ -98,6 +107,17 @@ class WebcrtSession:
_last_prompt_line: str = field(default="", repr=False) _last_prompt_line: str = field(default="", repr=False)
_last_audited_command: str = field(default="", repr=False) _last_audited_command: str = field(default="", repr=False)
_last_audited_at: float = field(default=0.0, 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: def touch(self) -> None:
self.last_activity = time.time() self.last_activity = time.time()
@ -256,39 +276,55 @@ class WebcrtSession:
self._log_fh = None self._log_fh = None
def append_session_log(self, text: str) -> 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 return
try: try:
self._log_fh.write(text) with self._log_lock:
self._log_fh.flush() fh.write(text)
max_bytes = int(getattr(settings, "webcrt_session_log_max_bytes", 0) or 0) if time.monotonic() - self._log_last_flush >= 0.25:
if max_bytes > 0: self.flush_session_log()
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()
except Exception: except Exception:
_log.warning("webcrt session log append failed session=%s", self.session_id, exc_info=True) _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 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: if fh is None:
return return
try: try:
@ -326,6 +362,57 @@ class WebcrtSession:
return "stale" return "stale"
return chunk # bytes | None 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( def write_stdin(
self, self,
data: str, data: str,
@ -351,24 +438,51 @@ class WebcrtSession:
text = map_network_cli_enter(text, self.conn) text = map_network_cli_enter(text, self.conn)
if not text: if not text:
return return
self._note_stdin_for_audit( if self.cli_hop_guard and (_HOP_EXIT_RE.search(text) or "\x04" in text):
text, self._hop_exit_until = time.monotonic() + 8.0
source=audit_source, live_cmds: list[AuditCmd] = []
audit_line=audit_override, if self._audit_live():
audit_lines=audit_list, # 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: with self._write_lock:
# Prefer raw channel I/O for interactive typing (char echo / backspace). # 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: try:
if channel is not None and hasattr(channel, "send") and callable(channel.send): if channel is not None and hasattr(channel, "send") and callable(channel.send):
payload = _encode_text(text, self.encoding) 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) view = memoryview(payload)
deadline = time.monotonic() + 30.0
while len(view): 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: if n <= 0:
time.sleep(0.01) time.sleep(0.005)
continue continue
view = view[n:] view = view[n:]
self.bytes_in += len(payload) self.bytes_in += len(payload)
@ -377,12 +491,79 @@ class WebcrtSession:
channel.write(payload) channel.write(payload)
self.bytes_in += len(payload) self.bytes_in += len(payload)
else: else:
self.conn.write_channel(text) conn.write_channel(text)
self.bytes_in += len(text) self.bytes_in += len(text)
except Exception: except Exception:
if self.closed or self.conn is None:
raise
self.conn.write_channel(text) self.conn.write_channel(text)
self.bytes_in += len(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( def _note_stdin_for_audit(
self, self,
@ -461,6 +642,26 @@ class WebcrtSession:
_log.debug("webcrt command audit failed session=%s", self.session_id, exc_info=True) _log.debug("webcrt command audit failed session=%s", self.session_id, exc_info=True)
def _note_stdout_for_audit(self, text: str) -> None: 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.""" """Track device prompts so the next typed line can be redacted if it is a password."""
chunk = str(text or "") chunk = str(text or "")
if not chunk: if not chunk:
@ -534,50 +735,74 @@ class WebcrtSession:
) )
self._reader.start() 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: def _reader_loop(self) -> None:
conn = self.conn conn = self.conn
if conn is None: if conn is None:
self.out_queue.put(None) self.out_queue.put(None)
self._wake()
return return
channel = getattr(conn, "remote_conn", None) channel = getattr(conn, "remote_conn", None)
hop_return = False hop_return = False
poll = max(0.002, float(getattr(settings, "webcrt_reader_poll_sec", 0.01) or 0.01)) 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: try:
while not self.closed: while not self.closed:
chunk = b"" chunk = b""
try: try:
if channel is not None and hasattr(channel, "recv_ready") and hasattr(channel, "recv"): if blocking:
# Paramiko SSH: prefer short blocking recv over fixed spin-sleep.
ready = False
try: try:
ready = bool(channel.recv_ready()) chunk = channel.recv(65536)
except Exception: except socket.timeout:
ready = False chunk = b""
if ready: else:
chunk = channel.recv(16384) 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: if not chunk:
break break
elif hasattr(channel, "exit_status_ready") and channel.exit_status_ready(): elif hasattr(channel, "exit_status_ready") and channel.exit_status_ready():
break break
else: else:
# Brief block: settimeout + recv wakes sooner than sleep(0.04). time.sleep(poll)
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
elif channel is not None and hasattr(channel, "read_very_eager"): elif channel is not None and hasattr(channel, "read_very_eager"):
# Telnet: do NOT use conn.read_channel() — Netmiko strips ANSI. # Telnet: do NOT use conn.read_channel() — Netmiko strips ANSI.
data = channel.read_very_eager() data = channel.read_very_eager()
@ -588,39 +813,52 @@ class WebcrtSession:
else _encode_text(str(data), self.encoding) else _encode_text(str(data), self.encoding)
) )
else: else:
time.sleep(poll) self._wait_readable(channel, 0.2)
continue
else: else:
text = conn.read_channel() text = conn.read_channel()
if text: if text:
chunk = _encode_text(str(text), self.encoding) chunk = _encode_text(str(text), self.encoding)
else: else:
time.sleep(poll) time.sleep(poll)
continue except EOFError:
break
except Exception as exc: except Exception as exc:
if self.closed: if self.closed:
break break
_log.debug("webcrt reader error session=%s: %s", self.session_id, exc) _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) time.sleep(0.05)
continue continue
now = time.monotonic()
if chunk: if chunk:
self.touch() self.last_activity = time.time()
self.bytes_out += len(chunk) self.bytes_out += len(chunk)
self.out_queue.put(chunk) self.out_queue.put(chunk)
self._wake()
decoded = ""
try: try:
decoded = _decode_bytes(chunk, self.encoding) decoded = dec.decode(bytes(chunk))
self.append_session_log(decoded) if decoded:
self._note_stdout_for_audit(decoded) self.append_session_log(decoded)
self._note_stdout_for_audit(decoded)
except Exception: except Exception:
pass _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): if self.cli_hop_guard and self._note_cli_hop_output(chunk, decoded):
hop_return = True hop_return = True
notice = ( notice = (
"\r\n*** WebCRT: 目标会话已结束，已断开代理连接 " "\r\n*** WebCRT: 目标会话已结束，已断开代理连接 "
"(target session ended; closing hop proxy) ***\r\n" "(target session ended; closing hop proxy) ***\r\n"
) )
self.out_queue.put(notice.encode("utf-8", errors="replace")) self.out_queue.put(notice.encode("utf-8", errors="replace"))
self._wake()
break 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: finally:
if hop_return and not self.closed: if hop_return and not self.closed:
try: try:
@ -630,22 +868,33 @@ class WebcrtSession:
except Exception: except Exception:
self.close("cli_hop_return") self.close("cli_hop_return")
self.out_queue.put(None) 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.""" """Accumulate stdout and return True when nested CLI hop has returned to proxy."""
try: if text is None:
text = _decode_bytes(chunk, self.encoding) try:
except Exception: text = _decode_bytes(chunk, self.encoding)
text = str(chunk) except Exception:
text = str(chunk)
self._hop_scan_buf = (self._hop_scan_buf + text)[-12000:] self._hop_scan_buf = (self._hop_scan_buf + text)[-12000:]
marker = str(self.cli_hop_prompt or "").strip() marker = str(self.cli_hop_prompt or "").strip()
last = extract_cli_prompt_marker(self._hop_scan_buf) last = extract_cli_prompt_marker(self._hop_scan_buf)
if last and (not marker or last != marker): if last and (not marker or last != marker):
self._cli_hop_seen_other_prompt = True 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( return should_close_cli_hop_session(
self._hop_scan_buf, self._hop_scan_buf,
self.cli_hop_prompt, self.cli_hop_prompt,
seen_other_prompt=self._cli_hop_seen_other_prompt, seen_other_prompt=armed,
) )
def run_post_login_commands(self) -> None: 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) _log.debug("post_login command failed session=%s", self.session_id, exc_info=True)
break 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: def close(self, reason: str = "closed") -> None:
if self.closed: if self.closed:
return return
@ -667,14 +929,27 @@ class WebcrtSession:
self.state = "closed" self.state = "closed"
self.close_reason = reason or "closed" self.close_reason = reason or "closed"
self._ready_event.set() self._ready_event.set()
self.close_sftp() # Enters still waiting for a device echo would be lost with the session.
try: try:
close_netmiko_connection(self.conn) pending_cmds = self._tracker.flush_all()
if pending_cmds:
self._emit_audit_cmds(pending_cmds)
except Exception: 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 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: try:
self.out_queue.put_nowait(None) self.out_queue.put_nowait(None)
except Exception: except Exception:
pass pass
self._wake(force=True)
self.close_session_log() self.close_session_log()

35
netx_api/webcrt_session_registry.py
View file

@ -34,6 +34,7 @@ from .webcrt_channel import (
_normalize_encoding, _normalize_encoding,
_prime_interactive_channel, _prime_interactive_channel,
_session_log_text, _session_log_text,
looks_like_network_prompt,
prepare_bootstrap_output, prepare_bootstrap_output,
uses_network_cli_keymap, 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 # 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. # (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. # 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: 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: except Exception:
early = "" early = ""
if early: if early:
sess.push_connect_echo(early) sess.push_connect_echo(early)
seed = f"{pre_log}{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( saw_password_change = bool(
re.search( re.search(
r"(?i)(change\s*now|please\s*choose|password\s+needs\s+to\s+be\s+changed).{0,80}:", 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 = "" primed = ""
combined = f"{seed}{primed}" combined = f"{seed}{primed}"
# Brief settle only — do not burn seconds here while the terminal shows a prompt. # 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: try:
more = _capture_raw_channel(conn, duration=settle_sec) more = _capture_raw_channel(conn, duration=settle_sec)
if more: if more:
@ -395,7 +398,7 @@ def _finish_connect(
if not echoed.strip(): if not echoed.strip():
bootstrap = prepare_bootstrap_output(combined) bootstrap = prepare_bootstrap_output(combined)
try: try:
leftover = _capture_raw_channel(conn, duration=0.1) leftover = _capture_raw_channel(conn, duration=0.05)
except Exception: except Exception:
leftover = "" leftover = ""
if ( if (
@ -407,7 +410,7 @@ def _finish_connect(
bootstrap = prepare_bootstrap_output(f"{bootstrap}{leftover}") bootstrap = prepare_bootstrap_output(f"{bootstrap}{leftover}")
else: else:
try: try:
leftover = _capture_raw_channel(conn, duration=0.08) leftover = _capture_raw_channel(conn, duration=0.04)
except Exception: except Exception:
leftover = "" leftover = ""
if ( if (
@ -421,6 +424,19 @@ def _finish_connect(
sess.conn = conn sess.conn = conn
sess.cli_hop_guard = bool(hop_guard) sess.cli_hop_guard = bool(hop_guard)
sess.cli_hop_prompt = str((hop_guard or {}).get("hop_prompt") or "") 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) sess.bootstrap_output = _encode_text(str(bootstrap or ""), sess.encoding)
# Nudge Enter on WS attach only when we still need a shell prompt. # 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). # 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) and not _looks_like_login_prompt(final_view)
) )
sess.open_session_log() 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 log_seed = echoed or bootstrap
if log_seed: if log_seed:
sess.append_session_log(log_seed if str(log_seed).endswith("\n") else str(log_seed) + "\n") 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. # Drop late prompt echoes that race into the queue right after reader start.
if not prompt_hint and final_view: if not prompt_hint and final_view:
prompt_hint = str(final_view).replace("\r\n", "\n").replace("\r", "\n").strip().split("\n")[-1].strip() 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: while time.time() < settle_deadline:
try: try:
chunk = sess.out_queue.get_nowait() 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. # A slow VM may emit its FIRST prompt after the reader starts.
# Only drop duplicates when a prompt was already delivered during login. # 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): 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 continue
if _looks_like_cli_prompt(text): if _looks_like_cli_prompt(text):
sess.needs_live_prompt = False 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") raise HTTPException(status_code=404, detail="webcrt_session_not_found")
# Allow re-attach after brief WS drop (React StrictMode remount / network blip). # 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. # Bump generation so the previous WS pump stops and does not steal echo bytes.
sess.attach_gen += 1 attach_gen = sess.bump_attach()
attach_gen = sess.attach_gen
sess.attached = True sess.attached = True
sess.detach_deadline = None sess.detach_deadline = None
sess.touch() sess.touch()

239
scripts/bench_webcrt.py Normal file
View file

@ -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 <id> [--prompt-re "<r1>|\\[~?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()))

189
scripts/e2e_webcrt_input_scenarios.py Normal file
View file

@ -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 <id> --vendor huawei
python scripts/e2e_webcrt_input_scenarios.py --ne-id <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())

167
scripts/fake_ssh_device.py Normal file
View file

@ -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<Tab>`` -> ``show ``), history on Up arrow, ``show clock`` (short output),
``bulk <MB>`` (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()

295
tests/test_webcrt_cmd_audit.py Normal file
View file

@ -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("<r1>display clock\x1b[13D\x1b[K"), "<r1>")
self.assertEqual(self._row("abcdef\x1b[3D\x1b[2P"), "abcf")
def test_huawei_history_redraw(self) -> None:
raw = "<r1>display clock\x1b[13D \x1b[13Ddis ip int brief \x1b[17D \x1b[17Ddisplay clock"
self.assertEqual(self._row(raw), "<r1>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("<r1>a\x1b[")
t.feed("1Db")
self.assertEqual(t.current_row(), "<r1>b")
class SplitPromptTests(unittest.TestCase):
def test_vendors(self) -> None:
self.assertEqual(split_prompt("<r1>display clock"), ("<r1>", "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", "<r1>"), ("in", (["show clock"], True, True)), ("out", "sh"), ("out", "ow clock\r\nT\r\n<r1>")])
self.assertEqual(got, ["<r1>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", "<r1>"), ("in", ([], False, True)), ("out", "\r\n<r1>")]), [])
def test_history_recall_same_frame_and_separate(self) -> None:
a = _run([("out", "<r1>"), ("in", ([], True, True)), ("out", "display clock\r\nx\r\n<r1>")])
b = _run([("out", "<r1>"), ("in", ([], True, False)), ("out", "display clock"),
("in", ([], False, True)), ("out", "\r\nx\r\n<r1>")])
self.assertEqual(a, ["<r1>display clock"])
self.assertEqual(b, ["<r1>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", "<r1>"), ("in", (["display ver\t"], True, True)),
("out", "\tdisX\x1b[1D \x1b[1Dplay ver\r\n<r1>display version \r\nVRP\r\n<r1>")])
self.assertEqual(got, ["<r1>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", "<r1>"), ("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<r1>")])
self.assertEqual(got, ["<r1>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", "<r1>"), ("in", (["display clock"], True, True)),
("in", ([], True, False)), ("in", ([], False, True)),
("out", "display clock\r\n2026\r\nSunday\r\n<r1>"),
("out", "display clock\r\n2026\r\nSunday\r\n<r1>")])
self.assertEqual(got, ["<r1>display clock", "<r1>display clock"])
def test_pager_enter_then_typed_line_recorded_once(self) -> None:
got = _run([("out", "<r1>"), ("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<r1>"),
("in", (["display clock"], True, True)), ("out", "display clock\r\nt\r\n<r1>")])
self.assertEqual(got, ["<r1>display cur", "<r1>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("<r1>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<r1>")
self.assertEqual([r.command for r in res], ["<r1>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>", "[~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)) + "<r1>")
self.assertEqual(t.current_row(), "<r1>")
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("<r1>")
sess.write_stdin("display clock\r", audit_line="<r1>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<r1>")
kwargs = mock_audit.call_args.kwargs
self.assertEqual(kwargs["command"], "<r1>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("<r1>")
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("<r1>display clock")
sess.write_stdin("\r", audit_source="prompt_sync")
sess._note_stdout_for_audit("\r\n<r1>")
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"<r1>\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), "<r1>你好\r\n")
if __name__ == "__main__":
unittest.main()

7
web/package-lock.json generated
View file

@ -12,6 +12,7 @@
"@heroui/styles": "^3.2.5", "@heroui/styles": "^3.2.5",
"@tanstack/react-query": "^5.100.8", "@tanstack/react-query": "^5.100.8",
"@xterm/addon-fit": "^0.11.0", "@xterm/addon-fit": "^0.11.0",
"@xterm/addon-webgl": "^0.19.0",
"@xterm/xterm": "^6.0.0", "@xterm/xterm": "^6.0.0",
"@xyflow/react": "^12.11.2", "@xyflow/react": "^12.11.2",
"dagre": "^0.8.5", "dagre": "^0.8.5",
@ -2065,6 +2066,12 @@
"integrity": "sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g==", "integrity": "sha512-jYcgT6xtVYhnhgxh3QgYDnnNMYTcf8ElbxxFzX0IZo+vabQqSPAjC3c1wJrKB5E19VwQei89QCiZZP86DCPF7g==",
"license": "MIT" "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": { "node_modules/@xterm/xterm": {
"version": "6.0.0", "version": "6.0.0",
"resolved": "https://registry.npmjs.org/@xterm/xterm/-/xterm-6.0.0.tgz", "resolved": "https://registry.npmjs.org/@xterm/xterm/-/xterm-6.0.0.tgz",

1
web/package.json
View file

@ -18,6 +18,7 @@
"@heroui/styles": "^3.2.5", "@heroui/styles": "^3.2.5",
"@tanstack/react-query": "^5.100.8", "@tanstack/react-query": "^5.100.8",
"@xterm/addon-fit": "^0.11.0", "@xterm/addon-fit": "^0.11.0",
"@xterm/addon-webgl": "^0.19.0",
"@xterm/xterm": "^6.0.0", "@xterm/xterm": "^6.0.0",
"@xyflow/react": "^12.11.2", "@xyflow/react": "^12.11.2",
"dagre": "^0.8.5", "dagre": "^0.8.5",

74
web/src/features/webcrt/components/WebTerminal.tsx
View file

@ -1,6 +1,7 @@
import { Button, Input } from "@heroui/react"; import { Button, Input } from "@heroui/react";
import { forwardRef, useEffect, useImperativeHandle, useRef, useState } from "react"; import { forwardRef, useEffect, useImperativeHandle, useRef, useState } from "react";
import { FitAddon } from "@xterm/addon-fit"; import { FitAddon } from "@xterm/addon-fit";
import { WebglAddon } from "@xterm/addon-webgl";
import { Terminal } from "@xterm/xterm"; import { Terminal } from "@xterm/xterm";
import "@xterm/xterm/css/xterm.css"; import "@xterm/xterm/css/xterm.css";
import { useI18n } from "../../../i18n"; import { useI18n } from "../../../i18n";
@ -250,13 +251,20 @@ function keyEventToStdin(e: KeyboardEvent): string | null {
return null; return null;
} }
function decodeBytes(buf: ArrayBuffer, _encoding: string): string { /**
// Server normalizes device encodings (e.g. GBK) to UTF-8 on the wire. * Server normalizes device encodings (e.g. GBK) to UTF-8 on the wire. Frames are cut at
try { * arbitrary byte boundaries, so decode as a stream (a split multi-byte char must not
return new TextDecoder("utf-8").decode(buf); * turn into U+FFFD).
} catch { */
return ""; 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"; const PREF_KEY = "netx.webcrt.termPrefs";
@ -540,6 +548,28 @@ export const WebTerminal = forwardRef<WebTerminalHandle, Props>(function WebTerm
const fit = new FitAddon(); const fit = new FitAddon();
term.loadAddon(fit); term.loadAddon(fit);
term.open(host); 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; termRef.current = term;
fitRef.current = fit; fitRef.current = fit;
@ -594,27 +624,31 @@ export const WebTerminal = forwardRef<WebTerminalHandle, Props>(function WebTerm
const isActiveSocket = () => !cancelled && !!ws && wsRef.current === ws; 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 writeBuf = "";
let writeRaf = 0; let writeRaf = 0;
let lastWriteAt = 0;
const decodeStdout = createStdoutDecoder();
const flushWriteBuf = () => { const flushWriteBuf = () => {
writeRaf = 0; if (writeRaf) {
cancelAnimationFrame(writeRaf);
writeRaf = 0;
}
const chunk = writeBuf; const chunk = writeBuf;
writeBuf = ""; writeBuf = "";
if (!chunk || cancelled) return; if (!chunk || cancelled) return;
lastWriteAt = performance.now();
term.write(applyKeywordHighlight(chunk, keywordHighlightRef.current)); term.write(applyKeywordHighlight(chunk, keywordHighlightRef.current));
}; };
const writeStdout = (raw: string) => { const writeStdout = (raw: string) => {
if (!raw || !isActiveSocket()) return; 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. // Always forward to parent (log buffer / frozen failure transcript); recording flag only gates download.
onStdoutRef.current?.(raw); onStdoutRef.current?.(raw);
writeBuf += raw; writeBuf += raw;
if (writeBuf.length >= 16384) { if (writeBuf.length >= 16384 || (!writeRaf && now - lastWriteAt >= 16)) {
if (writeRaf) {
cancelAnimationFrame(writeRaf);
writeRaf = 0;
}
flushWriteBuf(); flushWriteBuf();
return; return;
} }
@ -641,13 +675,13 @@ export const WebTerminal = forwardRef<WebTerminalHandle, Props>(function WebTerm
socket.onmessage = (ev) => { socket.onmessage = (ev) => {
if (!isActiveSocket()) return; if (!isActiveSocket()) return;
if (ev.data instanceof ArrayBuffer) { if (ev.data instanceof ArrayBuffer) {
writeStdout(decodeBytes(ev.data, encodingRef.current)); writeStdout(decodeStdout(ev.data));
return; return;
} }
if (typeof Blob !== "undefined" && ev.data instanceof Blob) { if (typeof Blob !== "undefined" && ev.data instanceof Blob) {
void ev.data.arrayBuffer().then((buf) => { void ev.data.arrayBuffer().then((buf) => {
if (!isActiveSocket()) return; if (!isActiveSocket()) return;
writeStdout(decodeBytes(buf, encodingRef.current)); writeStdout(decodeStdout(buf));
}); });
return; return;
} }
@ -902,6 +936,12 @@ export const WebTerminal = forwardRef<WebTerminalHandle, Props>(function WebTerm
return () => { return () => {
cancelled = true; cancelled = true;
try {
webgl?.dispose();
} catch {
/* ignore */
}
webgl = null;
if (writeRaf) { if (writeRaf) {
cancelAnimationFrame(writeRaf); cancelAnimationFrame(writeRaf);
writeRaf = 0; writeRaf = 0;