hansjone/netx
1
0
Fork 0
mirror of https://github.com/hansjone/netx.git synced 2026-10-09 04:20:45 +08:00
Code Issues Projects Releases 1 Packages Wiki Activity Actions 2

fix(ne-exec): allow read-only pipe filters in managed NE CLI

Browse source 
Whitelist include/exclude/begin/section/count/match/grep/one-line after show/display pipes while blocking redirect/append/tee/send.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
oliver 2026-07-14 16:24:47 +08:00
parent c5973eec69
commit 33af0efc98
2 changed files with 58 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

22
netx_api/ne_exec.py
View file

@ -35,6 +35,25 @@ _ALLOWED_PREFIX_RE = re.compile(r"(?i)^(show\s|display\s|ping\s|ping6\s)")
# Unicode / C1 line separators that can smuggle a second CLI after a show prefix.
_FORBIDDEN_LINE_SEPARATORS = ("\u2028", "\u2029", "\x85", "\x0b", "\x0c")
# Pipe segments allowed after show/display (output filtering only).
_ALLOWED_PIPE_SEGMENT_RE = re.compile(
r"(?i)^(include|exclude|begin|section|count|match|grep|one-line)(\s|$)"
)
_BLOCKED_PIPE_SEGMENT_RE = re.compile(r"(?i)\b(redirect|append|tee|send)\b")
def _validate_pipe_segments(cmd: str) -> None:
if "|" not in cmd:
return
parts = [p.strip() for p in cmd.split("|")]
if len(parts) < 2 or not parts[0] or any(not p for p in parts[1:]):
raise HTTPException(status_code=400, detail="command_pipe_not_allowed")
for segment in parts[1:]:
if _BLOCKED_PIPE_SEGMENT_RE.search(segment):
raise HTTPException(status_code=400, detail="command_pipe_not_allowed")
if not _ALLOWED_PIPE_SEGMENT_RE.match(segment):
raise HTTPException(status_code=400, detail="command_pipe_not_allowed")
def _validate_command(command: str) -> None:
cmd = str(command or "").strip()
@ -42,7 +61,7 @@ def _validate_command(command: str) -> None:
raise HTTPException(status_code=400, detail="empty_command")
if len(cmd) > 500:
raise HTTPException(status_code=400, detail="command_too_long")
if any(ch in cmd for ch in ("|", ";", "\n", "\r", "`")):
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")
@ -50,6 +69,7 @@ def _validate_command(command: str) -> None:
raise HTTPException(status_code=400, detail="command_blocked")
if not _ALLOWED_PREFIX_RE.match(cmd):
raise HTTPException(status_code=400, detail="command_not_allowed_prefix")
_validate_pipe_segments(cmd)
def _normalize_read_timeout(sec: int | None) -> int:

42
tests/test_ne_exec.py
View file

@ -33,16 +33,48 @@ class NeExecValidationTests(unittest.TestCase):
_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")
self.assertEqual(ctx.exception.status_code, 400)
self.assertEqual(ctx.exception.detail, "command_chars_not_allowed")
def test_allows_pipe_filter_subcommands(self) -> None:
for cmd in (
"show run | include hostname",
"show configuration | include hostname",
"display current-configuration | include sysname",
"show run | exclude ^!",
"show run | begin interface",
"show run | section ^router",
"show ip route | count",
"show run | match hostname",
"show run | grep hostname",
"show run | one-line",
"show run | include x | include y",
):
with self.subTest(cmd=cmd):
_validate_command(cmd)
def test_blocks_pipe_redirect_and_unknown(self) -> None:
for cmd in (
"show run | redirect tftp://1.1.1.1/config",
"show run | append flash:cfg.txt",
"show run | tee flash:cfg.txt",
"show run | send log",
"show run | unknown-filter x",
"show run |",
"show run | | include x",
):
with self.subTest(cmd=cmd):
with self.assertRaises(HTTPException) as ctx:
_validate_command(cmd)
self.assertEqual(ctx.exception.status_code, 400)
self.assertEqual(ctx.exception.detail, "command_pipe_not_allowed")
def test_blocks_configure(self) -> None:
with self.assertRaises(HTTPException):
_validate_command("configure terminal")
def test_blocks_configure_after_pipe(self) -> None:
with self.assertRaises(HTTPException) as ctx:
_validate_command("show run | configure terminal")
self.assertEqual(ctx.exception.detail, "command_blocked")
def test_blocks_non_show_prefix(self) -> None:
with self.assertRaises(HTTPException) as ctx:
_validate_command("interface GigabitEthernet0/0")