From 981347b5b1d6b9eb5dcb9a27d448fa0051fe3080 Mon Sep 17 00:00:00 2001 From: oliver Date: Thu, 28 May 2026 22:12:50 +0800 Subject: [PATCH] feat(ne): add managed NE read-only exec API Add a guarded managed-ne exec endpoint for oclaw ops tools to login managed devices and run read-only CLI safely. Include request schema and unit tests for command guardrails and execution flow. Co-authored-by: Cursor --- netx_api/managed_ne_router.py | 14 +++- netx_api/ne_exec.py | 127 ++++++++++++++++++++++++++++++++++ netx_api/ne_schemas.py | 8 +++ tests/test_ne_exec.py | 60 ++++++++++++++++ 4 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 netx_api/ne_exec.py create mode 100644 tests/test_ne_exec.py diff --git a/netx_api/managed_ne_router.py b/netx_api/managed_ne_router.py index 0f9d217..ce27379 100644 --- a/netx_api/managed_ne_router.py +++ b/netx_api/managed_ne_router.py @@ -8,7 +8,8 @@ from .db import get_db from .device_types import SUPPORTED_DEVICE_TYPES, SUPPORTED_VENDORS from .ne_connect import schedule_connect_tests from .ne_crypto import credentials_configured -from .ne_schemas import BatchHopApplyRequest, ConnectTestRequest, ManagedNeCreate, ManagedNeUpdate +from .ne_exec import execute_managed_ne_commands +from .ne_schemas import BatchHopApplyRequest, ConnectTestRequest, ManagedNeCreate, ManagedNeExecRequest, ManagedNeUpdate from .ne_service import ( batch_apply_hop_proxy, build_managed_ne_import_template, @@ -102,6 +103,17 @@ def api_batch_delete_managed_ne(body: ConnectTestRequest, db: Session = Depends( return batch_delete_managed_ne(db, body.ids) +@router.post("/exec") +def api_exec_managed_ne(body: ManagedNeExecRequest, db: Session = Depends(get_db)): + """Login to a managed NE and run read-only CLI (show/display/ping). For oclaw ops tools.""" + return execute_managed_ne_commands( + db, + body.ne_id, + body.commands, + read_timeout_sec=body.read_timeout_sec, + ) + + @router.post("/connect-test") def api_connect_test(body: ConnectTestRequest, db: Session = Depends(get_db)): ids = [str(x).strip() for x in body.ids if str(x).strip()] diff --git a/netx_api/ne_exec.py b/netx_api/ne_exec.py new file mode 100644 index 0000000..0ed9611 --- /dev/null +++ b/netx_api/ne_exec.py @@ -0,0 +1,127 @@ +"""Execute read-only CLI on netx managed network elements (for oclaw ops tools).""" + +from __future__ import annotations + +import re +from typing import Any + +from fastapi import HTTPException +from sqlalchemy.orm import Session + +from .config import settings +from .models import ManagedNE +from .ne_collect_runner import _collect_on_device +from .ne_crypto import credentials_configured +from .ne_service import get_device_credentials, row_to_out + +_EXEC_MAX_COMMANDS = 5 +_EXEC_MAX_OUTPUT = 32_000 +_EXEC_READ_TIMEOUT_DEFAULT = 60 +_EXEC_READ_TIMEOUT_MAX = 120 + +# Block obvious config-change / destructive patterns (case-insensitive). +_BLOCKED_RE = re.compile( + r"(?i)(" + 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")" +) + +_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")" +) + + +def _validate_command(command: str) -> None: + cmd = str(command or "").strip() + if not cmd: + 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", "`")): + 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): + raise HTTPException(status_code=400, detail="command_not_allowed_prefix") + + +def _normalize_read_timeout(sec: int | None) -> int: + raw = int(sec if sec is not None else _EXEC_READ_TIMEOUT_DEFAULT) + return max(10, min(_EXEC_READ_TIMEOUT_MAX, raw)) + + +def execute_managed_ne_commands( + db: Session, + ne_id: str, + commands: list[str], + *, + read_timeout_sec: int | None = None, +) -> dict[str, Any]: + if not credentials_configured(): + raise HTTPException(status_code=503, detail="credential_secret_key_not_configured") + nid = str(ne_id or "").strip() + if not nid: + raise HTTPException(status_code=400, detail="ne_id_required") + 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})") + for c in cmds: + _validate_command(c) + + row = db.get(ManagedNE, nid) + if not row: + raise HTTPException(status_code=404, detail="managed_ne_not_found") + + read_timeout = _normalize_read_timeout(read_timeout_sec) + creds = get_device_credentials(row) + meta = row_to_out(row).model_dump() + # Shallow copy for response (no secrets). + device = { + "id": meta["id"], + "name": meta["name"], + "vendor": meta["vendor"], + "device_type": meta["device_type"], + "ip_address": meta["ip_address"], + "port": meta["port"], + "protocol": meta["protocol"], + "connect_status": meta["connect_status"], + "hop_enabled": meta["hop_enabled"], + "hop_vendor": meta["hop_vendor"], + } + + prev_collect_timeout = int(settings.ne_collect_read_timeout_sec or 120) + try: + settings.ne_collect_read_timeout_sec = read_timeout + output = _collect_on_device(creds, cmds) + except Exception as exc: + return { + "ok": False, + "device": device, + "commands": cmds, + "read_timeout_sec": read_timeout, + "error": type(exc).__name__, + "detail": str(exc)[:2000], + } + finally: + settings.ne_collect_read_timeout_sec = prev_collect_timeout + + if len(output) > _EXEC_MAX_OUTPUT: + output = output[:_EXEC_MAX_OUTPUT] + "\n...[truncated]" + + return { + "ok": True, + "device": device, + "commands": cmds, + "read_timeout_sec": read_timeout, + "output": output, + } diff --git a/netx_api/ne_schemas.py b/netx_api/ne_schemas.py index f939a60..afc5b41 100644 --- a/netx_api/ne_schemas.py +++ b/netx_api/ne_schemas.py @@ -109,6 +109,14 @@ class ConnectTestRequest(BaseModel): ids: list[str] = Field(min_length=1) +class ManagedNeExecRequest(BaseModel): + """Run read-only show/display CLI on a managed NE (oclaw ops integration).""" + + ne_id: str + commands: list[str] = Field(min_length=1, max_length=5) + read_timeout_sec: int | None = Field(default=None, ge=10, le=120) + + class HopProxyConfig(BaseModel): """Shared jump-host (proxy) settings applied to one or many NEs.""" diff --git a/tests/test_ne_exec.py b/tests/test_ne_exec.py new file mode 100644 index 0000000..0068737 --- /dev/null +++ b/tests/test_ne_exec.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from fastapi import HTTPException + +from netx_api.ne_exec import _validate_command, execute_managed_ne_commands + + +class NeExecValidationTests(unittest.TestCase): + def test_allows_show(self) -> None: + _validate_command("show ip interface brief") + + 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_blocks_configure(self) -> None: + with self.assertRaises(HTTPException): + _validate_command("configure terminal") + + def test_blocks_non_show_prefix(self) -> None: + with self.assertRaises(HTTPException) as ctx: + _validate_command("interface GigabitEthernet0/0") + self.assertEqual(ctx.exception.detail, "command_not_allowed_prefix") + + +class NeExecRunTests(unittest.TestCase): + @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_success(self, _creds, _collect, _configured) -> None: + row = MagicMock() + row.id = "ne-1" + db = MagicMock() + db.get.return_value = row + with patch("netx_api.ne_exec.row_to_out") as row_out: + row_out.return_value.model_dump.return_value = { + "id": "ne-1", + "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", + } + out = execute_managed_ne_commands(db, "ne-1", ["show version"]) + self.assertTrue(out["ok"]) + self.assertEqual(out["output"], "ok-output") + self.assertEqual(out["commands"], ["show version"]) + + +if __name__ == "__main__": + unittest.main()