diff --git a/.env.example b/.env.example index 0153b53..d079720 100644 --- a/.env.example +++ b/.env.example @@ -3,6 +3,8 @@ NETX_DATABASE_URL=postgresql+psycopg://netx:netx@127.0.0.1:5432/netx NETX_CREDENTIAL_SECRET_KEY= NETX_NE_CONNECT_MAX_WORKERS=5 NETX_NE_CONNECT_TIMEOUT_SEC=30 +# Managed NE CLI: max commands per exec request (default 5, hard cap 50). Lab can set 10+. +NETX_NE_EXEC_MAX_COMMANDS=5 NETX_HOST=127.0.0.1 NETX_PORT=8890 NETX_VENDOR=ZTE diff --git a/docs/MCP.md b/docs/MCP.md index ba3a79a..9998966 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -69,6 +69,7 @@ pip install "git+https://github.com/hansjone/netx.git#subdirectory=packages/netx | `NETX_API_URL` | 否 | `http://127.0.0.1:8890` | netx REST 根地址，可指向远端 | | `NETX_API_TOKEN` | 否 | 空 | API 启用 Bearer 时填写 | | `NETX_LANG` | 否 | `zh` | `zh` / `en`，影响 API 文案 | +| `NETX_NE_EXEC_MAX_COMMANDS` | 否 | `5` | `execManagedNe` 单次最多命令数（硬上限 50）；API 与 MCP 需同设 | 本机默认端口时 **可不设任何变量**。 diff --git a/netx_api/config.py b/netx_api/config.py index 425137d..288e499 100644 --- a/netx_api/config.py +++ b/netx_api/config.py @@ -72,6 +72,8 @@ class Settings(BaseSettings): ne_collect_pending_stale_sec: int = 180 ne_collect_run_timeout_cap_sec: int = 600 ne_collection_data_dir: str = "data/ne_collections" + # Managed NE exec: max CLI commands per request (lab can raise; hard-capped in ne_exec). + ne_exec_max_commands: int = 5 settings = Settings() diff --git a/netx_api/managed_ne_router.py b/netx_api/managed_ne_router.py index fe91111..dd8b899 100644 --- a/netx_api/managed_ne_router.py +++ b/netx_api/managed_ne_router.py @@ -128,7 +128,7 @@ def api_delete_ume_synced_managed_ne(db: Session = Depends(get_db)): @router.post("/exec") def api_exec_managed_ne(body: ManagedNeExecRequest, db: Session = Depends(get_db)): - """Login to a managed NE or UME inventory NE and run read-only CLI (show/display/ping).""" + """Login to a managed NE or UME inventory NE and run read-only CLI (show/display/ping/traceroute).""" return execute_managed_ne_commands( db, body.commands, diff --git a/netx_api/ne_exec.py b/netx_api/ne_exec.py index 26b2f6e..eff5388 100644 --- a/netx_api/ne_exec.py +++ b/netx_api/ne_exec.py @@ -13,11 +13,16 @@ from .config import settings from .ne_collect_runner import _collect_on_device from .ne_crypto import credentials_configured -_EXEC_MAX_COMMANDS = 5 +_EXEC_MAX_COMMANDS_CAP = 50 _EXEC_MAX_OUTPUT = 32_000 _EXEC_READ_TIMEOUT_DEFAULT = 60 _EXEC_READ_TIMEOUT_MAX = 120 + +def _exec_max_commands() -> int: + raw = int(settings.ne_exec_max_commands or 5) + return max(1, min(_EXEC_MAX_COMMANDS_CAP, raw)) + # Block obvious config-change / destructive patterns (case-insensitive). _BLOCKED_RE = re.compile( r"(?i)(" @@ -29,8 +34,10 @@ _BLOCKED_RE = re.compile( r")" ) -# Read-only CLI: show/display plus ping reachability checks. -_ALLOWED_PREFIX_RE = re.compile(r"(?i)^(show\s|display\s|ping\s|ping6\s)") +# Read-only CLI: show/display plus ping/traceroute reachability checks. +_ALLOWED_PREFIX_RE = re.compile( + r"(?i)^(show\s|display\s|ping\s|ping6\s|traceroute\s|tracert\s|trace\s|trace6\s)" +) # Unicode / C1 line separators that can smuggle a second CLI after a show prefix. _FORBIDDEN_LINE_SEPARATORS = ("\u2028", "\u2029", "\x85", "\x0b", "\x0c") @@ -94,8 +101,9 @@ def execute_managed_ne_commands( cmds = [str(c).strip() for c in commands if str(c).strip()] if not cmds: raise HTTPException(status_code=400, detail="commands_required") - if len(cmds) > _EXEC_MAX_COMMANDS: - raise HTTPException(status_code=400, detail=f"too_many_commands (max {_EXEC_MAX_COMMANDS})") + max_cmds = _exec_max_commands() + if len(cmds) > max_cmds: + raise HTTPException(status_code=400, detail=f"too_many_commands (max {max_cmds})") for c in cmds: _validate_command(c) diff --git a/netx_api/ne_schemas.py b/netx_api/ne_schemas.py index f9efa76..c06a76d 100644 --- a/netx_api/ne_schemas.py +++ b/netx_api/ne_schemas.py @@ -117,7 +117,8 @@ class ManagedNeExecRequest(BaseModel): ne_id: str | None = None ume_ne_id: str | None = None - commands: list[str] = Field(min_length=1, max_length=5) + # Absolute ceiling; runtime limit is settings.ne_exec_max_commands (default 5). + commands: list[str] = Field(min_length=1, max_length=50) read_timeout_sec: int | None = Field(default=None, ge=10, le=120) diff --git a/packages/netx-mcp/src/netx_mcp/http_tools.py b/packages/netx-mcp/src/netx_mcp/http_tools.py index e610c81..3c32332 100644 --- a/packages/netx-mcp/src/netx_mcp/http_tools.py +++ b/packages/netx-mcp/src/netx_mcp/http_tools.py @@ -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"], diff --git a/packages/netx-mcp/tests/test_mcp_http.py b/packages/netx-mcp/tests/test_mcp_http.py index 1a2d512..11a6f2e 100644 --- a/packages/netx-mcp/tests/test_mcp_http.py +++ b/packages/netx-mcp/tests/test_mcp_http.py @@ -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) diff --git a/tests/test_ne_exec.py b/tests/test_ne_exec.py index 479232e..4db654a 100644 --- a/tests/test_ne_exec.py +++ b/tests/test_ne_exec.py @@ -20,11 +20,16 @@ class NeExecValidationTests(unittest.TestCase): _validate_command("ping6 2001::1") _validate_command("PING 10.0.0.1 vrf MGMT") + def test_allows_traceroute(self) -> None: + _validate_command("traceroute 192.168.0.1") + _validate_command("tracert 192.168.0.1") + _validate_command("trace 10.0.0.1") + _validate_command("trace6 2001::1") + _validate_command("TRACEROUTE 10.0.0.1 vpn-instance MGMT") + def test_blocks_non_allowed_prefix(self) -> None: for cmd in ( "get system info", - "traceroute 192.168.0.1", - "tracert 192.168.0.1", "terminal length 0", "?", ): @@ -156,6 +161,43 @@ class NeExecRunTests(unittest.TestCase): collect.assert_not_called() resolve.assert_not_called() + @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") + def test_execute_respects_max_commands_setting(self, resolve, collect, _configured) -> None: + db = MagicMock() + cmds = [f"show version {i}" for i in range(6)] + with patch("netx_api.ne_exec.settings") as mock_settings: + mock_settings.ne_exec_max_commands = 5 + with self.assertRaises(HTTPException) as ctx: + execute_managed_ne_commands(db, cmds, ne_id="ne-1") + self.assertIn("too_many_commands", str(ctx.exception.detail)) + collect.assert_not_called() + + resolve.return_value = ( + {"ip_address": "1.1.1.1"}, + { + "source": "managed", + "id": "ne-1", + "ume_ne_id": None, + "name": "R2", + "vendor": "Cisco", + "device_type": "cisco_ios", + "ip_address": "192.168.0.128", + "port": 22, + "protocol": "ssh", + "connect_status": "pass", + "hop_enabled": False, + "hop_vendor": "zte", + }, + ) + with patch("netx_api.ne_exec.settings") as mock_settings: + mock_settings.ne_exec_max_commands = 10 + mock_settings.ne_collect_read_timeout_sec = 120 + out = execute_managed_ne_commands(db, cmds, ne_id="ne-1") + self.assertTrue(out["ok"]) + collect.assert_called_once() + if __name__ == "__main__": unittest.main()