mirror of
https://github.com/hansjone/netx.git
synced 2026-10-09 00:50:46 +08:00
Skip WebCRT audit for empty Enter on prompt-only lines.
Do not reuse the previous command when the operator presses Enter on a blank prompt; validate prompt-only rows on both frontend and backend. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
parent
b35e56afb0
commit
7ca57278fe
4 changed files with 109 additions and 21 deletions
|
|
@ -539,6 +539,34 @@ def _is_prompt_command_line(line: str) -> bool:
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def is_auditable_command_line(line: str) -> bool:
|
||||||
|
"""False for empty Enter (prompt only, no command text)."""
|
||||||
|
s = normalize_audit_line(line)
|
||||||
|
if not s.strip():
|
||||||
|
return False
|
||||||
|
if _is_prompt_only_line(s):
|
||||||
|
return False
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def _is_prompt_only_line(line: str) -> bool:
|
||||||
|
"""True when line is a device prompt with no command typed."""
|
||||||
|
s = normalize_audit_line(line)
|
||||||
|
if not s:
|
||||||
|
return True
|
||||||
|
return bool(
|
||||||
|
re.match(
|
||||||
|
r"^(?:"
|
||||||
|
r"[\w.-]+(?:\([^)]+\))*[#>]\s*"
|
||||||
|
r"|<[^>]+>\s*"
|
||||||
|
r"|\[[^\]]+\]\s*"
|
||||||
|
r")$",
|
||||||
|
s,
|
||||||
|
flags=re.I,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def extract_last_prompt_command(text: str) -> str | None:
|
def extract_last_prompt_command(text: str) -> str | None:
|
||||||
"""Last prompt+command line in PTY transcript (tab-complete redraw aware).
|
"""Last prompt+command line in PTY transcript (tab-complete redraw aware).
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -370,16 +370,28 @@ class WebcrtSession:
|
||||||
if redacted and (buf_lines or audit_line):
|
if redacted and (buf_lines or audit_line):
|
||||||
self._password_mode = False
|
self._password_mode = False
|
||||||
if "\r" in text or "\n" in text:
|
if "\r" in text or "\n" in text:
|
||||||
from .webcrt_channel import extract_last_prompt_command, normalize_audit_line
|
from .webcrt_channel import (
|
||||||
|
extract_last_prompt_command,
|
||||||
|
is_auditable_command_line,
|
||||||
|
normalize_audit_line,
|
||||||
|
)
|
||||||
|
|
||||||
# Ground truth: xterm visible row at Enter (frontend). PTY stdout/stdin
|
# Ground truth: xterm visible row at Enter (frontend). PTY stdout/stdin
|
||||||
# still carry intermediate backspaces / tab redraws — do not prefer those.
|
# still carry intermediate backspaces / tab redraws — do not prefer those.
|
||||||
|
cmd: str | None = None
|
||||||
if audit_line and str(audit_line).strip():
|
if audit_line and str(audit_line).strip():
|
||||||
cmd = normalize_audit_line(audit_line)
|
candidate = normalize_audit_line(audit_line)
|
||||||
else:
|
if is_auditable_command_line(candidate):
|
||||||
raw = extract_last_prompt_command(self._stdout_tail) or self._last_prompt_line
|
cmd = candidate
|
||||||
cmd = raw if raw else (buf_lines[-1] if buf_lines else None)
|
if not cmd:
|
||||||
lines = [cmd] if cmd and str(cmd).strip() else []
|
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 []
|
||||||
self._last_prompt_line = ""
|
self._last_prompt_line = ""
|
||||||
else:
|
else:
|
||||||
lines = []
|
lines = []
|
||||||
|
|
@ -415,6 +427,12 @@ class WebcrtSession:
|
||||||
line = extract_last_prompt_command(self._stdout_tail)
|
line = extract_last_prompt_command(self._stdout_tail)
|
||||||
if line:
|
if line:
|
||||||
self._last_prompt_line = line
|
self._last_prompt_line = line
|
||||||
|
else:
|
||||||
|
from .webcrt_channel import _is_prompt_only_line, _strip_ansi
|
||||||
|
|
||||||
|
tail = _strip_ansi(self._stdout_tail).rsplit("\r", 1)[-1].split("\n")[-1]
|
||||||
|
if _is_prompt_only_line(tail):
|
||||||
|
self._last_prompt_line = ""
|
||||||
if looks_like_password_prompt(self._stdout_tail):
|
if looks_like_password_prompt(self._stdout_tail):
|
||||||
self._password_mode = True
|
self._password_mode = True
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,8 @@ from netx_api.webcrt_channel import (
|
||||||
normalize_audit_line,
|
normalize_audit_line,
|
||||||
extract_last_prompt_command,
|
extract_last_prompt_command,
|
||||||
finalize_audit_line,
|
finalize_audit_line,
|
||||||
|
is_auditable_command_line,
|
||||||
|
_is_prompt_only_line,
|
||||||
)
|
)
|
||||||
from netx_api.webcrt_session_model import WebcrtSession
|
from netx_api.webcrt_session_model import WebcrtSession
|
||||||
|
|
||||||
|
|
@ -120,6 +122,18 @@ class FinalizeAuditLineTests(unittest.TestCase):
|
||||||
self.assertEqual(finalize_audit_line(raw), "6150#display version")
|
self.assertEqual(finalize_audit_line(raw), "6150#display version")
|
||||||
|
|
||||||
|
|
||||||
|
class AuditableCommandLineTests(unittest.TestCase):
|
||||||
|
def test_prompt_only_not_auditable(self) -> None:
|
||||||
|
line = "AL5458-ACC-6120HS(config-if-loopback127)#"
|
||||||
|
self.assertTrue(_is_prompt_only_line(line))
|
||||||
|
self.assertFalse(is_auditable_command_line(line))
|
||||||
|
|
||||||
|
def test_command_with_prompt_is_auditable(self) -> None:
|
||||||
|
line = "AL5458-ACC-6120HS(config-if-loopback127)#ip address 1.1.1.11 255.255.255.255"
|
||||||
|
self.assertFalse(_is_prompt_only_line(line))
|
||||||
|
self.assertTrue(is_auditable_command_line(line))
|
||||||
|
|
||||||
|
|
||||||
class PasswordPromptTests(unittest.TestCase):
|
class PasswordPromptTests(unittest.TestCase):
|
||||||
def test_detects_password_prompt(self) -> None:
|
def test_detects_password_prompt(self) -> None:
|
||||||
self.assertTrue(looks_like_password_prompt("Password:"))
|
self.assertTrue(looks_like_password_prompt("Password:"))
|
||||||
|
|
@ -244,6 +258,30 @@ class SessionCommandAuditTests(unittest.TestCase):
|
||||||
kwargs = mock_audit.call_args.kwargs
|
kwargs = mock_audit.call_args.kwargs
|
||||||
self.assertEqual(kwargs["command"], final)
|
self.assertEqual(kwargs["command"], final)
|
||||||
|
|
||||||
|
@patch("netx_api.webcrt_session_model._audit")
|
||||||
|
def test_empty_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-empty",
|
||||||
|
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(config-if-loopback127)#ip address 1.1.1.11 255.255.255.255"
|
||||||
|
sess._note_stdout_for_audit("AL5458-ACC-6120HS(config-if-loopback127)#\r\n")
|
||||||
|
sess.write_stdin("\r", audit_line="AL5458-ACC-6120HS(config-if-loopback127)#")
|
||||||
|
mock_audit.assert_not_called()
|
||||||
|
|
||||||
@patch("netx_api.webcrt_session_model._audit")
|
@patch("netx_api.webcrt_session_model._audit")
|
||||||
def test_password_mode_redacts_command(self, mock_audit: MagicMock) -> None:
|
def test_password_mode_redacts_command(self, mock_audit: MagicMock) -> None:
|
||||||
conn = MagicMock()
|
conn = MagicMock()
|
||||||
|
|
|
||||||
|
|
@ -95,27 +95,31 @@ export function normalizeAuditLine(line: string): string {
|
||||||
return stripAnsi(line).replace(/\r/g, "").replace(/\s+$/, "");
|
return stripAnsi(line).replace(/\r/g, "").replace(/\s+$/, "");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** True when the row is only a device prompt (empty Enter — not auditable). */
|
||||||
|
export function isPromptOnlyLine(line: string): boolean {
|
||||||
|
const s = normalizeAuditLine(line);
|
||||||
|
if (!s) return true;
|
||||||
|
return /^(?:[\w.-]+(?:\([^)]+\))*[#>]\s*|<[^>]+>\s*|\[[^\]]+\]\s*)$/.test(s);
|
||||||
|
}
|
||||||
|
|
||||||
function currentCommandLine(term: Terminal): string {
|
function currentCommandLine(term: Terminal): string {
|
||||||
const buf = term.buffer.active;
|
const buf = term.buffer.active;
|
||||||
for (let y = buf.cursorY; y >= Math.max(0, buf.cursorY - 3); y -= 1) {
|
const cur = buf.getLine(buf.cursorY);
|
||||||
const line = buf.getLine(y);
|
if (cur) {
|
||||||
if (!line) continue;
|
return normalizeAuditLine(cur.translateToString(true));
|
||||||
const text = normalizeAuditLine(line.translateToString(true));
|
|
||||||
if (/[#>\]]\s*\S/.test(text) || /#[^\s]/.test(text)) return text;
|
|
||||||
}
|
}
|
||||||
const line = buf.getLine(buf.cursorY);
|
return "";
|
||||||
if (!line) return "";
|
|
||||||
return normalizeAuditLine(line.translateToString(true));
|
|
||||||
}
|
}
|
||||||
|
|
||||||
function auditLineForEnter(term: Terminal | null, explicitLine?: string): string | undefined {
|
function auditLineForEnter(term: Terminal | null, explicitLine?: string): string | undefined {
|
||||||
if (explicitLine != null && explicitLine.trim()) {
|
const raw =
|
||||||
const cmd = normalizeAuditLine(explicitLine);
|
explicitLine != null && explicitLine.trim()
|
||||||
return cmd.trim() ? cmd : undefined;
|
? normalizeAuditLine(explicitLine)
|
||||||
}
|
: term
|
||||||
if (!term) return undefined;
|
? currentCommandLine(term)
|
||||||
const cmd = currentCommandLine(term);
|
: "";
|
||||||
return cmd.trim() ? cmd : undefined;
|
if (!raw.trim() || isPromptOnlyLine(raw)) return undefined;
|
||||||
|
return raw;
|
||||||
}
|
}
|
||||||
|
|
||||||
function serializeTerminal(term: Terminal): string {
|
function serializeTerminal(term: Terminal): string {
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue