fix(ne-exec): allow only show/display CLI and harden agent injection

Co-authored-by: Cursor <cursoragent@cursor.com>

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(
+# Only vendor read-only query verbs (Cisco show / Huawei-ZTE display).
-    r"(?i)^("
+_ALLOWED_PREFIX_RE = re.compile(r"(?i)^(show\s|display\s)")
-    r"show\s|display\s|get\s|"
+
-    r"ping\s|ping6\s|traceroute\s|tracert\s|"
+# Unicode / C1 line separators that can smuggle a second CLI after a show prefix.
-    r"terminal\s+length\s|"
+_FORBIDDEN_LINE_SEPARATORS = ("\u2028", "\u2029", "\x85", "\x0b", "\x0c")
-    r"\?"
-    r")"
-)
 
 
 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):

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": {

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()