mirror of
https://github.com/hansjone/netx.git
synced 2026-10-15 04:53:20 +08:00
oliver
45de8144a3
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>
525 lines
21 KiB
Python
525 lines
21 KiB
Python
"""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
|