From 8bdaaaacbff795391b7c18c80b3205d6f3341f37 Mon Sep 17 00:00:00 2001 From: oliver Date: Tue, 6 Oct 2026 18:32:14 +0800 Subject: [PATCH] Add NE capability descriptors and native async exec jobs for agents. Expose recommended_mode/hints on getManagedNe, support background exec-jobs with getNeExecJob in MCP, and allow linux_shell on MikroTik as well as Linux. Co-authored-by: Cursor --- docs/MCP.md | 5 +- netx_api/auth_scopes.py | 2 + netx_api/config.py | 6 + netx_api/managed_ne_router.py | 159 ++++++++-- netx_api/ne_capability.py | 110 +++++++ netx_api/ne_exec_guard.py | 51 ++- netx_api/ne_exec_jobs.py | 297 ++++++++++++++++++ netx_api/ne_netmiko.py | 31 +- netx_api/ne_schemas.py | 13 + netx_api/ne_service_crud.py | 6 +- packages/netx-mcp/README.md | 2 +- packages/netx-mcp/src/netx_mcp/http_tools.py | 111 ++++++- packages/netx-mcp/tests/test_mcp_http.py | 3 +- tests/test_ne_capability_jobs.py | 96 ++++++ tests/test_ne_exec.py | 69 +++- tests/test_ne_netmiko_show.py | 19 +- web/src/i18n/en.ts | 6 +- web/src/i18n/zh.ts | 6 +- .../pages/managedNe/ManagedNeFormDialog.tsx | 6 +- web/src/pages/managedNe/formState.ts | 18 +- 20 files changed, 928 insertions(+), 88 deletions(-) create mode 100644 netx_api/ne_capability.py create mode 100644 netx_api/ne_exec_jobs.py create mode 100644 tests/test_ne_capability_jobs.py diff --git a/docs/MCP.md b/docs/MCP.md index a22fd4f..d07e90d 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -71,6 +71,9 @@ pip install "git+https://github.com/hansjone/netx.git#subdirectory=packages/netx | `NETX_MCP_TOKEN_FILE` | 否 | `data/auth/mcp_token` | 默认 token 文件路径 | | `NETX_LANG` | 否 | `zh` | `zh` / `en`，影响 API 文案 | | `NETX_NE_EXEC_MAX_COMMANDS` | 否 | `5` | `execManagedNe` 单次最多命令数（硬上限 50）；API 与 MCP 需同设 | +| `NETX_NE_EXEC_ASYNC_MIN_NES` | 否 | `4` | 批量 NE 数 ≥ 此值时自动走后台 job（`0`=仅 `async=true`） | +| `NETX_NE_EXEC_JOB_DIR` | 否 | `data/ne_exec_jobs` | 后台 exec job 状态目录 | +| `NETX_NE_EXEC_JOB_MAX_CONCURRENT` | 否 | `3` | 同时运行的 exec job 上限 | 本机默认端口时 **可不设任何变量**。启用登录后，先启动一次 netx API，会生成 `data/auth/mcp_token`；MCP 会自动带上该 token。若要把 token 写进 Cursor 配置： @@ -136,7 +139,7 @@ pip install "git+https://github.com/hansjone/netx.git#subdirectory=packages/netx | NMS 告警 | `queryNmsAlarms`, `aggregateNmsAlarms`, `runNmsDiagnostics` | | NMS 网元 | `queryNmsNeInventory`, `getNmsNe` | | NMS 原始/SQL | `queryNmsAlarmsRaw`, `aggregateNmsAlarmsRaw`, `listNmsAlarmFields`, `sqlQueryNms` | -| common（CLI + 路径） | `listManagedNe`, `getManagedNe`, `execManagedNe`, `listCliTargets`, `findTopologyPaths` | +| common（CLI + 路径） | `listManagedNe`, `getManagedNe`, `execManagedNe`, `getNeExecJob`, `listCliTargets`, `findTopologyPaths` | 参数优先 `nms_ne_id` / `nms_ne_ids`（保留 `ume_*` 别名）。拓扑 Fabric / 画布工具在 **[`netx-topology-mcp`](./MCP_TOPOLOGY.md)**。oclaw 中名称带前缀：`mcp__netx__`；DSH：`netx__`。 diff --git a/netx_api/auth_scopes.py b/netx_api/auth_scopes.py index c88373f..0a6b98f 100644 --- a/netx_api/auth_scopes.py +++ b/netx_api/auth_scopes.py @@ -118,6 +118,8 @@ def required_scope_for_request(method: str, path: str) -> str | None: if p.startswith("/v1/managed-ne"): path_tail = p.rstrip("/") + if "/exec-jobs" in p: + return SCOPE_NE_EXEC if m == "POST" and (path_tail.endswith("/exec") or path_tail.endswith("/exec-batch")): return SCOPE_NE_EXEC if m in ("POST", "PUT", "PATCH", "DELETE"): diff --git a/netx_api/config.py b/netx_api/config.py index d93654c..0fb1a12 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -146,6 +146,12 @@ class Settings(BaseSettings): ne_exec_max_commands: int = 5 # Opt-in: allow per-NE exec_policy (linux_shell/unrestricted). Default off — UI hidden. ne_exec_policy_enabled: bool = False + # Background exec jobs (MCP getNeExecJob / POST /v1/managed-ne/exec-jobs). + ne_exec_job_dir: str = "data/ne_exec_jobs" + ne_exec_job_max_concurrent: int = 3 + ne_exec_job_timeout_s: int = 900 + # Auto-async when batch NE count >= this (0 = only when async=true). + ne_exec_async_min_nes: int = 4 # WebCRT interactive terminal sessions (multi-operator concurrent terminals). webcrt_max_sessions: int = 40 # Per-user cap (0 = unlimited beyond global max). diff --git a/netx_api/managed_ne_router.py b/netx_api/managed_ne_router.py index fc8494d..a90bf62 100644 --- a/netx_api/managed_ne_router.py +++ b/netx_api/managed_ne_router.py @@ -12,13 +12,16 @@ from .db import get_db from .device_types import SUPPORTED_VENDORS from .ne_connect import schedule_connect_tests from .ne_crypto import credentials_configured +from .ne_capability import build_ne_capability from .ne_exec import execute_managed_ne_commands, execute_managed_ne_commands_batch +from .ne_exec_jobs import get_ne_exec_job, start_ne_exec_job from .ne_schemas import ( BatchAccountApplyRequest, BatchHopApplyRequest, ConnectTestRequest, ManagedNeCreate, ManagedNeExecBatchRequest, + ManagedNeExecJobCreate, ManagedNeExecRequest, ManagedNeUpdate, ) @@ -149,6 +152,39 @@ def api_delete_ume_synced_managed_ne(db: Session = Depends(get_db)): return delete_ume_synced_managed_ne(db).model_dump() +def _audit_exec( + db: Session, + *, + uid: str, + uname: str, + path: str, + out: dict, + body_ne_id: str = "", + body_ume_ne_id: str = "", + commands: list | None = None, +) -> None: + 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=path, + status_code=200 if out.get("ok") else 502, + detail={ + "ne_id": body_ne_id, + "ume_ne_id": body_ume_ne_id, + "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 commands or [])[:20], + "ok": bool(out.get("ok")), + "error": str(out.get("error") or "")[:500], + "output_len": len(str(out.get("output") or "")), + }, + ) + + @router.post("/exec") def api_exec_managed_ne( body: ManagedNeExecRequest, @@ -164,25 +200,15 @@ def api_exec_managed_ne( 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( + _audit_exec( db, - action="ne.exec", - actor_user_id=uid, - actor_username=uname, - method="POST", + uid=uid, + uname=uname, 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 "")), - }, + out=out, + body_ne_id=body.ne_id or "", + body_ume_ne_id=body.ume_ne_id or "", + commands=list(body.commands or []), ) return out @@ -235,6 +261,96 @@ def api_exec_managed_ne_batch( return out +@router.post("/exec-jobs") +def api_create_exec_job( + body: ManagedNeExecJobCreate, + ctx: Annotated[AuthContext, Depends(require_user)], + db: Session = Depends(get_db), +): + """Start a background exec (single or batch). Poll GET /exec-jobs/{job_id}.""" + from .db import SessionLocal + + uid, uname = _actor(ctx) + payload = body.model_dump(exclude_none=True) + is_batch = bool( + payload.get("targets") + or payload.get("ne_ids") + or payload.get("ume_ne_ids") + ) + kind = "exec-batch" if is_batch else "exec" + + def _runner() -> dict: + with SessionLocal() as job_db: + if is_batch: + targets = payload.get("targets") + out = execute_managed_ne_commands_batch( + targets=targets, + ne_ids=payload.get("ne_ids"), + ume_ne_ids=payload.get("ume_ne_ids"), + commands=payload.get("commands"), + read_timeout_sec=payload.get("read_timeout_sec"), + concurrency=payload.get("concurrency"), + ) + write_audit( + job_db, + action="ne.exec_batch", + actor_user_id=uid, + actor_username=uname, + method="POST", + path="/v1/managed-ne/exec-jobs", + status_code=200, + detail={ + "async": True, + "ne_ids": list(payload.get("ne_ids") or [])[:100], + "ume_ne_ids": list(payload.get("ume_ne_ids") or [])[:100], + "ok": bool(out.get("ok", True)), + }, + ) + return out + cmds = list(payload.get("commands") or []) + out = execute_managed_ne_commands( + job_db, + cmds, + ne_id=payload.get("ne_id"), + ume_ne_id=payload.get("ume_ne_id"), + read_timeout_sec=payload.get("read_timeout_sec"), + ) + _audit_exec( + job_db, + uid=uid, + uname=uname, + path="/v1/managed-ne/exec-jobs", + out=out, + body_ne_id=str(payload.get("ne_id") or ""), + body_ume_ne_id=str(payload.get("ume_ne_id") or ""), + commands=cmds, + ) + return out + + ack = start_ne_exec_job(kind=kind, arguments=payload, runner=_runner) + write_audit( + db, + action="ne.exec_job_start", + actor_user_id=uid, + actor_username=uname, + method="POST", + path="/v1/managed-ne/exec-jobs", + status_code=200 if ack.get("ok") else 429, + detail={"job_id": ack.get("job_id") or "", "kind": kind, "ok": bool(ack.get("ok"))}, + ) + return ack + + +@router.get("/exec-jobs/{job_id}") +def api_get_exec_job( + job_id: str, + ctx: Annotated[AuthContext, Depends(require_user)], +): + """Poll a background exec job started via POST /exec-jobs.""" + _ = ctx + return get_ne_exec_job(job_id) + + @router.post("/connect-test") def api_connect_test(body: ConnectTestRequest, db: Session = Depends(get_db)): ids = [str(x).strip() for x in body.ids if str(x).strip()] @@ -251,7 +367,14 @@ def api_connect_test(body: ConnectTestRequest, db: Session = Depends(get_db)): @router.get("/{ne_id}") def api_get_managed_ne(ne_id: str, db: Session = Depends(get_db)): - return get_managed_ne(db, ne_id).model_dump() + out = get_managed_ne(db, ne_id).model_dump() + out["capability"] = build_ne_capability( + device_type=out.get("device_type"), + exec_policy=out.get("exec_policy"), + vendor=out.get("vendor"), + hop_enabled=bool(out.get("hop_enabled")), + ) + return out @router.patch("/{ne_id}") diff --git a/netx_api/ne_capability.py b/netx_api/ne_capability.py new file mode 100644 index 0000000..fa44be3 --- /dev/null +++ b/netx_api/ne_capability.py @@ -0,0 +1,110 @@ +"""Agent-facing capability descriptor for managed NE exec. + +Derived from device_type + exec_policy + runtime settings — not stored in DB. +Callers (MCP / UI) should read this before choosing short CLI vs on-device script vs async job. +""" + +from __future__ import annotations + +from typing import Any + +from .ne_exec import _EXEC_BATCH_MAX_TARGETS, _EXEC_READ_TIMEOUT_DEFAULT, _EXEC_READ_TIMEOUT_MAX +from .ne_exec_guard import ( + EXEC_POLICY_LINUX_SHELL, + EXEC_POLICY_READONLY, + EXEC_POLICY_UNRESTRICTED, + _LINUX_SHELL_MAX_LEN, + _LINUX_SHELL_MAX_LINES, + _READONLY_MAX_LEN, + allows_open_exec_policy, + effective_exec_policy, + is_linux_device_type, + is_mikrotik_device_type, + normalize_exec_policy, +) + + +def _device_family(device_type: str | None) -> str: + if is_linux_device_type(device_type): + return "linux" + if is_mikrotik_device_type(device_type): + return "mikrotik" + return "network_cli" + + +def _max_commands() -> int: + from .config import settings + from .ne_exec import _EXEC_MAX_COMMANDS_CAP + + raw = int(getattr(settings, "ne_exec_max_commands", 5) or 5) + return max(1, min(_EXEC_MAX_COMMANDS_CAP, raw)) + + +def build_ne_capability( + *, + device_type: str | None, + exec_policy: str | None, + vendor: str | None = None, + hop_enabled: bool = False, +) -> dict[str, Any]: + """Return a stable capability object for agents and ops UIs.""" + stored = normalize_exec_policy(exec_policy) + effective = effective_exec_policy(exec_policy, device_type=device_type) + family = _device_family(device_type) + shell_ok = effective in (EXEC_POLICY_LINUX_SHELL, EXEC_POLICY_UNRESTRICTED) + open_eligible = allows_open_exec_policy(device_type) + max_cmds = _max_commands() + + if effective == EXEC_POLICY_READONLY: + recommended = "show_only" + hints = [ + "readonly: only show/display/ping/traceroute (optional pipe filters).", + "For shell/scripts set exec_policy=linux_shell on a Linux or MikroTik NE " + "(requires NETX_NE_EXEC_POLICY_ENABLED).", + ] + elif family == "linux": + recommended = "script_on_device" + hints = [ + "Prefer short on-device scripts for multi-step work; avoid huge inline heredocs in MCP args.", + "Long / multi-NE work: execManagedNe async=true then poll getNeExecJob.", + f"Max {max_cmds} commands per call; combine steps in one script when needed.", + ] + elif family == "mikrotik": + recommended = "short_cli" + hints = [ + "Prefer short RouterOS lines (comment=/find=); avoid one ultra-long line that may truncate.", + "Multiline scripts allowed under linux_shell; keep each logical change small and idempotent.", + "Long / multi-NE work: execManagedNe async=true then poll getNeExecJob.", + ] + else: + recommended = "show_only" + hints = ["network_cli: use vendor show/display; write changes are blocked under readonly."] + + if hop_enabled: + hints.append("hop_enabled: expect higher latency; raise read_timeout_sec toward 90–120 if needed.") + + return { + "vendor": str(vendor or "").strip() or None, + "device_type": str(device_type or "").strip() or None, + "device_family": family, + "exec_policy_stored": stored, + "exec_policy_effective": effective, + "allows_shell_scripts": bool(shell_ok), + "allows_multiline": bool(shell_ok), + "open_policy_eligible": bool(open_eligible), + "max_commands_per_call": max_cmds, + "max_command_len": _LINUX_SHELL_MAX_LEN if shell_ok else _READONLY_MAX_LEN, + "max_command_lines": _LINUX_SHELL_MAX_LINES if shell_ok else 1, + "read_timeout_sec": { + "default": _EXEC_READ_TIMEOUT_DEFAULT, + "max": _EXEC_READ_TIMEOUT_MAX, + }, + "supports_batch_exec": True, + "batch_max_targets": _EXEC_BATCH_MAX_TARGETS, + "supports_async_job": True, + "recommended_mode": recommended, + "hints": hints, + } + + +__all__ = ["build_ne_capability"] diff --git a/netx_api/ne_exec_guard.py b/netx_api/ne_exec_guard.py index e8eb8a6..1fc4f13 100644 --- a/netx_api/ne_exec_guard.py +++ b/netx_api/ne_exec_guard.py @@ -3,7 +3,9 @@ Policies (per managed NE ``exec_policy``): - ``readonly`` (default): network CLI only — show/display/ping/traceroute. -- ``linux_shell``: single-line shell; no network prefix/pipe rules; no write-deny list. +- ``linux_shell``: open shell/script on capable NEs (Linux, MikroTik RouterOS) — + pipes/&&/;/quotes/heredoc / RouterOS multiline scripts allowed; + no network prefix/pipe rules; no write-deny list. - ``unrestricted``: same as linux_shell (lab open); kept distinct for audit/UI. """ @@ -20,6 +22,11 @@ EXEC_POLICIES = frozenset( {EXEC_POLICY_READONLY, EXEC_POLICY_LINUX_SHELL, EXEC_POLICY_UNRESTRICTED} ) +# readonly: short show/ping lines. linux_shell: scripts / heredoc file writes for agents. +_READONLY_MAX_LEN = 500 +_LINUX_SHELL_MAX_LEN = 65_536 +_LINUX_SHELL_MAX_LINES = 2_000 + # Block obvious config-change / destructive patterns (case-insensitive). _BLOCKED_RE = re.compile( r"(?i)(" @@ -63,6 +70,16 @@ def is_linux_device_type(device_type: str | None) -> bool: return low in ("linux", "linux_ssh", "linux_telnet") or low.startswith("linux_") +def is_mikrotik_device_type(device_type: str | None) -> bool: + low = str(device_type or "").strip().lower() + return low in ("mikrotik_routeros", "mikrotik_switchos") or low.startswith("mikrotik_") + + +def allows_open_exec_policy(device_type: str | None) -> bool: + """Device types that may use linux_shell / unrestricted (multiline scripts).""" + return is_linux_device_type(device_type) or is_mikrotik_device_type(device_type) + + def exec_policy_feature_enabled() -> bool: """Global kill-switch: off → always readonly (UI hidden, API rejects open policies).""" from .config import settings @@ -71,11 +88,11 @@ def exec_policy_feature_enabled() -> bool: def effective_exec_policy(raw: str | None, *, device_type: str | None = None) -> str: - """Policy used at exec time (forces readonly when feature off or non-linux).""" + """Policy used at exec time (forces readonly when feature off or ineligible type).""" if not exec_policy_feature_enabled(): return EXEC_POLICY_READONLY pol = normalize_exec_policy(raw) - if pol != EXEC_POLICY_READONLY and not is_linux_device_type(device_type): + if pol != EXEC_POLICY_READONLY and not allows_open_exec_policy(device_type): return EXEC_POLICY_READONLY return pol @@ -85,14 +102,14 @@ def require_exec_policy_writable( *, device_type: str | None = None, ) -> str: - """Normalize for create/update; reject open policies when feature off or non-linux.""" + """Normalize for create/update; reject open policies when feature off or ineligible type.""" pol = normalize_exec_policy(raw) if pol == EXEC_POLICY_READONLY: return pol if not exec_policy_feature_enabled(): raise HTTPException(status_code=400, detail="exec_policy_feature_disabled") - if not is_linux_device_type(device_type): - raise HTTPException(status_code=400, detail="exec_policy_requires_linux_device_type") + if not allows_open_exec_policy(device_type): + raise HTTPException(status_code=400, detail="exec_policy_requires_shell_device_type") return pol @@ -109,11 +126,14 @@ def _validate_pipe_segments(cmd: str) -> None: raise HTTPException(status_code=400, detail="command_pipe_not_allowed") -def _validate_single_line(cmd: str) -> None: - if any(ch in cmd for ch in ("\n", "\r")): - raise HTTPException(status_code=400, detail="command_chars_not_allowed") +def _validate_linux_shell_command(cmd: str) -> None: + """Shell policy: allow multiline (heredoc / scripts); still reject exotic separators.""" if any(sep in cmd for sep in _FORBIDDEN_LINE_SEPARATORS): raise HTTPException(status_code=400, detail="command_chars_not_allowed") + # Count lines after normalizing CRLF; blank trailing newline from strip() is already gone. + line_count = cmd.count("\n") + 1 + if line_count > _LINUX_SHELL_MAX_LINES: + raise HTTPException(status_code=400, detail="command_too_many_lines") def _validate_readonly_command(cmd: str) -> None: @@ -133,12 +153,17 @@ def validate_ne_exec_command(command: str, *, policy: str = EXEC_POLICY_READONLY cmd = str(command or "").strip() if not cmd: raise HTTPException(status_code=400, detail="empty_command") - if len(cmd) > 500: - raise HTTPException(status_code=400, detail="command_too_long") pol = normalize_exec_policy(policy) + max_len = ( + _LINUX_SHELL_MAX_LEN + if pol in (EXEC_POLICY_LINUX_SHELL, EXEC_POLICY_UNRESTRICTED) + else _READONLY_MAX_LEN + ) + if len(cmd) > max_len: + raise HTTPException(status_code=400, detail="command_too_long") if pol in (EXEC_POLICY_LINUX_SHELL, EXEC_POLICY_UNRESTRICTED): - # One command string per slot; shell metacharacters (|;&&`$) allowed. - _validate_single_line(cmd) + # Shell metacharacters (|;&&`$'"<<) and newlines (heredoc) allowed. + _validate_linux_shell_command(cmd) return _validate_readonly_command(cmd) diff --git a/netx_api/ne_exec_jobs.py b/netx_api/ne_exec_jobs.py new file mode 100644 index 0000000..315d37f --- /dev/null +++ b/netx_api/ne_exec_jobs.py @@ -0,0 +1,297 @@ +"""Background jobs for managed-NE exec (NetX-native; Linux + RouterOS + batch). + +Agents poll GET /v1/managed-ne/exec-jobs/{job_id} (MCP: getNeExecJob). +Sync exec stays on POST /exec and /exec-batch; long or multi-NE work should use jobs. +""" + +from __future__ import annotations + +import json +import os +import threading +import time +import uuid +from pathlib import Path +from typing import Any, Callable + +_STATUS_RUNNING = "running" +_STATUS_SUCCEEDED = "succeeded" +_STATUS_FAILED = "failed" +_STATUS_TIMEOUT = "timeout" +_TERMINAL = {_STATUS_SUCCEEDED, _STATUS_FAILED, _STATUS_TIMEOUT} +_LOCK = threading.Lock() +_INFLIGHT = 0 +_DEFAULT_TIMEOUT_S = 900 +_DEFAULT_MAX_CONCURRENT = 3 +_DEFAULT_ASYNC_MIN_NES = 4 + + +def _env_int(name: str, default: int, *, min_v: int, max_v: int) -> int: + raw = str(os.getenv(name) or "").strip() + if not raw: + try: + from .config import settings + + attr = name.lower() + # Map common env → settings when present. + mapping = { + "NETX_NE_EXEC_JOB_MAX_CONCURRENT": "ne_exec_job_max_concurrent", + "NETX_NE_EXEC_ASYNC_MIN_NES": "ne_exec_async_min_nes", + "NETX_NE_EXEC_JOB_TIMEOUT_S": "ne_exec_job_timeout_s", + } + key = mapping.get(name) + if key and hasattr(settings, key): + return max(min_v, min(int(getattr(settings, key)), max_v)) + except Exception: + pass + return default + try: + n = int(raw) + except Exception: + return default + return max(min_v, min(int(n), max_v)) + + +def _jobs_dir() -> Path: + override = str(os.getenv("NETX_NE_EXEC_JOB_DIR") or "").strip() + if override: + p = Path(override).expanduser().resolve() + else: + try: + from .config import settings + + raw = str(getattr(settings, "ne_exec_job_dir", "") or "").strip() + if raw: + p = Path(raw).expanduser().resolve() + else: + p = Path("data/ne_exec_jobs").resolve() + except Exception: + p = Path("data/ne_exec_jobs").resolve() + p.mkdir(parents=True, exist_ok=True) + return p + + +def _job_path(job_id: str) -> Path: + return _jobs_dir() / f"{job_id}.json" + + +def _atomic_write(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(payload, ensure_ascii=False, default=str), encoding="utf-8") + tmp.replace(path) + + +def _read_job(job_id: str) -> dict[str, Any] | None: + path = _job_path(job_id) + if not path.is_file(): + return None + try: + data = json.loads(path.read_text(encoding="utf-8")) + except Exception: + return None + return data if isinstance(data, dict) else None + + +def _write_job(job: dict[str, Any]) -> None: + jid = str(job.get("job_id") or "").strip() + if not jid: + return + _atomic_write(_job_path(jid), job) + + +def async_min_ne_count() -> int: + """Auto-async when batch NE count >= this (0 disables auto; async=true still works).""" + return _env_int("NETX_NE_EXEC_ASYNC_MIN_NES", _DEFAULT_ASYNC_MIN_NES, min_v=0, max_v=50) + + +def max_concurrent_jobs() -> int: + return _env_int("NETX_NE_EXEC_JOB_MAX_CONCURRENT", _DEFAULT_MAX_CONCURRENT, min_v=1, max_v=16) + + +def default_job_timeout_s() -> int: + return _env_int("NETX_NE_EXEC_JOB_TIMEOUT_S", _DEFAULT_TIMEOUT_S, min_v=60, max_v=1800) + + +def count_exec_ne_targets(args: dict[str, Any] | None) -> int: + a = args if isinstance(args, dict) else {} + n = 0 + for key in ("ne_ids", "ume_ne_ids", "nms_ne_ids"): + val = a.get(key) + if isinstance(val, list): + n = max(n, len([x for x in val if str(x or "").strip()])) + targets = a.get("targets") + if isinstance(targets, list): + n = max(n, len([t for t in targets if isinstance(t, dict)])) + if n == 0 and ( + str(a.get("ne_id") or "").strip() + or str(a.get("ume_ne_id") or "").strip() + or str(a.get("nms_ne_id") or "").strip() + ): + return 1 + return int(n) + + +def _truthy_async_flag(raw: Any) -> bool | None: + if raw is None: + return None + if isinstance(raw, bool): + return raw + text = str(raw).strip().lower() + if not text: + return None + if text in {"1", "true", "yes", "on"}: + return True + if text in {"0", "false", "no", "off"}: + return False + return None + + +def should_run_exec_async(args: dict[str, Any] | None) -> bool: + a = dict(args or {}) + flag = _truthy_async_flag(a.get("async")) + if flag is False: + return False + if flag is True: + return True + min_n = async_min_ne_count() + if min_n <= 0: + return False + return count_exec_ne_targets(a) >= int(min_n) + + +def start_ne_exec_job( + *, + kind: str, + arguments: dict[str, Any], + runner: Callable[[], dict[str, Any]], + timeout_s: int | None = None, +) -> dict[str, Any]: + """Start background runner; returns immediate ack with job_id.""" + global _INFLIGHT + max_c = max_concurrent_jobs() + with _LOCK: + if _INFLIGHT >= max_c: + return { + "ok": False, + "error_code": "ne_exec_job_busy", + "error": "ne_exec_job_busy", + "hint": ( + f"Too many concurrent NE exec jobs (max {max_c}). " + "Poll getNeExecJob for running jobs or shrink the batch." + ), + } + _INFLIGHT += 1 + + job_id = uuid.uuid4().hex + timeout = int(timeout_s) if timeout_s is not None else default_job_timeout_s() + timeout = max(60, min(timeout, 1800)) + now = int(time.time() * 1000) + job = { + "job_id": job_id, + "status": _STATUS_RUNNING, + "kind": str(kind or "exec"), + "arguments": dict(arguments or {}), + "ne_count": count_exec_ne_targets(arguments), + "created_at_ms": now, + "updated_at_ms": now, + "timeout_s": timeout, + "result": None, + "error": "", + } + _write_job(job) + + def _worker() -> None: + global _INFLIGHT + started = time.time() + try: + result = runner() + if not isinstance(result, dict): + result = { + "ok": False, + "error": "invalid_runner_result", + "payload_type": type(result).__name__, + } + status = _STATUS_SUCCEEDED if result.get("ok") is not False else _STATUS_FAILED + if time.time() - started > timeout: + status = _STATUS_TIMEOUT + cur = _read_job(job_id) or job + cur.update( + { + "status": status, + "updated_at_ms": int(time.time() * 1000), + "result": result, + "error": str(result.get("error") or "") if status != _STATUS_SUCCEEDED else "", + "duration_ms": int((time.time() - started) * 1000), + } + ) + _write_job(cur) + except Exception as exc: + cur = _read_job(job_id) or job + cur.update( + { + "status": _STATUS_FAILED, + "updated_at_ms": int(time.time() * 1000), + "error": f"{type(exc).__name__}: {exc}", + "result": {"ok": False, "error": f"{type(exc).__name__}: {exc}"}, + "duration_ms": int((time.time() - started) * 1000), + } + ) + _write_job(cur) + finally: + with _LOCK: + _INFLIGHT = max(0, _INFLIGHT - 1) + + threading.Thread(target=_worker, name=f"ne-exec-{job_id[:8]}", daemon=True).start() + return { + "ok": True, + "async": True, + "status": _STATUS_RUNNING, + "job_id": job_id, + "ne_count": int(job["ne_count"]), + "poll_tool": "getNeExecJob", + "poll_path": f"/v1/managed-ne/exec-jobs/{job_id}", + "hint": ( + f"exec started in background (job_id={job_id}, ne_count={job['ne_count']}). " + "Tell the user the job_id and end the turn; later call getNeExecJob(job_id=...) " + "or ask the user to continue. Do not sleep/busy-wait in this turn." + ), + "example_poll": {"job_id": job_id}, + } + + +def get_ne_exec_job(job_id: str) -> dict[str, Any]: + jid = str(job_id or "").strip() + if not jid: + return {"ok": False, "error_code": "job_id_required", "error": "job_id_required"} + job = _read_job(jid) + if not job: + return {"ok": False, "error_code": "job_not_found", "error": "job_not_found", "job_id": jid} + status = str(job.get("status") or "") + out: dict[str, Any] = { + "ok": True, + "job_id": jid, + "status": status, + "kind": job.get("kind"), + "ne_count": job.get("ne_count"), + "created_at_ms": job.get("created_at_ms"), + "updated_at_ms": job.get("updated_at_ms"), + "duration_ms": job.get("duration_ms"), + "terminal": status in _TERMINAL, + } + if status in _TERMINAL: + out["result"] = job.get("result") + if job.get("error"): + out["error"] = job.get("error") + else: + out["hint"] = "Still running; poll getNeExecJob again later or ask the user to continue." + return out + + +__all__ = [ + "async_min_ne_count", + "count_exec_ne_targets", + "get_ne_exec_job", + "should_run_exec_async", + "start_ne_exec_job", +] diff --git a/netx_api/ne_netmiko.py b/netx_api/ne_netmiko.py index b722bfd..9b351e6 100644 --- a/netx_api/ne_netmiko.py +++ b/netx_api/ne_netmiko.py @@ -149,12 +149,15 @@ def _send_command_expect_prompt(conn: Any, cmd: str, *, read_timeout: int) -> st def send_show_command(conn: Any, command: str, *, read_timeout: int = 120) -> str: - """Send a show/display command via ``send_command`` (wait for device prompt). + """Send one CLI command via ``send_command`` (wait for device prompt). - Do not use ``send_command_timing`` as the primary path for Cisco config - collection: long idle during ``Building configuration...`` is treated as - end-of-output and truncates the config. Timing is only a fallback when - expect-prompt returns empty after a channel drain (IOSv leftover-prompt bug). + Never re-sends the same command on empty/short output: this path is shared by + show collection and open-policy config/scripts (MikroTik ``add``, etc.). A + leftover-prompt empty read must be fixed by draining *before* send, not by + retrying — retries duplicate side-effecting commands. + + Prefer ``send_command`` over ``send_command_timing`` for long config dumps: + idle during ``Building configuration...`` truncates timing reads. ``cmd_verify=False``: Netmiko's default echo check often raises ``Pattern not detected: 'show\\ lldp\\ ...'`` on IOSv / hop / slow echo paths. @@ -163,19 +166,7 @@ def send_show_command(conn: Any, command: str, *, read_timeout: int = 120) -> st if not cmd: return "" + # IOSv / hop: find_prompt / paging-off often leave an extra prompt in the + # channel; drain so send_command does not match that leftover immediately. drain_read_channel(conn) - out = _send_command_expect_prompt(conn, cmd, read_timeout=read_timeout) - if out.strip(): - return out - - # Leftover prompt matched before command echo — drain again and retry once. - drain_read_channel(conn) - out = _send_command_expect_prompt(conn, cmd, read_timeout=read_timeout) - if out.strip(): - return out - - drain_read_channel(conn) - try: - return str(conn.send_command_timing(cmd, read_timeout=read_timeout) or "") - except Exception: - return "" + return _send_command_expect_prompt(conn, cmd, read_timeout=read_timeout) diff --git a/netx_api/ne_schemas.py b/netx_api/ne_schemas.py index 59a4f46..0611390 100644 --- a/netx_api/ne_schemas.py +++ b/netx_api/ne_schemas.py @@ -176,6 +176,19 @@ class ManagedNeExecBatchRequest(BaseModel): concurrency: int | None = Field(default=4, ge=1, le=8) +class ManagedNeExecJobCreate(BaseModel): + """Start a background single or batch exec (poll GET /v1/managed-ne/exec-jobs/{job_id}).""" + + ne_id: str | None = None + ume_ne_id: str | None = None + targets: list[ManagedNeExecBatchTarget] | None = Field(default=None, max_length=20) + ne_ids: list[str] | None = Field(default=None, max_length=20) + ume_ne_ids: list[str] | None = Field(default=None, max_length=20) + commands: list[str] | None = Field(default=None, max_length=50) + read_timeout_sec: int | None = Field(default=None, ge=10, le=120) + concurrency: int | None = Field(default=4, ge=1, le=8) + + class HopProxyConfig(BaseModel): """Shared jump-host (proxy) settings applied to one or many NEs.""" diff --git a/netx_api/ne_service_crud.py b/netx_api/ne_service_crud.py index 83ab482..e8b253c 100644 --- a/netx_api/ne_service_crud.py +++ b/netx_api/ne_service_crud.py @@ -35,7 +35,7 @@ from .ne_service_common import ( ) from .ne_exec_guard import ( EXEC_POLICY_READONLY, - is_linux_device_type, + allows_open_exec_policy, require_exec_policy_writable, ) @@ -173,8 +173,8 @@ def update_managed_ne(db: Session, ne_id: str, body: ManagedNeUpdate) -> Managed str(data["exec_policy"]), device_type=row.device_type, ) - elif "device_type" in data and not is_linux_device_type(row.device_type): - # Leaving linux clears any previously open policy. + elif "device_type" in data and not allows_open_exec_policy(row.device_type): + # Leaving linux/mikrotik clears any previously open policy. row.exec_policy = EXEC_POLICY_READONLY if "password" in data and data["password"]: _require_crypto() diff --git a/packages/netx-mcp/README.md b/packages/netx-mcp/README.md index 6d700dd..72ce048 100644 --- a/packages/netx-mcp/README.md +++ b/packages/netx-mcp/README.md @@ -44,7 +44,7 @@ python -m netx_mcp `queryNmsAlarms`, `aggregateNmsAlarms`, `runNmsDiagnostics`, `queryNmsNeInventory`, `getNmsNe`, `queryNmsAlarmsRaw`, `aggregateNmsAlarmsRaw`, `listNmsAlarmFields`, `sqlQueryNms` **Managed CLI + paths**（问「能否登录」必须走这里，不要只查 inventory）： -`listManagedNe`, `getManagedNe`, `execManagedNe`, `listCliTargets`, `findTopologyPaths` +`listManagedNe`, `getManagedNe`, `execManagedNe`, `getNeExecJob`, `listCliTargets`, `findTopologyPaths` 参数优先 `nms_ne_id` / `nms_ne_ids`（保留 `ume_*` 别名）。 diff --git a/packages/netx-mcp/src/netx_mcp/http_tools.py b/packages/netx-mcp/src/netx_mcp/http_tools.py index 2bac6d2..65e3bcd 100644 --- a/packages/netx-mcp/src/netx_mcp/http_tools.py +++ b/packages/netx-mcp/src/netx_mcp/http_tools.py @@ -24,6 +24,57 @@ def exec_max_commands() -> int: return max(1, min(_EXEC_MAX_COMMANDS_CAP, raw)) +def _async_min_nes() -> int: + try: + raw = int(os.getenv("NETX_NE_EXEC_ASYNC_MIN_NES") or 4) + except ValueError: + raw = 4 + return max(0, min(50, raw)) + + +def _truthy_async_flag(raw: Any) -> bool | None: + if raw is None: + return None + if isinstance(raw, bool): + return raw + text = str(raw).strip().lower() + if text in {"1", "true", "yes", "on"}: + return True + if text in {"0", "false", "no", "off"}: + return False + return None + + +def _count_exec_ne_targets(args: dict[str, Any]) -> int: + n = 0 + for key in ("ne_ids", "nms_ne_ids", "ume_ne_ids"): + val = args.get(key) + if isinstance(val, list): + n = max(n, len([x for x in val if str(x or "").strip()])) + targets = args.get("targets") + if isinstance(targets, list): + n = max(n, len([t for t in targets if isinstance(t, dict)])) + if n == 0 and ( + str(args.get("ne_id") or "").strip() + or str(args.get("nms_ne_id") or "").strip() + or str(args.get("ume_ne_id") or "").strip() + ): + return 1 + return int(n) + + +def _should_run_exec_async(args: dict[str, Any]) -> bool: + flag = _truthy_async_flag(args.get("async")) + if flag is False: + return False + if flag is True: + return True + min_n = _async_min_nes() + if min_n <= 0: + return False + return _count_exec_ne_targets(args) >= min_n + + UME_RAW_GROUP_FIELDS = [ "alarm_alarm_key", "alarm_host_name", @@ -310,6 +361,8 @@ def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]: or (isinstance(ne_ids_raw, list) and ne_ids_raw) or (isinstance(nms_ne_ids_raw, list) and nms_ne_ids_raw) ) + want_async = _should_run_exec_async(args) + if multi: body: dict[str, Any] = {} if isinstance(targets_raw, list) and targets_raw: @@ -343,6 +396,12 @@ def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]: conc = args.get("concurrency") if conc is not None: body["concurrency"] = int(conc) + if want_async: + out = http_post_json("/v1/managed-ne/exec-jobs", body, timeout=60.0) + if not out.get("ok"): + return out + data = out.get("data") if isinstance(out.get("data"), dict) else out + return data if isinstance(data, dict) else {"ok": True, "data": data} # Wall clock: many NEs × per-cmd timeout; keep below oclaw MCP override. out = http_post_json("/v1/managed-ne/exec-batch", body, timeout=600.0) if not out.get("ok"): @@ -377,6 +436,12 @@ def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]: # Default 60s matches netx API default; slow show commands often exceed 30s. rts = args.get("read_timeout_sec") body["read_timeout_sec"] = int(rts) if rts is not None else 60 + if want_async: + out = http_post_json("/v1/managed-ne/exec-jobs", body, timeout=60.0) + if not out.get("ok"): + return out + data = out.get("data") if isinstance(out.get("data"), dict) else out + return data if isinstance(data, dict) else {"ok": True, "data": data} out = http_post_json("/v1/managed-ne/exec", body, timeout=300.0) if not out.get("ok"): return out @@ -386,6 +451,18 @@ def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]: return {"ok": True, "data": data} +def _get_ne_exec_job(args: dict[str, Any]) -> dict[str, Any]: + job_id = str(args.get("job_id") or "").strip() + if not job_id: + return { + "ok": False, + "error": "job_id_required", + "error_code": "job_id_required", + "hint": "Pass job_id from execManagedNe async ack.", + } + return http_json("GET", f"/v1/managed-ne/exec-jobs/{quote_ne_id(job_id)}", params=None) + + def _list_cli_targets(args: dict[str, Any]) -> dict[str, Any]: page = max(1, int(args.get("page") or 1)) page_size = min(500, max(1, int(args.get("page_size") or 50))) @@ -665,6 +742,8 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ "name": "getManagedNe", "description": ( "Get one **managed** NE by managed ne_id (from listManagedNe / listCliTargets source=managed). " + "Response includes capability (device_family, exec_policy_effective, recommended_mode, hints) " + "— read it before complex execManagedNe. " "Do NOT pass NMS inventory UUID here — use getNmsNe or execManagedNe(nms_ne_id=...) instead." ), "inputSchema": { @@ -686,8 +765,9 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ "description": ( f"Run CLI via netx (default read-only: show/display/ping/traceroute; " f"max {exec_max_commands()} commands per NE, NETX_NE_EXEC_MAX_COMMANDS). " - "Managed NE exec_policy=linux_shell|unrestricted allows single-line shell on that host " - "(check getManagedNe / listManagedNe). " + "Managed NE exec_policy=linux_shell|unrestricted allows shell/script on Linux or " + "MikroTik (routeros/switchos) hosts " + "(pipes/&&/;/quotes/heredoc / RouterOS multiline OK; check getManagedNe / listManagedNe). " "Single NE: ne_id OR nms_ne_id (+ alias ume_ne_id) + commands. " "Many NEs (batch-first, server concurrency default 4, max 20): " "(1) same CLI on all → ne_ids[]/nms_ne_ids[] + shared commands; " @@ -695,8 +775,9 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ "{nms_ne_id|ne_id, commands:[…]}, …] — do NOT fall back to one-NE loops. " "Do NOT loop one-NE execManagedNe for multi-NE work. " "Default read_timeout_sec=60; on timeout raise to 90–120 — do not blind-retry. " - "Large batches (≈4+ NEs) may auto-run async in oclaw: returns job_id immediately; " - "poll get_ne_exec_job. Pass async=true to force background, async=false to force sync." + "Long or multi-NE work: async=true (or auto when ≥4 NEs) returns job_id immediately; " + "poll getNeExecJob. Pass async=false to force sync. " + "Read getManagedNe.capability before choosing show_only vs script_on_device." ), "inputSchema": { "type": "object", @@ -774,8 +855,8 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ "async": { "type": "boolean", "description": ( - "oclaw-only: true=background job_id + get_ne_exec_job; " - "false=force sync; omit=auto for large batches (~4+ NEs)." + "true=background job_id + getNeExecJob; false=force sync; " + "omit=auto for large batches (~4+ NEs, NETX_NE_EXEC_ASYNC_MIN_NES)." ), }, }, @@ -783,6 +864,22 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ "additionalProperties": False, }, }, + { + "name": "getNeExecJob", + "description": ( + "Poll a background execManagedNe job (job_id from async ack). " + "When terminal=true, result holds the exec/exec-batch payload. " + "Do not busy-wait; end the turn and poll later if still running." + ), + "inputSchema": { + "type": "object", + "properties": { + "job_id": {"type": "string", "description": "Job id from execManagedNe async response."}, + }, + "required": ["job_id"], + "additionalProperties": False, + }, + }, { "name": "listCliTargets", "description": ( @@ -872,6 +969,7 @@ _HANDLERS: dict[str, Callable[[dict[str, Any]], dict[str, Any]]] = { "listManagedNe": _list_managed_ne, "getManagedNe": _get_managed_ne, "execManagedNe": _exec_managed_ne, + "getNeExecJob": _get_ne_exec_job, "listCliTargets": _list_cli_targets, "findTopologyPaths": _find_topology_paths, } @@ -890,6 +988,7 @@ TOOL_REQUIRED_SCOPE: dict[str, str] = { "listManagedNe": "ne:read", "getManagedNe": "ne:read", "execManagedNe": "ne:exec", + "getNeExecJob": "ne:exec", "listCliTargets": "ne:read", "findTopologyPaths": "ne:read", } diff --git a/packages/netx-mcp/tests/test_mcp_http.py b/packages/netx-mcp/tests/test_mcp_http.py index 11686c1..50e5b44 100644 --- a/packages/netx-mcp/tests/test_mcp_http.py +++ b/packages/netx-mcp/tests/test_mcp_http.py @@ -15,10 +15,11 @@ from netx_mcp.server import _fetch_scopes def test_http_mcp_tool_list_has_expected_tools() -> None: names = [str(t.get("name") or "") for t in HTTP_MCP_TOOLS] - assert len(names) == 14 + assert len(names) == 15 assert "queryNmsAlarms" in names assert "queryNmsAlarmsRaw" in names assert "execManagedNe" in names + assert "getNeExecJob" in names assert "listCliTargets" in names assert "findTopologyPaths" in names assert "getBizMonitorContext" not in names diff --git a/tests/test_ne_capability_jobs.py b/tests/test_ne_capability_jobs.py new file mode 100644 index 0000000..8279fb6 --- /dev/null +++ b/tests/test_ne_capability_jobs.py @@ -0,0 +1,96 @@ +"""Tests for NE capability descriptors and background exec jobs.""" + +from __future__ import annotations + +import tempfile +import time +import unittest +from pathlib import Path +from unittest.mock import patch + +from netx_api.ne_capability import build_ne_capability +from netx_api.ne_exec_jobs import ( + get_ne_exec_job, + should_run_exec_async, + start_ne_exec_job, +) + + +class NeCapabilityTests(unittest.TestCase): + def test_linux_shell_recommends_script(self) -> None: + with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=True): + cap = build_ne_capability( + device_type="linux", + exec_policy="linux_shell", + vendor="Other", + ) + self.assertEqual(cap["device_family"], "linux") + self.assertEqual(cap["exec_policy_effective"], "linux_shell") + self.assertTrue(cap["allows_shell_scripts"]) + self.assertTrue(cap["supports_async_job"]) + self.assertEqual(cap["recommended_mode"], "script_on_device") + + def test_mikrotik_shell_recommends_short_cli(self) -> None: + with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=True): + cap = build_ne_capability( + device_type="mikrotik_routeros", + exec_policy="linux_shell", + ) + self.assertEqual(cap["device_family"], "mikrotik") + self.assertEqual(cap["recommended_mode"], "short_cli") + self.assertTrue(cap["allows_multiline"]) + + def test_readonly_network_cli(self) -> None: + with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=True): + cap = build_ne_capability(device_type="zte_zxros", exec_policy="readonly") + self.assertEqual(cap["device_family"], "network_cli") + self.assertEqual(cap["recommended_mode"], "show_only") + self.assertFalse(cap["allows_shell_scripts"]) + + def test_feature_off_forces_readonly_effective(self) -> None: + with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=False): + cap = build_ne_capability(device_type="linux", exec_policy="linux_shell") + self.assertEqual(cap["exec_policy_stored"], "linux_shell") + self.assertEqual(cap["exec_policy_effective"], "readonly") + self.assertFalse(cap["allows_shell_scripts"]) + + +class NeExecJobTests(unittest.TestCase): + def test_should_run_async_flags(self) -> None: + self.assertTrue(should_run_exec_async({"async": True, "ne_id": "a"})) + self.assertFalse(should_run_exec_async({"async": False, "ne_ids": ["a", "b", "c", "d", "e"]})) + with patch("netx_api.ne_exec_jobs.async_min_ne_count", return_value=4): + self.assertTrue( + should_run_exec_async({"ne_ids": ["1", "2", "3", "4"]}) + ) + self.assertFalse( + should_run_exec_async({"ne_ids": ["1", "2"]}) + ) + + def test_job_roundtrip(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + with patch.dict("os.environ", {"NETX_NE_EXEC_JOB_DIR": tmp}): + ack = start_ne_exec_job( + kind="exec", + arguments={"ne_id": "abc", "commands": ["show version"]}, + runner=lambda: {"ok": True, "output": "hi"}, + ) + self.assertTrue(ack.get("ok")) + jid = str(ack.get("job_id") or "") + self.assertTrue(jid) + done = None + for _ in range(50): + polled = get_ne_exec_job(jid) + if polled.get("terminal"): + done = polled + break + time.sleep(0.02) + self.assertIsNotNone(done) + assert done is not None + self.assertEqual(done.get("status"), "succeeded") + self.assertEqual((done.get("result") or {}).get("output"), "hi") + self.assertTrue(Path(tmp, f"{jid}.json").is_file()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_ne_exec.py b/tests/test_ne_exec.py index 3cfa409..305dff2 100644 --- a/tests/test_ne_exec.py +++ b/tests/test_ne_exec.py @@ -105,6 +105,10 @@ class NeExecValidationTests(unittest.TestCase): "systemctl status sshd", "cat /etc/os-release && uname -a", "df -h; free -m", + "printf 'a\\nb' > /tmp/x", + "python3 -c \"print('hi')\"", + "cat <<'EOF' > /tmp/x\nhello world\nEOF", + "tee /tmp/x < None: + def test_effective_policy_forces_readonly_for_ineligible_types(self) -> None: from netx_api.ne_exec_guard import effective_exec_policy with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=True): self.assertEqual(effective_exec_policy("linux_shell", device_type="zte_zxros"), "readonly") self.assertEqual(effective_exec_policy("linux_shell", device_type="linux"), "linux_shell") self.assertEqual(effective_exec_policy("unrestricted", device_type="linux_ssh"), "unrestricted") + self.assertEqual( + effective_exec_policy("linux_shell", device_type="mikrotik_routeros"), + "linux_shell", + ) + self.assertEqual( + effective_exec_policy("unrestricted", device_type="mikrotik_switchos"), + "unrestricted", + ) def test_require_writable_rejects_when_feature_off(self) -> None: from netx_api.ne_exec_guard import require_exec_policy_writable @@ -134,7 +146,7 @@ class NeExecValidationTests(unittest.TestCase): require_exec_policy_writable("linux_shell", device_type="linux") self.assertEqual(ctx.exception.detail, "exec_policy_feature_disabled") - def test_require_writable_rejects_non_linux(self) -> None: + def test_require_writable_rejects_ineligible_types(self) -> None: from netx_api.ne_exec_guard import require_exec_policy_writable with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=True): @@ -142,15 +154,37 @@ class NeExecValidationTests(unittest.TestCase): require_exec_policy_writable("linux_shell", device_type="linux"), "linux_shell", ) + self.assertEqual( + require_exec_policy_writable("linux_shell", device_type="mikrotik_routeros"), + "linux_shell", + ) with self.assertRaises(HTTPException) as ctx: require_exec_policy_writable("linux_shell", device_type="cisco_ios") - self.assertEqual(ctx.exception.detail, "exec_policy_requires_linux_device_type") + self.assertEqual(ctx.exception.detail, "exec_policy_requires_shell_device_type") - def test_linux_shell_blocks_newline(self) -> None: + def test_linux_shell_allows_multiline_and_chained(self) -> None: + # Newlines are intentional for heredoc / small scripts (agent-friendly). + _validate_command("ls\nrm -rf /", policy="linux_shell") + _validate_command("cat <<'EOF' > /tmp/a\nx\nEOF", policy="unrestricted") + + def test_linux_shell_blocks_unicode_line_separator(self) -> None: with self.assertRaises(HTTPException) as ctx: - _validate_command("ls\nrm -rf /", policy="linux_shell") + _validate_command("ls\u2028rm -rf /", policy="linux_shell") self.assertEqual(ctx.exception.detail, "command_chars_not_allowed") + def test_linux_shell_blocks_too_many_lines(self) -> None: + body = "\n".join(["echo x"] * 2001) + with self.assertRaises(HTTPException) as ctx: + _validate_command(body, policy="linux_shell") + self.assertEqual(ctx.exception.detail, "command_too_many_lines") + + def test_linux_shell_allows_longer_than_readonly_cap(self) -> None: + cmd = "echo " + ("a" * 600) + with self.assertRaises(HTTPException) as ctx: + _validate_command(cmd, policy="readonly") + self.assertEqual(ctx.exception.detail, "command_too_long") + _validate_command(cmd, policy="linux_shell") + def test_readonly_still_blocks_linux_cmds(self) -> None: with self.assertRaises(HTTPException) as ctx: _validate_command("ls -la", policy="readonly") @@ -285,6 +319,31 @@ class NeExecRunTests(unittest.TestCase): self.assertEqual(out["device"]["exec_policy"], "linux_shell") collect.assert_called_once() + @patch("netx_api.ne_exec.credentials_configured", return_value=True) + @patch("netx_api.ne_exec._collect_on_device", return_value="ros-ok") + @patch("netx_api.ne_exec.resolve_cli_target") + def test_execute_mikrotik_shell_policy_allows_multiline(self, resolve, collect, _configured) -> None: + resolve.return_value = ( + _ready_creds(), + { + "source": "managed", + "id": "mt-1", + "exec_policy": "linux_shell", + "name": "rb5009", + "device_type": "mikrotik_routeros", + "ip_address": "10.0.0.10", + }, + ) + db = MagicMock() + script = ":local x 1\n:put $x\n/ip address print" + with patch("netx_api.config.settings") as mock_settings: + mock_settings.ne_exec_policy_enabled = True + out = execute_managed_ne_commands(db, [script], ne_id="mt-1") + self.assertTrue(out["ok"]) + self.assertEqual(out["output"], "ros-ok") + self.assertEqual(out["device"]["exec_policy"], "linux_shell") + collect.assert_called_once() + @patch("netx_api.ne_exec.credentials_configured", return_value=True) @patch("netx_api.ne_exec._collect_on_device", return_value="ok-output") @patch("netx_api.ne_exec.resolve_cli_target") diff --git a/tests/test_ne_netmiko_show.py b/tests/test_ne_netmiko_show.py index dd1b285..2add1bd 100644 --- a/tests/test_ne_netmiko_show.py +++ b/tests/test_ne_netmiko_show.py @@ -1,9 +1,9 @@ -"""Tests for Netmiko show helpers (IOSv leftover-prompt drain / retry).""" +"""Tests for Netmiko show helpers (IOSv leftover-prompt drain; no re-send).""" from __future__ import annotations import unittest -from unittest.mock import MagicMock, call, patch +from unittest.mock import MagicMock, patch from netx_api.ne_netmiko import drain_read_channel, send_show_command @@ -24,25 +24,24 @@ class DrainReadChannelTests(unittest.TestCase): class SendShowCommandTests(unittest.TestCase): @patch("netx_api.ne_netmiko.drain_read_channel") - def test_returns_first_nonempty_send_command(self, drain: MagicMock) -> None: + def test_drains_then_sends_once(self, drain: MagicMock) -> None: conn = MagicMock() conn.send_command.return_value = "*12:00:00 UTC" out = send_show_command(conn, "show clock", read_timeout=30) self.assertEqual(out, "*12:00:00 UTC") conn.send_command.assert_called_once() conn.send_command_timing.assert_not_called() - drain.assert_called() + drain.assert_called_once_with(conn) @patch("netx_api.ne_netmiko.drain_read_channel") - def test_retries_then_falls_back_to_timing_when_empty(self, drain: MagicMock) -> None: + def test_empty_output_does_not_resend(self, drain: MagicMock) -> None: conn = MagicMock() conn.send_command.return_value = "" - conn.send_command_timing.return_value = "Cisco IOS Software" out = send_show_command(conn, "show version", read_timeout=30) - self.assertEqual(out, "Cisco IOS Software") - self.assertEqual(conn.send_command.call_count, 2) - conn.send_command_timing.assert_called_once_with("show version", read_timeout=30) - self.assertGreaterEqual(drain.call_count, 2) + self.assertEqual(out, "") + conn.send_command.assert_called_once() + conn.send_command_timing.assert_not_called() + drain.assert_called_once_with(conn) if __name__ == "__main__": diff --git a/web/src/i18n/en.ts b/web/src/i18n/en.ts index 39e2828..8e14743 100644 --- a/web/src/i18n/en.ts +++ b/web/src/i18n/en.ts @@ -1640,9 +1640,9 @@ const en = { }, execPolicy: { readonly: "Read-only (show/display/ping)", - linuxShell: "Linux shell (single line)", - unrestricted: "Lab open (single line)", - hint: "Applies to MCP/API execManagedNe only. Default is read-only; shell / lab-open require a Linux device type.", + linuxShell: "Shell / script (multiline / heredoc OK)", + unrestricted: "Lab open (multiline / heredoc OK)", + hint: "Applies to MCP/API execManagedNe only. Default is read-only; Linux and MikroTik (routeros/switchos) may use shell / lab-open for multiline scripts and heredoc.", }, source: { manual: "Manual", diff --git a/web/src/i18n/zh.ts b/web/src/i18n/zh.ts index f2259b6..89e55aa 100644 --- a/web/src/i18n/zh.ts +++ b/web/src/i18n/zh.ts @@ -1627,9 +1627,9 @@ const zh = { }, execPolicy: { readonly: "只读（show/display/ping）", - linuxShell: "Linux shell（单行）", - unrestricted: "实验室全开（单行）", - hint: "仅影响 MCP/API 的 execManagedNe。默认只读；仅 Linux 设备类型可选 shell / 实验室全开。", + linuxShell: "Shell/脚本（允许多行/heredoc）", + unrestricted: "实验室全开（允许多行/heredoc）", + hint: "仅影响 MCP/API 的 execManagedNe。默认只读；Linux 与 MikroTik（routeros/switchos）可选 shell / 实验室全开，支持多行脚本与 heredoc。", }, source: { manual: "手工", diff --git a/web/src/pages/managedNe/ManagedNeFormDialog.tsx b/web/src/pages/managedNe/ManagedNeFormDialog.tsx index bad22db..0bb5b78 100644 --- a/web/src/pages/managedNe/ManagedNeFormDialog.tsx +++ b/web/src/pages/managedNe/ManagedNeFormDialog.tsx @@ -14,7 +14,7 @@ import { buildManagedNeSaveBody, emptyManagedNeForm, formFromManagedNe, - isLinuxDeviceType, + allowsOpenExecPolicy, type ManagedNeFormState, } from "./formState"; @@ -139,7 +139,7 @@ export function ManagedNeFormDialog({ setForm((prev) => ({ ...prev, device_type, - exec_policy: isLinuxDeviceType(device_type) ? prev.exec_policy : "readonly", + exec_policy: allowsOpenExecPolicy(device_type) ? prev.exec_policy : "readonly", })); }} > @@ -162,7 +162,7 @@ export function ManagedNeFormDialog({ } > - {isLinuxDeviceType(form.device_type) ? ( + {allowsOpenExecPolicy(form.device_type) ? ( <> diff --git a/web/src/pages/managedNe/formState.ts b/web/src/pages/managedNe/formState.ts index 17ea13e..dfbf42d 100644 --- a/web/src/pages/managedNe/formState.ts +++ b/web/src/pages/managedNe/formState.ts @@ -46,6 +46,22 @@ export function isLinuxDeviceType(deviceType: string | undefined | null): boolea return low === "linux" || low === "linux_ssh" || low === "linux_telnet" || low.startsWith("linux_"); } +export function isMikrotikDeviceType(deviceType: string | undefined | null): boolean { + const low = String(deviceType || "") + .trim() + .toLowerCase(); + return ( + low === "mikrotik_routeros" || + low === "mikrotik_switchos" || + low.startsWith("mikrotik_") + ); +} + +/** Linux + MikroTik may use linux_shell / unrestricted (multiline scripts). */ +export function allowsOpenExecPolicy(deviceType: string | undefined | null): boolean { + return isLinuxDeviceType(deviceType) || isMikrotikDeviceType(deviceType); +} + export function emptyManagedNeForm(): ManagedNeFormState { return { name: "", @@ -173,7 +189,7 @@ export function buildManagedNeSaveBody( ...(form.hop_password ? { hop_password: form.hop_password } : {}), }; if (opts.execPolicyEnabled) { - body.exec_policy = isLinuxDeviceType(form.device_type) ? form.exec_policy : "readonly"; + body.exec_policy = allowsOpenExecPolicy(form.device_type) ? form.exec_policy : "readonly"; } if (form.hop_enabled) { if (!form.hop_host.trim()) throw new Error(opts.hopHostRequired);