From accd28c9a1873ae0909d1cf5e25a50fccfac0236 Mon Sep 17 00:00:00 2001 From: oliver Date: Sat, 1 Aug 2026 19:01:44 +0800 Subject: [PATCH] Harden WebCRT interactive sessions for more CRT-like device login. Improve async connect, binary stdout coalescing, paste throttling with progress, SSH keepalive, login prompt settle, and optional SFTP while keeping core terminal UX focused. Co-authored-by: Cursor --- netx_api/config.py | 11 + netx_api/ne_session_factory.py | 22 ++ netx_api/webcrt_router.py | 249 ++++++++++-- netx_api/webcrt_service.py | 604 +++++++++++++++++++++++------ netx_api/webcrt_sftp.py | 189 +++++++++ tests/test_webcrt.py | 47 ++- web/WEB.md | 10 +- web/src/components/WebTerminal.tsx | 388 ++++++++++++++++-- web/src/i18n/en.ts | 21 + web/src/i18n/zh.ts | 21 + web/src/index.css | 281 +++++++++++++- web/src/pages/WebcrtPage.tsx | 200 ++++++++-- web/src/services/api.ts | 65 ++++ 13 files changed, 1884 insertions(+), 224 deletions(-) create mode 100644 netx_api/webcrt_sftp.py diff --git a/netx_api/config.py b/netx_api/config.py index dac3a2e..52ad0f3 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -88,6 +88,17 @@ class Settings(BaseSettings): webcrt_connect_timeout_sec: int = 90 webcrt_attach_timeout_sec: int = 60 webcrt_data_dir: str = "data/webcrt" + # SSH transport keepalive interval (seconds); 0 disables. + webcrt_keepalive_sec: int = 30 + # Device anti-idle CLI nudge (0 = off). Keep off: NEs close idle VTY themselves. + webcrt_anti_idle_sec: int = 0 + webcrt_anti_idle_payload: str = " " + # Cap stdout queue depth (drop oldest when full) to protect memory. + webcrt_out_queue_max: int = 2000 + # Persist per-session transcripts under webcrt_data_dir/sessions/. + webcrt_session_log_enabled: bool = True + # Reader: short blocking wait instead of fixed 40ms spin (seconds). + webcrt_reader_poll_sec: float = 0.01 # Local app login / audit (lab defaults; override in production) auth_enabled: bool = True # Stable default so JWT survives restarts without .env. Override in production. diff --git a/netx_api/ne_session_factory.py b/netx_api/ne_session_factory.py index ee4d1b5..7756af7 100644 --- a/netx_api/ne_session_factory.py +++ b/netx_api/ne_session_factory.py @@ -311,6 +311,7 @@ def _netmiko_over_ssh_client( session_timeout: int | None, session_log: Any = None, interactive: bool = False, + keepalive: int | None = None, ) -> ConnectHandler: """Netmiko session over an already-authenticated SSH client (bastion protocol proxy).""" base_cls = _netmiko_driver_class(device_type) @@ -340,6 +341,7 @@ def _netmiko_over_ssh_client( enable_secret=enable_secret, session_timeout=session_timeout, session_log=session_log, + keepalive=keepalive, ) return _PreauthSession(**dev) @@ -354,6 +356,7 @@ def _base_connect_kwargs( enable_secret: str, session_timeout: int | None = None, session_log: Any = None, + keepalive: int | None = None, ) -> dict[str, Any]: timeout = int(settings.ne_connect_timeout_sec or 30) dev: dict[str, Any] = { @@ -368,6 +371,9 @@ def _base_connect_kwargs( } if session_timeout is not None: dev["session_timeout"] = session_timeout + if keepalive is not None and int(keepalive) > 0: + # Paramiko/Netmiko SSH transport keepalive (seconds between null packets). + dev["keepalive"] = int(keepalive) secret = str(enable_secret or "").strip() if secret: dev["secret"] = secret @@ -382,6 +388,7 @@ def _connect_direct( session_timeout: int | None = None, session_log: Any = None, interactive: bool = False, + keepalive: int | None = None, ) -> ConnectHandler: device_type = normalize_netmiko_device_type(creds["device_type"], creds["protocol"]) dev = _base_connect_kwargs( @@ -393,6 +400,7 @@ def _connect_direct( enable_secret=str(creds.get("enable_secret") or ""), session_timeout=session_timeout, session_log=session_log, + keepalive=keepalive, ) return _build_netmiko_connection(dev, interactive=interactive) @@ -602,6 +610,7 @@ def _connect_via_cli_hop( cols: int | None = None, rows: int | None = None, interactive: bool = False, + keepalive: int | None = None, ) -> ConnectHandler: """Login to ZTE/Huawei/Cisco hop NE, run CLI jump command, then target secondary auth.""" hop_host = str(creds.get("hop_host") or "").strip() @@ -621,6 +630,7 @@ def _connect_via_cli_hop( enable_secret="", session_timeout=session_timeout or 180, session_log=session_log, + keepalive=keepalive, ) conn = _build_netmiko_connection(hop_dev, interactive=interactive) try: @@ -674,6 +684,7 @@ def _connect_via_bastion( session_timeout: int | None = None, session_log: Any = None, interactive: bool = False, + keepalive: int | None = None, ) -> ConnectHandler: """SSH to bastion with composite username; bastion proxies to target (protocol proxy).""" hop_host = str(creds.get("hop_host") or "").strip() @@ -707,6 +718,7 @@ def _connect_via_bastion( session_timeout=session_timeout or 180, session_log=session_log, interactive=interactive, + keepalive=keepalive, ) except Exception: if ssh_client is not None: @@ -738,6 +750,7 @@ def _connect_via_linux_hop( session_timeout: int | None = None, session_log: Any = None, interactive: bool = False, + keepalive: int | None = None, ) -> ConnectHandler: """SSH to Linux bastion, then direct-tcpip tunnel to target (classic ProxyJump-style).""" hop_host = str(creds.get("hop_host") or "").strip() @@ -791,6 +804,7 @@ def _connect_via_linux_hop( enable_secret=str(creds.get("enable_secret") or ""), session_timeout=session_timeout, session_log=session_log, + keepalive=keepalive, ) dev["sock"] = channel conn = _build_netmiko_connection(dev, interactive=interactive) @@ -822,12 +836,16 @@ def open_netmiko_connection( cols: int | None = None, rows: int | None = None, interactive: bool = False, + keepalive: int | None = None, ) -> ConnectHandler: """Open a Netmiko connection to the target NE (direct or via configured hop). ``interactive=True`` (WebCRT) skips Netmiko's automatic ``terminal length`` / ``terminal width`` (and vendor equivalents). Collection / MCP keep the default. """ + ka = keepalive + if ka is None and interactive: + ka = int(getattr(settings, "webcrt_keepalive_sec", 0) or 0) or None if creds.get("hop_enabled"): vendor = _hop_vendor(creds) if vendor == "linux": @@ -836,6 +854,7 @@ def open_netmiko_connection( session_timeout=session_timeout, session_log=session_log, interactive=interactive, + keepalive=ka, ) if vendor == "bastion": return _connect_via_bastion( @@ -843,6 +862,7 @@ def open_netmiko_connection( session_timeout=session_timeout, session_log=session_log, interactive=interactive, + keepalive=ka, ) return _connect_via_cli_hop( creds, @@ -851,10 +871,12 @@ def open_netmiko_connection( cols=cols, rows=rows, interactive=interactive, + keepalive=ka, ) return _connect_direct( creds, session_timeout=session_timeout, session_log=session_log, interactive=interactive, + keepalive=ka, ) diff --git a/netx_api/webcrt_router.py b/netx_api/webcrt_router.py index 5e15a35..6b98dbc 100644 --- a/netx_api/webcrt_router.py +++ b/netx_api/webcrt_router.py @@ -7,7 +7,7 @@ import json import logging from typing import Any -from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect +from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile, WebSocket, WebSocketDisconnect from pydantic import BaseModel, Field from sqlalchemy.orm import Session @@ -21,6 +21,9 @@ from .webcrt_service import ( get_session, list_sessions, mark_attached, + wait_session_ready, + _decode_bytes, + _normalize_encoding, ) _log = logging.getLogger("netx.webcrt.router") @@ -33,6 +36,22 @@ class WebcrtSessionCreate(BaseModel): ume_ne_id: str | None = Field(default=None) cols: int = Field(default=80, ge=20, le=500) rows: int = Field(default=24, ge=5, le=200) + encoding: str = Field(default="utf-8") + post_login_commands: list[str] = Field(default_factory=list) + # Default async so UI can open WS while connect runs; tests may force sync via service API. + async_connect: bool = Field(default=True) + + +class WebcrtSftpListBody(BaseModel): + ne_id: str | None = Field(default=None) + ume_ne_id: str | None = Field(default=None) + path: str = Field(default=".") + + +class WebcrtSftpDownloadBody(BaseModel): + ne_id: str | None = Field(default=None) + ume_ne_id: str | None = Field(default=None) + path: str def _client_label(request: Request | None = None, websocket: WebSocket | None = None) -> str: @@ -66,6 +85,9 @@ def api_create_session( cols=body.cols, rows=body.rows, client=_client_label(request=request), + encoding=body.encoding, + post_login_commands=list(body.post_login_commands or [])[:20], + async_connect=bool(body.async_connect), ) @@ -74,6 +96,61 @@ def api_close_session(session_id: str, request: Request) -> dict[str, Any]: return close_session(session_id, reason="client_delete", client=_client_label(request=request)) +@router.post("/sftp/list") +def api_sftp_list(body: WebcrtSftpListBody, db: Session = Depends(get_db)) -> dict[str, Any]: + from .webcrt_sftp import sftp_list + + mid = str(body.ne_id or "").strip() + uid = str(body.ume_ne_id or "").strip() + if bool(mid) == bool(uid): + raise HTTPException(status_code=400, detail="exactly_one_of_ne_id_or_ume_ne_id_required") + return sftp_list(db, managed_ne_id=mid or None, ume_ne_id=uid or None, path=body.path) + + +@router.post("/sftp/download") +def api_sftp_download(body: WebcrtSftpDownloadBody, db: Session = Depends(get_db)) -> Any: + from fastapi.responses import Response + + from .webcrt_sftp import sftp_download + + mid = str(body.ne_id or "").strip() + uid = str(body.ume_ne_id or "").strip() + if bool(mid) == bool(uid): + raise HTTPException(status_code=400, detail="exactly_one_of_ne_id_or_ume_ne_id_required") + data, filename = sftp_download(db, managed_ne_id=mid or None, ume_ne_id=uid or None, path=body.path) + return Response( + content=data, + media_type="application/octet-stream", + headers={"Content-Disposition": f'attachment; filename="{filename}"'}, + ) + + +@router.post("/sftp/upload") +async def api_sftp_upload( + db: Session = Depends(get_db), + ne_id: str | None = Form(default=None), + ume_ne_id: str | None = Form(default=None), + remote_path: str = Form(...), + file: UploadFile = File(...), +) -> dict[str, Any]: + from .webcrt_sftp import sftp_upload + + mid = str(ne_id or "").strip() + uid = str(ume_ne_id or "").strip() + if bool(mid) == bool(uid): + raise HTTPException(status_code=400, detail="exactly_one_of_ne_id_or_ume_ne_id_required") + content = await file.read() + if len(content) > 8 * 1024 * 1024: + raise HTTPException(status_code=413, detail="sftp_file_too_large") + return sftp_upload( + db, + managed_ne_id=mid or None, + ume_ne_id=uid or None, + remote_path=remote_path, + data=content, + ) + + @router.websocket("/sessions/{session_id}/ws") async def websocket_session(websocket: WebSocket, session_id: str) -> None: if bool(settings.auth_enabled): @@ -90,8 +167,6 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: if resolved is None: await websocket.close(code=4401) return - websocket.state.auth_user = resolved[0] - websocket.state.auth_via = resolved[1] await websocket.accept() attach_gen = 0 @@ -105,12 +180,13 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: await websocket.send_json( { "type": "status", - "state": "connected", + "state": "connecting" if sess.state == "connecting" else "connected", "session_id": sess.session_id, "ne_id": sess.ne_id, "ne_name": sess.ne_name, "ne_ip": sess.ne_ip, "protocol": sess.protocol, + "encoding": sess.encoding, "cols": sess.cols, "rows": sess.rows, "device_type": sess.device_type, @@ -119,30 +195,122 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: } ) + # Wait for async connect without blocking the event loop. + if sess.state == "connecting": + loop = asyncio.get_running_loop() + try: + await loop.run_in_executor( + None, + lambda: wait_session_ready( + session_id, + timeout=max(30, int(settings.webcrt_connect_timeout_sec or 90)) + 15, + ), + ) + sess = get_session(session_id) or sess + except HTTPException as exc: + await websocket.send_json( + {"type": "status", "state": "error", "message": str(exc.detail)} + ) + await websocket.close(code=4502) + return + + await websocket.send_json( + { + "type": "status", + "state": "connected", + "session_id": sess.session_id, + "ne_id": sess.ne_id, + "ne_name": sess.ne_name, + "ne_ip": sess.ne_ip, + "protocol": sess.protocol, + "encoding": sess.encoding, + "cols": sess.cols, + "rows": sess.rows, + "device_type": sess.device_type, + "vendor": sess.vendor, + "cli_hop": bool(sess.cli_hop_guard), + "connect_ms": ( + int((sess.connect_finished_at - sess.connect_started_at) * 1000) + if sess.connect_finished_at + else None + ), + } + ) + # Replay full login transcript (kept for StrictMode remount / brief reconnect). bootstrap = bytes(sess.bootstrap_output or b"") if bootstrap: try: - await websocket.send_json( - {"type": "stdout", "data": bootstrap.decode("utf-8", errors="replace")} - ) + if _normalize_encoding(sess.encoding) != "utf-8": + bootstrap = _decode_bytes(bootstrap, sess.encoding).encode("utf-8", errors="replace") + await websocket.send_bytes(bootstrap) except Exception: - _log.debug("webcrt bootstrap send failed session=%s", session_id, exc_info=True) + try: + await websocket.send_json( + {"type": "stdout", "data": _decode_bytes(bytes(sess.bootstrap_output or b""), sess.encoding)} + ) + except Exception: + _log.debug("webcrt bootstrap send failed session=%s", session_id, exc_info=True) stop = asyncio.Event() + stdin_buf: list[str] = [] + stdin_flush_task: asyncio.Task[None] | None = None + + async def flush_stdin() -> None: + nonlocal stdin_buf + if not stdin_buf: + return + data = "".join(stdin_buf) + stdin_buf = [] + try: + await asyncio.get_running_loop().run_in_executor(None, sess.write_stdin, data) + except Exception as exc: + await websocket.send_json( + {"type": "status", "state": "error", "message": f"write_failed:{exc}"} + ) + stop.set() + + async def schedule_stdin_flush() -> None: + await asyncio.sleep(0.008) + await flush_stdin() async def pump_stdout() -> None: loop = asyncio.get_running_loop() + pending: list[bytes] = [] + last_flush = loop.time() + + def _to_browser_bytes(raw: bytes) -> bytes: + if _normalize_encoding(sess.encoding) == "utf-8": + return raw + return _decode_bytes(raw, sess.encoding).encode("utf-8", errors="replace") + + async def _flush_pending() -> bool: + nonlocal pending, last_flush + if not pending: + return True + blob = _to_browser_bytes(b"".join(pending)) + pending = [] + last_flush = loop.time() + try: + await websocket.send_bytes(blob) + return True + except Exception: + return False + while not stop.is_set(): chunk = await loop.run_in_executor( - None, lambda: sess.take_stdout(attach_gen, timeout=0.25) + None, lambda: sess.take_stdout(attach_gen, timeout=0.05) ) if chunk == "stale": - # Newer WS owns the session (StrictMode remount); exit without stealing bytes. break if chunk == "empty": + if pending and (loop.time() - last_flush) >= 0.016: + if not await _flush_pending(): + stop.set() + break continue if chunk is None: + await _flush_pending() stop.set() try: await websocket.send_json( @@ -155,12 +323,11 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: except Exception: pass break - try: - text = chunk.decode("utf-8", errors="replace") - await websocket.send_json({"type": "stdout", "data": text}) - except Exception: - stop.set() - break + pending.append(chunk) + if sum(len(p) for p in pending) >= 8192 or (loop.time() - last_flush) >= 0.016: + if not await _flush_pending(): + stop.set() + break reader_task = asyncio.create_task(pump_stdout()) if sess.needs_live_prompt: @@ -171,35 +338,56 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: _log.debug("webcrt live prompt sync failed session=%s", session_id, exc_info=True) try: while not stop.is_set(): - raw = await websocket.receive_text() + msg_raw = await websocket.receive() + if msg_raw.get("type") == "websocket.disconnect": + break + if "bytes" in msg_raw and msg_raw["bytes"] is not None: + # Binary stdin: decode with session encoding. + try: + text = _decode_bytes(bytes(msg_raw["bytes"]), sess.encoding) + except Exception: + continue + stdin_buf.append(text) + if stdin_flush_task is None or stdin_flush_task.done(): + stdin_flush_task = asyncio.create_task(schedule_stdin_flush()) + continue + raw = msg_raw.get("text") + if raw is None: + continue try: msg = json.loads(raw) except json.JSONDecodeError: - # Treat plain text as stdin. msg = {"type": "stdin", "data": raw} mtype = str(msg.get("type") or "").strip().lower() if mtype == "stdin": data = msg.get("data") if data is None: continue - try: - await asyncio.get_running_loop().run_in_executor( - None, sess.write_stdin, str(data) - ) - except Exception as exc: - await websocket.send_json( - {"type": "status", "state": "error", "message": f"write_failed:{exc}"} - ) - break + stdin_buf.append(str(data)) + # Coalesce high-frequency keystrokes briefly. + if len(stdin_buf) >= 8: + if stdin_flush_task and not stdin_flush_task.done(): + stdin_flush_task.cancel() + await flush_stdin() + elif stdin_flush_task is None or stdin_flush_task.done(): + stdin_flush_task = asyncio.create_task(schedule_stdin_flush()) elif mtype == "resize": cols = int(msg.get("cols") or sess.cols) rows = int(msg.get("rows") or sess.rows) await asyncio.get_running_loop().run_in_executor(None, sess.resize, cols, rows) + elif mtype == "break": + try: + await asyncio.get_running_loop().run_in_executor(None, sess.send_break) + except Exception as exc: + await websocket.send_json( + {"type": "status", "state": "error", "message": f"break_failed:{exc}"} + ) elif mtype == "ping": sess.touch() await websocket.send_json({"type": "pong"}) elif mtype == "close": stop.set() + await flush_stdin() close_session( session_id, reason="client_close", @@ -212,13 +400,18 @@ async def websocket_session(websocket: WebSocket, session_id: str) -> None: _log.exception("webcrt ws error session=%s", session_id) finally: stop.set() + if stdin_flush_task and not stdin_flush_task.done(): + stdin_flush_task.cancel() + try: + await flush_stdin() + except Exception: + pass reader_task.cancel() try: await reader_task except Exception: pass # Keep device session briefly so React remount / blip can re-attach. - # Only the current attach_gen may detach — older StrictMode sockets must not. if get_session(session_id) is not None: detach_session( session_id, diff --git a/netx_api/webcrt_service.py b/netx_api/webcrt_service.py index 4f6aa46..76489b4 100644 --- a/netx_api/webcrt_service.py +++ b/netx_api/webcrt_service.py @@ -121,7 +121,7 @@ def map_network_cli_enter(data: str, conn: ConnectHandler | None) -> str: return text.replace("\r\n", ret).replace("\r", ret) -def _drain_channel(conn: ConnectHandler, *, rounds: int = 10, wait: float = 0.12) -> str: +def _drain_channel(conn: ConnectHandler, *, rounds: int = 6, wait: float = 0.06) -> str: """Read whatever is already sitting on the channel after login.""" chunks: list[str] = [] empty_streak = 0 @@ -173,7 +173,11 @@ def normalize_cli_transcript(text: str) -> str: lines = s.split("\n") while lines and not str(lines[-1]).strip(): lines.pop() - while len(lines) >= 2 and lines[-1] == lines[-2] and _looks_like_cli_prompt(lines[-1]): + # Drop blank lines immediately before a final prompt (banner\n\nR2# -> banner\nR2#). + while len(lines) >= 2 and not str(lines[-2]).strip() and _looks_like_cli_prompt(lines[-1]): + lines.pop(-2) + # Collapse trailing duplicate prompt lines (slow VMs often echo R2# several times). + while len(lines) >= 2 and str(lines[-1]).strip() == str(lines[-2]).strip() and _looks_like_cli_prompt(lines[-1]): lines.pop() return "\n".join(lines) @@ -187,25 +191,131 @@ def prepare_bootstrap_output(text: str) -> str: return normalize_cli_transcript(text) -def _prime_interactive_channel(conn: ConnectHandler) -> None: - """Send one RETURN after Netmiko login so the interactive channel is fully ready.""" +def _drain_raw_channel(conn: ConnectHandler, *, duration: float = 0.5) -> None: + """Discard leftover bytes on the live channel (SSH/Telnet) after login priming.""" + channel = getattr(conn, "remote_conn", None) + if channel is None: + try: + _drain_channel(conn, rounds=max(2, int(duration / 0.05)), wait=0.05) + except Exception: + pass + return + end = time.time() + max(0.1, float(duration)) + while time.time() < end: + got = False + try: + if hasattr(channel, "recv_ready") and hasattr(channel, "recv") and channel.recv_ready(): + channel.recv(65535) + got = True + elif hasattr(channel, "read_very_eager"): + data = channel.read_very_eager() + if data: + got = True + else: + part = conn.read_channel() + if part: + got = True + except Exception: + break + if not got: + time.sleep(0.04) + + +def _prime_interactive_channel(conn: ConnectHandler, *, already_prompted: bool = False) -> None: + """Send one RETURN after Netmiko login so the interactive channel is fully ready. + + Skip the sync Enter when the login transcript already ends with a CLI prompt — + otherwise slow Cisco VMs accumulate duplicate ``R2#`` lines in the bootstrap. + """ try: - _drain_channel(conn, rounds=4, wait=0.05) + _drain_raw_channel(conn, duration=0.15) except Exception: pass - try: - conn.write_channel(channel_return(conn)) - except Exception: + if not already_prompted: try: - conn.write_channel("\n") + conn.write_channel(channel_return(conn)) except Exception: - return + try: + conn.write_channel("\n") + except Exception: + return + try: + _drain_channel(conn, rounds=6, wait=0.08) + except Exception: + pass try: - _drain_channel(conn, rounds=8, wait=0.1) + _drain_raw_channel(conn, duration=0.35) except Exception: pass +def _is_prompt_only_echo(text: str, prompt_hint: str = "") -> bool: + """True when chunk is only whitespace / CR / a repeated prompt (safe to drop after bootstrap).""" + s = str(text or "").replace("\r\n", "\n").replace("\r", "\n").strip() + if not s: + return True + hint = str(prompt_hint or "").strip() + if hint and s == hint: + return True + # Single-line prompt echo only. + if "\n" not in s and _looks_like_cli_prompt(s): + return True + if hint and all(line.strip() in ("", hint) for line in s.split("\n")): + return True + return False + + +def _normalize_encoding(name: str) -> str: + enc = str(name or "utf-8").strip().lower().replace("_", "-") + if enc in ("gbk", "gb2312", "gb18030", "cp936"): + return "gbk" + return "utf-8" + + +def _decode_bytes(data: bytes, encoding: str) -> str: + enc = _normalize_encoding(encoding) + try: + return data.decode(enc, errors="replace") + except Exception: + return data.decode("utf-8", errors="replace") + + +def _encode_text(text: str, encoding: str) -> bytes: + enc = _normalize_encoding(encoding) + try: + return text.encode(enc, errors="replace") + except Exception: + return text.encode("utf-8", errors="replace") + + +class _BoundedByteQueue: + """Thread-safe queue that drops oldest chunks when full (backpressure).""" + + def __init__(self, maxsize: int = 2000) -> None: + self._q: queue.Queue[bytes | None] = queue.Queue() + self._max = max(8, int(maxsize or 2000)) + self._lock = threading.Lock() + self.dropped = 0 + + def put(self, item: bytes | None) -> None: + with self._lock: + while self._q.qsize() >= self._max: + try: + self._q.get_nowait() + self.dropped += 1 + except queue.Empty: + break + self._q.put(item) + + def put_nowait(self, item: bytes | None) -> None: + self.put(item) + + def get_nowait(self) -> bytes | None: + return self._q.get_nowait() + + def qsize(self) -> int: + return self._q.qsize() + def _utc_now() -> datetime: return datetime.now(timezone.utc) @@ -221,6 +331,12 @@ def webcrt_data_root() -> Path: return root.resolve() +def _session_log_path(session_id: str) -> Path: + folder = webcrt_data_root() / "sessions" + folder.mkdir(parents=True, exist_ok=True) + return folder / f"{session_id}.log" + + def _audit(event: str, **fields: Any) -> None: record = {"ts": _utc_iso(), "event": event, **fields} try: @@ -244,6 +360,7 @@ class WebcrtSession: device_type: str = "" vendor: str = "" cli_keymap: bool = True + encoding: str = "utf-8" conn: ConnectHandler | None = None created_at: float = field(default_factory=time.time) last_activity: float = field(default_factory=time.time) @@ -251,24 +368,69 @@ class WebcrtSession: detach_deadline: float | None = None closed: bool = False close_reason: str = "" + # connecting | ready | error | closed + state: str = "ready" + connect_error: str = "" + connect_started_at: float = field(default_factory=time.time) + connect_finished_at: float | None = None bootstrap_output: bytes = b"" needs_live_prompt: bool = True # React StrictMode remounts open a second WS before the first fully tears down. # Only the newest attach_gen may consume out_queue / mark detach. attach_gen: int = 0 - out_queue: queue.Queue[bytes | None] = field(default_factory=queue.Queue) + out_queue: _BoundedByteQueue = field( + default_factory=lambda: _BoundedByteQueue(int(getattr(settings, "webcrt_out_queue_max", 2000) or 2000)) + ) # Vendor CLI hop (Huawei/ZTE/Cisco): close when nested target session returns to hop. cli_hop_guard: bool = False cli_hop_prompt: str = "" + post_login_commands: list[str] = field(default_factory=list) + bytes_in: int = 0 + bytes_out: int = 0 _reader: threading.Thread | None = field(default=None, repr=False) _write_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) _stdout_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) _hop_scan_buf: str = field(default="", repr=False) _cli_hop_seen_other_prompt: bool = field(default=False, repr=False) + _log_fh: Any = field(default=None, repr=False) + _ready_event: threading.Event = field(default_factory=threading.Event, repr=False) def touch(self) -> None: self.last_activity = time.time() + def open_session_log(self) -> None: + if not bool(getattr(settings, "webcrt_session_log_enabled", True)): + return + if self._log_fh is not None: + return + try: + self._log_fh = _session_log_path(self.session_id).open("a", encoding="utf-8", errors="replace") + self._log_fh.write(f"# session={self.session_id} ne={self.ne_id} ip={self.ne_ip} ts={_utc_iso()}\n") + self._log_fh.flush() + except Exception: + _log.debug("webcrt session log open failed", exc_info=True) + self._log_fh = None + + def append_session_log(self, text: str) -> None: + if not text or self._log_fh is None: + return + try: + self._log_fh.write(text) + self._log_fh.flush() + except Exception: + pass + + def close_session_log(self) -> None: + fh = self._log_fh + self._log_fh = None + if fh is None: + return + try: + fh.write(f"\n# closed reason={self.close_reason} ts={_utc_iso()}\n") + fh.close() + except Exception: + pass + def take_stdout(self, attach_gen: int, *, timeout: float = 0.25) -> bytes | None | str: """Exclusive stdout take for one WS attach generation. @@ -295,7 +457,7 @@ class WebcrtSession: return chunk # bytes | None if time.time() >= deadline: return "empty" - time.sleep(0.02) + time.sleep(0.005) def write_stdin(self, data: str) -> None: if self.closed or self.conn is None: @@ -315,12 +477,10 @@ class WebcrtSession: return with self._write_lock: # Prefer raw channel I/O for interactive typing (char echo / backspace). - # Netmiko write_channel is fine for automation but can feel "half-duplex" - # on some Telnet/VRP sessions when used keystroke-by-keystroke. channel = getattr(self.conn, "remote_conn", None) try: if channel is not None and hasattr(channel, "send") and callable(channel.send): - payload = text.encode(getattr(self.conn, "encoding", None) or "utf-8", errors="replace") + payload = _encode_text(text, self.encoding) # Paramiko may write partially when the window is full. view = memoryview(payload) while len(view): @@ -329,14 +489,47 @@ class WebcrtSession: time.sleep(0.01) continue view = view[n:] + self.bytes_in += len(payload) elif channel is not None and hasattr(channel, "write") and callable(channel.write): - encoding = getattr(self.conn, "encoding", None) or "utf-8" - channel.write(text.encode(encoding, errors="replace") if isinstance(text, str) else text) + payload = _encode_text(text, self.encoding) + channel.write(payload) + self.bytes_in += len(payload) else: self.conn.write_channel(text) + self.bytes_in += len(text) except Exception: self.conn.write_channel(text) + self.bytes_in += len(text) self.touch() + + 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: + raise RuntimeError("session_closed") + channel = getattr(self.conn, "remote_conn", None) + with self._write_lock: + sent = False + if channel is not None and hasattr(channel, "send_break") and callable(channel.send_break): + try: + channel.send_break(0) + sent = True + except Exception: + _log.debug("send_break failed session=%s", self.session_id, exc_info=True) + if not sent and channel is not None and hasattr(channel, "send") and callable(channel.send): + # Telnet IAC BREAK = 255 243 + try: + channel.send(b"\xff\xf3") + sent = True + except Exception: + pass + if not sent: + # Fallback: Ctrl-C often interrupts device CLI more-pages. + try: + self.conn.write_channel("\x03") + except Exception: + raise RuntimeError("break_failed") + self.touch() + def resize(self, cols: int, rows: int) -> None: if self.closed or self.conn is None: return @@ -369,53 +562,76 @@ class WebcrtSession: return channel = getattr(conn, "remote_conn", None) hop_return = False + poll = max(0.002, float(getattr(settings, "webcrt_reader_poll_sec", 0.01) or 0.01)) try: while not self.closed: chunk = b"" try: if channel is not None and hasattr(channel, "recv_ready") and hasattr(channel, "recv"): - # Paramiko SSH: raw bytes keep ANSI / backspace echo intact. - if channel.recv_ready(): - chunk = channel.recv(4096) + # Paramiko SSH: prefer short blocking recv over fixed spin-sleep. + ready = False + try: + ready = bool(channel.recv_ready()) + except Exception: + ready = False + if ready: + chunk = channel.recv(16384) if not chunk: break elif hasattr(channel, "exit_status_ready") and channel.exit_status_ready(): break else: - time.sleep(0.04) - continue + # Brief block: settimeout + recv wakes sooner than sleep(0.04). + prev_timeout = None + try: + prev_timeout = channel.gettimeout() + except Exception: + prev_timeout = None + try: + channel.settimeout(poll) + chunk = channel.recv(16384) + except Exception: + chunk = b"" + finally: + try: + channel.settimeout(prev_timeout) + except Exception: + pass + if not chunk: + continue elif channel is not None and hasattr(channel, "read_very_eager"): - # Telnet: do NOT use conn.read_channel() — Netmiko strips ANSI - # escape codes, which removes Huawei backspace echo (\x1b[1D \x1b[1D). + # Telnet: do NOT use conn.read_channel() — Netmiko strips ANSI. data = channel.read_very_eager() if data: chunk = ( data if isinstance(data, (bytes, bytearray)) - else str(data).encode( - getattr(conn, "encoding", None) or "utf-8", - errors="replace", - ) + else _encode_text(str(data), self.encoding) ) else: - time.sleep(0.04) + time.sleep(poll) continue else: text = conn.read_channel() if text: - chunk = text.encode("utf-8", errors="replace") + chunk = _encode_text(str(text), self.encoding) else: - time.sleep(0.04) + time.sleep(poll) continue except Exception as exc: if self.closed: break _log.debug("webcrt reader error session=%s: %s", self.session_id, exc) - time.sleep(0.1) + time.sleep(0.05) continue if chunk: self.touch() + self.bytes_out += len(chunk) self.out_queue.put(chunk) + try: + self.append_session_log(_decode_bytes(chunk, self.encoding)) + except Exception: + pass if self.cli_hop_guard and self._note_cli_hop_output(chunk): hop_return = True notice = ( @@ -426,7 +642,6 @@ class WebcrtSession: break finally: if hop_return and not self.closed: - # Prefer registry close for audit + remove; fall back to local close. try: close_session(self.session_id, reason="cli_hop_return") except Exception: @@ -436,12 +651,10 @@ class WebcrtSession: def _note_cli_hop_output(self, chunk: bytes) -> bool: """Accumulate stdout and return True when nested CLI hop has returned to proxy.""" try: - text = chunk.decode("utf-8", errors="replace") + text = _decode_bytes(chunk, self.encoding) except Exception: text = str(chunk) self._hop_scan_buf = (self._hop_scan_buf + text)[-12000:] - # Track a prompt that differs from the hop so same-sysname labs still need - # an explicit nested-close message before we tear down. marker = str(self.cli_hop_prompt or "").strip() last = extract_cli_prompt_marker(self._hop_scan_buf) if last and (not marker or last != marker): @@ -452,11 +665,25 @@ class WebcrtSession: seen_other_prompt=self._cli_hop_seen_other_prompt, ) + def run_post_login_commands(self) -> None: + cmds = [str(c).rstrip("\r\n") for c in (self.post_login_commands or []) if str(c).strip()] + if not cmds or self.closed or self.conn is None: + return + for cmd in cmds[:20]: + try: + self.write_stdin(cmd + "\r") + time.sleep(0.15) + except Exception: + _log.debug("post_login command failed session=%s", self.session_id, exc_info=True) + break + def close(self, reason: str = "closed") -> None: if self.closed: return self.closed = True + self.state = "closed" self.close_reason = reason or "closed" + self._ready_event.set() try: close_netmiko_connection(self.conn) except Exception: @@ -466,6 +693,7 @@ class WebcrtSession: self.out_queue.put_nowait(None) except Exception: pass + self.close_session_log() def _ensure_reaper() -> None: @@ -490,16 +718,32 @@ def _reaper_loop() -> None: def _reap_sessions() -> None: idle = max(60, int(settings.webcrt_idle_timeout_sec or 1800)) attach = max(10, int(settings.webcrt_attach_timeout_sec or 60)) + anti_idle = max(0, int(getattr(settings, "webcrt_anti_idle_sec", 0) or 0)) + anti_payload = str(getattr(settings, "webcrt_anti_idle_payload", " ") or " ") now = time.time() to_close: list[tuple[WebcrtSession, str]] = [] + to_nudge: list[WebcrtSession] = [] with _sessions_lock: for sess in list(_sessions.values()): if sess.closed: _sessions.pop(sess.session_id, None) continue + if sess.state == "connecting": + # Connecting sessions use connect timeout, not attach timeout alone. + connect_budget = max(30, int(settings.webcrt_connect_timeout_sec or 90)) + 30 + if (now - sess.connect_started_at) > connect_budget: + to_close.append((sess, "connect_timeout")) + continue if sess.attached: if (now - sess.last_activity) > idle: to_close.append((sess, "idle_timeout")) + elif ( + anti_idle > 0 + and sess.state == "ready" + and sess.conn is not None + and (now - sess.last_activity) >= anti_idle + ): + to_nudge.append(sess) continue # Not attached: either never attached, or briefly detached for reconnect. if sess.detach_deadline is not None: @@ -509,6 +753,18 @@ def _reap_sessions() -> None: to_close.append((sess, "attach_timeout")) elif (now - sess.last_activity) > idle: to_close.append((sess, "idle_timeout")) + for sess in to_nudge: + try: + # Touch without changing visible prompt when payload is empty/null-ish. + payload = anti_payload + if payload == "\\0": + payload = "\x00" + if payload: + sess.write_stdin(payload) + else: + sess.touch() + except Exception: + _log.debug("webcrt anti-idle failed session=%s", sess.session_id, exc_info=True) for sess, reason in to_close: close_session(sess.session_id, reason=reason) @@ -526,6 +782,21 @@ def get_session(session_id: str) -> WebcrtSession | None: return sess +def wait_session_ready(session_id: str, *, timeout: float = 120.0) -> WebcrtSession: + """Block until async connect finishes (ready or error). Used by tests and WS.""" + deadline = time.time() + max(1.0, float(timeout)) + while time.time() < deadline: + sess = get_session(session_id) + if sess is None: + raise HTTPException(status_code=404, detail="webcrt_session_not_found") + if sess.state == "ready": + return sess + if sess.state == "error": + raise HTTPException(status_code=502, detail=sess.connect_error or "connect_failed") + sess._ready_event.wait(timeout=0.25) + raise HTTPException(status_code=504, detail="connect_timeout") + + def _webcrt_creds_ready(creds: dict[str, Any]) -> bool: """True when WebCRT can open a session with the resolved credentials. @@ -547,6 +818,133 @@ def _webcrt_creds_ready(creds: dict[str, Any]) -> bool: return bool(str(creds.get("password") or "")) +def _finish_connect( + sess: WebcrtSession, + *, + creds: dict[str, Any], + device: dict[str, Any], + connect_timeout: int, + client: str, +) -> None: + log_buf = io.BytesIO() + try: + conn = open_netmiko_connection( + creds, + session_timeout=connect_timeout, + session_log=log_buf, + cols=sess.cols, + rows=sess.rows, + interactive=True, + ) + except Exception as exc: + partial = _session_log_text(log_buf).strip() + from .ne_cli_errors import format_cli_failure + + classified = format_cli_failure(exc, partial) + detail = f"connect_failed:{classified}" + if partial: + detail = f"{detail}\n--- device transcript ---\n{partial[-4000:]}" + sess.state = "error" + sess.connect_error = detail + sess.connect_finished_at = time.time() + sess._ready_event.set() + _audit( + "session_open_failed", + session_id=sess.session_id, + ne_id=sess.ne_id, + ne_ip=sess.ne_ip, + source=str(device.get("source") or ""), + client=client or "", + error=str(exc)[:500], + transcript_len=len(partial), + ) + return + + channel = getattr(conn, "remote_conn", None) + if channel is not None and hasattr(channel, "resize_pty"): + try: + channel.resize_pty(width=sess.cols, height=sess.rows) + except Exception: + pass + + pre_log = _session_log_text(log_buf) + already_prompted = _looks_like_cli_prompt(pre_log) + _prime_interactive_channel(conn, already_prompted=already_prompted) + bootstrap = prepare_bootstrap_output(_session_log_text(log_buf)) + if not bootstrap.strip(): + try: + more = _drain_channel(conn, rounds=6, wait=0.08) + except Exception: + more = "" + if more: + bootstrap = prepare_bootstrap_output(more) + # Slow Cisco VMs leave extra prompt bytes; discard before the live reader starts. + try: + _drain_raw_channel(conn, duration=0.55) + except Exception: + pass + + hop_guard = get_cli_hop_guard(conn) + sess.conn = conn + sess.cli_hop_guard = bool(hop_guard) + sess.cli_hop_prompt = str((hop_guard or {}).get("hop_prompt") or "") + sess.bootstrap_output = _encode_text(str(bootstrap or ""), sess.encoding) + # Bootstrap already ends at a prompt → never nudge another Enter on WS attach. + sess.needs_live_prompt = not _looks_like_cli_prompt(bootstrap) + sess.open_session_log() + if bootstrap: + sess.append_session_log(bootstrap if bootstrap.endswith("\n") else bootstrap + "\n") + sess.start_reader() + # Drop late prompt echoes that race into the queue right after reader start. + prompt_hint = "" + if bootstrap: + prompt_hint = str(bootstrap).replace("\r\n", "\n").replace("\r", "\n").strip().split("\n")[-1].strip() + settle_deadline = time.time() + 0.45 + while time.time() < settle_deadline: + try: + chunk = sess.out_queue.get_nowait() + except queue.Empty: + time.sleep(0.02) + continue + if chunk is None: + sess.out_queue.put(None) + break + try: + text = _decode_bytes(chunk, sess.encoding) + except Exception: + text = "" + if _is_prompt_only_echo(text, prompt_hint): + continue + # Non-prompt data: put back and stop settling. + sess.out_queue.put(chunk) + break + try: + sess.run_post_login_commands() + except Exception: + _log.debug("post_login failed session=%s", sess.session_id, exc_info=True) + sess.state = "ready" + sess.connect_finished_at = time.time() + sess._ready_event.set() + elapsed_ms = int((sess.connect_finished_at - sess.connect_started_at) * 1000) + _audit( + "session_created", + session_id=sess.session_id, + ne_id=sess.ne_id, + ne_name=sess.ne_name, + ne_ip=sess.ne_ip, + protocol=sess.protocol, + encoding=sess.encoding, + source=str(device.get("source") or ""), + 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), + cli_hop_prompt=str((hop_guard or {}).get("hop_prompt") or ""), + client=client or "", + connect_ms=elapsed_ms, + active=active_session_count(), + ) + + def create_session( db: Session, *, @@ -555,6 +953,9 @@ def create_session( cols: int = 80, rows: int = 24, client: str = "", + encoding: str = "utf-8", + post_login_commands: list[str] | None = None, + async_connect: bool = True, ) -> dict[str, Any]: from .cli_resolve import resolve_cli_target @@ -588,63 +989,8 @@ def create_session( device_type = str(device.get("device_type") or creds.get("device_type") or "") vendor = str(device.get("vendor") or creds.get("vendor") or "") cli_keymap = uses_network_cli_keymap(device_type, vendor) + enc = _normalize_encoding(encoding) - # Capture real Telnet/SSH login I/O (banner, Username/Password, prompts). - log_buf = io.BytesIO() - try: - conn = open_netmiko_connection( - creds, - session_timeout=connect_timeout, - session_log=log_buf, - cols=c, - rows=r, - interactive=True, - ) - except Exception as exc: - partial = _session_log_text(log_buf).strip() - _audit( - "session_open_failed", - session_id=session_id, - ne_id=target_id, - ne_ip=target_ip, - source=str(device.get("source") or ""), - client=client or "", - error=str(exc)[:500], - transcript_len=len(partial), - ) - from .ne_cli_errors import format_cli_failure - - classified = format_cli_failure(exc, partial) - detail = f"connect_failed:{classified}" - if partial: - # Keep detail bounded; UI surfaces this on open failure. - detail = f"{detail}\n--- device transcript ---\n{partial[-4000:]}" - raise HTTPException(status_code=502, detail=detail) from exc - - channel = getattr(conn, "remote_conn", None) - if channel is not None and hasattr(channel, "resize_pty"): - try: - channel.resize_pty(width=c, height=r) - except Exception: - pass - - # Prefer session_log (full login transcript including final prompt). - _prime_interactive_channel(conn) - bootstrap = prepare_bootstrap_output(_session_log_text(log_buf)) - if not bootstrap.strip(): - try: - more = _drain_channel(conn, rounds=6, wait=0.1) - except Exception: - more = "" - if more: - bootstrap = prepare_bootstrap_output(more) - # Discard any unread bytes so the live reader starts clean. - try: - _drain_channel(conn, rounds=3, wait=0.05) - except Exception: - pass - - hop_guard = get_cli_hop_guard(conn) sess = WebcrtSession( session_id=session_id, ne_id=target_id, @@ -656,34 +1002,51 @@ def create_session( device_type=device_type, vendor=vendor, cli_keymap=cli_keymap, - conn=conn, - bootstrap_output=str(bootstrap or "").encode("utf-8", errors="replace"), - # Only nudge a live prompt when transcript has no recognizable prompt yet. - needs_live_prompt=not _looks_like_cli_prompt(bootstrap), - cli_hop_guard=bool(hop_guard), - cli_hop_prompt=str((hop_guard or {}).get("hop_prompt") or ""), + encoding=enc, + state="connecting", + post_login_commands=list(post_login_commands or [])[:20], ) - # Keep bootstrap for WS attach replay; do not rely solely on out_queue (StrictMode remount). - sess.start_reader() - with _sessions_lock: _sessions[session_id] = sess _audit( - "session_created", + "session_connecting", session_id=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 ""), - 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), - cli_hop_prompt=str((hop_guard or {}).get("hop_prompt") or ""), + encoding=enc, client=client or "", - active=active_session_count(), + async_connect=bool(async_connect), ) + + if async_connect: + t = threading.Thread( + target=_finish_connect, + kwargs={ + "sess": sess, + "creds": creds, + "device": device, + "connect_timeout": connect_timeout, + "client": client or "", + }, + name=f"webcrt-connect-{session_id[:8]}", + daemon=True, + ) + t.start() + else: + _finish_connect( + sess, + creds=creds, + device=device, + connect_timeout=connect_timeout, + client=client or "", + ) + if sess.state == "error": + with _sessions_lock: + _sessions.pop(session_id, None) + raise HTTPException(status_code=502, detail=sess.connect_error or "connect_failed") + return { "session_id": session_id, "ne_id": sess.ne_id, @@ -693,8 +1056,10 @@ def create_session( "protocol": sess.protocol, "cols": sess.cols, "rows": sess.rows, + "encoding": enc, + "state": sess.state, "ws_path": f"/v1/webcrt/sessions/{session_id}/ws", - "cli_hop": bool(hop_guard), + "cli_hop": bool(sess.cli_hop_guard), } @@ -715,6 +1080,7 @@ def mark_attached(session_id: str) -> tuple[WebcrtSession, int]: ne_id=sess.ne_id, ne_ip=sess.ne_ip, attach_gen=attach_gen, + state=sess.state, ) return sess, attach_gen @@ -769,6 +1135,9 @@ def close_session(session_id: str, *, reason: str = "closed", client: str = "") ne_ip=sess.ne_ip, reason=reason, client=client or "", + bytes_in=sess.bytes_in, + bytes_out=sess.bytes_out, + queue_dropped=getattr(sess.out_queue, "dropped", 0), active=active_session_count(), ) return {"ok": True, "session_id": session_id, "closed": True, "reason": reason} @@ -783,9 +1152,20 @@ def list_sessions() -> dict[str, Any]: "ne_name": s.ne_name, "ne_ip": s.ne_ip, "protocol": s.protocol, + "encoding": s.encoding, + "state": s.state, "attached": s.attached, "created_at": datetime.fromtimestamp(s.created_at, tz=timezone.utc).isoformat(), "last_activity": datetime.fromtimestamp(s.last_activity, tz=timezone.utc).isoformat(), + "bytes_in": s.bytes_in, + "bytes_out": s.bytes_out, + "queue_depth": s.out_queue.qsize(), + "queue_dropped": getattr(s.out_queue, "dropped", 0), + "connect_ms": ( + int((s.connect_finished_at - s.connect_started_at) * 1000) + if s.connect_finished_at + else None + ), } for s in _sessions.values() if not s.closed @@ -794,5 +1174,7 @@ def list_sessions() -> dict[str, Any]: "total": len(items), "max_sessions": max(1, int(settings.webcrt_max_sessions or 20)), "idle_timeout_sec": max(60, int(settings.webcrt_idle_timeout_sec or 1800)), + "keepalive_sec": int(getattr(settings, "webcrt_keepalive_sec", 30) or 0), + "anti_idle_sec": int(getattr(settings, "webcrt_anti_idle_sec", 0) or 0), "items": items, } diff --git a/netx_api/webcrt_sftp.py b/netx_api/webcrt_sftp.py new file mode 100644 index 0000000..8a44448 --- /dev/null +++ b/netx_api/webcrt_sftp.py @@ -0,0 +1,189 @@ +"""Lightweight SFTP helpers for WebCRT (SSH targets only; separate from interactive PTY).""" + +from __future__ import annotations + +import logging +import posixpath +from typing import Any + +import paramiko +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from .cli_resolve import resolve_cli_target +from .config import settings +from .ne_crypto import CredentialCryptoError +from .webcrt_service import _webcrt_creds_ready + +_log = logging.getLogger("netx.webcrt.sftp") + + +def _require_ssh_direct(creds: dict[str, Any], device: dict[str, Any]) -> None: + protocol = str(device.get("protocol") or creds.get("protocol") or "ssh").lower() + if protocol != "ssh": + raise HTTPException(status_code=400, detail="sftp_requires_ssh") + if creds.get("hop_enabled"): + # Keep v1 simple: SFTP only for direct SSH (no hop/proxy jump). + raise HTTPException(status_code=400, detail="sftp_hop_not_supported") + + +def _open_sftp(creds: dict[str, Any]) -> tuple[paramiko.SSHClient, paramiko.SFTPClient]: + timeout = int(settings.ne_connect_timeout_sec or 30) + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + try: + client.connect( + str(creds["ip_address"]), + port=int(creds.get("port") or 22), + username=str(creds["username"]), + password=str(creds["password"]), + timeout=timeout, + banner_timeout=timeout, + auth_timeout=timeout, + look_for_keys=False, + allow_agent=False, + ) + sftp = client.open_sftp() + return client, sftp + except HTTPException: + try: + client.close() + except Exception: + pass + raise + except Exception as exc: + try: + client.close() + except Exception: + pass + raise HTTPException(status_code=502, detail=f"sftp_connect_failed:{exc}") from exc + + +def _resolve(db: Session, *, managed_ne_id: str | None, ume_ne_id: str | None) -> tuple[dict[str, Any], dict[str, Any]]: + try: + creds, device = resolve_cli_target(db, managed_ne_id=managed_ne_id, ume_ne_id=ume_ne_id) + except HTTPException: + raise + except CredentialCryptoError as exc: + raise HTTPException(status_code=400, detail=str(exc) or "credential_crypto_error") from exc + except Exception as exc: + raise HTTPException(status_code=400, detail=f"credential_error:{exc}") from exc + if not _webcrt_creds_ready(creds): + raise HTTPException(status_code=400, detail="credentials_incomplete") + _require_ssh_direct(creds, device) + return creds, device + + +def sftp_list( + db: Session, + *, + managed_ne_id: str | None, + ume_ne_id: str | None, + path: str = ".", +) -> dict[str, Any]: + creds, device = _resolve(db, managed_ne_id=managed_ne_id, ume_ne_id=ume_ne_id) + remote = str(path or ".").strip() or "." + client, sftp = _open_sftp(creds) + try: + entries = [] + for attr in sftp.listdir_attr(remote): + mode = int(getattr(attr, "st_mode", 0) or 0) + is_dir = bool(mode & 0o40000) + entries.append( + { + "name": attr.filename, + "size": int(getattr(attr, "st_size", 0) or 0), + "mtime": int(getattr(attr, "st_mtime", 0) or 0), + "is_dir": is_dir, + } + ) + entries.sort(key=lambda x: (not x["is_dir"], str(x["name"]).lower())) + return { + "ne_id": str(device.get("id") or ""), + "ne_name": str(device.get("name") or ""), + "path": remote, + "items": entries, + } + except HTTPException: + raise + except Exception as exc: + raise HTTPException(status_code=502, detail=f"sftp_list_failed:{exc}") from exc + finally: + try: + sftp.close() + except Exception: + pass + try: + client.close() + except Exception: + pass + + +def sftp_download( + db: Session, + *, + managed_ne_id: str | None, + ume_ne_id: str | None, + path: str, +) -> tuple[bytes, str]: + creds, _device = _resolve(db, managed_ne_id=managed_ne_id, ume_ne_id=ume_ne_id) + remote = str(path or "").strip() + if not remote or remote.endswith("/"): + raise HTTPException(status_code=400, detail="sftp_path_required") + client, sftp = _open_sftp(creds) + try: + with sftp.open(remote, "rb") as fh: + data = fh.read(8 * 1024 * 1024 + 1) + if len(data) > 8 * 1024 * 1024: + raise HTTPException(status_code=413, detail="sftp_file_too_large") + return data, posixpath.basename(remote) or "download.bin" + except HTTPException: + raise + except Exception as exc: + raise HTTPException(status_code=502, detail=f"sftp_download_failed:{exc}") from exc + finally: + try: + sftp.close() + except Exception: + pass + try: + client.close() + except Exception: + pass + + +def sftp_upload( + db: Session, + *, + managed_ne_id: str | None, + ume_ne_id: str | None, + remote_path: str, + data: bytes, +) -> dict[str, Any]: + creds, device = _resolve(db, managed_ne_id=managed_ne_id, ume_ne_id=ume_ne_id) + remote = str(remote_path or "").strip() + if not remote: + raise HTTPException(status_code=400, detail="sftp_path_required") + client, sftp = _open_sftp(creds) + try: + with sftp.open(remote, "wb") as fh: + fh.write(data) + return { + "ok": True, + "ne_id": str(device.get("id") or ""), + "path": remote, + "size": len(data), + } + except HTTPException: + raise + except Exception as exc: + raise HTTPException(status_code=502, detail=f"sftp_upload_failed:{exc}") from exc + finally: + try: + sftp.close() + except Exception: + pass + try: + client.close() + except Exception: + pass diff --git a/tests/test_webcrt.py b/tests/test_webcrt.py index bd93fac..3f5dec8 100644 --- a/tests/test_webcrt.py +++ b/tests/test_webcrt.py @@ -15,10 +15,13 @@ class _FakeConn: def __init__(self) -> None: self.written: list[str] = [] self.RETURN = "\n" - self.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty"]) + self.remote_conn = MagicMock(spec=["recv_ready", "recv", "exit_status_ready", "resize_pty", "send_break", "send"]) self.remote_conn.recv_ready.return_value = False self.remote_conn.exit_status_ready.return_value = False self.remote_conn.resize_pty = MagicMock() + self.remote_conn.send_break = MagicMock() + # No send by default so write_stdin uses write_channel in unit tests. + del self.remote_conn.send def write_channel(self, data: str) -> None: self.written.append(data) @@ -68,6 +71,9 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(conn.written[-1], "\x1b[D") sess.resize(120, 40) conn.remote_conn.resize_pty.assert_called_with(width=120, height=40) + conn.remote_conn.send_break = MagicMock() + sess.send_break() + conn.remote_conn.send_break.assert_called() sess.close("test") self.assertTrue(sess.closed) @@ -90,6 +96,11 @@ class WebcrtServiceTests(unittest.TestCase): self.assertEqual(svc.normalize_cli_transcript("banner\nR2#R2#"), "banner\nR2#") self.assertEqual(svc.prepare_bootstrap_output("login\nR2#\nR2#"), "login\nR2#") self.assertTrue(svc.prepare_bootstrap_output("login\nR2#").endswith("R2#")) + # Slow VM / prime Enter can leave three identical prompts. + self.assertEqual(svc.prepare_bootstrap_output("banner\nR2#\nR2#\nR2#"), "banner\nR2#") + self.assertEqual(svc.prepare_bootstrap_output("banner\nR2#\n\nR2#"), "banner\nR2#") + self.assertTrue(svc._is_prompt_only_echo("\r\nR2#\r\n", "R2#")) + self.assertFalse(svc._is_prompt_only_echo("R2#show clock\r\n", "R2#")) @patch.object(svc, "_audit") @patch.object(svc, "open_netmiko_connection") @@ -120,11 +131,12 @@ class WebcrtServiceTests(unittest.TestCase): db = MagicMock() with patch.object(svc.settings, "webcrt_max_sessions", 1): - out = svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test") + out = svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test", async_connect=False) self.assertIn("session_id", out) + self.assertEqual(out.get("state"), "ready") self.assertFalse(out.get("cli_hop")) with self.assertRaises(HTTPException) as ctx: - svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test") + svc.create_session(db, ne_id="ne-a", cols=80, rows=24, client="test", async_connect=False) self.assertEqual(ctx.exception.status_code, 429) @patch.object(svc, "_audit") @@ -172,7 +184,7 @@ class WebcrtServiceTests(unittest.TestCase): mock_open.side_effect = _open_with_log db = MagicMock() - out = svc.create_session(db, ne_id="ne-hop", cols=100, rows=30, client="test") + out = svc.create_session(db, ne_id="ne-hop", cols=100, rows=30, client="test", async_connect=False) mock_open.assert_called_once() called_creds = mock_open.call_args.args[0] self.assertTrue(called_creds["hop_enabled"]) @@ -222,13 +234,15 @@ class WebcrtServiceTests(unittest.TestCase): ) mock_open.side_effect = lambda *a, **k: _FakeConn() mock_guard.return_value = {"hop_prompt": "", "hop_vendor": "huawei", "hop_host": "1.1.1.1"} - out = svc.create_session(MagicMock(), ne_id="ne-cli-hop", cols=100, rows=30, client="test") - self.assertTrue(out.get("cli_hop")) - self.assertEqual(mock_open.call_args.kwargs.get("cols"), 100) - self.assertEqual(mock_open.call_args.kwargs.get("rows"), 30) + out = svc.create_session( + MagicMock(), ne_id="ne-cli-hop", cols=100, rows=30, client="test", async_connect=False + ) sess = svc.get_session(out["session_id"]) assert sess is not None self.assertTrue(sess.cli_hop_guard) + self.assertTrue(out.get("cli_hop") or sess.cli_hop_guard) + self.assertEqual(mock_open.call_args.kwargs.get("cols"), 100) + self.assertEqual(mock_open.call_args.kwargs.get("rows"), 30) self.assertEqual(sess.cli_hop_prompt, "") svc.close_session(out["session_id"], reason="test") @@ -265,7 +279,9 @@ class WebcrtServiceTests(unittest.TestCase): }, ) mock_open.return_value = _FakeConn() - out = svc.create_session(MagicMock(), ne_id="ne-bastion", cols=80, rows=24, client="test") + out = svc.create_session( + MagicMock(), ne_id="ne-bastion", cols=80, rows=24, client="test", async_connect=False + ) mock_open.assert_called_once() self.assertEqual(out["ne_id"], "ne-bastion") svc.close_session(out["session_id"], reason="test") @@ -466,6 +482,19 @@ class WebcrtServiceTests(unittest.TestCase): sess._note_cli_hop_output(b"Connection closed by foreign host\r\n\r\n") ) + def test_bounded_queue_drops_oldest(self) -> None: + q = svc._BoundedByteQueue(maxsize=8) + for i in range(10): + q.put(str(i).encode()) + self.assertGreaterEqual(q.dropped, 2) + first = q.get_nowait() + self.assertEqual(first, b"2") + + def test_normalize_encoding(self) -> None: + self.assertEqual(svc._normalize_encoding("GBK"), "gbk") + self.assertEqual(svc._normalize_encoding("utf8"), "utf-8") + self.assertEqual(svc._encode_text("测", "gbk")[:1], b"\xb2") + if __name__ == "__main__": unittest.main() diff --git a/web/WEB.md b/web/WEB.md index 8444020..6715dbe 100644 --- a/web/WEB.md +++ b/web/WEB.md @@ -139,12 +139,14 @@ src/ ## WebCRT -- API：`POST /v1/webcrt/sessions`（`ne_id` 或 `ume_ne_id`）、`WS /v1/webcrt/sessions/{id}/ws`、`DELETE /v1/webcrt/sessions/{id}` +- API：`POST /v1/webcrt/sessions`（`ne_id` 或 `ume_ne_id`；可选 `encoding`/`post_login_commands`/`async_connect`）、`WS /v1/webcrt/sessions/{id}/ws`、`DELETE /v1/webcrt/sessions/{id}` +- SFTP（直连 SSH）：`POST /v1/webcrt/sftp/list|download|upload` - 目标列表复用 `/v1/cli/targets`（托管 + UME，搜索分页；`source=all|managed|ume`） - 凭据：托管走网元自身账号；UME 走 CLI 连接模板（`resolve_cli_target`） -- 前端：CRT 风格左右分栏（会话管理 + 多标签终端） -- 审计：`NETX_WEBCRT_DATA_DIR`（默认 `data/webcrt/audit.jsonl`） -- 限流：`NETX_WEBCRT_MAX_SESSIONS`、`NETX_WEBCRT_IDLE_TIMEOUT_SEC` +- 前端：CRT 风格左右分栏（收藏/最近 + 多标签终端）；粘贴节流、选区复制、Break、Ctrl+F、Button Bar、广播、编码/字号 +- WS：stdout 二进制帧合批；stdin 短合批；建连可异步（POST 立即返回，进度经 WS `connecting`→`connected`） +- 审计 / 会话日志：`NETX_WEBCRT_DATA_DIR`（`audit.jsonl` + `sessions/*.log`） +- 限流 / 保活：`NETX_WEBCRT_MAX_SESSIONS`、`NETX_WEBCRT_IDLE_TIMEOUT_SEC`、`NETX_WEBCRT_KEEPALIVE_SEC`、`NETX_WEBCRT_ANTI_IDLE_SEC` ## Toast diff --git a/web/src/components/WebTerminal.tsx b/web/src/components/WebTerminal.tsx index c7ac2c4..34572e7 100644 --- a/web/src/components/WebTerminal.tsx +++ b/web/src/components/WebTerminal.tsx @@ -1,4 +1,4 @@ -import { forwardRef, useEffect, useImperativeHandle, useRef } from "react"; +import { forwardRef, useEffect, useImperativeHandle, useRef, useState } from "react"; import { FitAddon } from "@xterm/addon-fit"; import { Terminal } from "@xterm/xterm"; import "@xterm/xterm/css/xterm.css"; @@ -6,9 +6,14 @@ import "@xterm/xterm/css/xterm.css"; export type WebTerminalHandle = { clear: () => void; copyAll: () => Promise; + copySelection: () => Promise; getText: () => string; fit: () => void; focus: () => void; + sendBreak: () => void; + sendText: (data: string, opts?: { throttle?: boolean }) => void; + findNext: (term: string) => void; + findPrevious: (term: string) => void; }; type Props = { @@ -16,6 +21,9 @@ type Props = { title?: string; recording?: boolean; autoFocus?: boolean; + encoding?: string; + pasteDelayMs?: number; + copyOnSelect?: boolean; onStatus?: (state: string, message?: string) => void; onReady?: () => void; onStdout?: (data: string) => void; @@ -35,11 +43,10 @@ function serializeTerminal(term: Terminal): string { function isSidebarSearchTarget(target: EventTarget | null): boolean { if (!(target instanceof HTMLElement)) return false; if (target.tagName === "INPUT" || target.tagName === "TEXTAREA" || target.tagName === "SELECT") { - // xterm's hidden textarea must still receive keys normally. if (target.classList.contains("xterm-helper-textarea")) return false; return true; } - return Boolean(target.closest(".webcrt-sidebar__search")); + return Boolean(target.closest(".webcrt-sidebar__search") || target.closest(".webcrt-ctx") || target.closest(".webcrt-find")); } function isXtermTextarea(target: EventTarget | null): boolean { @@ -49,7 +56,7 @@ function isXtermTextarea(target: EventTarget | null): boolean { /** Map a browser key event to the bytes xterm/onData would normally emit. */ function keyEventToStdin(e: KeyboardEvent): string | null { if (e.ctrlKey || e.altKey || e.metaKey) return null; - if (e.key === "Backspace") return "\x08"; // BS — common SecureCRT/VT default + if (e.key === "Backspace") return "\x08"; if (e.key === "Enter") return "\r"; if (e.key === "Tab") return "\t"; if (e.key === "Escape") return "\x1b"; @@ -64,19 +71,65 @@ function keyEventToStdin(e: KeyboardEvent): string | null { return null; } +function decodeBytes(buf: ArrayBuffer, _encoding: string): string { + // Server normalizes device encodings (e.g. GBK) to UTF-8 on the wire. + try { + return new TextDecoder("utf-8").decode(buf); + } catch { + return ""; + } +} + +const PREF_KEY = "netx.webcrt.termPrefs"; + +function loadPrefs(): { copyOnSelect: boolean; pasteDelayMs: number } { + try { + const raw = localStorage.getItem(PREF_KEY); + if (!raw) return { copyOnSelect: true, pasteDelayMs: 40 }; + const j = JSON.parse(raw) as Partial<{ copyOnSelect: boolean; pasteDelayMs: number }>; + return { + copyOnSelect: j.copyOnSelect !== false, + pasteDelayMs: Math.max(0, Math.min(200, Number(j.pasteDelayMs) || 40)), + }; + } catch { + return { copyOnSelect: true, pasteDelayMs: 40 }; + } +} + export const WebTerminal = forwardRef(function WebTerminal( - { wsUrl, title, recording, autoFocus = true, onStatus, onReady, onStdout }, + { + wsUrl, + title, + recording, + autoFocus = true, + encoding = "utf-8", + pasteDelayMs, + copyOnSelect, + onStatus, + onReady, + onStdout, + }, ref, ) { const hostRef = useRef(null); const termRef = useRef(null); const fitRef = useRef(null); const wsRef = useRef(null); + const findIndexRef = useRef(0); const onStatusRef = useRef(onStatus); const onReadyRef = useRef(onReady); const onStdoutRef = useRef(onStdout); const recordingRef = useRef(!!recording); const autoFocusRef = useRef(autoFocus); + const encodingRef = useRef(encoding); + const pasteDelayRef = useRef(pasteDelayMs ?? loadPrefs().pasteDelayMs); + const copyOnSelectRef = useRef(copyOnSelect ?? loadPrefs().copyOnSelect); + const pasteQueueRef = useRef>(Promise.resolve()); + const [ctxMenu, setCtxMenu] = useState<{ x: number; y: number } | null>(null); + const [findOpen, setFindOpen] = useState(false); + const [findQuery, setFindQuery] = useState(""); + const [pasteStatus, setPasteStatus] = useState<{ done: number; total: number } | null>(null); + const findInputRef = useRef(null); useEffect(() => { onStatusRef.current = onStatus; @@ -92,6 +145,18 @@ export const WebTerminal = forwardRef(function WebTerm autoFocusRef.current = autoFocus; }, [autoFocus]); + useEffect(() => { + encodingRef.current = encoding; + }, [encoding]); + + useEffect(() => { + if (pasteDelayMs != null) pasteDelayRef.current = pasteDelayMs; + }, [pasteDelayMs]); + + useEffect(() => { + if (copyOnSelect != null) copyOnSelectRef.current = copyOnSelect; + }, [copyOnSelect]); + const focusTerminal = () => { try { termRef.current?.focus(); @@ -100,6 +165,46 @@ export const WebTerminal = forwardRef(function WebTerm } }; + const sendJson = (payload: Record) => { + const ws = wsRef.current; + if (ws && ws.readyState === WebSocket.OPEN) { + ws.send(JSON.stringify(payload)); + } + }; + + const sendStdinImmediate = (data: string) => { + if (!data) return; + sendJson({ type: "stdin", data }); + }; + + /** SecureCRT-like paste: send line-by-line with delay; show progress while sending. */ + const sendStdinThrottled = (data: string) => { + if (!data) return; + const delay = pasteDelayRef.current; + if (delay <= 0 || data.length < 8) { + sendStdinImmediate(data); + return; + } + pasteQueueRef.current = pasteQueueRef.current.then(async () => { + const lines = data.replace(/\r\n/g, "\n").replace(/\r/g, "\n").split("\n"); + const total = Math.max(1, lines.length); + setPasteStatus({ done: 0, total }); + try { + for (let i = 0; i < lines.length; i += 1) { + const line = lines[i]; + const chunk = i < lines.length - 1 ? `${line}\r` : line; + if (chunk) sendStdinImmediate(chunk); + setPasteStatus({ done: i + 1, total }); + if (i < lines.length - 1) { + await new Promise((r) => window.setTimeout(r, delay)); + } + } + } finally { + setPasteStatus(null); + } + }); + }; + useImperativeHandle(ref, () => ({ clear: () => { termRef.current?.clear(); @@ -112,6 +217,13 @@ export const WebTerminal = forwardRef(function WebTerm } return text; }, + copySelection: async () => { + const text = termRef.current?.getSelection() || ""; + if (text && navigator.clipboard?.writeText) { + await navigator.clipboard.writeText(text); + } + return text; + }, getText: () => (termRef.current ? serializeTerminal(termRef.current) : ""), fit: () => { try { @@ -121,6 +233,39 @@ export const WebTerminal = forwardRef(function WebTerm } }, focus: focusTerminal, + sendBreak: () => sendJson({ type: "break" }), + sendText: (data: string, opts?: { throttle?: boolean }) => { + if (opts?.throttle) sendStdinThrottled(data); + else sendStdinImmediate(data); + }, + findNext: (q: string) => { + const term = termRef.current; + if (!term || !q) return; + const text = serializeTerminal(term).toLowerCase(); + const needle = q.toLowerCase(); + let idx = text.indexOf(needle, findIndexRef.current + 1); + if (idx < 0) idx = text.indexOf(needle); + if (idx >= 0) { + findIndexRef.current = idx; + // Approximate scroll: each buffer line ~1 row. + const line = text.slice(0, idx).split("\n").length - 1; + term.scrollToLine(Math.max(0, line - 2)); + } + }, + findPrevious: (q: string) => { + const term = termRef.current; + if (!term || !q) return; + const text = serializeTerminal(term).toLowerCase(); + const needle = q.toLowerCase(); + const before = text.slice(0, Math.max(0, findIndexRef.current)); + let idx = before.lastIndexOf(needle); + if (idx < 0) idx = text.lastIndexOf(needle); + if (idx >= 0) { + findIndexRef.current = idx; + const line = text.slice(0, idx).split("\n").length - 1; + term.scrollToLine(Math.max(0, line - 2)); + } + }, })); useEffect(() => { @@ -131,10 +276,12 @@ export const WebTerminal = forwardRef(function WebTerm cursorBlink: true, fontSize: 13, fontFamily: 'Consolas, "Courier New", monospace', + scrollback: 10000, theme: { background: "#0b1220", foreground: "#e2e8f0", cursor: "#e2e8f0", + selectionBackground: "#334155", }, convertEol: true, }); @@ -155,7 +302,6 @@ export const WebTerminal = forwardRef(function WebTerm if (!autoFocusRef.current) return; focusTerminal(); }; - // Focus immediately so the first keystroke is not lost to the sidebar/body. maybeFocus(); requestAnimationFrame(() => { doFit(); @@ -167,22 +313,10 @@ export const WebTerminal = forwardRef(function WebTerm }); const ws = new WebSocket(wsUrl); + ws.binaryType = "arraybuffer"; wsRef.current = ws; onStatusRef.current?.("connecting"); - const sendJson = (payload: Record) => { - if (ws.readyState === WebSocket.OPEN) { - ws.send(JSON.stringify(payload)); - } - }; - - const sendStdin = (data: string) => { - if (!data) return; - sendJson({ type: "stdin", data }); - }; - - // Display only what the device echoes (chars, Tab completion, BS erase, etc.). - // No local echo / local erase — those desync on Tab complete and prompt redraw. const writeStdout = (raw: string) => { if (raw) term.write(raw); if (recordingRef.current) onStdoutRef.current?.(raw); @@ -201,6 +335,18 @@ export const WebTerminal = forwardRef(function WebTerm }; ws.onmessage = (ev) => { + if (ev.data instanceof ArrayBuffer) { + writeStdout(decodeBytes(ev.data, encodingRef.current)); + maybeFocus(); + return; + } + if (typeof Blob !== "undefined" && ev.data instanceof Blob) { + void ev.data.arrayBuffer().then((buf) => { + writeStdout(decodeBytes(buf, encodingRef.current)); + maybeFocus(); + }); + return; + } try { const msg = JSON.parse(String(ev.data || "{}")) as { type?: string; @@ -215,7 +361,7 @@ export const WebTerminal = forwardRef(function WebTerm } if (msg.type === "status") { onStatusRef.current?.(String(msg.state || ""), msg.message); - if (msg.state === "connected") { + if (msg.state === "connected" || msg.state === "connecting") { maybeFocus(); return; } @@ -243,27 +389,41 @@ export const WebTerminal = forwardRef(function WebTerm } }; - // When focused, xterm onData sends keystrokes. const dataDisposable = term.onData((data) => { - // Normalize Backspace: xterm emits DEL(0x7f); devices expect BS(0x08) like default SecureCRT. const normalized = data.replace(/\x7f/g, "\x08"); - sendStdin(normalized); + // Large pastes from xterm arrive as one onData blob. + if (normalized.length > 32 || normalized.includes("\r") || normalized.includes("\n")) { + sendStdinThrottled(normalized); + } else { + sendStdinImmediate(normalized); + } + }); + + const selDisposable = term.onSelectionChange(() => { + if (!copyOnSelectRef.current) return; + const sel = term.getSelection(); + if (sel && navigator.clipboard?.writeText) { + void navigator.clipboard.writeText(sel).catch(() => undefined); + } }); - // When NOT focused (sidebar still focused after click), capture keys and forward - // so the first character / Backspace are not lost to the browser. const onKeyDownCapture = (e: KeyboardEvent) => { if (!autoFocusRef.current) return; if (host.closest("[hidden]")) return; if (isSidebarSearchTarget(e.target)) return; - // Always block browser "Backspace = history back" while this session pane is active. + if ((e.ctrlKey || e.metaKey) && (e.key === "f" || e.key === "F")) { + e.preventDefault(); + setFindOpen(true); + window.setTimeout(() => findInputRef.current?.focus(), 0); + return; + } + if (e.key === "Backspace") { e.preventDefault(); } if (isXtermTextarea(e.target)) { - // Let xterm onData handle it; we only prevented browser back above. maybeFocus(); return; } @@ -274,10 +434,31 @@ export const WebTerminal = forwardRef(function WebTerm e.preventDefault(); e.stopPropagation(); maybeFocus(); - sendStdin(data); + sendStdinImmediate(data); }; window.addEventListener("keydown", onKeyDownCapture, true); + const onContextMenu = (e: MouseEvent) => { + e.preventDefault(); + setCtxMenu({ x: e.clientX, y: e.clientY }); + }; + host.addEventListener("contextmenu", onContextMenu); + + const onPasteCapture = (e: ClipboardEvent) => { + if (!autoFocusRef.current) return; + if (host.closest("[hidden]")) return; + if (isSidebarSearchTarget(e.target)) return; + const text = e.clipboardData?.getData("text") || ""; + if (!text) return; + // Intercept paste so we always apply delay (even if focus is odd). + if (document.activeElement === term.textarea || host.contains(document.activeElement) || autoFocusRef.current) { + e.preventDefault(); + sendStdinThrottled(text); + maybeFocus(); + } + }; + window.addEventListener("paste", onPasteCapture, true); + const pingTimer = window.setInterval(() => { sendJson({ type: "ping" }); }, 25000); @@ -297,8 +478,11 @@ export const WebTerminal = forwardRef(function WebTerm window.clearInterval(pingTimer); window.removeEventListener("resize", onWinResize); window.removeEventListener("keydown", onKeyDownCapture, true); + window.removeEventListener("paste", onPasteCapture, true); + host.removeEventListener("contextmenu", onContextMenu); ro?.disconnect(); dataDisposable.dispose(); + selDisposable.dispose(); try { ws.close(); } catch { @@ -311,13 +495,147 @@ export const WebTerminal = forwardRef(function WebTerm }; }, [wsUrl, title]); + const pasteFromClipboard = async () => { + try { + const text = await navigator.clipboard.readText(); + sendStdinThrottled(text); + } catch { + /* ignore */ + } + setCtxMenu(null); + focusTerminal(); + }; + + const copySelection = async () => { + const text = termRef.current?.getSelection() || ""; + if (text && navigator.clipboard?.writeText) { + await navigator.clipboard.writeText(text); + } + setCtxMenu(null); + }; + return ( -
{ - focusTerminal(); - }} - /> +
+ {pasteStatus ? ( +
+ 粘贴中 {pasteStatus.done}/{pasteStatus.total} 行… +
+ ) : null} + {findOpen ? ( +
+ setFindQuery(e.target.value)} + onKeyDown={(e) => { + if (e.key === "Enter") { + e.preventDefault(); + const term = termRef.current; + if (!term || !findQuery) return; + const text = serializeTerminal(term).toLowerCase(); + const needle = findQuery.toLowerCase(); + if (e.shiftKey) { + const before = text.slice(0, Math.max(0, findIndexRef.current)); + let idx = before.lastIndexOf(needle); + if (idx < 0) idx = text.lastIndexOf(needle); + if (idx >= 0) { + findIndexRef.current = idx; + term.scrollToLine(Math.max(0, text.slice(0, idx).split("\n").length - 3)); + } + } else { + let idx = text.indexOf(needle, findIndexRef.current + 1); + if (idx < 0) idx = text.indexOf(needle); + if (idx >= 0) { + findIndexRef.current = idx; + term.scrollToLine(Math.max(0, text.slice(0, idx).split("\n").length - 3)); + } + } + } else if (e.key === "Escape") { + setFindOpen(false); + focusTerminal(); + } + }} + /> + + + +
+ ) : null} +
{ + focusTerminal(); + setCtxMenu(null); + }} + /> + {ctxMenu ? ( +
e.stopPropagation()} + > + + + +
+ ) : null} +
); }); diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index fd0e087..296fd23 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -939,6 +939,25 @@ const en = { copied: "Copied to clipboard", copyEmpty: "Terminal is empty", copyFailed: "Copy failed", + break: "Break", + breakSent: "Break sent", + broadcast: "Broadcast", + broadcastSent: "Sent to all sessions", + postLogin: "Post-login cmds", + }, + favorites: "Favorites", + recent: "Recent", + favorite: "Toggle favorite", + encoding: "Encoding", + fontSize: "Font", + broadcastPh: "Send to all tabs…", + buttonBarHint: "One per line: label|command", + postLoginPrompt: "Post-login commands (one per line, this device)", + postLoginSaved: "Post-login commands saved", + sftp: { + refresh: "Refresh", + upload: "Upload", + uploaded: "Upload complete", }, err: { sessionLimit: "Session limit reached; close other terminals first", @@ -946,6 +965,8 @@ const en = { credsIncomplete: "Incomplete username/password on this NE", cliProfile: "UME NEs require a CLI connect profile (default account / hop)", connectFailed: "Device login failed: {{detail}}", + sftpHop: "SFTP does not support hop devices yet (direct SSH only)", + sftpSsh: "SFTP requires SSH", }, }, topology: { diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index 4ab5e6b..3f3616a 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -936,6 +936,25 @@ const zh = { copied: "已复制到剪贴板", copyEmpty: "终端内容为空", copyFailed: "复制失败", + break: "Break", + breakSent: "已发送 Break", + broadcast: "广播", + broadcastSent: "已广播到全部会话", + postLogin: "登录后命令", + }, + favorites: "收藏", + recent: "最近", + favorite: "收藏/取消", + encoding: "编码", + fontSize: "字号", + broadcastPh: "发送到全部标签…", + buttonBarHint: "每行：按钮名|命令", + postLoginPrompt: "登录后自动执行（每行一条，仅本设备）", + postLoginSaved: "登录后命令已保存", + sftp: { + refresh: "刷新", + upload: "上传", + uploaded: "上传成功", }, err: { sessionLimit: "并发会话已达上限，请先关闭其他终端", @@ -943,6 +962,8 @@ const zh = { credsIncomplete: "网元账号或密码不完整", cliProfile: "UME 网元需要先配置 CLI 连接模板（默认账号/跳板）", connectFailed: "登录设备失败：{{detail}}", + sftpHop: "SFTP 暂不支持跳板设备，请使用直连 SSH", + sftpSsh: "SFTP 仅支持 SSH 协议", }, }, topology: { diff --git a/web/src/index.css b/web/src/index.css index e825315..3e6d0d9 100644 --- a/web/src/index.css +++ b/web/src/index.css @@ -2289,7 +2289,8 @@ pre { display: none !important; } -.webcrt-main__pane .webcrt-term { +.webcrt-main__pane .webcrt-term, +.webcrt-main__pane .webcrt-term-wrap { flex: 1; width: 100%; height: auto !important; @@ -2332,6 +2333,284 @@ pre { display: none; } +.webcrt-term-wrap { + position: relative; + flex: 1; + min-height: 0; + display: flex; + flex-direction: column; + height: 100%; +} + +.webcrt-term-wrap .webcrt-term { + flex: 1; + min-height: 0; +} + +.webcrt-paste-status { + position: absolute; + top: 8px; + right: 10px; + z-index: 4; + padding: 4px 10px; + font-size: 12px; + color: #e2e8f0; + background: rgba(15, 23, 42, 0.92); + border: 1px solid #475569; + border-radius: 2px; + pointer-events: none; +} + +.webcrt-find { + display: flex; + gap: 4px; + align-items: center; + padding: 4px 8px; + background: #1e293b; + border-bottom: 1px solid #334155; + z-index: 2; +} + +.webcrt-find input { + flex: 1; + min-width: 0; + background: #0f172a; + border: 1px solid #475569; + color: #e2e8f0; + padding: 2px 8px; + font-size: 12px; +} + +.webcrt-find button { + background: #334155; + border: 1px solid #475569; + color: #e2e8f0; + padding: 2px 8px; + cursor: pointer; +} + +.webcrt-ctx { + position: fixed; + z-index: 50; + min-width: 120px; + background: #1e293b; + border: 1px solid #475569; + box-shadow: 0 8px 24px rgba(0, 0, 0, 0.35); + display: flex; + flex-direction: column; + padding: 4px 0; +} + +.webcrt-ctx button { + background: transparent; + border: 0; + color: #e2e8f0; + text-align: left; + padding: 6px 14px; + cursor: pointer; + font-size: 12px; +} + +.webcrt-ctx button:hover { + background: #334155; +} + +.webcrt-tree__row { + display: flex; + align-items: stretch; + gap: 0; +} + +.webcrt-tree__fav { + flex: 0 0 22px; + display: flex; + align-items: center; + justify-content: center; + background: transparent; + border: 0; + color: #94a3b8; + cursor: pointer; + padding: 0; +} + +.webcrt-tree__fav:hover { + color: #fbbf24; +} + +.webcrt-tree__row .webcrt-tree__item { + flex: 1; + min-width: 0; +} + +.webcrt-sidebar__pins { + max-height: 28%; + overflow: auto; + border-bottom: 1px solid #334155; + padding-bottom: 4px; +} + +.webcrt-sidebar__pin-title { + font-size: 11px; + color: #94a3b8; + padding: 6px 10px 2px; + text-transform: uppercase; + letter-spacing: 0.04em; +} + +.webcrt-sidebar__prefs { + display: flex; + gap: 8px; + padding: 8px 10px; + border-top: 1px solid #334155; + font-size: 12px; + color: #cbd5e1; +} + +.webcrt-sidebar__prefs label { + display: flex; + flex-direction: column; + gap: 2px; + flex: 1; +} + +.webcrt-sidebar__prefs select, +.webcrt-sidebar__prefs input { + background: #0f172a; + border: 1px solid #475569; + color: #e2e8f0; + padding: 2px 6px; + font-size: 12px; +} + +.webcrt-buttonbar { + display: flex; + flex-wrap: wrap; + gap: 4px; + align-items: center; + padding: 4px 8px; + background: #1e293b; + border-bottom: 1px solid #334155; +} + +.webcrt-buttonbar__btn { + background: #334155; + border: 1px solid #475569; + color: #e2e8f0; + padding: 2px 8px; + font-size: 11px; + cursor: pointer; + border-radius: 2px; +} + +.webcrt-buttonbar__btn:hover { + background: #475569; +} + +.webcrt-broadcast { + display: flex; + gap: 4px; + margin-left: auto; + align-items: center; +} + +.webcrt-broadcast input { + width: 160px; + background: #0f172a; + border: 1px solid #475569; + color: #e2e8f0; + padding: 2px 6px; + font-size: 11px; +} + +.webcrt-broadcast button { + background: #0ea5e9; + border: 0; + color: #0f172a; + padding: 2px 8px; + font-size: 11px; + cursor: pointer; +} + +.webcrt-buttonbar-edit { + padding: 6px 8px; + background: #0f172a; + border-bottom: 1px solid #334155; +} + +.webcrt-buttonbar-edit textarea { + width: 100%; + background: #1e293b; + border: 1px solid #475569; + color: #e2e8f0; + font-family: Consolas, monospace; + font-size: 12px; + padding: 6px; +} + +.webcrt-sftp { + position: absolute; + right: 0; + top: 0; + bottom: 0; + width: min(320px, 40%); + background: #0f172a; + border-left: 1px solid #334155; + z-index: 3; + display: flex; + flex-direction: column; +} + +.webcrt-sftp__bar { + display: flex; + gap: 4px; + padding: 6px; + border-bottom: 1px solid #334155; +} + +.webcrt-sftp__bar input { + flex: 1; + min-width: 0; + background: #1e293b; + border: 1px solid #475569; + color: #e2e8f0; + padding: 2px 6px; + font-size: 12px; +} + +.webcrt-sftp__bar button, +.webcrt-sftp__upload { + background: #334155; + border: 1px solid #475569; + color: #e2e8f0; + padding: 2px 8px; + font-size: 11px; + cursor: pointer; +} + +.webcrt-sftp__list { + list-style: none; + margin: 0; + padding: 0; + overflow: auto; + flex: 1; +} + +.webcrt-sftp__list button { + width: 100%; + text-align: left; + background: transparent; + border: 0; + border-bottom: 1px solid #1e293b; + color: #e2e8f0; + padding: 6px 10px; + font-size: 12px; + cursor: pointer; +} + +.webcrt-sftp__list button:hover { + background: #1e293b; +} + @media (max-width: 900px) { .webcrt-shell { grid-template-columns: 1fr; diff --git a/web/src/pages/WebcrtPage.tsx b/web/src/pages/WebcrtPage.tsx index 29d7662..16d4ef8 100644 --- a/web/src/pages/WebcrtPage.tsx +++ b/web/src/pages/WebcrtPage.tsx @@ -9,6 +9,9 @@ import { createWebcrtSession, fetchCliTargets, fetchManagedNeById, + webcrtSftpDownload, + webcrtSftpList, + webcrtSftpUpload, webcrtWsUrl, } from "../services/api"; import { pageCount } from "../utils/display"; @@ -50,6 +53,8 @@ function webcrtErrorMessage(err: unknown, t: (key: string, vars?: Record([]); const [activeTabKey, setActiveTabKey] = useState(""); + const [sftpOpen, setSftpOpen] = useState(false); + const [sftpPath, setSftpPath] = useState("."); + const [sftpItems, setSftpItems] = useState>([]); const connectingKeysRef = useRef>(new Set()); const tabsRef = useRef([]); tabsRef.current = tabs; @@ -145,8 +153,18 @@ export function WebcrtPage() { const rows = Math.max(24, Math.floor((window.innerHeight - 180) / 18)); const body = target.source === "ume" - ? { ume_ne_id: target.ume_ne_id || target.id, cols, rows } - : { ne_id: target.id, cols, rows }; + ? { + ume_ne_id: target.ume_ne_id || target.id, + cols, + rows, + async_connect: true, + } + : { + ne_id: target.id, + cols, + rows, + async_connect: true, + }; const sess = await createWebcrtSession(body); const wsUrl = webcrtWsUrl(sess.session_id); updateTab(key, { @@ -228,7 +246,9 @@ export function WebcrtPage() { const tab = tabsRef.current.find((x) => x.key === activeTabKey); if (!tab) return; try { - const text = await termRefs.current.get(tab.key)?.copyAll(); + const handle = termRefs.current.get(tab.key); + const sel = await handle?.copySelection(); + const text = sel || (await handle?.copyAll()); if (text) showOk(t("webcrt.actions.copied")); else showError(t("webcrt.actions.copyEmpty")); } catch { @@ -236,6 +256,22 @@ export function WebcrtPage() { } }, [activeTabKey, showOk, showError, t]); + const refreshSftp = useCallback(async () => { + const tab = tabsRef.current.find((x) => x.key === activeTabKey); + if (!tab) return; + try { + const body = + tab.target.source === "ume" + ? { ume_ne_id: tab.target.ume_ne_id || tab.target.id, path: sftpPath } + : { ne_id: tab.target.id, path: sftpPath }; + const res = await webcrtSftpList(body); + setSftpItems(res.items || []); + setSftpPath(res.path || sftpPath); + } catch (err) { + showError(webcrtErrorMessage(err, t)); + } + }, [activeTabKey, showError, sftpPath, t]); + // Auto-connect from /webcrt?ne_id=... useEffect(() => { if (!presetNeId) return; @@ -292,6 +328,35 @@ export function WebcrtPage() { const activeTab = tabs.find((x) => x.key === activeTabKey) || null; + const renderDeviceRow = (row: CliTargetItem) => { + const key = targetKey(row); + const tab = tabs.find((x) => x.key === key); + const isConnecting = tab?.status === "connecting" || connectingKeysRef.current.has(key); + return ( +
  • + +
    • + ); + }; + return (
    {t("common.pagerMeta", { total, page, pages })} @@ -435,6 +468,16 @@ export function WebcrtPage() { ⧉ {t("webcrt.actions.copy")} +
    ) : null}
    @@ -476,6 +519,8 @@ export function WebcrtPage() { if (activeTabKey === tab.key) { window.setTimeout(() => termRefs.current.get(tab.key)?.focus(), 40); } + } else if (state === "connecting") { + updateTab(tab.key, { status: "connecting" }); } else if (state === "error") updateTab(tab.key, { status: "error" }); else if (state === "closed") updateTab(tab.key, { status: "closed" }); }} @@ -483,6 +528,89 @@ export function WebcrtPage() { ) : null}
    ))} + {sftpOpen && activeTab ? ( +
    +
    + setSftpPath(e.target.value)} /> + + +
    +
      + {sftpItems.map((it) => ( +
    • + +
      • + ))} +
    +
    + ) : null}
    ) : ( diff --git a/web/src/services/api.ts b/web/src/services/api.ts index 1984681..4ceacc1 100644 --- a/web/src/services/api.ts +++ b/web/src/services/api.ts @@ -459,6 +459,8 @@ export type WebcrtSessionCreateResult = { cols: number; rows: number; ws_path: string; + encoding?: string; + state?: string; /** True when session used a vendor CLI hop (nested stelnet/telnet). */ cli_hop?: boolean; }; @@ -468,6 +470,9 @@ export const createWebcrtSession = (body: { ume_ne_id?: string; cols?: number; rows?: number; + encoding?: string; + post_login_commands?: string[]; + async_connect?: boolean; }) => apiPost("/v1/webcrt/sessions", body); export const closeWebcrtSession = (sessionId: string) => @@ -475,6 +480,66 @@ export const closeWebcrtSession = (sessionId: string) => `/v1/webcrt/sessions/${encodeURIComponent(sessionId)}`, ); +export type WebcrtSftpListResult = { + ne_id: string; + ne_name: string; + path: string; + items: Array<{ name: string; size: number; mtime: number; is_dir: boolean }>; +}; + +export const webcrtSftpList = (body: { ne_id?: string; ume_ne_id?: string; path?: string }) => + apiPost("/v1/webcrt/sftp/list", body); + +export async function webcrtSftpDownload(body: { + ne_id?: string; + ume_ne_id?: string; + path: string; +}): Promise { + const path = "/v1/webcrt/sftp/download"; + const res = await fetch(path, { + method: "POST", + headers: authHeaders({ "Content-Type": "application/json" }), + body: JSON.stringify(body), + }); + if (res.status === 401) { + handleUnauthorized(path); + throw new Error("401 unauthorized"); + } + if (!res.ok) { + const text = await res.text(); + throw new Error(text || `HTTP ${res.status}`); + } + return res.blob(); +} + +export async function webcrtSftpUpload(body: { + ne_id?: string; + ume_ne_id?: string; + remote_path: string; + file: File; +}): Promise<{ ok: boolean; path: string; size: number }> { + const path = "/v1/webcrt/sftp/upload"; + const fd = new FormData(); + if (body.ne_id) fd.append("ne_id", body.ne_id); + if (body.ume_ne_id) fd.append("ume_ne_id", body.ume_ne_id); + fd.append("remote_path", body.remote_path); + fd.append("file", body.file); + const res = await fetch(path, { + method: "POST", + headers: authHeaders(), + body: fd, + }); + if (res.status === 401) { + handleUnauthorized(path); + throw new Error("401 unauthorized"); + } + if (!res.ok) { + const text = await res.text(); + throw new Error(text || `HTTP ${res.status}`); + } + return (await res.json()) as { ok: boolean; path: string; size: number }; +} + export const webcrtWsUrl = (sessionId: string): string => { const proto = window.location.protocol === "https:" ? "wss:" : "ws:"; const path = `/v1/webcrt/sessions/${encodeURIComponent(sessionId)}/ws`;