From 4295b6d677b0ba5a2319b8c426714de507efed73 Mon Sep 17 00:00:00 2001 From: oliver Date: Wed, 2 Sep 2026 22:02:52 +0800 Subject: [PATCH] Add WebCRT R1 audit e2e helper that reads credentials from the environment. Co-authored-by: Cursor --- scripts/e2e_webcrt_audit_r1.py | 282 +++++++++++++++++++++++++++++++++ 1 file changed, 282 insertions(+) create mode 100644 scripts/e2e_webcrt_audit_r1.py diff --git a/scripts/e2e_webcrt_audit_r1.py b/scripts/e2e_webcrt_audit_r1.py new file mode 100644 index 0000000..8388429 --- /dev/null +++ b/scripts/e2e_webcrt_audit_r1.py @@ -0,0 +1,282 @@ +#!/usr/bin/env python3 +"""E2E WebCRT audit test against local netx + a lab device (default R1). + +Credentials and target come from the environment (no secrets in-repo):: + + set NETX_PASSWORD=... + set NETX_BASE=http://127.0.0.1:8890 + set NETX_USER=admin + set NETX_NE_IP=192.168.0.127 + set NETX_NE_ID=117df1e739d84b46859d425414b07c09 + python scripts/e2e_webcrt_audit_r1.py +""" + +from __future__ import annotations + +import asyncio +import json +import os +import sys +import time +from typing import Any + +import requests + +try: + import websockets +except ImportError: + print("pip install websockets", file=sys.stderr) + raise + +BASE = os.environ.get("NETX_BASE", "http://127.0.0.1:8890").rstrip("/") +R1_IP = os.environ.get("NETX_NE_IP", "192.168.0.127") +R1_NE_ID = os.environ.get("NETX_NE_ID", "117df1e739d84b46859d425414b07c09") +USER = os.environ.get("NETX_USER", "admin") +PWD = os.environ.get("NETX_PASSWORD") or os.environ.get("NETX_PWD", "") + + +def login() -> str: + r = requests.post( + f"{BASE}/v1/auth/login", + json={"username": USER, "password": PWD}, + timeout=15, + ) + r.raise_for_status() + return str(r.json()["access_token"]) + + +def headers(tok: str) -> dict[str, str]: + return {"Authorization": f"Bearer {tok}"} + + +def create_session(tok: str, ne_id: str) -> dict[str, Any]: + r = requests.post( + f"{BASE}/v1/webcrt/sessions", + headers=headers(tok), + json={"ne_id": ne_id, "cols": 120, "rows": 40, "async_connect": True}, + timeout=30, + ) + r.raise_for_status() + return r.json() + + +def ws_ticket(tok: str) -> str: + r = requests.post(f"{BASE}/v1/webcrt/ws-ticket", headers=headers(tok), timeout=10) + r.raise_for_status() + return str(r.json()["ticket"]) + + +def recent_commands(tok: str, session_id: str) -> list[str]: + r = requests.get( + f"{BASE}/v1/audit-logs", + headers=headers(tok), + params={"action": "webcrt.command", "limit": 40}, + timeout=15, + ) + r.raise_for_status() + out: list[str] = [] + for row in r.json().get("items", []): + detail = row.get("detail") or {} + if str(detail.get("session_id") or "") != session_id and str(row.get("path") or ""): + # Prefer session filter via detail when present + pass + if str(detail.get("session_id") or "") and str(detail.get("session_id")) != session_id: + continue + # Also accept jsonl-sourced detail without session when command looks recent + cmd = str(detail.get("command") or "").strip() + if cmd: + out.append(cmd) + # Fallback: read audit.jsonl for this session + if not out: + try: + with open("data/webcrt/audit.jsonl", encoding="utf-8") as fh: + for line in fh: + try: + obj = json.loads(line) + except Exception: + continue + if obj.get("event") != "command": + continue + if obj.get("session_id") != session_id: + continue + cmd = str(obj.get("command") or "").strip() + if cmd: + out.append(cmd) + except OSError: + pass + return out + + +async def ws_run(session_id: str, ticket: str) -> str: + url = f"{BASE.replace('http://', 'ws://').replace('https://', 'wss://')}/v1/webcrt/sessions/{session_id}/ws?ws_ticket={ticket}" + transcript = "" + last_prompt = "" + + async with websockets.connect(url, max_size=8 * 1024 * 1024) as ws: + + async def recv_until_prompt(timeout: float = 20.0) -> None: + nonlocal transcript, last_prompt + deadline = time.time() + timeout + saw_ready = False + while time.time() < deadline: + try: + raw = await asyncio.wait_for(ws.recv(), timeout=1.0) + except asyncio.TimeoutError: + if "" in transcript or "[~r1]" in transcript or "[r1]" in transcript: + return + if saw_ready and time.time() + 3 > deadline: + return + continue + if isinstance(raw, bytes): + chunk = raw.decode("utf-8", errors="replace") + else: + msg = json.loads(raw) + if msg.get("type") == "status" and msg.get("state") == "error": + raise RuntimeError(f"session error: {msg.get('message')}") + if msg.get("type") == "status" and msg.get("state") == "ready": + saw_ready = True + continue + if msg.get("type") != "stdout": + continue + chunk = str(msg.get("data") or "") + transcript += chunk + plain = chunk.replace("\r", "\n") + for ln in plain.split("\n"): + s = ln.strip() + if s.startswith("") or s.startswith("[~r1]") or s.startswith("[r1]"): + last_prompt = s + if "" in transcript or "[~r1]" in transcript or "[r1]" in transcript: + return + + async def drain(ms: float = 0.6) -> None: + nonlocal transcript, last_prompt + end = time.time() + ms + while time.time() < end: + try: + raw = await asyncio.wait_for(ws.recv(), timeout=0.15) + except asyncio.TimeoutError: + continue + if isinstance(raw, bytes): + chunk = raw.decode("utf-8", errors="replace") + else: + msg = json.loads(raw) + if msg.get("type") != "stdout": + continue + chunk = str(msg.get("data") or "") + transcript += chunk + for ln in chunk.replace("\r", "\n").split("\n"): + s = ln.strip() + if s.startswith("") or s.startswith("[~r1]") or s.startswith("[r1]"): + last_prompt = s + + async def send_stdin(data: str, audit_line: str | None = None) -> None: + payload: dict[str, Any] = {"type": "stdin", "data": data} + if audit_line: + payload["audit_line"] = audit_line[:512] + await ws.send(json.dumps(payload)) + + await recv_until_prompt(45.0) + await drain(1.5) + if "" not in transcript and "[~r1]" not in transcript and "[r1]" not in transcript: + # Nudge device for prompt (same as UI prompt_sync Enter). + await send_stdin("\r") + await drain(2.0) + if "" not in transcript and "[~r1]" not in transcript and "[r1]" not in transcript: + raise RuntimeError(f"no R1 prompt in transcript tail: {transcript[-500:]!r}") + + # --- Case 1: Tab completion with frontend audit_line (xterm ground truth) --- + await send_stdin("dis ip int\t") + await drain(1.2) + visible = last_prompt if "display" in last_prompt.lower() or "interface" in last_prompt.lower() else "display ip interface brief" + # Huawei often expands to "display ip interface brief" + if "display" not in visible.lower(): + # keep whatever device showed on the prompt line + visible = last_prompt or visible + await send_stdin("\r", visible) + await drain(2.0) + + # --- Case 2: Backspace/insert-style edit — wrong stdin, correct audit_line --- + # Simulate operator seeing "display version" after fixing typos. + await send_stdin("disX\x08play version\r", "display version") + await drain(2.0) + + # --- Case 3: Mid-line insert disagreement — audit_line must win --- + await send_stdin("dis version\r", "display version") + await drain(2.0) + + # --- Case 4: Tab without audit_line — rely on device echo / prompt_hint --- + await send_stdin("dis ll n b\t") + await drain(1.5) + await send_stdin("\r") # no audit_line + await drain(2.5) + + await ws.send(json.dumps({"type": "close"})) + return transcript + + +def main() -> int: + if not PWD: + print( + "Set NETX_PASSWORD (or NETX_PWD) before running this script.", + file=sys.stderr, + ) + return 2 + print("login...") + tok = login() + ne_id = R1_NE_ID + print(f"ne_id={ne_id} ip={R1_IP}") + + sess = create_session(tok, ne_id) + sid = str(sess["session_id"]) + print(f"session={sid} ne_ip={sess.get('ne_ip') or sess.get('ip')}") + + ticket = ws_ticket(tok) + print("websocket test...") + transcript = asyncio.run(ws_run(sid, ticket)) + print("transcript_tail:", repr(transcript[-300:])) + + time.sleep(1.0) + cmds = recent_commands(tok, sid) + # Also always print jsonl session commands + jsonl_cmds: list[str] = [] + try: + with open("data/webcrt/audit.jsonl", encoding="utf-8") as fh: + for line in fh: + try: + obj = json.loads(line) + except Exception: + continue + if obj.get("event") == "command" and obj.get("session_id") == sid: + jsonl_cmds.append(str(obj.get("command") or "")) + except OSError: + pass + cmds = jsonl_cmds or cmds + + print("\n=== session webcrt.command audits ===") + for c in cmds: + print(repr(c)) + + failures: list[str] = [] + joined = "\n".join(cmds) + if "\t" in joined: + failures.append("literal Tab leaked into audit") + if "dislayplay" in joined or "disX" in joined: + failures.append("backspace/edit corruption recorded") + if not any("display ip interface" in c.lower() or "ip interface brief" in c.lower() for c in cmds): + failures.append("tab+audit_line did not record display ip interface brief") + if not any("display version" in c for c in cmds): + failures.append("display version not audited correctly") + if any(c.strip() in ("dis version", "dis version") for c in cmds): + failures.append("audit_line ignored for insert/disagreement case") + if not any("dis ll n" in c for c in cmds): + failures.append("missing dis ll n b audit") + + if failures: + print("\nFAIL:", "; ".join(failures), file=sys.stderr) + return 2 + print("\nOK: audit scenarios look correct") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())