diff --git a/_local/system.env.example b/_local/system.env.example index bb32695d..93f488ef 100644 --- a/_local/system.env.example +++ b/_local/system.env.example @@ -433,10 +433,11 @@ TAVILY_API_KEY= # 【前端】多数上述项在 Admin 设置表单中有对应勾选或输入框。 # ----------------------------------------------------------------------------- -# 二十二、netx（network_ops 内部工具：UME 告警 / 网元清单 / 诊断） +# 二十二、netx（network_ops 内部工具：UME 告警 / 纳管网元 CLI / 诊断） # ----------------------------------------------------------------------------- # OCLAW_NETX_BASE_URL netx HTTP 根地址（无尾部斜杠）；runtime/tools/experts/network_ops/netx_tools.py -# 调用 /v1/alarms、/v1/alarms/aggregate、/v1/diagnostics。 +# UME：/v1/ume/alarms、/v1/ume/inventory/ne 等；纳管网元：/v1/managed-ne、/v1/managed-ne/exec。 +# netx 侧需配置 NETX_CREDENTIAL_SECRET_KEY（设备密码 Fernet 加密）。 # 默认 http://127.0.0.1:8890（代码内兜底；与本文件一致时可省略）。 # OCLAW_NETX_API_TOKEN 可选；netx 若启用 Bearer 鉴权，填与 netx 侧一致的 token。 # OCLAW_OPS_NETX_CONTEXT_INJECT 默认 1；ops 专家每轮把 netx 最新导入 batch_id 注入 system（类似附件锚点）。 diff --git a/runtime/tools/experts/network_ops/netx_tools.py b/runtime/tools/experts/network_ops/netx_tools.py index 68a4a269..a0df3873 100644 --- a/runtime/tools/experts/network_ops/netx_tools.py +++ b/runtime/tools/experts/network_ops/netx_tools.py @@ -98,6 +98,23 @@ def _localize_netx_payload(data: dict[str, Any], *, lang: str) -> dict[str, Any] return data +def _http_post_json(path: str, body: dict[str, Any], *, timeout: float = 180.0) -> dict[str, Any]: + base = _netx_base_url() + url = f"{base}{path}" + try: + with httpx.Client(timeout=timeout, trust_env=False) as client: + resp = client.post(url, json=body, headers=_netx_headers()) + text = resp.text + if not resp.is_success: + return {"ok": False, "error": f"netx_http_{resp.status_code}", "detail": text[:800]} + data = resp.json() if text else {} + if isinstance(data, dict): + data = _localize_netx_payload(data, lang=str(NETX_TOOL_LANG.get() or "zh")) + return {"ok": True, "data": data if isinstance(data, dict) else {"raw": data}} + except Exception as exc: + return {"ok": False, "error": "netx_request_failed", "detail": str(exc)[:800]} + + def _http_json(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: base = _netx_base_url() url = f"{base}{path}" @@ -970,6 +987,146 @@ def netx_aggregate_ume_alarms_raw_tool() -> ToolSpec: ) +def netx_list_managed_ne_tool() -> ToolSpec: + """List netx managed NEs (inventory for CLI login targets).""" + + def handler(args: dict[str, Any]) -> dict[str, Any]: + page = max(1, int(args.get("page") or 1)) + page_size = min(500, max(1, int(args.get("page_size") or 50))) + params: dict[str, Any] = {"page": page, "page_size": page_size} + if str(args.get("keyword") or "").strip(): + params["keyword"] = str(args.get("keyword")).strip() + if str(args.get("vendor") or "").strip(): + params["vendor"] = str(args.get("vendor")).strip() + if str(args.get("connect_status") or "").strip(): + params["connect_status"] = str(args.get("connect_status")).strip() + return _http_json("GET", "/v1/managed-ne", params=params) + + return ToolSpec( + name="netx_list_managed_ne", + description=( + "列出 netx「网元管理」中已纳管的设备（GET /v1/managed-ne）。" + "返回 id、name、ip、vendor、device_type、connect_status 等（不含密码）。" + "keyword 可匹配名称/IP/用户名/标签；connect_status 可选 unknown/testing/pass/fail。" + "登录查配置前先用本工具定位 ne_id，再 netx_get_managed_ne / netx_exec_managed_ne。" + ), + parameters={ + "type": "object", + "properties": { + "keyword": {"type": "string", "description": "名称/IP/用户名/标签包含（可选）"}, + "vendor": {"type": "string", "description": "厂商过滤（可选）"}, + "connect_status": { + "type": "string", + "enum": ["unknown", "testing", "pass", "fail"], + "description": "连通性状态过滤（可选）", + }, + "page": {"type": "integer", "minimum": 1, "default": 1}, + "page_size": {"type": "integer", "minimum": 1, "maximum": 500, "default": 50}, + }, + "required": [], + "additionalProperties": False, + }, + handler=handler, + tags=frozenset({"netx", "ops", "managed_ne", "inventory", "read_only"}), + risk_level="low", + read_only=True, + ) + + +def netx_get_managed_ne_tool() -> ToolSpec: + """Single managed NE metadata from netx.""" + + def handler(args: dict[str, Any]) -> dict[str, Any]: + ne_id = str(args.get("ne_id") or "").strip() + if not ne_id: + return {"ok": False, "error": "ne_id_required", "error_code": "ne_id_required"} + return _http_json("GET", f"/v1/managed-ne/{ne_id}", params=None) + + return ToolSpec( + name="netx_get_managed_ne", + description=( + "读取 netx 单条纳管网元详情（GET /v1/managed-ne/{ne_id}）。" + "含 connect_status、connect_message、connect_detail（连通测试日志）、跳板 hop_* 配置摘要。" + ), + parameters={ + "type": "object", + "properties": { + "ne_id": {"type": "string", "description": "网元 UUID（列表 items[].id）"}, + }, + "required": ["ne_id"], + "additionalProperties": False, + }, + handler=handler, + tags=frozenset({"netx", "ops", "managed_ne", "read_only"}), + risk_level="low", + read_only=True, + ) + + +def netx_exec_managed_ne_tool() -> ToolSpec: + """Run read-only CLI on a managed NE via netx.""" + + def handler(args: dict[str, Any]) -> dict[str, Any]: + ne_id = str(args.get("ne_id") or "").strip() + if not ne_id: + return {"ok": False, "error": "ne_id_required", "error_code": "ne_id_required"} + raw_cmds = args.get("commands") + if not isinstance(raw_cmds, list) or not raw_cmds: + return {"ok": False, "error": "commands_required", "error_code": "commands_required"} + commands = [str(c).strip() for c in raw_cmds if str(c).strip()] + if not commands: + return {"ok": False, "error": "commands_required", "error_code": "commands_required"} + if len(commands) > 5: + return {"ok": False, "error": "too_many_commands", "error_code": "too_many_commands"} + body: dict[str, Any] = {"ne_id": ne_id, "commands": commands} + rts = args.get("read_timeout_sec") + if rts is not None: + body["read_timeout_sec"] = int(rts) + out = _http_post_json("/v1/managed-ne/exec", body, timeout=300.0) + if not out.get("ok"): + return out + data = out.get("data") or {} + if isinstance(data, dict) and data.get("ok") is False: + return {"ok": False, "data": data, "error": str(data.get("error") or "exec_failed")} + return {"ok": True, "data": data} + + return ToolSpec( + name="netx_exec_managed_ne", + description=( + "经 netx 登录「网元管理」中的设备并执行只读 CLI（POST /v1/managed-ne/exec）。" + "每条命令须以 show / display / get / ping / traceroute / terminal length / ? 开头；" + "禁止管道符、分号及改配置类命令；单次最多 5 条；默认读超时 60s。" + "返回合并输出（含命令回显）；失败时含 error/detail。" + "先 netx_list_managed_ne 解析 ne_id；若 connect_status 非 pass 可先 netx_get_managed_ne 看 connect_detail。" + ), + parameters={ + "type": "object", + "properties": { + "ne_id": {"type": "string", "description": "纳管网元 UUID"}, + "commands": { + "type": "array", + "items": {"type": "string"}, + "minItems": 1, + "maxItems": 5, + "description": "只读 CLI 列表，如 show version、display interface brief", + }, + "read_timeout_sec": { + "type": "integer", + "minimum": 10, + "maximum": 120, + "description": "单条命令 Netmiko 读超时（秒），默认 60", + }, + }, + "required": ["ne_id", "commands"], + "additionalProperties": False, + }, + handler=handler, + tags=frozenset({"netx", "ops", "managed_ne", "cli", "exec"}), + risk_level="medium", + read_only=False, + ) + + __all__ = [ # "netx_query_alarms_tool", # "netx_aggregate_alarms_tool", @@ -987,4 +1144,7 @@ __all__ = [ "netx_aggregate_ume_alarms_raw_tool", "netx_list_ume_alarm_fields_tool", "netx_sql_query_ume_tool", + "netx_list_managed_ne_tool", + "netx_get_managed_ne_tool", + "netx_exec_managed_ne_tool", ] diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.en.md b/runtime/workspaces/ops/ROLE_SYSTEM.en.md index b67b3ed1..f56a8f89 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.en.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.en.md @@ -31,8 +31,9 @@ You are the ops specialist (network operations expert). - If `host_name` is empty, fall back to `user_label` / `ne_name` with a "host_name missing" note — never bare `ne_id`. - NE stats/aggregates: prefer `group_by=alarm_host_name` or `group_by=ne_host_name`; do not group by `alarm_ne_id` / `ne_ne_id` for user output. -## Required skill +## Required skills - For every netx/UME **alarm or NE** request, load and follow skill: `ops-netx-ume-playbook` (skill text may be Chinese; **user-facing output must still match the user's language**). +- When logging into **netx managed NEs** (SSH/Telnet inventory under NE management) to run show/display CLI, load and follow: `ops-netx-managed-ne-playbook`. ## netx detail and statistics (internal tools) @@ -44,4 +45,9 @@ Each turn may append a **UME alarm runtime anchor** at the end of system context - `netx_query_ume_ne_inventory`: synced NE list (`keyword`). - `netx_get_ume_ne`: single NE by `ne_id` (includes `raw_json`). +## netx managed NE (device CLI) + +- `netx_list_managed_ne` / `netx_get_managed_ne`: managed inventory and connect-test detail. +- `netx_exec_managed_ne`: read-only CLI via netx login (show/display/ping; no config changes). + Uses `OCLAW_NETX_BASE_URL` / `OCLAW_NETX_API_TOKEN`. Disable anchor inject: `OCLAW_OPS_NETX_CONTEXT_INJECT=0`. diff --git a/runtime/workspaces/ops/ROLE_SYSTEM.md b/runtime/workspaces/ops/ROLE_SYSTEM.md index 5c681e35..b47f6f97 100644 --- a/runtime/workspaces/ops/ROLE_SYSTEM.md +++ b/runtime/workspaces/ops/ROLE_SYSTEM.md @@ -26,6 +26,7 @@ ## 必须加载技能 - 每次处理 netx/UME **告警或网元** 问题时，必须加载并遵循技能：`ops-netx-ume-playbook`。 +- 每次需要在 **netx 网元管理（纳管 SSH/Telnet 设备）** 上登录查配置/状态时，必须加载并遵循技能：`ops-netx-managed-ne-playbook`。 ## netx 明细与统计（内部工具） @@ -37,4 +38,9 @@ - `netx_query_ume_ne_inventory`：分页查询已同步的 UME 网元清单（可选 `keyword`）。 - `netx_get_ume_ne`：按 `ne_id`（UUID）取单网元详情（含 `raw_json`）。 +## netx 纳管网元（登录设备查 CLI） + +- `netx_list_managed_ne` / `netx_get_managed_ne`：网元管理清单与连通详情。 +- `netx_exec_managed_ne`：经 netx 登录设备执行只读 CLI（show/display/ping；禁止改配置）。 + 工具走 netx（`OCLAW_NETX_BASE_URL` / `OCLAW_NETX_API_TOKEN`）。关闭自动锚点：环境变量 `OCLAW_OPS_NETX_CONTEXT_INJECT=0`。 diff --git a/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md new file mode 100644 index 00000000..ec55d3df --- /dev/null +++ b/skills/_workspace/ops/ops-netx-managed-ne-playbook/SKILL.md @@ -0,0 +1,41 @@ +--- +name: ops-netx-managed-ne-playbook +description: 面向 ops 专家的 netx 纳管网元（网元管理）作业手册。覆盖设备清单、连通状态、经 netx 登录设备执行只读 CLI。 +--- + +# Ops Netx 纳管网元作业手册 + +## 强制使用范围 + +凡是需要在 **netx 网元管理** 中已录入的设备上**登录并查询**（show/display/ping 等）时，必须加载并遵循本技能。 + +与 UME 网元清单（`ops-netx-ume-playbook`）不同：本手册针对 **SSH/Telnet 纳管设备**（含 ZTE/华为/思科跳板），不是 UME REST 同步清单。 + +## 工具选择顺序 + +1. **定位设备** + - `netx_list_managed_ne`：`keyword`（名称/IP）、`connect_status=pass` 过滤 + - `netx_get_managed_ne`：单条详情、`connect_detail`（连通失败排障） +2. **登录查信息** + - `netx_exec_managed_ne`：`ne_id` + `commands`（最多 5 条只读命令） + +## CLI 约束（服务端强制） + +- 允许前缀：`show `、`display `、`get `、`ping `、`traceroute `、`terminal length `、`?` +- 禁止：`|`、`;`、换行拼接、改配置类（configure/write/copy/reload/delete 等） +- 示例： + - 思科：`show version`、`show configuration | include hostname` **不可**（含 `|`）→ 改用 `show configuration` 或连通测试已解析的 hostname + - 华为：`display version`、`display interface brief` + - ZTE：`show version`、`show interface` + +## 排障流程 + +1. `connect_status` 为 `fail`：先 `netx_get_managed_ne` 阅读 `connect_detail`，勿反复盲 exec +2. 设备经跳板：详情中确认 `hop_enabled`、`hop_vendor`、模板是否正确 +3. 超时：对慢命令提高 `read_timeout_sec`（最大 120），或减少单次命令条数 + +## 输出约定 + +- 结论 + **工具返回 output 摘录**（勿编造 CLI 结果） +- 标明 `ip_address`、`name`、`ne_id`（对用户展示优先 name/IP，ne_id 作关联键） +- 英文会话：用户可见回复不得含汉字（CLI 原文可摘录但需说明为设备原文） diff --git a/tests/test_netx_managed_ne_tools.py b/tests/test_netx_managed_ne_tools.py new file mode 100644 index 00000000..cb09ed57 --- /dev/null +++ b/tests/test_netx_managed_ne_tools.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from typing import Any + +import pytest + + +def test_netx_list_managed_ne_forwards_params(monkeypatch: pytest.MonkeyPatch) -> None: + import runtime.tools.experts.network_ops.netx_tools as nt + + calls: list[tuple[str, str, dict[str, Any] | None]] = [] + + def fake(method: str, path: str, *, params: dict[str, Any] | None = None) -> dict[str, Any]: + calls.append((method, path, params)) + return {"ok": True, "data": {"total": 0, "items": []}} + + monkeypatch.setattr(nt, "_http_json", fake) + spec = nt.netx_list_managed_ne_tool() + out = spec.handler({"keyword": "192.168", "connect_status": "pass", "page": 1, "page_size": 20}) + assert out.get("ok") is True + assert calls[0] == ("GET", "/v1/managed-ne", {"page": 1, "page_size": 20, "keyword": "192.168", "connect_status": "pass"}) + + +def test_netx_exec_managed_ne_posts_body(monkeypatch: pytest.MonkeyPatch) -> None: + import runtime.tools.experts.network_ops.netx_tools as nt + + bodies: list[dict[str, Any]] = [] + + def fake_post(path: str, body: dict[str, Any], *, timeout: float = 180.0) -> dict[str, Any]: + bodies.append(body) + return {"ok": True, "data": {"ok": True, "output": "R2#show version\n..."}} + + monkeypatch.setattr(nt, "_http_post_json", fake_post) + spec = nt.netx_exec_managed_ne_tool() + out = spec.handler({"ne_id": "abc", "commands": ["show version"], "read_timeout_sec": 90}) + assert out.get("ok") is True + assert bodies[0]["ne_id"] == "abc" + assert bodies[0]["commands"] == ["show version"] + assert bodies[0]["read_timeout_sec"] == 90 + + +def test_netx_exec_requires_commands(monkeypatch: pytest.MonkeyPatch) -> None: + import runtime.tools.experts.network_ops.netx_tools as nt + + monkeypatch.setattr(nt, "_http_post_json", lambda *a, **k: {"ok": True, "data": {}}) + spec = nt.netx_exec_managed_ne_tool() + out = spec.handler({"ne_id": "abc"}) + assert out.get("ok") is False + assert out.get("error_code") == "commands_required"