diff --git a/netx_api/ne_exec.py b/netx_api/ne_exec.py index 0ed9611..a42138d 100644 --- a/netx_api/ne_exec.py +++ b/netx_api/ne_exec.py @@ -25,18 +25,16 @@ _BLOCKED_RE = re.compile( r"configure\s+terminal|conf\s+t\b|" r"\bwrite\s+(memory|erase)|\bcopy\s+run|\bcopy\s+startup|" r"\breload\b|\breboot\b|\berase\b|\bformat\b|\bdelete\b|" - r"\bcommit\b|\brollback\b|startup-config" + r"\bcommit\b|\brollback\b|startup-config|" + r"\bsystem-view\b|\bip\s+address\b|\bvlan\s+\d" r")" ) -_ALLOWED_PREFIX_RE = re.compile( - r"(?i)^(" - r"show\s|display\s|get\s|" - r"ping\s|ping6\s|traceroute\s|tracert\s|" - r"terminal\s+length\s|" - r"\?" - r")" -) +# Only vendor read-only query verbs (Cisco show / Huawei-ZTE display). +_ALLOWED_PREFIX_RE = re.compile(r"(?i)^(show\s|display\s)") + +# Unicode / C1 line separators that can smuggle a second CLI after a show prefix. +_FORBIDDEN_LINE_SEPARATORS = ("\u2028", "\u2029", "\x85", "\x0b", "\x0c") def _validate_command(command: str) -> None: @@ -47,6 +45,8 @@ def _validate_command(command: str) -> None: raise HTTPException(status_code=400, detail="command_too_long") if any(ch in cmd for ch in ("|", ";", "\n", "\r", "`")): raise HTTPException(status_code=400, detail="command_chars_not_allowed") + if any(sep in cmd for sep in _FORBIDDEN_LINE_SEPARATORS): + raise HTTPException(status_code=400, detail="command_chars_not_allowed") if _BLOCKED_RE.search(cmd): raise HTTPException(status_code=400, detail="command_blocked") if not _ALLOWED_PREFIX_RE.match(cmd): diff --git a/packages/netx-mcp/src/netx_mcp/http_tools.py b/packages/netx-mcp/src/netx_mcp/http_tools.py index ed4d620..85b431c 100644 --- a/packages/netx-mcp/src/netx_mcp/http_tools.py +++ b/packages/netx-mcp/src/netx_mcp/http_tools.py @@ -391,7 +391,7 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [ }, { "name": "execManagedNe", - "description": "Run read-only CLI on a managed NE via netx (show/display/ping; max 5 commands).", + "description": "Run read-only CLI on a managed NE via netx (show/display only; max 5 commands).", "inputSchema": { "type": "object", "properties": { diff --git a/tests/test_ne_exec.py b/tests/test_ne_exec.py index 0068737..a5f0836 100644 --- a/tests/test_ne_exec.py +++ b/tests/test_ne_exec.py @@ -12,6 +12,24 @@ class NeExecValidationTests(unittest.TestCase): def test_allows_show(self) -> None: _validate_command("show ip interface brief") + def test_allows_display(self) -> None: + _validate_command("display interface brief") + + def test_blocks_ping_and_other_non_show_display(self) -> None: + for cmd in ( + "ping 192.168.0.1", + "ping6 2001::1", + "get system info", + "traceroute 192.168.0.1", + "tracert 192.168.0.1", + "terminal length 0", + "?", + ): + with self.subTest(cmd=cmd): + with self.assertRaises(HTTPException) as ctx: + _validate_command(cmd) + self.assertEqual(ctx.exception.detail, "command_not_allowed_prefix") + def test_blocks_pipe(self) -> None: with self.assertRaises(HTTPException) as ctx: _validate_command("show run | include hostname") @@ -27,6 +45,37 @@ class NeExecValidationTests(unittest.TestCase): _validate_command("interface GigabitEthernet0/0") self.assertEqual(ctx.exception.detail, "command_not_allowed_prefix") + def test_blocks_newline_chained_show_and_configure(self) -> None: + with self.assertRaises(HTTPException) as ctx: + _validate_command("show interface\nconfigure terminal") + self.assertEqual(ctx.exception.detail, "command_chars_not_allowed") + + def test_blocks_unicode_line_separator_chained_commands(self) -> None: + with self.assertRaises(HTTPException) as ctx: + _validate_command("show interface\u2028system-view") + self.assertEqual(ctx.exception.detail, "command_chars_not_allowed") + + def test_blocks_system_view_on_one_line(self) -> None: + with self.assertRaises(HTTPException) as ctx: + _validate_command("system-view") + self.assertEqual(ctx.exception.detail, "command_blocked") + + def test_blocks_ip_address_config(self) -> None: + with self.assertRaises(HTTPException) as ctx: + _validate_command("ip address 1.1.1.1 255.255.255.0") + self.assertEqual(ctx.exception.detail, "command_blocked") + + def test_agent_batch_rejects_configure_before_connect(self) -> None: + cmds = [ + "show interface", + "configure terminal", + "interface ge1/1", + "ip address 1.1.1.1 255.255.255.0", + ] + for c in cmds[1:]: + with self.assertRaises(HTTPException): + _validate_command(c) + class NeExecRunTests(unittest.TestCase): @patch("netx_api.ne_exec.credentials_configured", return_value=True) @@ -55,6 +104,21 @@ class NeExecRunTests(unittest.TestCase): self.assertEqual(out["output"], "ok-output") self.assertEqual(out["commands"], ["show version"]) + @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.get_device_credentials", return_value={"ip_address": "1.1.1.1"}) + def test_execute_skips_device_when_any_command_invalid(self, _creds, collect, _configured) -> None: + db = MagicMock() + with self.assertRaises(HTTPException) as ctx: + execute_managed_ne_commands( + db, + "ne-1", + ["show interface", "configure terminal"], + ) + self.assertEqual(ctx.exception.detail, "command_blocked") + collect.assert_not_called() + db.get.assert_not_called() + if __name__ == "__main__": unittest.main()