feat(ne-exec): make max CLI commands configurable and allow traceroute

Add NETX_NE_EXEC_MAX_COMMANDS (default 5, cap 50) and whitelist traceroute/tracert/trace/trace6 prefixes for lab/ops use.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-07-28 10:57:55 +08:00
parent a6a5abc9ef
commit 88479245bb
9 changed files with 116 additions and 15 deletions

View file

@ -2,10 +2,24 @@
from __future__ import annotations
import os
from typing import Any, Callable
from .http_client import http_json, http_post_json, mcp_from_handler_result, quote_ne_id
_EXEC_MAX_COMMANDS_CAP = 50
_EXEC_MAX_COMMANDS_DEFAULT = 5
def exec_max_commands() -> int:
"""Mirror netx API NETX_NE_EXEC_MAX_COMMANDS (default 5, hard cap 50)."""
try:
raw = int(os.getenv("NETX_NE_EXEC_MAX_COMMANDS") or _EXEC_MAX_COMMANDS_DEFAULT)
except ValueError:
raw = _EXEC_MAX_COMMANDS_DEFAULT
return max(1, min(_EXEC_MAX_COMMANDS_CAP, raw))
UME_RAW_GROUP_FIELDS = [
"alarm_alarm_key",
"alarm_host_name",
@ -241,7 +255,7 @@ def _exec_managed_ne(args: dict[str, Any]) -> dict[str, Any]:
commands = [str(c).strip() for c in raw_cmds if str(c).strip()]
if not commands:
return {"ok": False, "error": "commands_required", "error_code": "commands_required"}
if len(commands) > 5:
if len(commands) > exec_max_commands():
return {"ok": False, "error": "too_many_commands", "error_code": "too_many_commands"}
body: dict[str, Any] = {"commands": commands}
if ne_id:
@ -418,13 +432,22 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [
},
{
"name": "execManagedNe",
"description": "Run read-only CLI via netx (show/display/ping; max 5 commands). Use ne_id (managed NE) OR ume_ne_id (UME inventory).",
"description": (
f"Run read-only CLI via netx (show/display/ping/traceroute; "
f"max {exec_max_commands()} commands, NETX_NE_EXEC_MAX_COMMANDS). "
"Use ne_id (managed NE) OR ume_ne_id (UME inventory)."
),
"inputSchema": {
"type": "object",
"properties": {
"ne_id": {"type": "string"},
"ume_ne_id": {"type": "string"},
"commands": {"type": "array", "items": {"type": "string"}, "minItems": 1, "maxItems": 5},
"commands": {
"type": "array",
"items": {"type": "string"},
"minItems": 1,
"maxItems": exec_max_commands(),
},
"read_timeout_sec": {"type": "integer", "minimum": 10, "maximum": 120},
},
"required": ["commands"],

View file

@ -12,12 +12,15 @@ import pytest
from netx_mcp.http_tools import HTTP_MCP_TOOLS, call_http_tool
def test_http_mcp_tool_list_has_twelve_tools() -> None:
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) == 12
assert len(names) == 13
assert "queryUmeAlarms" in names
assert "queryUmeAlarmsRaw" in names
assert "execManagedNe" in names
assert "listCliTargets" in names
exec_tool = next(t for t in HTTP_MCP_TOOLS if t.get("name") == "execManagedNe")
assert exec_tool["inputSchema"]["properties"]["commands"]["maxItems"] >= 5
def test_call_query_ume_alarms_forwards_http() -> None:
@ -59,6 +62,25 @@ def test_call_exec_managed_ne_posts_body() -> None:
assert payload["ok"] is True
def test_call_exec_managed_ne_respects_max_commands_env(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("NETX_NE_EXEC_MAX_COMMANDS", "10")
cmds = [f"show version {i}" for i in range(6)]
with patch("netx_mcp.http_tools.http_post_json") as mock_post:
mock_post.return_value = {"ok": True, "data": {"ok": True, "output": "ok"}}
out = call_http_tool("execManagedNe", {"ne_id": "abc", "commands": cmds})
mock_post.assert_called_once()
text = out["content"][0]["text"]
payload = json.loads(text)
assert payload["ok"] is True
monkeypatch.setenv("NETX_NE_EXEC_MAX_COMMANDS", "5")
out = call_http_tool("execManagedNe", {"ne_id": "abc", "commands": cmds})
text = out["content"][0]["text"]
payload = json.loads(text)
assert payload.get("ok") is False
assert payload.get("error_code") == "too_many_commands"
def test_stdio_initialize_and_tools_list() -> None:
proc = subprocess.Popen(
[sys.executable, "-m", "netx_mcp"],
@ -81,7 +103,7 @@ def test_stdio_initialize_and_tools_list() -> None:
list_line = proc.stdout.readline()
list_resp = json.loads(list_line)
tools = list_resp["result"]["tools"]
assert len(tools) == 12
assert len(tools) == 13
proc.terminate()
proc.wait(timeout=5)