diff --git a/.env.example b/.env.example
index 545d844..c9083fd 100644
--- a/.env.example
+++ b/.env.example
@@ -5,6 +5,8 @@ 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
+# Opt-in: show/allow per-NE exec_policy (linux_shell|unrestricted). Default false.
+NETX_NE_EXEC_POLICY_ENABLED=false
NETX_HOST=127.0.0.1
NETX_PORT=8890
NETX_VENDOR=ZTE
diff --git a/alembic/versions/20260921_exec_policy.py b/alembic/versions/20260921_exec_policy.py
new file mode 100644
index 0000000..60b5194
--- /dev/null
+++ b/alembic/versions/20260921_exec_policy.py
@@ -0,0 +1,35 @@
+"""Add managed_ne.exec_policy for per-NE CLI exec gates.
+
+Revision ID: 20260921_exec_policy
+Revises: 20260812_ne_collect_trigger
+Create Date: 2026-09-21
+"""
+
+from __future__ import annotations
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "20260921_exec_policy"
+down_revision: Union[str, Sequence[str], None] = "20260812_ne_collect_trigger"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ from netx_api.schema_patches import apply_hop_schema_safety_net
+
+ apply_hop_schema_safety_net(op.get_bind())
+
+
+def downgrade() -> None:
+ bind = op.get_bind()
+ dialect = str(getattr(bind.dialect, "name", "") or "").lower()
+ if dialect.startswith("postgres"):
+ op.execute("ALTER TABLE managed_ne DROP COLUMN IF EXISTS exec_policy")
+ else:
+ try:
+ op.drop_column("managed_ne", "exec_policy")
+ except Exception:
+ pass
diff --git a/netx_api/app_startup.py b/netx_api/app_startup.py
index c4e563e..bd7c397 100644
--- a/netx_api/app_startup.py
+++ b/netx_api/app_startup.py
@@ -43,6 +43,10 @@ def run_api_startup() -> None:
"""Full API boot sequence previously inlined in ``main.on_startup``."""
assert_secure_defaults_or_exit()
_configure_ume_diag_logging()
+ _log.info(
+ "startup: ne_exec_policy_enabled=%s",
+ bool(getattr(settings, "ne_exec_policy_enabled", False)),
+ )
Base.metadata.create_all(bind=engine)
alembic_ok = True
diff --git a/netx_api/cli_resolve.py b/netx_api/cli_resolve.py
index 4227cbc..ebfea95 100644
--- a/netx_api/cli_resolve.py
+++ b/netx_api/cli_resolve.py
@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
from .device_types import SUPPORTED_DEVICE_TYPES
from .models import CliConnectProfile, ManagedNE, UmeCliOverride, UmeInventoryNE
from .ne_crypto import decrypt_secret
+from .ne_exec_guard import EXEC_POLICY_READONLY, normalize_exec_policy
from .ne_service import get_device_credentials, row_to_out
_BUILTIN_NE_TYPE_RULES: list[tuple[re.Pattern[str], str, str]] = [
@@ -141,6 +142,7 @@ def resolve_cli_target(
"port": meta["port"],
"protocol": meta["protocol"],
"connect_status": meta["connect_status"],
+ "exec_policy": normalize_exec_policy(meta.get("exec_policy")),
"hop_enabled": meta["hop_enabled"],
"hop_vendor": meta["hop_vendor"],
}
@@ -198,6 +200,7 @@ def resolve_cli_target(
"port": int(profile.port or 22),
"protocol": str(profile.protocol or "ssh"),
"connect_status": connect_status,
+ "exec_policy": EXEC_POLICY_READONLY,
"hop_enabled": bool(profile.hop_enabled),
"hop_vendor": str(profile.hop_vendor or ""),
"cli_profile_id": str(profile.id),
diff --git a/netx_api/config.py b/netx_api/config.py
index 3354c0c..ccbdc2a 100644
--- a/netx_api/config.py
+++ b/netx_api/config.py
@@ -1,10 +1,20 @@
from __future__ import annotations
+from pathlib import Path
+
from pydantic_settings import BaseSettings, SettingsConfigDict
+# Always load repo-root `.env` (cwd-independent). Optional cwd `.env` overrides last.
+_NETX_ROOT = Path(__file__).resolve().parents[1]
+_ENV_FILES = (str(_NETX_ROOT / ".env"), ".env")
+
class Settings(BaseSettings):
- model_config = SettingsConfigDict(env_file=".env", env_prefix="NETX_", extra="ignore")
+ model_config = SettingsConfigDict(
+ env_file=_ENV_FILES,
+ env_prefix="NETX_",
+ extra="ignore",
+ )
database_url: str = "postgresql+psycopg://netx:netx@127.0.0.1:5432/netx"
host: str = "127.0.0.1"
@@ -114,6 +124,8 @@ class Settings(BaseSettings):
biz_state_heavy_workers: int = 4
# Managed NE exec: max CLI commands per request (lab can raise; hard-capped in ne_exec).
ne_exec_max_commands: int = 5
+ # Opt-in: allow per-NE exec_policy (linux_shell/unrestricted). Default off — UI hidden.
+ ne_exec_policy_enabled: bool = False
# WebCRT interactive terminal sessions (multi-operator concurrent terminals).
webcrt_max_sessions: int = 40
# Per-user cap (0 = unlimited beyond global max).
diff --git a/netx_api/managed_ne_router.py b/netx_api/managed_ne_router.py
index 4ae5bf1..fc8494d 100644
--- a/netx_api/managed_ne_router.py
+++ b/netx_api/managed_ne_router.py
@@ -75,9 +75,14 @@ def api_list_managed_ne(
@router.get("/meta/device-types")
def api_device_types():
# Include generic/linux so LLDP/WebCRT placeholders can be edited without a bogus select value.
+ from .config import settings
from .device_types import WEBCRT_DEVICE_TYPES
- return {"device_types": list(WEBCRT_DEVICE_TYPES), "vendors": list(SUPPORTED_VENDORS)}
+ return {
+ "device_types": list(WEBCRT_DEVICE_TYPES),
+ "vendors": list(SUPPORTED_VENDORS),
+ "exec_policy_enabled": bool(getattr(settings, "ne_exec_policy_enabled", False)),
+ }
@router.get("/meta/credentials-configured")
@@ -150,7 +155,7 @@ def api_exec_managed_ne(
ctx: Annotated[AuthContext, Depends(require_user)],
db: Session = Depends(get_db),
):
- """Login to a managed NE or UME inventory NE and run read-only CLI (show/display/ping/traceroute)."""
+ """Login to a managed NE or UME inventory NE and run CLI (policy from managed NE exec_policy)."""
uid, uname = _actor(ctx)
out = execute_managed_ne_commands(
db,
diff --git a/netx_api/models/managed_ne.py b/netx_api/models/managed_ne.py
index 0fc9fde..355840a 100644
--- a/netx_api/models/managed_ne.py
+++ b/netx_api/models/managed_ne.py
@@ -19,6 +19,8 @@ class ManagedNE(Base):
name: Mapped[str] = mapped_column(String(256), default="", index=True)
vendor: Mapped[str] = mapped_column(String(64), default="Other", index=True)
device_type: Mapped[str] = mapped_column(String(128), default="")
+ # MCP/API execManagedNe command gate: readonly | linux_shell | unrestricted
+ exec_policy: Mapped[str] = mapped_column(String(32), default="readonly")
# Not unique: WebCRT sessions may share a host IP with distinct session names.
# Inventory create/update still enforces uniqueness in ne_service.
ip_address: Mapped[str] = mapped_column(String(128), index=True)
diff --git a/netx_api/ne_exec.py b/netx_api/ne_exec.py
index 62206c7..fa4592c 100644
--- a/netx_api/ne_exec.py
+++ b/netx_api/ne_exec.py
@@ -14,7 +14,12 @@ from .config import settings
from .db import SessionLocal
from .ne_collect_runner import _collect_on_device
from .ne_crypto import credentials_configured
-from .ne_exec_guard import _validate_command, validate_ne_exec_command
+from .ne_exec_guard import (
+ _validate_command,
+ effective_exec_policy,
+ normalize_exec_policy,
+ validate_ne_exec_command,
+)
_EXEC_MAX_COMMANDS_CAP = 50
_EXEC_MAX_OUTPUT = 32_000
@@ -28,6 +33,8 @@ __all__ = [
"_validate_command",
"execute_managed_ne_commands",
"execute_managed_ne_commands_batch",
+ "effective_exec_policy",
+ "normalize_exec_policy",
"validate_ne_exec_command",
]
@@ -62,10 +69,16 @@ def execute_managed_ne_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_ne_exec_command(c)
creds, device = resolve_cli_target(db, managed_ne_id=mid or None, ume_ne_id=uid or None)
+ exec_policy = effective_exec_policy(
+ (device or {}).get("exec_policy") or (creds or {}).get("exec_policy"),
+ device_type=(device or {}).get("device_type") or (creds or {}).get("device_type"),
+ )
+ if isinstance(device, dict):
+ device["exec_policy"] = exec_policy
+ for c in cmds:
+ validate_ne_exec_command(c, policy=exec_policy)
skip = cli_creds_skip_reason(creds, interactive=False)
if skip:
return {
diff --git a/netx_api/ne_exec_guard.py b/netx_api/ne_exec_guard.py
index dd408a6..e8eb8a6 100644
--- a/netx_api/ne_exec_guard.py
+++ b/netx_api/ne_exec_guard.py
@@ -1,4 +1,11 @@
-"""NE CLI command allow/deny gates (read-only exec for ops tools)."""
+"""NE CLI command allow/deny gates (execManagedNe / ops tools).
+
+Policies (per managed NE ``exec_policy``):
+
+- ``readonly`` (default): network CLI only — show/display/ping/traceroute.
+- ``linux_shell``: single-line shell; no network prefix/pipe rules; no write-deny list.
+- ``unrestricted``: same as linux_shell (lab open); kept distinct for audit/UI.
+"""
from __future__ import annotations
@@ -6,6 +13,13 @@ import re
from fastapi import HTTPException
+EXEC_POLICY_READONLY = "readonly"
+EXEC_POLICY_LINUX_SHELL = "linux_shell"
+EXEC_POLICY_UNRESTRICTED = "unrestricted"
+EXEC_POLICIES = frozenset(
+ {EXEC_POLICY_READONLY, EXEC_POLICY_LINUX_SHELL, EXEC_POLICY_UNRESTRICTED}
+)
+
# Block obvious config-change / destructive patterns (case-insensitive).
_BLOCKED_RE = re.compile(
r"(?i)("
@@ -39,6 +53,49 @@ _ALLOWED_PIPE_SEGMENT_RE = re.compile(
_BLOCKED_PIPE_SEGMENT_RE = re.compile(r"(?i)\b(redirect|append|tee|send)\b")
+def normalize_exec_policy(raw: str | None) -> str:
+ p = str(raw or "").strip().lower()
+ return p if p in EXEC_POLICIES else EXEC_POLICY_READONLY
+
+
+def is_linux_device_type(device_type: str | None) -> bool:
+ low = str(device_type or "").strip().lower()
+ return low in ("linux", "linux_ssh", "linux_telnet") or low.startswith("linux_")
+
+
+def exec_policy_feature_enabled() -> bool:
+ """Global kill-switch: off → always readonly (UI hidden, API rejects open policies)."""
+ from .config import settings
+
+ return bool(getattr(settings, "ne_exec_policy_enabled", False))
+
+
+def effective_exec_policy(raw: str | None, *, device_type: str | None = None) -> str:
+ """Policy used at exec time (forces readonly when feature off or non-linux)."""
+ if not exec_policy_feature_enabled():
+ return EXEC_POLICY_READONLY
+ pol = normalize_exec_policy(raw)
+ if pol != EXEC_POLICY_READONLY and not is_linux_device_type(device_type):
+ return EXEC_POLICY_READONLY
+ return pol
+
+
+def require_exec_policy_writable(
+ raw: str | None,
+ *,
+ device_type: str | None = None,
+) -> str:
+ """Normalize for create/update; reject open policies when feature off or non-linux."""
+ pol = normalize_exec_policy(raw)
+ if pol == EXEC_POLICY_READONLY:
+ return pol
+ if not exec_policy_feature_enabled():
+ raise HTTPException(status_code=400, detail="exec_policy_feature_disabled")
+ if not is_linux_device_type(device_type):
+ raise HTTPException(status_code=400, detail="exec_policy_requires_linux_device_type")
+ return pol
+
+
def _validate_pipe_segments(cmd: str) -> None:
if "|" not in cmd:
return
@@ -52,13 +109,14 @@ def _validate_pipe_segments(cmd: str) -> None:
raise HTTPException(status_code=400, detail="command_pipe_not_allowed")
-def validate_ne_exec_command(command: str) -> None:
- """Raise HTTPException if command is empty, smuggled, blocked, or not allowlisted."""
- 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")
+def _validate_single_line(cmd: str) -> None:
+ 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")
+
+
+def _validate_readonly_command(cmd: str) -> None:
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):
@@ -70,5 +128,20 @@ def validate_ne_exec_command(command: str) -> None:
_validate_pipe_segments(cmd)
+def validate_ne_exec_command(command: str, *, policy: str = EXEC_POLICY_READONLY) -> None:
+ """Raise HTTPException if command is empty, smuggled, blocked, or not allowlisted."""
+ 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")
+ pol = normalize_exec_policy(policy)
+ if pol in (EXEC_POLICY_LINUX_SHELL, EXEC_POLICY_UNRESTRICTED):
+ # One command string per slot; shell metacharacters (|;&&`$) allowed.
+ _validate_single_line(cmd)
+ return
+ _validate_readonly_command(cmd)
+
+
# Back-compat alias used by tests / callers.
_validate_command = validate_ne_exec_command
diff --git a/netx_api/ne_schemas.py b/netx_api/ne_schemas.py
index 3b65fff..59a4f46 100644
--- a/netx_api/ne_schemas.py
+++ b/netx_api/ne_schemas.py
@@ -6,8 +6,10 @@ from typing import Literal
from pydantic import BaseModel, Field, field_validator
from .device_types import SUPPORTED_VENDORS
+from .ne_exec_guard import EXEC_POLICIES, EXEC_POLICY_READONLY
ConnectStatus = Literal["unknown", "testing", "pass", "fail"]
+ExecPolicy = Literal["readonly", "linux_shell", "unrestricted"]
class ManagedNeCreate(BaseModel):
@@ -21,6 +23,7 @@ class ManagedNeCreate(BaseModel):
password: str = ""
tags: str = ""
remark: str = ""
+ exec_policy: ExecPolicy = "readonly"
hop_enabled: bool = False
hop_vendor: str = "zte"
hop_host: str = ""
@@ -44,6 +47,16 @@ class ManagedNeCreate(BaseModel):
return item
return "Other"
+ @field_validator("exec_policy", mode="before")
+ @classmethod
+ def normalize_exec_policy_create(cls, v: object) -> str:
+ if v is None or str(v).strip() == "":
+ return EXEC_POLICY_READONLY
+ raw = str(v).strip().lower()
+ if raw not in EXEC_POLICIES:
+ raise ValueError("unsupported_exec_policy")
+ return raw
+
class ManagedNeUpdate(BaseModel):
name: str | None = None
@@ -56,6 +69,7 @@ class ManagedNeUpdate(BaseModel):
password: str | None = None
tags: str | None = None
remark: str | None = None
+ exec_policy: ExecPolicy | None = None
hop_enabled: bool | None = None
hop_vendor: str | None = None
hop_host: str | None = None
@@ -81,6 +95,18 @@ class ManagedNeUpdate(BaseModel):
return item
return "Other"
+ @field_validator("exec_policy", mode="before")
+ @classmethod
+ def normalize_exec_policy_update(cls, v: object) -> str | None:
+ if v is None:
+ return None
+ raw = str(v).strip().lower()
+ if not raw:
+ return EXEC_POLICY_READONLY
+ if raw not in EXEC_POLICIES:
+ raise ValueError("unsupported_exec_policy")
+ return raw
+
class ManagedNeOut(BaseModel):
id: str
@@ -102,6 +128,7 @@ class ManagedNeOut(BaseModel):
# Provenance: "" | ume_sync | webcrt | lldp | …
source: str = ""
source_ref: str = ""
+ exec_policy: ExecPolicy = "readonly"
hop_enabled: bool = False
hop_vendor: str = "zte"
hop_host: str = ""
@@ -121,7 +148,7 @@ class ConnectTestRequest(BaseModel):
class ManagedNeExecRequest(BaseModel):
- """Run read-only show/display CLI on a managed NE or UME inventory NE (oclaw ops integration)."""
+ """Run CLI on a managed NE or UME inventory NE (gates follow managed NE exec_policy)."""
ne_id: str | None = None
ume_ne_id: str | None = None
diff --git a/netx_api/ne_service_common.py b/netx_api/ne_service_common.py
index 5570ab2..9e74f0b 100644
--- a/netx_api/ne_service_common.py
+++ b/netx_api/ne_service_common.py
@@ -20,6 +20,7 @@ from .ne_crypto import CredentialCryptoError, credentials_configured, decrypt_se
from .ne_schemas import ManagedNeCreate, ManagedNeOut, ManagedNeUpdate
from .ne_hop_templates import expand_bastion_hop_fields, normalize_hop_host
from .ne_session_factory import default_bastion_username_template, default_hop_command_template
+from .ne_exec_guard import normalize_exec_policy
from .timeutil import utcnow_naive
IMPORT_COLUMNS = (
@@ -241,6 +242,7 @@ def row_to_out(row: ManagedNE) -> ManagedNeOut:
remark=str(row.remark or ""),
source=str(row.source or ""),
source_ref=str(row.source_ref or ""),
+ exec_policy=normalize_exec_policy(getattr(row, "exec_policy", None)), # type: ignore[arg-type]
hop_enabled=bool(row.hop_enabled),
hop_vendor=str(row.hop_vendor or "zte"),
hop_host=str(row.hop_host or ""),
@@ -273,6 +275,7 @@ def get_device_credentials(row: ManagedNE) -> dict[str, Any]:
"password": decrypt_secret(row.password_enc),
"enable_secret": decrypt_secret(row.enable_secret_enc),
"name": str(row.name or ""),
+ "exec_policy": normalize_exec_policy(getattr(row, "exec_policy", None)),
"hop_enabled": hop_enabled,
"hop_vendor": str(row.hop_vendor or "zte"),
"hop_host": str(row.hop_host or ""),
diff --git a/netx_api/ne_service_crud.py b/netx_api/ne_service_crud.py
index 34a5915..83ab482 100644
--- a/netx_api/ne_service_crud.py
+++ b/netx_api/ne_service_crud.py
@@ -33,6 +33,11 @@ from .ne_service_common import (
_validate_hop_on_create,
row_to_out,
)
+from .ne_exec_guard import (
+ EXEC_POLICY_READONLY,
+ is_linux_device_type,
+ require_exec_policy_writable,
+)
# Inventory create stays strict; updates must also accept WebCRT/LLDP placeholder types
# (generic/linux) so operators can open the form and promote them to zte_zxros etc.
@@ -116,6 +121,10 @@ def create_managed_ne(db: Session, body: ManagedNeCreate) -> ManagedNeOut:
remark=str(body.remark or "").strip(),
source="",
source_ref="",
+ exec_policy=require_exec_policy_writable(
+ getattr(body, "exec_policy", None),
+ device_type=body.device_type,
+ ),
created_at=now,
updated_at=now,
)
@@ -159,6 +168,14 @@ def update_managed_ne(db: Session, ne_id: str, body: ManagedNeUpdate) -> Managed
row.tags = str(data["tags"]).strip()
if "remark" in data and data["remark"] is not None:
row.remark = str(data["remark"]).strip()
+ if "exec_policy" in data and data["exec_policy"] is not None:
+ row.exec_policy = require_exec_policy_writable(
+ str(data["exec_policy"]),
+ device_type=row.device_type,
+ )
+ elif "device_type" in data and not is_linux_device_type(row.device_type):
+ # Leaving linux clears any previously open policy.
+ row.exec_policy = EXEC_POLICY_READONLY
if "password" in data and data["password"]:
_require_crypto()
row.password_enc = encrypt_secret(str(data["password"]))
diff --git a/netx_api/schema_patches.py b/netx_api/schema_patches.py
index 30a0226..ab4a31d 100644
--- a/netx_api/schema_patches.py
+++ b/netx_api/schema_patches.py
@@ -200,7 +200,7 @@ def apply_collection_schema_safety_net(conn: Connection) -> None:
def apply_hop_schema_safety_net(conn: Connection) -> None:
- """Always-on hop columns (Alembic head stamp skips legacy domain patches)."""
+ """Always-on hop / exec_policy columns (Alembic head stamp skips legacy domain patches)."""
_run_sql(
conn,
"ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS hop_enter_system_view BOOLEAN DEFAULT FALSE",
@@ -209,6 +209,10 @@ def apply_hop_schema_safety_net(conn: Connection) -> None:
conn,
"ALTER TABLE cli_connect_profile ADD COLUMN IF NOT EXISTS hop_enter_system_view BOOLEAN DEFAULT FALSE",
)
+ _run_sql(
+ conn,
+ "ALTER TABLE managed_ne ADD COLUMN IF NOT EXISTS exec_policy VARCHAR(32) DEFAULT 'readonly'",
+ )
def apply_key_alert_schema_patches(
@@ -539,14 +543,48 @@ def apply_all_legacy_startup_ddl(engine: Engine) -> None:
def run_alembic_upgrade_to_head() -> None:
- """Programmatic ``alembic upgrade head`` (optional on API start)."""
+ """Programmatic ``alembic upgrade head`` (optional on API start).
+
+ Skip the full Alembic command when already at head — avoids noisy
+ ``Context impl`` logs and lock waits when nothing to apply.
+ """
+ import time
from pathlib import Path
from alembic import command
from alembic.config import Config
+ from alembic.script import ScriptDirectory
+ from sqlalchemy import create_engine, text
+ from .config import settings
+
+ t0 = time.monotonic()
root = Path(__file__).resolve().parents[1]
cfg = Config(str(root / "alembic.ini"))
- # env.py reads settings.database_url; keep ini placeholder overwritten there.
+ cfg.set_main_option("sqlalchemy.url", settings.database_url)
+
+ script = ScriptDirectory.from_config(cfg)
+ head = script.get_current_head()
+ engine = create_engine(settings.database_url, pool_pre_ping=True)
+ current: str | None = None
+ try:
+ with engine.connect() as conn:
+ try:
+ row = conn.execute(text("SELECT version_num FROM alembic_version")).fetchone()
+ current = str(row[0]) if row and row[0] is not None else None
+ except Exception:
+ current = None
+ finally:
+ engine.dispose()
+
+ if head and current == head:
+ _log.info(
+ "alembic already at head (%s), skip upgrade (%.2fs)",
+ head,
+ time.monotonic() - t0,
+ )
+ return
+
+ _log.info("alembic upgrading %s -> %s …", current, head)
command.upgrade(cfg, "head")
- _log.info("alembic upgrade head completed")
+ _log.info("alembic upgrade head completed (%.2fs)", time.monotonic() - t0)
diff --git a/packages/netx-mcp/src/netx_mcp/http_tools.py b/packages/netx-mcp/src/netx_mcp/http_tools.py
index bed447d..2bac6d2 100644
--- a/packages/netx-mcp/src/netx_mcp/http_tools.py
+++ b/packages/netx-mcp/src/netx_mcp/http_tools.py
@@ -684,8 +684,10 @@ HTTP_MCP_TOOLS: list[dict[str, Any]] = [
{
"name": "execManagedNe",
"description": (
- f"Run read-only CLI via netx (show/display/ping/traceroute; "
+ f"Run CLI via netx (default read-only: show/display/ping/traceroute; "
f"max {exec_max_commands()} commands per NE, NETX_NE_EXEC_MAX_COMMANDS). "
+ "Managed NE exec_policy=linux_shell|unrestricted allows single-line shell on that host "
+ "(check getManagedNe / listManagedNe). "
"Single NE: ne_id OR nms_ne_id (+ alias ume_ne_id) + commands. "
"Many NEs (batch-first, server concurrency default 4, max 20): "
"(1) same CLI on all → ne_ids[]/nms_ne_ids[] + shared commands; "
diff --git a/scripts/start_netx.ps1 b/scripts/start_netx.ps1
index 9da9eed..8e75164 100644
--- a/scripts/start_netx.ps1
+++ b/scripts/start_netx.ps1
@@ -3,7 +3,11 @@ param(
[int]$Port = 8890,
[int]$WebPort = 5173,
[switch]$SkipInstall = $false,
+ [Alias("Bg")]
[switch]$Background = $false,
+ # Common typo for -Background (otherwise PowerShell ignores intent / or fails).
+ [Parameter(DontShow)]
+ [switch]$Backgtound = $false,
[switch]$WithWeb = $false,
# Keep collectors inside the API process (legacy). Default: split API + worker.
[switch]$InlineSchedulers = $false
@@ -11,6 +15,11 @@ param(
$ErrorActionPreference = "Stop"
+if ($Backgtound -and -not $Background) {
+ Write-Host "[WARN] -Backgtound is a typo; treating as -Background." -ForegroundColor Yellow
+ $Background = $true
+}
+
$projectRoot = Split-Path -Parent $PSScriptRoot
Set-Location $projectRoot
@@ -234,10 +243,28 @@ if ($Background) {
Write-Host "Err = $webErrFile"
}
Write-Host ""
- Write-Host "==> Background services started; this script exits (API/worker/web keep running)." -ForegroundColor Cyan
+ Write-Host "==> Background services started; this script will exit now." -ForegroundColor Cyan
+ Write-Host " API/worker/web keep running in the background — you can close this terminal." -ForegroundColor Cyan
+ Write-Host " API log: $logFile" -ForegroundColor DarkGray
+ Write-Host " Err log: $errFile" -ForegroundColor DarkGray
+ if (-not $InlineSchedulers) {
+ Write-Host " Worker: $workerLogFile" -ForegroundColor DarkGray
+ }
+ if ($WithWeb) {
+ Write-Host " Web log: $webLogFile" -ForegroundColor DarkGray
+ Write-Host " UI: $webUrl/" -ForegroundColor Green
+ }
+ Write-Host " Stop: .\scripts\stop_netx.ps1" -ForegroundColor DarkGray
exit 0
}
+if (-not $Background) {
+ Write-Host "==> Foreground mode (no -Background): this terminal stays attached to the API." -ForegroundColor Yellow
+ Write-Host " Schedule/UME logs will keep printing here until you Ctrl+C." -ForegroundColor Yellow
+ Write-Host " For detach: .\scripts\start_netx.ps1 -Background -WithWeb" -ForegroundColor Yellow
+ Write-Host ""
+}
+
if ($WithWeb) {
Write-Host "==> Starting Vite dev server in background (foreground API mode)"
$webRoot = Join-Path $projectRoot "web"
diff --git a/tests/test_ne_exec.py b/tests/test_ne_exec.py
index ae3164d..3cfa409 100644
--- a/tests/test_ne_exec.py
+++ b/tests/test_ne_exec.py
@@ -98,6 +98,64 @@ class NeExecValidationTests(unittest.TestCase):
_validate_command("interface GigabitEthernet0/0")
self.assertEqual(ctx.exception.detail, "command_not_allowed_prefix")
+ def test_linux_shell_allows_shell_commands(self) -> None:
+ for cmd in (
+ "ls -la /var/log",
+ "ip addr | grep eth0",
+ "systemctl status sshd",
+ "cat /etc/os-release && uname -a",
+ "df -h; free -m",
+ ):
+ with self.subTest(cmd=cmd):
+ _validate_command(cmd, policy="linux_shell")
+ _validate_command(cmd, policy="unrestricted")
+
+ def test_effective_policy_forces_readonly_when_feature_off(self) -> None:
+ from netx_api.ne_exec_guard import effective_exec_policy
+
+ with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=False):
+ self.assertEqual(effective_exec_policy("linux_shell", device_type="linux"), "readonly")
+ self.assertEqual(effective_exec_policy("unrestricted", device_type="linux"), "readonly")
+
+ def test_effective_policy_forces_readonly_for_non_linux(self) -> None:
+ from netx_api.ne_exec_guard import effective_exec_policy
+
+ with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=True):
+ self.assertEqual(effective_exec_policy("linux_shell", device_type="zte_zxros"), "readonly")
+ self.assertEqual(effective_exec_policy("linux_shell", device_type="linux"), "linux_shell")
+ self.assertEqual(effective_exec_policy("unrestricted", device_type="linux_ssh"), "unrestricted")
+
+ def test_require_writable_rejects_when_feature_off(self) -> None:
+ from netx_api.ne_exec_guard import require_exec_policy_writable
+
+ with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=False):
+ self.assertEqual(require_exec_policy_writable("readonly"), "readonly")
+ with self.assertRaises(HTTPException) as ctx:
+ require_exec_policy_writable("linux_shell", device_type="linux")
+ self.assertEqual(ctx.exception.detail, "exec_policy_feature_disabled")
+
+ def test_require_writable_rejects_non_linux(self) -> None:
+ from netx_api.ne_exec_guard import require_exec_policy_writable
+
+ with patch("netx_api.ne_exec_guard.exec_policy_feature_enabled", return_value=True):
+ self.assertEqual(
+ require_exec_policy_writable("linux_shell", device_type="linux"),
+ "linux_shell",
+ )
+ with self.assertRaises(HTTPException) as ctx:
+ require_exec_policy_writable("linux_shell", device_type="cisco_ios")
+ self.assertEqual(ctx.exception.detail, "exec_policy_requires_linux_device_type")
+
+ def test_linux_shell_blocks_newline(self) -> None:
+ with self.assertRaises(HTTPException) as ctx:
+ _validate_command("ls\nrm -rf /", policy="linux_shell")
+ self.assertEqual(ctx.exception.detail, "command_chars_not_allowed")
+
+ def test_readonly_still_blocks_linux_cmds(self) -> None:
+ with self.assertRaises(HTTPException) as ctx:
+ _validate_command("ls -la", policy="readonly")
+ 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")
@@ -182,6 +240,16 @@ class NeExecRunTests(unittest.TestCase):
@patch("netx_api.ne_exec._collect_on_device", return_value="ok-output")
@patch("netx_api.ne_exec.resolve_cli_target")
def test_execute_skips_device_when_any_command_invalid(self, resolve, collect, _configured) -> None:
+ resolve.return_value = (
+ _ready_creds(),
+ {
+ "source": "managed",
+ "id": "ne-1",
+ "exec_policy": "readonly",
+ "name": "R2",
+ "ip_address": "192.168.0.128",
+ },
+ )
db = MagicMock()
with self.assertRaises(HTTPException) as ctx:
execute_managed_ne_commands(
@@ -191,7 +259,54 @@ class NeExecRunTests(unittest.TestCase):
)
self.assertEqual(ctx.exception.detail, "command_blocked")
collect.assert_not_called()
- resolve.assert_not_called()
+ resolve.assert_called_once()
+
+ @patch("netx_api.ne_exec.credentials_configured", return_value=True)
+ @patch("netx_api.ne_exec._collect_on_device", return_value="shell-ok")
+ @patch("netx_api.ne_exec.resolve_cli_target")
+ def test_execute_linux_shell_policy_allows_shell(self, resolve, collect, _configured) -> None:
+ resolve.return_value = (
+ _ready_creds(),
+ {
+ "source": "managed",
+ "id": "linux-1",
+ "exec_policy": "linux_shell",
+ "name": "lab",
+ "device_type": "linux",
+ "ip_address": "10.0.0.9",
+ },
+ )
+ db = MagicMock()
+ with patch("netx_api.config.settings") as mock_settings:
+ mock_settings.ne_exec_policy_enabled = True
+ out = execute_managed_ne_commands(db, ["ls -la /tmp"], ne_id="linux-1")
+ self.assertTrue(out["ok"])
+ self.assertEqual(out["output"], "shell-ok")
+ self.assertEqual(out["device"]["exec_policy"], "linux_shell")
+ collect.assert_called_once()
+
+ @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_ignores_db_policy_when_feature_off(self, resolve, collect, _configured) -> None:
+ resolve.return_value = (
+ _ready_creds(),
+ {
+ "source": "managed",
+ "id": "linux-1",
+ "exec_policy": "linux_shell",
+ "name": "lab",
+ "device_type": "linux",
+ "ip_address": "10.0.0.9",
+ },
+ )
+ db = MagicMock()
+ with patch("netx_api.config.settings") as mock_settings:
+ mock_settings.ne_exec_policy_enabled = False
+ with self.assertRaises(HTTPException) as ctx:
+ execute_managed_ne_commands(db, ["ls -la /tmp"], ne_id="linux-1")
+ self.assertEqual(ctx.exception.detail, "command_not_allowed_prefix")
+ collect.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")
diff --git a/web/src/pages/managedNe/ManagedNeFormDialog.tsx b/web/src/pages/managedNe/ManagedNeFormDialog.tsx
index a7d4490..bad22db 100644
--- a/web/src/pages/managedNe/ManagedNeFormDialog.tsx
+++ b/web/src/pages/managedNe/ManagedNeFormDialog.tsx
@@ -14,6 +14,7 @@ import {
buildManagedNeSaveBody,
emptyManagedNeForm,
formFromManagedNe,
+ isLinuxDeviceType,
type ManagedNeFormState,
} from "./formState";
@@ -56,6 +57,7 @@ export function ManagedNeFormDialog({
}, [open, editingId]);
const vendors = metaQuery.data?.vendors ?? [];
+ const execPolicyEnabled = Boolean(metaQuery.data?.exec_policy_enabled);
const deviceTypes = useMemo(() => {
const base = metaQuery.data?.device_types ?? [];
const cur = String(form.device_type || "").trim();
@@ -70,6 +72,7 @@ export function ManagedNeFormDialog({
hopHostRequired: t("managedNe.hop.hostRequired"),
hopUserRequired: t("managedNe.hop.userRequired"),
hopPasswordRequired: t("managedNe.hop.passwordRequired"),
+ execPolicyEnabled: Boolean(metaQuery.data?.exec_policy_enabled),
});
if (editing) {
return updateManagedNe(editing.id, body);
@@ -131,7 +134,14 @@ export function ManagedNeFormDialog({
label={t("managedNe.col.deviceType")}
required
value={form.device_type}
- onChange={(e) => setForm({ ...form, device_type: e.target.value })}
+ onChange={(e) => {
+ const device_type = e.target.value;
+ setForm((prev) => ({
+ ...prev,
+ device_type,
+ exec_policy: isLinuxDeviceType(device_type) ? prev.exec_policy : "readonly",
+ }));
+ }}
>
{deviceTypes.map((dt) => (
))}
+ {execPolicyEnabled ? (
+ <>
+
+ setForm({
+ ...form,
+ exec_policy: e.target.value as ManagedNeFormState["exec_policy"],
+ })
+ }
+ >
+
+ {isLinuxDeviceType(form.device_type) ? (
+ <>
+
+
+
+ ) : null}
+
+ {t("managedNe.execPolicy.hint")}
+
+ ) : null}
{
const body: Record = {
name: form.name,
@@ -152,6 +172,9 @@ export function buildManagedNeSaveBody(
...(form.password ? { password: form.password } : {}),
...(form.hop_password ? { hop_password: form.hop_password } : {}),
};
+ if (opts.execPolicyEnabled) {
+ body.exec_policy = isLinuxDeviceType(form.device_type) ? form.exec_policy : "readonly";
+ }
if (form.hop_enabled) {
if (!form.hop_host.trim()) throw new Error(opts.hopHostRequired);
if (!form.hop_username.trim()) throw new Error(opts.hopUserRequired);
diff --git a/web/src/services/api.ts b/web/src/services/api.ts
index 0089a09..ef451f4 100644
--- a/web/src/services/api.ts
+++ b/web/src/services/api.ts
@@ -495,6 +495,7 @@ export const fetchManagedNeMeta = () =>
device_types: types.device_types,
vendors: types.vendors,
credentials_configured: creds.configured,
+ exec_policy_enabled: Boolean(types.exec_policy_enabled),
}));
export type ManagedNeStats = {
diff --git a/web/src/types.ts b/web/src/types.ts
index aabeee5..a97027f 100644
--- a/web/src/types.ts
+++ b/web/src/types.ts
@@ -211,6 +211,8 @@ export type ManagedNeItem = {
hop_vrf: string;
hop_target_auth_mode: string;
hop_enter_system_view?: boolean;
+ /** MCP/API CLI gate: readonly | linux_shell | unrestricted */
+ exec_policy?: "readonly" | "linux_shell" | "unrestricted";
created_at: string;
updated_at: string;
};
@@ -225,6 +227,8 @@ export type ManagedNeListResponse = {
export type ManagedNeMeta = {
device_types: string[];
vendors: string[];
+ /** When false (default), exec_policy UI/API open policies are disabled. */
+ exec_policy_enabled?: boolean;
};
export type ManagedNeImportResult = {