diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index f4c2db9..60ec4a6 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -502,6 +502,21 @@ def normalize_audit_line(line: str) -> str: return s.replace("\r", "").rstrip() +def finalize_audit_line(line: str) -> str: + """Apply echoed backspaces then normalize (PTY stdout fallback only).""" + s = _strip_ansi(str(line or "")) + out: list[str] = [] + for ch in s: + if ch in ("\b", "\x7f"): + if out: + out.pop() + continue + if ord(ch) < 32 and ch != "\t": + continue + out.append(ch) + return normalize_audit_line("".join(out)) + + def _strip_ansi(text: str) -> str: return re.sub(r"\x1b\[[0-9;?]*[A-Za-z]|\x1b\].*?\x07|\x1b.", "", str(text or "")) @@ -537,11 +552,11 @@ def extract_last_prompt_command(text: str) -> str | None: tail = s.rsplit("\r", 1)[-1] tail_line = tail.split("\n")[-1].rstrip() if _is_prompt_command_line(tail_line): - return tail_line[:512] + return finalize_audit_line(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 finalize_audit_line(line)[:512] return None diff --git a/netx_api/webcrt_session_model.py b/netx_api/webcrt_session_model.py index 18d4ed2..f7625ba 100644 --- a/netx_api/webcrt_session_model.py +++ b/netx_api/webcrt_session_model.py @@ -372,13 +372,14 @@ class WebcrtSession: if "\r" in text or "\n" in text: from .webcrt_channel import extract_last_prompt_command, normalize_audit_line - cmd = extract_last_prompt_command(self._stdout_tail) or self._last_prompt_line - if not cmd and audit_line: + # Ground truth: xterm visible row at Enter (frontend). PTY stdout/stdin + # still carry intermediate backspaces / tab redraws — do not prefer those. + if audit_line and str(audit_line).strip(): cmd = normalize_audit_line(audit_line) - if cmd and str(cmd).strip(): - lines = [cmd] else: - lines = buf_lines + raw = extract_last_prompt_command(self._stdout_tail) or self._last_prompt_line + cmd = raw if raw else (buf_lines[-1] if buf_lines else None) + lines = [cmd] if cmd and str(cmd).strip() else [] self._last_prompt_line = "" else: lines = [] diff --git a/tests/test_webcrt_audit.py b/tests/test_webcrt_audit.py index 51eef31..a98ee48 100644 --- a/tests/test_webcrt_audit.py +++ b/tests/test_webcrt_audit.py @@ -12,6 +12,7 @@ from netx_api.webcrt_channel import ( looks_like_password_prompt, normalize_audit_line, extract_last_prompt_command, + finalize_audit_line, ) from netx_api.webcrt_session_model import WebcrtSession @@ -113,6 +114,12 @@ class ExtractPromptCommandTests(unittest.TestCase): ) +class FinalizeAuditLineTests(unittest.TestCase): + def test_applies_echoed_backspaces(self) -> None: + raw = "6150#disX\bplay version" + self.assertEqual(finalize_audit_line(raw), "6150#display version") + + class PasswordPromptTests(unittest.TestCase): def test_detects_password_prompt(self) -> None: self.assertTrue(looks_like_password_prompt("Password:")) @@ -208,8 +215,8 @@ class SessionCommandAuditTests(unittest.TestCase): self.assertFalse(kwargs["redacted"]) @patch("netx_api.webcrt_session_model._audit") - 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.""" + def test_audit_line_wins_over_stdout_backspaces(self, mock_audit: MagicMock) -> None: + """Frontend xterm row at Enter is authoritative; PTY stdout has edit noise.""" conn = MagicMock() conn.RETURN = "\n" conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) @@ -217,7 +224,7 @@ class SessionCommandAuditTests(unittest.TestCase): conn.write_channel = MagicMock() sess = WebcrtSession( - session_id="s-stdout", + session_id="s-xterm", ne_id="ne1", ne_name="lab", ne_ip="1.2.3.4", @@ -230,15 +237,12 @@ class SessionCommandAuditTests(unittest.TestCase): conn=conn, ) 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" + "AL5458-ACC-6120HS(config-if-loopback127)#ip address 11\b\b\bip ad\b\bip address 1.1.1.11 32" ) - sess.write_stdin("\r", audit_line="AL5458-ACC-6120HS(config-if-loopback127)#ip ad") + final = "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 255.255.255.255" + sess.write_stdin("\r", audit_line=final) kwargs = mock_audit.call_args.kwargs - self.assertEqual( - kwargs["command"], - "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 32", - ) + self.assertEqual(kwargs["command"], final) @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 0aec745..f204e0a 100644 --- a/web/src/components/WebTerminal.tsx +++ b/web/src/components/WebTerminal.tsx @@ -778,7 +778,11 @@ export const WebTerminal = forwardRef(function WebTerm e.preventDefault(); e.stopPropagation(); maybeFocus(); - sendStdinImmediate(data); + if (e.key === "Enter") { + sendStdinWithAudit("\r", auditLineForEnter(term)); + return; + } + sendStdinWithAudit(data); }; window.addEventListener("keydown", onKeyDownCapture, true);