diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index cbfd7eb..1503997 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -539,14 +539,49 @@ def _is_prompt_command_line(line: str) -> bool: ) +def _has_cli_prompt_prefix(line: str) -> bool: + s = normalize_audit_line(line) + return bool( + re.match( + r"^(?:" + r"[\w.-]+(?:\([^)]+\))*[#>]" + r"|>[\w.-]+" + r"|<[^>]+>" + r"|\[[^\]]+\]" + r")", + s, + flags=re.I, + ) + ) + + +def _is_device_output_line(line: str) -> bool: + """Device error/warning echo — never an operator-typed command.""" + s = normalize_audit_line(line).strip() + if not s: + return False + low = s.lower() + if low.startswith("%error") or low.startswith("%warning"): + return True + if "invalid input detected" in low: + return True + if re.match(r"^\^+\s*$", s): + return True + if re.match(r"^enter configuration commands", low): + return True + return False + + def is_auditable_command_line(line: str) -> bool: - """False for empty Enter (prompt only, no command text).""" + """False for empty Enter, device output, or lines without a CLI prompt prefix.""" s = normalize_audit_line(line) if not s.strip(): return False if _is_prompt_only_line(s): return False - return True + if _is_device_output_line(s): + return False + return _has_cli_prompt_prefix(s) def _is_prompt_only_line(line: str) -> bool: diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index 18fcf20..089053c 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -801,7 +801,10 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: if sess.needs_live_prompt: sess.needs_live_prompt = False try: - await asyncio.get_running_loop().run_in_executor(webcrt_io_executor(), sess.write_stdin, "\r") + await asyncio.get_running_loop().run_in_executor( + webcrt_io_executor(), + lambda: sess.write_stdin("\r", audit_source="prompt_sync"), + ) except Exception: _log.debug("webcrt live prompt sync failed session=%s", session_id, exc_info=True) try: diff --git a/netx_api/webcrt_session_model.py b/netx_api/webcrt_session_model.py index 0cfffa5..aeb9bd5 100644 --- a/netx_api/webcrt_session_model.py +++ b/netx_api/webcrt_session_model.py @@ -385,28 +385,26 @@ class WebcrtSession: if redacted and (buf_lines or audit_line): self._password_mode = False if "\r" in text or "\n" in text: - from .webcrt_channel import ( - extract_last_prompt_command, - is_auditable_command_line, - normalize_audit_line, - ) + from .webcrt_channel import is_auditable_command_line, normalize_audit_line - # Ground truth: xterm visible row at Enter (frontend). PTY stdout/stdin - # still carry intermediate backspaces / tab redraws — do not prefer those. + src = str(source or "stdin") cmd: str | None = None - if audit_line and str(audit_line).strip(): + if redacted and (audit_line or buf_lines): + lines = ["***"] + elif src == "prompt_sync": + lines = [] + elif audit_line and str(audit_line).strip(): candidate = normalize_audit_line(audit_line) if is_auditable_command_line(candidate): cmd = candidate - if not cmd: - raw = extract_last_prompt_command(self._stdout_tail) - if raw and is_auditable_command_line(raw): - cmd = raw - elif buf_lines and str(buf_lines[-1]).strip(): - last = str(buf_lines[-1]).strip() - if is_auditable_command_line(last): - cmd = last - lines = [cmd] if cmd else [] + lines = [cmd] if cmd else [] + elif src == "post_login": + if buf_lines and str(buf_lines[-1]).strip(): + cmd = str(buf_lines[-1]).strip() + lines = [cmd] if cmd else [] + else: + # Interactive WebCRT: only trust frontend audit_line at Enter. + lines = [] self._last_prompt_line = "" else: lines = [] diff --git a/tests/test_webcrt_audit.py b/tests/test_webcrt_audit.py index 2c88eb0..ddc0ae8 100644 --- a/tests/test_webcrt_audit.py +++ b/tests/test_webcrt_audit.py @@ -14,6 +14,7 @@ from netx_api.webcrt_channel import ( extract_last_prompt_command, finalize_audit_line, is_auditable_command_line, + _is_device_output_line, _is_prompt_only_line, ) from netx_api.webcrt_session_model import WebcrtSession @@ -133,6 +134,14 @@ class AuditableCommandLineTests(unittest.TestCase): self.assertFalse(_is_prompt_only_line(line)) self.assertTrue(is_auditable_command_line(line)) + def test_device_error_not_auditable(self) -> None: + line = "%Error 140303: Invalid input detected at '^' marker." + self.assertTrue(_is_device_output_line(line)) + self.assertFalse(is_auditable_command_line(line)) + + def test_plain_stdin_without_prompt_not_auditable(self) -> None: + self.assertFalse(is_auditable_command_line("display version")) + class PasswordPromptTests(unittest.TestCase): def test_detects_password_prompt(self) -> None: @@ -218,12 +227,12 @@ class SessionCommandAuditTests(unittest.TestCase): owner_username="bob", conn=conn, ) - sess.write_stdin("display version\r") + sess.write_stdin("\r", audit_line="6150#display version") mock_audit.assert_called() event = mock_audit.call_args[0][0] self.assertEqual(event, "command") kwargs = mock_audit.call_args.kwargs - self.assertEqual(kwargs["command"], "display version") + self.assertEqual(kwargs["command"], "6150#display version") self.assertEqual(kwargs["owner_username"], "bob") self.assertEqual(kwargs["ne_name"], "lab") self.assertFalse(kwargs["redacted"]) @@ -306,6 +315,55 @@ class SessionCommandAuditTests(unittest.TestCase): sess.write_stdin("\r", audit_line=line) self.assertEqual(mock_audit.call_count, 1) + @patch("netx_api.webcrt_session_model._audit") + def test_prompt_sync_enter_not_audited(self, mock_audit: MagicMock) -> None: + 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-sync", + ne_id="ne1", + ne_name="lab", + ne_ip="1.2.3.4", + protocol="ssh", + cols=80, + rows=24, + cli_keymap=False, + conn=conn, + ) + sess._last_prompt_line = "AL5458-ACC-6120HS#configure terminal" + sess._note_stdout_for_audit("AL5458-ACC-6120HS#configure terminal\r\n") + sess.write_stdin("\r", audit_source="prompt_sync") + mock_audit.assert_not_called() + + @patch("netx_api.webcrt_session_model._audit") + def test_device_error_audit_line_rejected(self, mock_audit: MagicMock) -> None: + 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-err", + ne_id="ne1", + ne_name="lab", + ne_ip="1.2.3.4", + protocol="ssh", + cols=80, + rows=24, + cli_keymap=False, + conn=conn, + ) + sess.write_stdin( + "\r", + audit_line="%Error 140303: Invalid input detected at '^' marker.", + ) + mock_audit.assert_not_called() + @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 2cdcfa9..56c4304 100644 --- a/web/src/components/WebTerminal.tsx +++ b/web/src/components/WebTerminal.tsx @@ -102,13 +102,55 @@ export function isPromptOnlyLine(line: string): boolean { return /^(?:[\w.-]+(?:\([^)]+\))*[#>]\s*|<[^>]+>\s*|\[[^\]]+\]\s*)$/.test(s); } +const CLI_PROMPT_PREFIX = /^(?:[\w.-]+(?:\([^)]+\))*[#>]|>[\w.-]+|<[^>]+>|\[[^\]]+\])/; + +export function hasCliPromptPrefix(line: string): boolean { + return CLI_PROMPT_PREFIX.test(normalizeAuditLine(line)); +} + +/** Device error/warning lines must never be audited as operator commands. */ +export function isDeviceOutputLine(line: string): boolean { + const s = normalizeAuditLine(line).trim(); + if (!s) return false; + const low = s.toLowerCase(); + if (low.startsWith("%error") || low.startsWith("%warning")) return true; + if (low.includes("invalid input detected")) return true; + if (/^\^+\s*$/.test(s)) return true; + if (low.startsWith("enter configuration commands")) return true; + return false; +} + +export function isAuditableCommandLine(line: string): boolean { + const s = normalizeAuditLine(line); + if (!s.trim() || isPromptOnlyLine(s) || isDeviceOutputLine(s)) return false; + return hasCliPromptPrefix(s); +} + function currentCommandLine(term: Terminal): string { const buf = term.buffer.active; - const cur = buf.getLine(buf.cursorY); - if (cur) { - return normalizeAuditLine(cur.translateToString(true)); + const y = buf.cursorY; + const rows: string[] = []; + let sawPrompt = false; + + for (let i = y; i >= Math.max(0, y - 5); i -= 1) { + const text = normalizeAuditLine(buf.getLine(i)?.translateToString(true) ?? ""); + if (!text.trim()) { + if (rows.length) break; + continue; + } + if (isDeviceOutputLine(text)) break; + rows.unshift(text); + if (hasCliPromptPrefix(text)) { + sawPrompt = true; + break; + } } - return ""; + + if (!sawPrompt && rows.length === 0) { + const cur = buf.getLine(y); + return cur ? normalizeAuditLine(cur.translateToString(true)) : ""; + } + return rows.join("").trimEnd(); } function auditLineForEnter(term: Terminal | null, explicitLine?: string): string | undefined { @@ -118,7 +160,7 @@ function auditLineForEnter(term: Terminal | null, explicitLine?: string): string : term ? currentCommandLine(term) : ""; - if (!raw.trim() || isPromptOnlyLine(raw)) return undefined; + if (!isAuditableCommandLine(raw)) return undefined; return raw; }