#!/usr/bin/env python3 # -*- coding: utf-8 -*- """作业18 · 扫描 Agent 工具 OpenAPI 与 _config，生成出站域名白名单草案 用法： python hw18_01_出站域名白名单.py python hw18_01_出站域名白名单.py --repo-root D:/workbuddy/192.168.110.207 输出： data/homework18/agent_egress_whitelist_draft.md data/homework18/agent_egress_whitelist_draft.json """ from __future__ import annotations import argparse import json import os import re from datetime import datetime from pathlib import Path from typing import Any from urllib.parse import urlparse # 手工补充：工具 API 二次出站（FastAPI 后端 SSH/外呼，不在 OpenAPI servers 里） BACKEND_EGRESS = [ { "host": "116.62.10.183", "port": 22, "scheme": "ssh", "domain": "116.62.10.183", "source": "作业5/7 后端 SSH", "consumer": "8779/8780 FastAPI → GPU 机", "tools": "get_vllm_status, get_gpu_info, get_request_queue, restart_vllm_container, detect_vllm_anomaly", "risk": "L1-高", "category": "backend_side_channel", }, { "host": "116.62.10.183", "port": 8000, "scheme": "http", "domain": "116.62.10.183", "source": "作业5 后端 curl metrics", "consumer": "8779 FastAPI", "tools": "get_request_queue", "risk": "L2-中", "category": "backend_side_channel", }, { "host": "oapi.dingtalk.com", "port": 443, "scheme": "https", "domain": "oapi.dingtalk.com", "source": "作业7 notify / 作业6 周报", "consumer": "8780 FastAPI", "tools": "send_recovery_notification", "risk": "L3-外网", "category": "backend_side_channel", }, ] # Dify 平台侧（非 Agent 工具 HTTP，但 Agent 对话依赖）— 附录 PLATFORM_EGRESS = [ {"domain": "116.62.10.183", "port": 8000, "purpose": "vLLM 推理（Dify→模型）", "risk": "L2"}, {"domain": "192.168.110.207", "port": 9997, "purpose": "Xinference embedding/rerank", "risk": "L2"}, {"domain": "192.168.110.207", "port": 3000, "purpose": "Langfuse trace 上报", "risk": "L2"}, {"domain": "192.168.110.207", "port": 8088, "purpose": "Weaviate 中继检索", "risk": "L2"}, {"domain": "marketplace.dify.ai", "port": 443, "purpose": "插件市场（可选）", "risk": "L3"}, ] def out_dir() -> Path: p = Path(os.environ.get("HW18_OUT", Path(__file__).resolve().parents[2] / "data" / "homework18")) p.mkdir(parents=True, exist_ok=True) return p def parse_openapi_servers(path: Path) -> list[dict[str, Any]]: text = path.read_text(encoding="utf-8") urls = re.findall(r"^\s*-\s*url:\s*(\S+)", text, re.MULTILINE) tools = re.findall(r"operationId:\s*(\S+)", text) title_m = re.search(r"^\s*title:\s*(.+)$", text, re.MULTILINE) title = title_m.group(1).strip() if title_m else path.stem rows = [] for url in urls: u = urlparse(url.strip('"')) host = u.hostname or "" port = u.port or (443 if u.scheme == "https" else 80) rows.append( { "host": host, "port": port, "scheme": u.scheme or "http", "domain": host, "source": str(path.relative_to(path.parents[2]) if len(path.parents) > 2 else path.name), "openapi_title": title, "tools": ", ".join(tools) if tools else "—", "risk": "L2-中" if host.startswith("192.168.") or host.startswith("10.") else "L3-外网", "category": "agent_openapi_tool", } ) return rows def scan_repo(root: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for yaml in sorted(root.glob("patches_cursor/**/*openapi*.yaml")): rows.extend(parse_openapi_servers(yaml)) # 作业2 工作流 HTTP（无 openapi 文件） rows.append( { "host": "192.168.110.207", "port": 8778, "scheme": "http", "domain": "192.168.110.207", "source": "作业2 工作流 HTTP 节点", "openapi_title": "Ticket API", "tools": "POST /ticket（工作流，非 Agent OpenAPI）", "risk": "L2-中", "category": "workflow_http", } ) rows.extend(BACKEND_EGRESS) return dedupe_rows(rows) def dedupe_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: seen: set[tuple] = set() out: list[dict[str, Any]] = [] for r in rows: key = (r.get("domain"), r.get("port"), r.get("category"), r.get("tools", "")[:40]) if key in seen: continue seen.add(key) out.append(r) return out def to_markdown(rows: list[dict[str, Any]], root: Path) -> str: agent_rows = [r for r in rows if r["category"] in ("agent_openapi_tool", "workflow_http")] backend_rows = [r for r in rows if r["category"] == "backend_side_channel"] lines = [ "# Agent 工具出站域名白名单草案（作业18 · 自动生成）", "", f"- 生成时间：`{datetime.now().isoformat(timespec='seconds')}`", f"- 扫描根目录：`{root}`", "- **状态**：草案 — 新增域名须走 §审核 签字", "", "## 1. Agent OpenAPI 工具 / 工作流出站（Dify → ssrf_proxy → 目标）", "", "| 域名/IP | 端口 | 协议 | 来源 | 工具/路径 | 风险 | 白名单 |", "|---------|------|------|------|-----------|------|--------|", ] for r in agent_rows: wl = "✅ 建议纳入" if r["risk"].startswith("L2") else "⚠️ 需审批" lines.append( f"| `{r['domain']}` | {r['port']} | {r['scheme']} | {r.get('openapi_title', r['source'])} | {r['tools']} | {r['risk']} | {wl} |" ) lines.extend( [ "", "## 2. 工具 API 二次出站（207 FastAPI 后端 → 其它系统）", "", "> Agent 只访问 207:877x；**真正 SSH/钉钉/vLLM metrics 由后端发起**，白名单与审计必须覆盖。", "", "| 域名/IP | 端口 | 协议 | 消费者 | 关联工具 | 风险 | 白名单 |", "|---------|------|------|--------|----------|------|--------|", ] ) for r in backend_rows: wl = "🔒 高权" if r["risk"].startswith("L1") else ("✅ 建议纳入" if "dingtalk" not in r["domain"] else "⚠️ 外网需审批") lines.append( f"| `{r['domain']}` | {r['port']} | {r['scheme']} | {r['consumer']} | {r['tools']} | {r['risk']} | {wl} |" ) lines.extend( [ "", "## 3. 建议 SSRF / 防火墙白名单条目（草案）", "", "```text", "# 内网 Agent 工具 API（207）", "192.168.110.207:8777 # 作业1 getServerStatus", "192.168.110.207:8778 # 作业2 工单工作流", "192.168.110.207:8779 # 作业5 vLLM 健康", "192.168.110.207:8780 # 作业7 自愈", "", "# 后端二次出站", "116.62.10.183:22 # SSH（仅 8779/8780 源 IP 207）", "116.62.10.183:8000 # vLLM metrics", "oapi.dingtalk.com:443 # 钉钉通知", "```", "", "## 4. 平台依赖出站（附录，非工具 HTTP）", "", "| 域名/IP | 端口 | 用途 | 风险 |", "|---------|------|------|------|", ] ) for p in PLATFORM_EGRESS: lines.append(f"| `{p['domain']}` | {p['port']} | {p['purpose']} | {p['risk']} |") lines.extend( [ "", "## 5. 禁止 / 默认拒绝", "", "- 公网任意域名（除已审批的 `oapi.dingtalk.com`）", "- 云元数据 `169.254.169.254`", "- 内网其它网段未登记 CIDR", "- Agent Shell 未在白名单内的 egress（见专题七）", "", "审核失败策略见：`作业18_出站审核与业务策略_cursor.md`", "", ] ) return "\n".join(lines) def main() -> None: parser = argparse.ArgumentParser(description="生成 Agent 出站白名单草案") parser.add_argument("--repo-root", default=str(Path(__file__).resolve().parents[2])) args = parser.parse_args() root = Path(args.repo_root) rows = scan_repo(root) od = out_dir() md = to_markdown(rows, root) md_path = od / "agent_egress_whitelist_draft.md" json_path = od / "agent_egress_whitelist_draft.json" md_path.write_text(md, encoding="utf-8") json_path.write_text( json.dumps( {"generated_at": datetime.now().isoformat(), "entries": rows, "platform_appendix": PLATFORM_EGRESS}, ensure_ascii=False, indent=2, ), encoding="utf-8", ) print(f"Wrote {md_path}") print(f"Wrote {json_path}") print(f"Agent/工作流条目: {len([r for r in rows if r['category'] != 'backend_side_channel'])}") print(f"后端二次出站: {len([r for r in rows if r['category'] == 'backend_side_channel'])}") if __name__ == "__main__": main()