diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index d0838a3..fb39f07 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -496,6 +496,12 @@ def looks_like_password_prompt(text: str) -> bool: return bool(_PASSWORD_PROMPT_RE.search(parts[-1])) +def normalize_audit_line(line: str) -> str: + """Normalize xterm-visible input line for audit (keep device prompt prefix).""" + s = re.sub(r"\x1b\[[0-9;?]*[A-Za-z]|\x1b\].*?\x07|\x1b.", "", str(line or "")) + return s.replace("\r", "").rstrip() + + def feed_command_line_buffer(buf: str, data: str, *, max_line: int = 512) -> tuple[str, list[str]]: """Accumulate stdin into completed command lines (Enter / CR / LF). @@ -505,22 +511,41 @@ def feed_command_line_buffer(buf: str, data: str, *, max_line: int = 512) -> tup cur = str(buf or "") completed: list[str] = [] limit = max(64, min(int(max_line or 512), 4096)) - for ch in str(data or ""): + raw = str(data or "") + i = 0 + while i < len(raw): + ch = raw[i] + if ch == "\x1b" and i + 1 < len(raw): + # Skip CSI / SS3 cursor-key sequences (Delete, arrows, etc.). + if raw[i + 1] == "[": + j = i + 2 + while j < len(raw) and raw[j] not in "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz~": + j += 1 + i = j + 1 if j < len(raw) else len(raw) + continue + if raw[i + 1] == "O" and i + 2 < len(raw): + i += 3 + continue if ch in ("\r", "\n"): if cur: completed.append(cur[:limit]) cur = "" + i += 1 continue if ch in ("\b", "\x7f"): cur = cur[:-1] if cur else "" + i += 1 continue if ch == "\x03": # Ctrl-C — abandon current line cur = "" + i += 1 continue if ord(ch) < 32 and ch != "\t": + i += 1 continue if len(cur) < limit: cur += ch + i += 1 return cur, completed diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index 0247732..18fcf20 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -693,16 +693,22 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: stop = asyncio.Event() stdin_buf: list[str] = [] + stdin_audit_line: str | None = None stdin_flush_task: asyncio.Task[None] | None = None async def flush_stdin() -> None: - nonlocal stdin_buf + nonlocal stdin_buf, stdin_audit_line if not stdin_buf: return data = "".join(stdin_buf) + audit_line = stdin_audit_line stdin_buf = [] + stdin_audit_line = None try: - await asyncio.get_running_loop().run_in_executor(webcrt_io_executor(), sess.write_stdin, data) + await asyncio.get_running_loop().run_in_executor( + webcrt_io_executor(), + lambda: sess.write_stdin(data, audit_line=audit_line), + ) except Exception as exc: await websocket.send_json( {"type": "status", "state": "error", "message": f"write_failed:{exc}"} @@ -825,6 +831,9 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: data = msg.get("data") if data is None: continue + audit_raw = msg.get("audit_line") + if audit_raw is not None and str(audit_raw).strip(): + stdin_audit_line = str(audit_raw).strip()[:512] stdin_buf.append(str(data)) # Coalesce high-frequency keystrokes briefly. if len(stdin_buf) >= 8: diff --git a/netx_api/webcrt_session_model.py b/netx_api/webcrt_session_model.py index ba6c8c2..767dc2d 100644 --- a/netx_api/webcrt_session_model.py +++ b/netx_api/webcrt_session_model.py @@ -310,12 +310,13 @@ class WebcrtSession: return "stale" return chunk # bytes | None - def write_stdin(self, data: str, *, audit_source: str = "stdin") -> None: + def write_stdin(self, data: str, *, audit_source: str = "stdin", audit_line: str | None = None) -> None: if self.closed or self.conn is None: raise RuntimeError("session_closed") text = str(data or "") if not text: return + audit_override = str(audit_line).strip() if audit_line is not None else None if self.cli_keymap: text = map_network_cli_keys( text, @@ -326,7 +327,7 @@ class WebcrtSession: text = map_network_cli_enter(text, self.conn) if not text: return - self._note_stdin_for_audit(text, source=audit_source) + self._note_stdin_for_audit(text, source=audit_source, audit_line=audit_override) with self._write_lock: # Prefer raw channel I/O for interactive typing (char echo / backspace). channel = getattr(self.conn, "remote_conn", None) @@ -354,13 +355,26 @@ class WebcrtSession: self.bytes_in += len(text) self.touch() - def _note_stdin_for_audit(self, text: str, *, source: str = "stdin") -> None: + def _note_stdin_for_audit( + self, + text: str, + *, + source: str = "stdin", + audit_line: str | None = None, + ) -> None: """Extract completed command lines from stdin and emit webcrt.command audits.""" with self._cmd_buf_lock: - self._cmd_buf, lines = feed_command_line_buffer(self._cmd_buf, text) + self._cmd_buf, buf_lines = feed_command_line_buffer(self._cmd_buf, text) redacted = bool(self._password_mode) - if redacted and lines: + if redacted and (buf_lines or audit_line): self._password_mode = False + if audit_line is not None and ("\r" in text or "\n" in text): + from .webcrt_channel import normalize_audit_line + + cmd = normalize_audit_line(audit_line) + lines = [cmd] if cmd.strip() else [] + else: + lines = buf_lines for cmd in lines: if not str(cmd).strip(): continue diff --git a/tests/test_webcrt_audit.py b/tests/test_webcrt_audit.py index cc0567e..ba24d16 100644 --- a/tests/test_webcrt_audit.py +++ b/tests/test_webcrt_audit.py @@ -10,6 +10,7 @@ from netx_api.webcrt_channel import ( _audit, feed_command_line_buffer, looks_like_password_prompt, + normalize_audit_line, ) from netx_api.webcrt_session_model import WebcrtSession @@ -55,6 +56,33 @@ class FeedCommandLineBufferTests(unittest.TestCase): self.assertEqual(len(lines), 1) self.assertEqual(len(lines[0]), 512) + def test_ignores_delete_key_sequence(self) -> None: + buf, lines = feed_command_line_buffer("", "dis\x1b[3~play\n") + self.assertEqual(lines, ["display"]) + + +class NormalizeAuditLineTests(unittest.TestCase): + def test_keeps_zte_hash_prompt(self) -> None: + self.assertEqual( + normalize_audit_line("AL5458-ACC-6120HS#display version"), + "AL5458-ACC-6120HS#display version", + ) + + def test_keeps_huawei_angle_prompt(self) -> None: + self.assertEqual( + normalize_audit_line("display version"), + "display version", + ) + + def test_keeps_bracket_prompt(self) -> None: + self.assertEqual(normalize_audit_line("[6150]show run"), "[6150]show run") + + def test_strips_ansi(self) -> None: + self.assertEqual( + normalize_audit_line("\x1b[31m6150#show run\x1b[0m"), + "6150#show run", + ) + class PasswordPromptTests(unittest.TestCase): def test_detects_password_prompt(self) -> None: @@ -150,6 +178,32 @@ class SessionCommandAuditTests(unittest.TestCase): self.assertEqual(kwargs["ne_name"], "lab") self.assertFalse(kwargs["redacted"]) + @patch("netx_api.webcrt_session_model._audit") + def test_audit_line_overrides_stdin_buffer(self, mock_audit: MagicMock) -> None: + """Tab-completed command: xterm-visible line wins over stdin keystrokes.""" + conn = MagicMock() + conn.RETURN = "\n" + conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) + del conn.remote_conn.send + conn.write_channel = MagicMock() + + sess = WebcrtSession( + session_id="s-tab", + 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, + ) + sess.write_stdin("dis\t\r", audit_line="AL5458-ACC-6120HS#display version") + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["command"], "AL5458-ACC-6120HS#display version") + @patch("netx_api.webcrt_session_model._audit") def test_password_mode_redacts_command(self, mock_audit: MagicMock) -> None: conn = MagicMock() diff --git a/web/src/components/WebTerminal.tsx b/web/src/components/WebTerminal.tsx index 108ead8..a76194b 100644 --- a/web/src/components/WebTerminal.tsx +++ b/web/src/components/WebTerminal.tsx @@ -86,6 +86,32 @@ type Props = { initialOutput?: string; }; +function stripAnsi(text: string): string { + return text.replace(/\x1b\[[0-9;?]*[A-Za-z]|\x1b\][^\x07]*(\x07|$)|\x1b./g, ""); +} + +/** Visible xterm row on Enter — keep device prompt prefix, drop ANSI only. */ +export function normalizeAuditLine(line: string): string { + return stripAnsi(line).replace(/\r/g, "").replace(/\s+$/, ""); +} + +function currentCommandLine(term: Terminal): string { + const buf = term.buffer.active; + const line = buf.getLine(buf.cursorY); + if (!line) return ""; + return normalizeAuditLine(line.translateToString(true)); +} + +function auditLineForEnter(term: Terminal | null, explicitLine?: string): string | undefined { + if (explicitLine != null && explicitLine.trim()) { + const cmd = normalizeAuditLine(explicitLine); + return cmd.trim() ? cmd : undefined; + } + if (!term) return undefined; + const cmd = currentCommandLine(term); + return cmd.trim() ? cmd : undefined; +} + function serializeTerminal(term: Terminal): string { const buf = term.buffer.active; const lines: string[] = []; @@ -97,6 +123,16 @@ function serializeTerminal(term: Terminal): string { return lines.join("\n").replace(/\s+$/g, ""); } + const buf = term.buffer.active; + const lines: string[] = []; + for (let i = 0; i < buf.length; i += 1) { + const line = buf.getLine(i); + if (!line) continue; + lines.push(line.translateToString(true)); + } + return lines.join("\n").replace(/\s+$/g, ""); +} + /** Line-wise search — avoids serializing the full scrollback on every Find. */ function findBufferLine( term: Terminal, @@ -292,9 +328,18 @@ export const WebTerminal = forwardRef(function WebTerm } }; - const sendStdinImmediate = (data: string) => { + const sendStdinWithAudit = (data: string, explicitAuditLine?: string) => { if (!data) return; - sendJson({ type: "stdin", data }); + const payload: Record = { type: "stdin", data }; + if (data.includes("\r") || data.includes("\n")) { + const auditLine = auditLineForEnter(termRef.current, explicitAuditLine); + if (auditLine) payload.audit_line = auditLine.slice(0, 512); + } + sendJson(payload); + }; + + const sendStdinImmediate = (data: string) => { + sendStdinWithAudit(data); }; /** Wait until device echoes (stdout after sentAt) or maxMs elapses — whichever first. */ @@ -324,11 +369,11 @@ export const WebTerminal = forwardRef(function WebTerm * Line-by-line paste paced by device echo. * pasteDelayMs is a *maximum* wait per line; fast responses advance immediately. */ - const sendStdinThrottled = (data: string) => { + const sendStdinThrottled = (data: string, explicitAuditLine?: string) => { if (!data) return; const maxDelay = pasteDelayRef.current; if (maxDelay <= 0 || data.length < 8) { - sendStdinImmediate(data); + sendStdinWithAudit(data, explicitAuditLine); return; } pasteQueueRef.current = pasteQueueRef.current.then(async () => { @@ -340,7 +385,7 @@ export const WebTerminal = forwardRef(function WebTerm const line = lines[i]; const chunk = i < lines.length - 1 ? `${line}\r` : line; const sentAt = performance.now(); - if (chunk) sendStdinImmediate(chunk); + if (chunk) sendStdinWithAudit(chunk, line.trim() || undefined); setPasteStatus({ done: i + 1, total }); if (i < lines.length - 1) { await waitForEchoOrTimeout(maxDelay, sentAt); @@ -660,11 +705,16 @@ export const WebTerminal = forwardRef(function WebTerm const dataDisposable = term.onData((data) => { const normalized = data.replace(/\x7f/g, "\x08"); + // Capture visible line before Enter moves the cursor to the next row. + const auditLine = + normalized.includes("\r") || normalized.includes("\n") + ? auditLineForEnter(term) + : undefined; // Large pastes from xterm arrive as one onData blob. if (normalized.length > 32 || normalized.includes("\r") || normalized.includes("\n")) { - sendStdinThrottled(normalized); + sendStdinThrottled(normalized, auditLine); } else { - sendStdinImmediate(normalized); + sendStdinWithAudit(normalized, auditLine); } });