diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index fb39f07..f4c2db9 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -502,6 +502,49 @@ def normalize_audit_line(line: str) -> str: return s.replace("\r", "").rstrip() +def _strip_ansi(text: str) -> str: + return re.sub(r"\x1b\[[0-9;?]*[A-Za-z]|\x1b\].*?\x07|\x1b.", "", str(text or "")) + + +def _is_prompt_command_line(line: str) -> bool: + """True when line looks like ``hostname#command`` (non-empty command tail).""" + s = str(line or "").strip() + if not s: + return False + return bool( + re.match( + r"^(?:" + r"[\w.-]+(?:\([^)]+\))*[#>]\s*\S" + r"|<[^>]+>\s*\S" + r"|\[[^\]]+\]\s*\S" + r")", + s, + flags=re.I, + ) + ) + + +def extract_last_prompt_command(text: str) -> str | None: + """Last prompt+command line in PTY transcript (tab-complete redraw aware). + + Network devices often refresh the current input with ``\\r`` after tab; the + final segment after the last carriage return is the ground truth for audit. + """ + s = _strip_ansi(text) + if not s.strip(): + return None + # Prefer the tail after the last in-line refresh (tab completion / prompt rewrite). + tail = s.rsplit("\r", 1)[-1] + tail_line = tail.split("\n")[-1].rstrip() + if _is_prompt_command_line(tail_line): + return tail_line[:512] + for frag in reversed(re.split(r"[\r\n]+", s)): + line = frag.strip() + if line and _is_prompt_command_line(line): + return line[:512] + return None + + 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). diff --git a/netx_api/webcrt_session_model.py b/netx_api/webcrt_session_model.py index 767dc2d..18d4ed2 100644 --- a/netx_api/webcrt_session_model.py +++ b/netx_api/webcrt_session_model.py @@ -95,6 +95,7 @@ class WebcrtSession: _cmd_buf_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) _password_mode: bool = field(default=False, repr=False) _stdout_tail: str = field(default="", repr=False) + _last_prompt_line: str = field(default="", repr=False) def touch(self) -> None: self.last_activity = time.time() @@ -368,13 +369,19 @@ class WebcrtSession: redacted = bool(self._password_mode) 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 + if "\r" in text or "\n" in text: + from .webcrt_channel import extract_last_prompt_command, normalize_audit_line - cmd = normalize_audit_line(audit_line) - lines = [cmd] if cmd.strip() else [] + cmd = extract_last_prompt_command(self._stdout_tail) or self._last_prompt_line + if not cmd and audit_line: + cmd = normalize_audit_line(audit_line) + if cmd and str(cmd).strip(): + lines = [cmd] + else: + lines = buf_lines + self._last_prompt_line = "" else: - lines = buf_lines + lines = [] for cmd in lines: if not str(cmd).strip(): continue @@ -401,7 +408,12 @@ class WebcrtSession: if not chunk: return with self._cmd_buf_lock: - self._stdout_tail = (self._stdout_tail + chunk)[-4000:] + self._stdout_tail = (self._stdout_tail + chunk)[-8000:] + from .webcrt_channel import extract_last_prompt_command + + line = extract_last_prompt_command(self._stdout_tail) + if line: + self._last_prompt_line = line if looks_like_password_prompt(self._stdout_tail): self._password_mode = True diff --git a/tests/test_webcrt_audit.py b/tests/test_webcrt_audit.py index ba24d16..51eef31 100644 --- a/tests/test_webcrt_audit.py +++ b/tests/test_webcrt_audit.py @@ -11,6 +11,7 @@ from netx_api.webcrt_channel import ( feed_command_line_buffer, looks_like_password_prompt, normalize_audit_line, + extract_last_prompt_command, ) from netx_api.webcrt_session_model import WebcrtSession @@ -84,6 +85,34 @@ class NormalizeAuditLineTests(unittest.TestCase): ) +class ExtractPromptCommandTests(unittest.TestCase): + def test_tab_redraw_after_carriage_return(self) -> None: + """ZTE tab completion redraws the line with \\r — audit must use that tail.""" + transcript = ( + "AL5458-ACC-6120HS(config-if-loopback127)#ip ad" + "\rAL5458-ACC-6120HS(config-if-loopback127)#ip address " + "1.1.1.11 32" + ) + self.assertEqual( + extract_last_prompt_command(transcript), + "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 32", + ) + + def test_config_mode_interface(self) -> None: + transcript = "AL5458-ACC-6120HS(config)#interface loopback127" + self.assertEqual( + extract_last_prompt_command(transcript), + "AL5458-ACC-6120HS(config)#interface loopback127", + ) + + def test_show_partial_command(self) -> None: + transcript = "AL5458-ACC-6120HS(config-if-loopback127)#show th" + self.assertEqual( + extract_last_prompt_command(transcript), + "AL5458-ACC-6120HS(config-if-loopback127)#show th", + ) + + class PasswordPromptTests(unittest.TestCase): def test_detects_password_prompt(self) -> None: self.assertTrue(looks_like_password_prompt("Password:")) @@ -179,8 +208,8 @@ class SessionCommandAuditTests(unittest.TestCase): 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.""" + def test_stdout_prompt_line_wins_over_stdin_on_enter(self, mock_audit: MagicMock) -> None: + """Tab-completed command is in device stdout, not stdin keystrokes.""" conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) @@ -188,7 +217,7 @@ class SessionCommandAuditTests(unittest.TestCase): conn.write_channel = MagicMock() sess = WebcrtSession( - session_id="s-tab", + session_id="s-stdout", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", @@ -200,9 +229,16 @@ class SessionCommandAuditTests(unittest.TestCase): owner_username="bob", conn=conn, ) - sess.write_stdin("dis\t\r", audit_line="AL5458-ACC-6120HS#display version") + sess._note_stdout_for_audit( + "AL5458-ACC-6120HS(config-if-loopback127)#ip ad" + "\rAL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 32" + ) + sess.write_stdin("\r", audit_line="AL5458-ACC-6120HS(config-if-loopback127)#ip ad") kwargs = mock_audit.call_args.kwargs - self.assertEqual(kwargs["command"], "AL5458-ACC-6120HS#display version") + self.assertEqual( + kwargs["command"], + "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 32", + ) @patch("netx_api.webcrt_session_model._audit") def test_password_mode_redacts_command(self, mock_audit: MagicMock) -> None: diff --git a/web/src/components/WebTerminal.tsx b/web/src/components/WebTerminal.tsx index 2b616d6..0aec745 100644 --- a/web/src/components/WebTerminal.tsx +++ b/web/src/components/WebTerminal.tsx @@ -97,6 +97,12 @@ export function normalizeAuditLine(line: string): string { function currentCommandLine(term: Terminal): string { const buf = term.buffer.active; + for (let y = buf.cursorY; y >= Math.max(0, buf.cursorY - 3); y -= 1) { + const line = buf.getLine(y); + if (!line) continue; + const text = normalizeAuditLine(line.translateToString(true)); + if (/[#>\]]\s*\S/.test(text) || /#[^\s]/.test(text)) return text; + } const line = buf.getLine(buf.cursorY); if (!line) return ""; return normalizeAuditLine(line.translateToString(true)); diff --git a/web/src/pages/AuditPage.tsx b/web/src/pages/AuditPage.tsx index 16e21b4..43bea5e 100644 --- a/web/src/pages/AuditPage.tsx +++ b/web/src/pages/AuditPage.tsx @@ -87,8 +87,8 @@ export function auditSummary( : neIds.length; return t("audit.summary.neExecBatch", { n: String(Number.isFinite(targetCount) ? targetCount : 0), - ok: String(d.ok_count ?? 0), - fail: String(d.fail_count ?? 0), + ok: String(d.ok_count != null ? Number(d.ok_count) : 0), + fail: String(d.fail_count != null ? Number(d.fail_count) : 0), }); } if (action.startsWith("auth.")) {