From 8915455b528710d30c4369dc27ba06dae239b386 Mon Sep 17 00:00:00 2001 From: oliver Date: Tue, 1 Sep 2026 21:36:44 +0800 Subject: [PATCH] Record WebCRT device login/commands and NE exec in operation audit. Make session lifecycle and typed CLI lines searchable in audit_log with password redaction, and surface summaries plus device filters in the ops UI. Co-authored-by: Cursor --- netx_api/audit_async.py | 10 +- netx_api/managed_ne_router.py | 82 ++++++++++- netx_api/webcrt_channel.py | 111 +++++++++++++++ netx_api/webcrt_session_model.py | 54 +++++++- netx_api/webcrt_session_registry.py | 13 ++ tests/test_ne_exec_audit.py | 78 +++++++++++ tests/test_webcrt.py | 6 +- tests/test_webcrt_audit.py | 208 ++++++++++++++++++++++++++++ web/src/i18n/en.ts | 18 +++ web/src/i18n/zh.ts | 18 +++ web/src/index.css | 46 ++++++ web/src/pages/AuditPage.tsx | 179 ++++++++++++++++++++---- 12 files changed, 783 insertions(+), 40 deletions(-) create mode 100644 tests/test_ne_exec_audit.py create mode 100644 tests/test_webcrt_audit.py diff --git a/netx_api/audit_async.py b/netx_api/audit_async.py index dabf289..85ecb08 100644 --- a/netx_api/audit_async.py +++ b/netx_api/audit_async.py @@ -99,8 +99,14 @@ def enqueue_audit( ) -> None: global _dropped act = str(action or "") - # Always persist auth / security-relevant events. - if act.startswith("auth.") or act.startswith("users.") or act.startswith("api_tokens.") or act.startswith("webcrt."): + # Always persist auth / security / device-op events. + if ( + act.startswith("auth.") + or act.startswith("users.") + or act.startswith("api_tokens.") + or act.startswith("webcrt.") + or act.startswith("ne.") + ): pass elif act.startswith("http.") and not _sample_ok(): return diff --git a/netx_api/managed_ne_router.py b/netx_api/managed_ne_router.py index be14851..4ae5bf1 100644 --- a/netx_api/managed_ne_router.py +++ b/netx_api/managed_ne_router.py @@ -1,9 +1,13 @@ from __future__ import annotations -from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile +from typing import Annotated + +from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile from fastapi.responses import Response from sqlalchemy.orm import Session +from .auth_deps import AuthContext, require_user +from .auth_service import write_audit from .db import get_db from .device_types import SUPPORTED_VENDORS from .ne_connect import schedule_connect_tests @@ -39,6 +43,16 @@ from .models import ManagedNE router = APIRouter(prefix="/v1/managed-ne", tags=["managed-ne"]) +def _actor(ctx: AuthContext | None = None, request: Request | None = None) -> tuple[str, str]: + if ctx is not None and ctx.user is not None: + return str(ctx.user.id or ""), str(ctx.user.username or "") + if request is not None: + user = getattr(request.state, "auth_user", None) + if user: + return str(getattr(user, "id", "") or ""), str(getattr(user, "username", "") or "") + return "", "" + + @router.get("") def api_list_managed_ne( keyword: str | None = Query(default=None), @@ -131,24 +145,55 @@ def api_delete_ume_synced_managed_ne(db: Session = Depends(get_db)): @router.post("/exec") -def api_exec_managed_ne(body: ManagedNeExecRequest, db: Session = Depends(get_db)): +def api_exec_managed_ne( + body: ManagedNeExecRequest, + ctx: Annotated[AuthContext, Depends(require_user)], + db: Session = Depends(get_db), +): """Login to a managed NE or UME inventory NE and run read-only CLI (show/display/ping/traceroute).""" - return execute_managed_ne_commands( + uid, uname = _actor(ctx) + out = execute_managed_ne_commands( db, body.commands, ne_id=body.ne_id, ume_ne_id=body.ume_ne_id, read_timeout_sec=body.read_timeout_sec, ) + device = out.get("device") if isinstance(out.get("device"), dict) else {} + write_audit( + db, + action="ne.exec", + actor_user_id=uid, + actor_username=uname, + method="POST", + path="/v1/managed-ne/exec", + status_code=200 if out.get("ok") else 502, + detail={ + "ne_id": body.ne_id or "", + "ume_ne_id": body.ume_ne_id or "", + "ne_name": str(device.get("name") or device.get("ne_name") or ""), + "ne_ip": str(device.get("ip_address") or device.get("ip") or device.get("mgmt_ip") or ""), + "commands": list(out.get("commands") or body.commands or [])[:20], + "ok": bool(out.get("ok")), + "error": str(out.get("error") or "")[:500], + "output_len": len(str(out.get("output") or "")), + }, + ) + return out @router.post("/exec-batch") -def api_exec_managed_ne_batch(body: ManagedNeExecBatchRequest): +def api_exec_managed_ne_batch( + body: ManagedNeExecBatchRequest, + ctx: Annotated[AuthContext, Depends(require_user)], + db: Session = Depends(get_db), +): """Run read-only CLI on many NEs concurrently (field multi-NE sweeps).""" + uid, uname = _actor(ctx) targets = None if body.targets: targets = [t.model_dump() for t in body.targets] - return execute_managed_ne_commands_batch( + out = execute_managed_ne_commands_batch( targets=targets, ne_ids=body.ne_ids, ume_ne_ids=body.ume_ne_ids, @@ -156,6 +201,33 @@ def api_exec_managed_ne_batch(body: ManagedNeExecBatchRequest): read_timeout_sec=body.read_timeout_sec, concurrency=body.concurrency, ) + items = out.get("items") if isinstance(out, dict) else None + ok_n = 0 + fail_n = 0 + if isinstance(items, list): + for row in items: + if isinstance(row, dict) and row.get("ok"): + ok_n += 1 + else: + fail_n += 1 + write_audit( + db, + action="ne.exec_batch", + actor_user_id=uid, + actor_username=uname, + method="POST", + path="/v1/managed-ne/exec-batch", + status_code=200, + detail={ + "ne_ids": list(body.ne_ids or [])[:100], + "ume_ne_ids": list(body.ume_ne_ids or [])[:100], + "target_count": len(targets or []) + len(body.ne_ids or []) + len(body.ume_ne_ids or []), + "commands": list(body.commands or [])[:20], + "ok_count": ok_n, + "fail_count": fail_n, + }, + ) + return out @router.post("/connect-test") diff --git a/netx_api/webcrt_channel.py b/netx_api/webcrt_channel.py index f9635fd..d0838a3 100644 --- a/netx_api/webcrt_channel.py +++ b/netx_api/webcrt_channel.py @@ -468,7 +468,86 @@ def read_session_log_tail(session_id: str, *, max_bytes: int = 49152) -> str: return "" +# Lifecycle + command events also land in audit_log (ops UI). Attach/detach/sftp stay file-only. +_DB_AUDIT_EVENTS = frozenset( + { + "session_connecting", + "session_created", + "session_open_failed", + "session_closed", + "command", + } +) + +_PASSWORD_PROMPT_RE = re.compile( + r"(?:enter\s+)?(?:password|密码|passwd)\s*[:>]\s*$", + re.IGNORECASE, +) + + +def looks_like_password_prompt(text: str) -> bool: + """True when device stdout tail asks for a password (interactive auth).""" + s = str(text or "").replace("\r\n", "\n").replace("\r", "\n") + # Drop ANSI so prompt detection is stable. + s = re.sub(r"\x1b\[[0-9;?]*[A-Za-z]|\x1b\].*?\x07|\x1b.", "", s) + parts = [ln.strip() for ln in s.split("\n") if ln.strip()] + if not parts: + return False + return bool(_PASSWORD_PROMPT_RE.search(parts[-1])) + + +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). + + Handles backspace, ignores most control chars, truncates over-long lines. + Returns ``(new_buffer, completed_lines)``. + """ + cur = str(buf or "") + completed: list[str] = [] + limit = max(64, min(int(max_line or 512), 4096)) + for ch in str(data or ""): + if ch in ("\r", "\n"): + if cur: + completed.append(cur[:limit]) + cur = "" + continue + if ch in ("\b", "\x7f"): + cur = cur[:-1] if cur else "" + continue + if ch == "\x03": # Ctrl-C — abandon current line + cur = "" + continue + if ord(ch) < 32 and ch != "\t": + continue + if len(cur) < limit: + cur += ch + return cur, completed + + def _audit(event: str, **fields: Any) -> None: + """Write WebCRT audit to jsonl; dual-write selected events into audit_log.""" + # Enrich actor / device fields from the live session when callers omit them. + sid = str(fields.get("session_id") or "").strip() + if sid and ( + not fields.get("owner_user_id") + or not fields.get("owner_username") + or not fields.get("ne_name") + or "protocol" not in fields + ): + try: + from .webcrt_session_registry import get_session + + sess = get_session(sid) + if sess is not None: + fields.setdefault("owner_user_id", sess.owner_user_id) + fields.setdefault("owner_username", sess.owner_username) + fields.setdefault("ne_id", sess.ne_id) + fields.setdefault("ne_name", sess.ne_name) + fields.setdefault("ne_ip", sess.ne_ip) + fields.setdefault("protocol", sess.protocol) + except Exception: + pass + record = {"ts": _utc_iso(), "event": event, **fields} try: path = webcrt_data_root() / "audit.jsonl" @@ -477,3 +556,35 @@ def _audit(event: str, **fields: Any) -> None: except Exception: _log.exception("webcrt audit write failed") _log.info("webcrt.%s %s", event, {k: v for k, v in fields.items() if k != "detail"}) + + if str(event or "") not in _DB_AUDIT_EVENTS: + return + try: + from .audit_async import enqueue_audit + + actor_uid = str(fields.get("owner_user_id") or fields.get("actor_user_id") or "") + actor_name = str(fields.get("owner_username") or fields.get("actor_username") or "") + detail = { + k: v + for k, v in fields.items() + if k + not in { + "owner_user_id", + "owner_username", + "actor_user_id", + "actor_username", + } + } + enqueue_audit( + action=f"webcrt.{event}", + actor_user_id=actor_uid, + actor_username=actor_name, + method="", + path=f"/v1/webcrt/sessions/{sid}" if sid else "/v1/webcrt", + status_code=0, + client_ip=str(fields.get("client_ip") or ""), + user_agent=str(fields.get("client") or "")[:512], + detail=detail, + ) + except Exception: + _log.exception("webcrt audit_log enqueue failed event=%s", event) diff --git a/netx_api/webcrt_session_model.py b/netx_api/webcrt_session_model.py index e43abb3..ba6c8c2 100644 --- a/netx_api/webcrt_session_model.py +++ b/netx_api/webcrt_session_model.py @@ -18,10 +18,13 @@ from .ne_session_factory import ( ) from .webcrt_channel import ( _BoundedByteQueue, + _audit, _decode_bytes, _encode_text, _session_log_path, _utc_iso, + feed_command_line_buffer, + looks_like_password_prompt, map_network_cli_enter, map_network_cli_keys, ) @@ -87,6 +90,11 @@ class WebcrtSession: sftp_ready: bool = False _sftp: Any = field(default=None, repr=False) _sftp_lock: threading.RLock = field(default_factory=threading.RLock, repr=False) + # Interactive command audit: line buffer + password-prompt redaction. + _cmd_buf: str = field(default="", repr=False) + _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) def touch(self) -> None: self.last_activity = time.time() @@ -302,7 +310,7 @@ class WebcrtSession: return "stale" return chunk # bytes | None - def write_stdin(self, data: str) -> None: + def write_stdin(self, data: str, *, audit_source: str = "stdin") -> None: if self.closed or self.conn is None: raise RuntimeError("session_closed") text = str(data or "") @@ -318,6 +326,7 @@ class WebcrtSession: text = map_network_cli_enter(text, self.conn) if not text: return + self._note_stdin_for_audit(text, source=audit_source) with self._write_lock: # Prefer raw channel I/O for interactive typing (char echo / backspace). channel = getattr(self.conn, "remote_conn", None) @@ -345,6 +354,43 @@ class WebcrtSession: self.bytes_in += len(text) self.touch() + def _note_stdin_for_audit(self, text: str, *, source: str = "stdin") -> None: + """Extract completed command lines from stdin and emit webcrt.command audits.""" + with self._cmd_buf_lock: + self._cmd_buf, lines = feed_command_line_buffer(self._cmd_buf, text) + redacted = bool(self._password_mode) + if redacted and lines: + self._password_mode = False + for cmd in lines: + if not str(cmd).strip(): + continue + try: + _audit( + "command", + session_id=self.session_id, + ne_id=self.ne_id, + ne_name=self.ne_name, + ne_ip=self.ne_ip, + protocol=self.protocol, + owner_user_id=self.owner_user_id, + owner_username=self.owner_username, + command="***" if redacted else str(cmd)[:512], + redacted=bool(redacted), + source=str(source or "stdin")[:32], + ) + except Exception: + _log.debug("webcrt command audit failed session=%s", self.session_id, exc_info=True) + + def _note_stdout_for_audit(self, text: str) -> None: + """Track device prompts so the next typed line can be redacted if it is a password.""" + chunk = str(text or "") + if not chunk: + return + with self._cmd_buf_lock: + self._stdout_tail = (self._stdout_tail + chunk)[-4000:] + if looks_like_password_prompt(self._stdout_tail): + self._password_mode = True + def send_break(self) -> None: """Send SSH break / Telnet IAC BREAK to interrupt paging or hung commands.""" if self.closed or self.conn is None: @@ -472,7 +518,9 @@ class WebcrtSession: self.bytes_out += len(chunk) self.out_queue.put(chunk) try: - self.append_session_log(_decode_bytes(chunk, self.encoding)) + decoded = _decode_bytes(chunk, self.encoding) + self.append_session_log(decoded) + self._note_stdout_for_audit(decoded) except Exception: pass if self.cli_hop_guard and self._note_cli_hop_output(chunk): @@ -516,7 +564,7 @@ class WebcrtSession: return for cmd in cmds[:20]: try: - self.write_stdin(cmd + "\r") + self.write_stdin(cmd + "\r", audit_source="post_login") time.sleep(0.15) except Exception: _log.debug("post_login command failed session=%s", self.session_id, exc_info=True) diff --git a/netx_api/webcrt_session_registry.py b/netx_api/webcrt_session_registry.py index 2f73fde..cc38cb0 100644 --- a/netx_api/webcrt_session_registry.py +++ b/netx_api/webcrt_session_registry.py @@ -261,8 +261,12 @@ def _finish_connect( "session_open_failed", session_id=sess.session_id, ne_id=sess.ne_id, + ne_name=sess.ne_name, ne_ip=sess.ne_ip, + protocol=sess.protocol, source=str(device.get("source") or ""), + owner_user_id=sess.owner_user_id, + owner_username=sess.owner_username, client=client or "", error=str(exc)[:500], transcript_len=len(partial), @@ -500,6 +504,8 @@ def _finish_connect( protocol=sess.protocol, encoding=sess.encoding, source=str(device.get("source") or ""), + owner_user_id=sess.owner_user_id, + owner_username=sess.owner_username, hop_enabled=bool(creds.get("hop_enabled")), hop_vendor=str(creds.get("hop_vendor") or "") if creds.get("hop_enabled") else "", cli_hop_guard=bool(hop_guard), @@ -608,9 +614,12 @@ def create_session( "session_connecting", session_id=session_id, ne_id=sess.ne_id, + ne_name=sess.ne_name, ne_ip=sess.ne_ip, protocol=sess.protocol, encoding=enc, + owner_user_id=sess.owner_user_id, + owner_username=sess.owner_username, client=client or "", async_connect=bool(async_connect), ) @@ -746,7 +755,11 @@ def close_session(session_id: str, *, reason: str = "closed", client: str = "") "session_closed", session_id=session_id, ne_id=sess.ne_id, + ne_name=sess.ne_name, ne_ip=sess.ne_ip, + protocol=sess.protocol, + owner_user_id=sess.owner_user_id, + owner_username=sess.owner_username, reason=reason, client=client or "", bytes_in=sess.bytes_in, diff --git a/tests/test_ne_exec_audit.py b/tests/test_ne_exec_audit.py new file mode 100644 index 0000000..1368488 --- /dev/null +++ b/tests/test_ne_exec_audit.py @@ -0,0 +1,78 @@ +"""Device exec audit coverage for managed-ne /exec endpoints.""" + +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from netx_api.auth_deps import AuthContext, require_user +from netx_api.db import get_db +from netx_api.managed_ne_router import router as managed_ne_router +from netx_api.models import AppUser + + +class ManagedNeExecAuditTests(unittest.TestCase): + def setUp(self) -> None: + self.app = FastAPI() + self.app.include_router(managed_ne_router) + + fake = AppUser(id="u1", username="alice", role="admin", password_hash="x") + fake.is_active = True + + def _user() -> AuthContext: + return AuthContext(user=fake, auth_via="disabled", scopes=frozenset({"ne:exec"})) + + def _db(): + yield MagicMock() + + self.app.dependency_overrides[require_user] = _user + self.app.dependency_overrides[get_db] = _db + self.client = TestClient(self.app) + + def tearDown(self) -> None: + self.app.dependency_overrides.clear() + + @patch("netx_api.managed_ne_router.write_audit") + @patch("netx_api.managed_ne_router.execute_managed_ne_commands") + def test_exec_writes_audit(self, mock_exec: MagicMock, mock_audit: MagicMock) -> None: + mock_exec.return_value = { + "ok": True, + "device": {"name": "core-sw", "ip_address": "10.0.0.1"}, + "commands": ["display version"], + "output": "VRP", + } + r = self.client.post( + "/v1/managed-ne/exec", + json={"ne_id": "ne1", "commands": ["display version"]}, + ) + self.assertEqual(r.status_code, 200) + mock_audit.assert_called_once() + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["action"], "ne.exec") + self.assertEqual(kwargs["actor_username"], "alice") + self.assertEqual(kwargs["detail"]["ne_name"], "core-sw") + self.assertEqual(kwargs["detail"]["commands"], ["display version"]) + + @patch("netx_api.managed_ne_router.write_audit") + @patch("netx_api.managed_ne_router.execute_managed_ne_commands_batch") + def test_exec_batch_writes_audit(self, mock_batch: MagicMock, mock_audit: MagicMock) -> None: + mock_batch.return_value = { + "items": [{"ok": True}, {"ok": False}], + } + r = self.client.post( + "/v1/managed-ne/exec-batch", + json={"ne_ids": ["a", "b"], "commands": ["display clock"]}, + ) + self.assertEqual(r.status_code, 200) + mock_audit.assert_called_once() + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["action"], "ne.exec_batch") + self.assertEqual(kwargs["detail"]["ok_count"], 1) + self.assertEqual(kwargs["detail"]["fail_count"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index e63093c..4d93477 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -44,7 +44,8 @@ class WebcrtServiceTests(unittest.TestCase): def tearDown(self) -> None: self.setUp() - def test_session_write_resize_and_close(self) -> None: + @patch("netx_api.webcrt_session_model._audit") + def test_session_write_resize_and_close(self, _mock_audit: MagicMock) -> None: conn = _FakeConn() sess = svc.WebcrtSession( session_id="s1", @@ -342,7 +343,8 @@ class WebcrtServiceTests(unittest.TestCase): self.assertIn("", echo) self.assertFalse(sess.needs_live_prompt) before = list(fake.written) - sess.write_stdin("\n") + with patch("netx_api.webcrt_session_model._audit"): + sess.write_stdin("\n") self.assertEqual(fake.written[len(before) :], ["\n"]) svc.close_session(out["session_id"], reason="test") diff --git a/tests/test_webcrt_audit.py b/tests/test_webcrt_audit.py new file mode 100644 index 0000000..cc0567e --- /dev/null +++ b/tests/test_webcrt_audit.py @@ -0,0 +1,208 @@ +"""Tests for WebCRT / device operation audit (command lines + audit_log dual-write).""" + +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from netx_api.webcrt_channel import ( + _DB_AUDIT_EVENTS, + _audit, + feed_command_line_buffer, + looks_like_password_prompt, +) +from netx_api.webcrt_session_model import WebcrtSession + + +class FeedCommandLineBufferTests(unittest.TestCase): + def test_enter_emits_line(self) -> None: + buf, lines = feed_command_line_buffer("", "display version\r") + self.assertEqual(buf, "") + self.assertEqual(lines, ["display version"]) + + def test_backspace(self) -> None: + buf, lines = feed_command_line_buffer("", "disX\x08play\n") + self.assertEqual(buf, "") + self.assertEqual(lines, ["display"]) + + def test_multiline_paste(self) -> None: + buf, lines = feed_command_line_buffer("", "a\nb\nc\n") + self.assertEqual(buf, "") + self.assertEqual(lines, ["a", "b", "c"]) + + def test_partial_stays_in_buffer(self) -> None: + buf, lines = feed_command_line_buffer("", "sho") + self.assertEqual(buf, "sho") + self.assertEqual(lines, []) + buf, lines = feed_command_line_buffer(buf, "w ver\n") + self.assertEqual(buf, "") + self.assertEqual(lines, ["show ver"]) + + def test_empty_line_skipped(self) -> None: + buf, lines = feed_command_line_buffer("", "\r\n") + self.assertEqual(buf, "") + self.assertEqual(lines, []) + + def test_ctrl_c_clears(self) -> None: + buf, lines = feed_command_line_buffer("half", "\x03show\n") + self.assertEqual(buf, "") + self.assertEqual(lines, ["show"]) + + def test_truncates_long_line(self) -> None: + long = "x" * 600 + buf, lines = feed_command_line_buffer("", long + "\n", max_line=512) + self.assertEqual(buf, "") + self.assertEqual(len(lines), 1) + self.assertEqual(len(lines[0]), 512) + + +class PasswordPromptTests(unittest.TestCase): + def test_detects_password_prompt(self) -> None: + self.assertTrue(looks_like_password_prompt("Password:")) + self.assertTrue(looks_like_password_prompt("Please enter password:")) + self.assertTrue(looks_like_password_prompt("请输入密码:")) + self.assertFalse(looks_like_password_prompt("")) + self.assertFalse(looks_like_password_prompt("Username:")) + + +class AuditDualWriteTests(unittest.TestCase): + def test_db_events_set(self) -> None: + self.assertIn("session_created", _DB_AUDIT_EVENTS) + self.assertIn("command", _DB_AUDIT_EVENTS) + self.assertNotIn("session_attached", _DB_AUDIT_EVENTS) + + @patch("netx_api.webcrt_channel.webcrt_data_root") + @patch("netx_api.audit_async.enqueue_audit") + def test_lifecycle_enqueues_audit_log(self, mock_enq: MagicMock, mock_root: MagicMock) -> None: + root = MagicMock() + path = MagicMock() + mock_root.return_value = root + root.__truediv__ = MagicMock(return_value=path) + path.open = MagicMock() + fh = MagicMock() + path.open.return_value.__enter__ = MagicMock(return_value=fh) + path.open.return_value.__exit__ = MagicMock(return_value=False) + + _audit( + "session_created", + session_id="sid1", + ne_id="ne1", + ne_name="core-sw", + ne_ip="10.0.0.1", + protocol="ssh", + owner_user_id="u1", + owner_username="alice", + ) + mock_enq.assert_called_once() + kwargs = mock_enq.call_args.kwargs + self.assertEqual(kwargs["action"], "webcrt.session_created") + self.assertEqual(kwargs["actor_username"], "alice") + self.assertEqual(kwargs["actor_user_id"], "u1") + self.assertEqual(kwargs["detail"]["ne_name"], "core-sw") + self.assertEqual(kwargs["detail"]["ne_ip"], "10.0.0.1") + + @patch("netx_api.webcrt_channel.webcrt_data_root") + @patch("netx_api.audit_async.enqueue_audit") + def test_attach_does_not_enqueue(self, mock_enq: MagicMock, mock_root: MagicMock) -> None: + root = MagicMock() + path = MagicMock() + mock_root.return_value = root + root.__truediv__ = MagicMock(return_value=path) + path.open = MagicMock() + fh = MagicMock() + path.open.return_value.__enter__ = MagicMock(return_value=fh) + path.open.return_value.__exit__ = MagicMock(return_value=False) + + _audit("session_attached", session_id="sid1", ne_id="ne1") + mock_enq.assert_not_called() + + +class SessionCommandAuditTests(unittest.TestCase): + @patch("netx_api.webcrt_session_model._audit") + def test_write_stdin_audits_completed_command(self, mock_audit: MagicMock) -> None: + conn = MagicMock() + conn.RETURN = "\n" + conn.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) + # Force write_channel path (no send). + del conn.remote_conn.send + conn.write_channel = MagicMock() + + sess = WebcrtSession( + session_id="s-cmd", + ne_id="ne1", + ne_name="lab", + ne_ip="1.2.3.4", + protocol="ssh", + cols=80, + rows=24, + cli_keymap=False, + owner_user_id="u1", + owner_username="bob", + conn=conn, + ) + sess.write_stdin("display version\r") + mock_audit.assert_called() + event = mock_audit.call_args[0][0] + self.assertEqual(event, "command") + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["command"], "display version") + self.assertEqual(kwargs["owner_username"], "bob") + self.assertEqual(kwargs["ne_name"], "lab") + self.assertFalse(kwargs["redacted"]) + + @patch("netx_api.webcrt_session_model._audit") + def test_password_mode_redacts_command(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-pw", + ne_id="ne1", + ne_name="lab", + ne_ip="1.2.3.4", + protocol="ssh", + cols=80, + rows=24, + cli_keymap=False, + conn=conn, + ) + sess._note_stdout_for_audit("Password:") + self.assertTrue(sess._password_mode) + sess.write_stdin("secret-pass\r") + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["command"], "***") + self.assertTrue(kwargs["redacted"]) + self.assertFalse(sess._password_mode) + + @patch("netx_api.webcrt_session_model._audit") + def test_post_login_source(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-pl", + ne_id="ne1", + ne_name="lab", + ne_ip="1.2.3.4", + protocol="ssh", + cols=80, + rows=24, + cli_keymap=False, + conn=conn, + post_login_commands=["screen-length 0 temporary"], + ) + with patch("netx_api.webcrt_session_model.time.sleep"): + sess.run_post_login_commands() + kwargs = mock_audit.call_args.kwargs + self.assertEqual(kwargs["source"], "post_login") + self.assertEqual(kwargs["command"], "screen-length 0 temporary") + + +if __name__ == "__main__": + unittest.main() diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 0d433a8..78f13e8 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -1867,6 +1867,24 @@ const en = { audit: { title: "Audit & ops", logsTitle: "Operation logs", + colSummary: "Summary", + colDetail: "Detail", + showDetail: "Expand", + hideDetail: "Collapse", + filterAll: "All", + filterWebcrt: "Device terminal", + filterNeExec: "Device exec", + filterAuth: "Auth", + summary: { + connecting: "Connecting to {{device}}", + loginOk: "Logged in to {{device}}", + loginFail: "Login failed {{device}}: {{error}}", + logout: "Disconnected {{device}}", + logoutReason: "Disconnected {{device}} ({{reason}})", + command: "{{device}} · cmd: {{command}}", + neExec: "{{device}} · exec: {{commands}}", + neExecBatch: "Batch exec {{n}} NE(s) (ok {{ok}} / fail {{fail}})", + }, nav: { tasks: "Task overview", logs: "Operation logs", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index d1a8168..abed4f5 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -1845,6 +1845,24 @@ const zh = { audit: { title: "操作审计", logsTitle: "操作日志", + colSummary: "摘要", + colDetail: "详情", + showDetail: "展开", + hideDetail: "收起", + filterAll: "全部", + filterWebcrt: "设备终端", + filterNeExec: "设备执行", + filterAuth: "登录认证", + summary: { + connecting: "正在登录 {{device}}", + loginOk: "登录 {{device}}", + loginFail: "登录失败 {{device}}：{{error}}", + logout: "断开 {{device}}", + logoutReason: "断开 {{device}}（{{reason}}）", + command: "{{device}} · 命令: {{command}}", + neExec: "{{device}} · 执行: {{commands}}", + neExecBatch: "批量执行 {{n}} 台（成功 {{ok}} / 失败 {{fail}}）", + }, nav: { tasks: "任务概览", logs: "操作日志", diff --git a/web/src/index.css b/web/src/index.css index 45b62c5..1b36169 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -10664,3 +10664,49 @@ option { border-color: rgba(148, 163, 184, 0.2) !important; color: #e2e8f0 !important; } + +/* Operation audit logs */ +.system-page .audit-quick-filters button.is-active { + background: rgba(37, 99, 235, 0.18); + border-color: rgba(59, 130, 246, 0.55); + color: #1e3a8a; + font-weight: 600; +} + +.system-page .audit-summary-cell { + max-width: 420px; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.system-page button.linkish { + background: transparent; + border: none; + color: #2563eb; + cursor: pointer; + padding: 0; + height: auto; + font: inherit; + text-decoration: underline; +} + +.system-page .audit-detail-row td { + background: rgba(15, 23, 42, 0.04); + padding: 10px 12px 14px; +} + +.system-page .audit-detail-pre { + margin: 0; + max-height: 240px; + overflow: auto; + white-space: pre-wrap; + word-break: break-word; + font-family: ui-monospace, SFMono-Regular, Menlo, Consolas, monospace; + font-size: 12px; + line-height: 1.45; +} + +.system-page .audit-detail-meta { + margin: 8px 0 0; +} diff --git a/web/src/pages/AuditPage.tsx b/web/src/pages/AuditPage.tsx index c2c70e3..08fe923 100644 --- a/web/src/pages/AuditPage.tsx +++ b/web/src/pages/AuditPage.tsx @@ -1,4 +1,4 @@ -import { useMemo, useState } from "react"; +import { Fragment, useMemo, useState } from "react"; import { useQuery } from "@tanstack/react-query"; import { useAuth } from "../auth/AuthContext"; import { useI18n } from "../i18n"; @@ -17,6 +17,8 @@ type AuditItem = { detail: Record; }; +type QuickFilter = "" | "webcrt." | "ne.exec" | "auth."; + function statusClass(code: number): string { if (code >= 500) return "pt-list-status--failed"; if (code >= 400) return "pt-list-status--warning"; @@ -24,21 +26,89 @@ function statusClass(code: number): string { return "pt-list-status--other"; } +function detailStr(detail: Record, key: string): string { + const v = detail?.[key]; + if (v == null) return ""; + return String(v).trim(); +} + +function deviceLabel(detail: Record): string { + const name = detailStr(detail, "ne_name"); + const ip = detailStr(detail, "ne_ip"); + if (name && ip) return `${name} (${ip})`; + return name || ip || detailStr(detail, "ne_id") || ""; +} + +export function auditSummary( + action: string, + detail: Record, + t: (key: string, vars?: Record) => string, +): string { + const d = detail || {}; + const device = deviceLabel(d); + if (action === "webcrt.session_connecting") { + return t("audit.summary.connecting", { device: device || t("common.empty") }); + } + if (action === "webcrt.session_created") { + return t("audit.summary.loginOk", { device: device || t("common.empty") }); + } + if (action === "webcrt.session_open_failed") { + const err = detailStr(d, "error") || t("common.empty"); + return t("audit.summary.loginFail", { device: device || t("common.empty"), error: err.slice(0, 80) }); + } + if (action === "webcrt.session_closed") { + const reason = detailStr(d, "reason"); + return reason + ? t("audit.summary.logoutReason", { device: device || t("common.empty"), reason }) + : t("audit.summary.logout", { device: device || t("common.empty") }); + } + if (action === "webcrt.command") { + const cmd = detailStr(d, "command") || t("common.empty"); + const redacted = Boolean(d.redacted); + return t("audit.summary.command", { + device: device || t("common.empty"), + command: redacted ? "***" : cmd, + }); + } + if (action === "ne.exec") { + const cmds = Array.isArray(d.commands) ? d.commands.map(String).filter(Boolean) : []; + return t("audit.summary.neExec", { + device: device || t("common.empty"), + commands: cmds.slice(0, 3).join("; ") || t("common.empty"), + }); + } + if (action === "ne.exec_batch") { + return t("audit.summary.neExecBatch", { + n: String(d.target_count ?? ((d.ne_ids as unknown[]) || []).length || 0), + ok: String(d.ok_count ?? 0), + fail: String(d.fail_count ?? 0), + }); + } + if (action.startsWith("auth.")) { + return action.replace(/^auth\./, ""); + } + return ""; +} + export function AuditPage() { const { t } = useI18n(); const { ready, isAdmin } = useAuth(); const [page, setPage] = useState(1); const [username, setUsername] = useState(""); const [action, setAction] = useState(""); + const [quick, setQuick] = useState(""); + const [expanded, setExpanded] = useState(null); + + const effectiveAction = action.trim() || quick; const query = useQuery({ - queryKey: ["auditLogs", page, username, action], + queryKey: ["auditLogs", page, username, effectiveAction], queryFn: () => { const p = new URLSearchParams(); p.set("page", String(page)); p.set("page_size", "50"); if (username.trim()) p.set("username", username.trim()); - if (action.trim()) p.set("action", action.trim()); + if (effectiveAction) p.set("action", effectiveAction); return apiGet<{ total: number; page: number; page_size: number; items: AuditItem[] }>( `/v1/audit-logs?${p.toString()}`, ); @@ -49,7 +119,13 @@ export function AuditPage() { const items = useMemo(() => query.data?.items || [], [query.data]); const total = query.data?.total || 0; const pages = Math.max(1, Math.ceil(total / 50)); - const hasFilters = Boolean(username.trim() || action.trim()); + const hasFilters = Boolean(username.trim() || action.trim() || quick); + + const setQuickFilter = (next: QuickFilter) => { + setPage(1); + setQuick(next); + if (next) setAction(""); + }; return (
@@ -60,6 +136,26 @@ export function AuditPage() {

{isAdmin ? t("auth.auditHintAdmin") : t("auth.auditHintUser")}

+
+ {( + [ + ["", "audit.filterAll"], + ["webcrt.", "audit.filterWebcrt"], + ["ne.exec", "audit.filterNeExec"], + ["auth.", "audit.filterAuth"], + ] as const + ).map(([value, labelKey]) => ( + + ))} +
+
{isAdmin ? ( { setPage(1); setAction(e.target.value); + if (e.target.value.trim()) setQuick(""); }} /> + + + {open ? ( + + + 
{JSON.stringify(row.detail || {}, null, 2)}
+ {row.method || row.path ? ( +

+ {row.method} {row.path} +

+ ) : null} + + + ) : null} + + ); + })}