Audit WebCRT commands from device stdout after tab completion.

ZTE tab redraws the input line via carriage return; parse the last prompt line from PTY stdout on Enter instead of stdin keystrokes. Also remove remaining ?? in AuditPage.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-09-02 10:53:15 +08:00
parent a9c7e81793
commit 6ea14e6d5e
5 changed files with 110 additions and 13 deletions

View file

@ -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).

View file

@ -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