Fix log audit record bug

This commit is contained in:
oliver 2026-09-03 14:19:51 +08:00
parent 4295b6d677
commit 3612dd6e45
2 changed files with 136 additions and 1 deletions

View file

@ -670,6 +670,11 @@ def _stdout_has_inplace_edit(text: str) -> bool:
Only inspects the last fragment (live input line). Older history-edit CSI still Only inspects the last fragment (live input line). Older history-edit CSI still
sitting in ``stdout_tail`` must not make bare Enter look like an in-place edit. sitting in ``stdout_tail`` must not make bare Enter look like an in-place edit.
Excludes the common ``---- More ----`` wipe (``ESC[16D``). Excludes the common ``---- More ----`` wipe (``ESC[16D``).
Huawei rewrites the input row with CSI cursor moves (``ESC[<n>D/C/P/@``). ZTE
instead erases-to-EOL (``ESC[K``) and reprints, so the ``[DCP@]`` scan alone
misses ZTE history recalls — also treat a live prompt+command fragment that
carries ``ESC[K`` as an in-place redraw.
""" """
s = str(text or "")[-4000:] s = str(text or "")[-4000:]
if not s: if not s:
@ -694,6 +699,11 @@ def _stdout_has_inplace_edit(text: str) -> bool:
return True return True
if cmd == "D" and n != 16: if cmd == "D" and n != 16:
return True return True
# ZTE redraws the live input row with ESC[K (erase-to-EOL) + reprint / backspace
# rather than CSI cursor moves. A prompt+command fragment carrying ESC[K is an
# in-place edit (a trailing prompt cleanup alone renders as prompt-only above).
if "\x1b[K" in frag and _command_tail(rendered).strip():
return True
return False return False
@ -868,7 +878,18 @@ def render_pty_line(text: str) -> str:
cursor = 0 cursor = 0
i += 1 i += 1
continue continue
if ch in ("\b", "\x7f"): if ch == "\b":
# Backspace (0x08) in PTY output is cursor-left WITHOUT erasing —
# ZTE line redraws move the cursor via long backspace runs then
# overwrite/ESC[K to repaint. A destructive delete here mangles
# the recalled command text, so render_pty_line ends up with a
# fragment ("p") instead of the full "show ip interface brief".
if cursor > 0:
cursor -= 1
i += 1
continue
if ch == "\x7f":
# DEL (0x7f) deletes the cell to the left of the cursor.
if cursor > 0: if cursor > 0:
cursor -= 1 cursor -= 1
if cursor < len(cells): if cursor < len(cells):
@ -1066,6 +1087,15 @@ def pick_audit_command(
return echo_n return echo_n
if len(et) == len(bt) and et != bt: if len(et) == len(bt) and et != bt:
return echo_n return echo_n
# Echo longer with the xterm hint as a token-prefix of the fully
# redrawn device line: the xterm visible row lagged a partial prefix
# at Enter (ZTE reprints via ESC[K, so the snapshot can still read
# ``show ip`` while the device already shows ``show ip interface
# brief``). Trust the complete device echo.
if len(et) > len(bt) and all(
et[i].lower() == bt[i].lower() for i in range(len(bt))
):
return echo_n
return base_n return base_n
return base_n return base_n

View file

@ -286,6 +286,111 @@ class ResolveAuditCommandsTests(unittest.TestCase):
) )
self.assertEqual(cmd, "[~r1]display ip interface brief") self.assertEqual(cmd, "[~r1]display ip interface brief")
def test_zte_history_insert_partial_hint_prefers_echo(self) -> None:
"""ZTE redraws via ESC[K + backspace (no CSI cursor moves).
Up-arrow recalls ``show interface brief``, the operator inserts ``ip`` so
the device redraws ``show ip interface brief``. The xterm visible row can
lag a partial ``show ip`` snapshot at Enter — the full device echo must win.
"""
from netx_api.webcrt_channel import _stdout_has_inplace_edit
prompt = "AL5458-ACC-6120HS#"
# ZTE recall + ESC[K reprint of the fully-edited line.
stdout = (
"\r\r\n" + prompt + "\x1b[Kshow interface brief"
+ "\x08\x08\x08\x1b[Kshow interface brief"
+ "\r" + prompt + "\x1b[Kshow ip interface brief"
)
# ZTE redraw carries no [DCP@] CSI — without the ESC[K path this is False.
self.assertTrue(_stdout_has_inplace_edit(stdout))
# Lagging partial xterm hint; device echo is the full line.
cmd = pick_audit_command(
"ip",
prompt + "show ip",
prompt_hint=prompt + "show interface brief",
stdout_tail=stdout,
source="stdin",
)
self.assertEqual(cmd, prompt + "show ip interface brief")
# Same even when the frontend managed to send the full hint.
cmd2 = pick_audit_command(
"ip",
prompt + "show ip interface brief",
prompt_hint=prompt + "show interface brief",
stdout_tail=stdout,
source="stdin",
)
self.assertEqual(cmd2, prompt + "show ip interface brief")
# Resolve path (Enter flush) must also pick the full line.
out = resolve_audit_commands(
["ip"],
audit_line=prompt + "show ip",
prompt_hint=prompt + "show interface brief",
stdout_tail=stdout,
source="stdin",
)
self.assertEqual(out, [prompt + "show ip interface brief"])
def test_zte_real_backspace_redraw_reconstructs_full_command(self) -> None:
"""Real ZTE device redraws via backspace runs (not CSI cursor moves).
Up-arrow recalls ``show interface brief``, left-arrows move the cursor
after ``show``, then ``ip`` is inserted. The device redraws using
long ``\\x08`` backspace sequences + ``ESC[K`` reprints — it never
echoes the full ``show ip interface brief`` as one contiguous string.
``render_pty_line`` must treat ``\\x08`` as non-destructive cursor-left
(standard terminal behaviour) so the recalled text survives the redraw
and the full edited line is reconstructed.
"""
import os
from netx_api.webcrt_channel import (
_live_input_line,
_stdout_has_inplace_edit,
extract_last_prompt_command,
render_pty_line,
)
fixture = os.path.join(os.path.dirname(__file__), "fixtures",
"zte_history_edit_stdout.txt")
with open(fixture, encoding="utf-8") as f:
stdout = f.read()
prompt = "AL5458-ACC-6120HS#"
expected = prompt + "show ip interface brief"
# render_pty_line must reconstruct the full edited command.
self.assertEqual(render_pty_line(stdout.split("\n")[-1]), expected)
# extract_last_prompt_command must find the full line.
self.assertEqual(extract_last_prompt_command(stdout), expected)
# Live input line must show the full edited command.
self.assertEqual(_live_input_line(stdout), expected)
# In-place edit detected (ESC[K redraw).
self.assertTrue(_stdout_has_inplace_edit(stdout))
# Even with a lagging partial xterm hint ("show ip"), the full device
# echo must win.
cmd = pick_audit_command(
"ip",
prompt + "show ip",
prompt_hint=prompt + "show interface brief",
stdout_tail=stdout,
source="stdin",
)
self.assertEqual(cmd, expected)
# Resolve path (Enter flush) must also pick the full line.
out = resolve_audit_commands(
["ip"],
audit_line=prompt + "show ip",
prompt_hint=prompt + "show interface brief",
stdout_tail=stdout,
source="stdin",
)
self.assertEqual(out, [expected])
def test_history_midline_delete_prefers_device_echo(self) -> None: def test_history_midline_delete_prefers_device_echo(self) -> None:
"""Up-arrow then delete middle ``ip`` — must not keep stale longer audit_line.""" """Up-arrow then delete middle ``ip`` — must not keep stale longer audit_line."""
frag = ( frag = (