agent-gate-sec 0.2.0

AgentGate - AI Agent 数据采集与安全分析中间层服务

AgentGate

AI Agent 数据采集与安全分析中间层服务

AgentGate 是一个独立常驻服务，位于 AI Agent 与 AgentMoss 之间：接收 Agent 的工具调用数据，采集系统上下文，转换格式后提交 AgentMoss 安全分析，回传 Allow/Deny 决策。

┌──────────────┐     HTTP/Unix Socket     ┌──────────────────┐     HTTP/Unix Socket     ┌─────────────┐
│   AI Agent   │ ── AgentGateInput ──────▶│    AgentGate      │ ── AnalyzeRequest ────▶│ AgentMoss   │
│  (任意Agent)  │                          │  (独立服务)        │                          │ (安全分析)   │
│              │◀─ AgentGateResponse ────│  · 系统上下文采集   │◀─ AnalyzeResponse ────│             │
└──────────────┘                          │  · 格式标准化      │                          └─────────────┘
                                          │  · 决策缓存        │
                                          │  · Session 存储   │
                                          └──────────────────┘

---

1. 安装

git clone [e-mail]:kenhkl/AgentGate.git
cd AgentGate
pip install --break-system-packages -e ".[dev]"

2. 启动服务

AgentGate 依赖 AgentMoss 做安全分析，需要先启动 AgentMoss，再启动 AgentGate。

2.1 启动 AgentMoss

cd /home/hkl/gitcode/AgentMoss

# HTTP 模式（默认端口 9090）
python3 -m agent_moss server --mode http --port 9090

验证：

curl http://127.0.0.1:9090/api/v1/health
# → {"status": "ok", ...}

2.2 启动 AgentGate

cd /home/hkl/gitcode/AgentGate

# HTTP 模式（默认端口 9100）
agent-gate server --mode http --port 9100

# 或指定数据库路径
agent-gate server --mode http --port 9100 --db-path /tmp/agent_gate.db

# Unix Socket 模式（同机更低延迟）
agent-gate server --mode socket --socket /var/run/agent_gate/agent_gate.sock

验证：

curl http://127.0.0.1:9100/api/v1/health
# → {"status": "ok", "version": "0.1.0"}

2.3 数据存储路径

db_path 默认为 "auto"，根据运行环境自动选择：

场景	实际路径
开发（git clone 中运行）	data/agent_gate.db
用户安装（pip install）	~/.local/share/agent_gate/agent_gate.db
系统服务（systemd）	显式 --db-path /var/lib/agent_gate/agent_gate.db

---

3. 验证联调效果

3.1 端到端测试（自动启动双服务）

python3 examples/opendesk/test_e2e.py

自动启动 AgentMoss + AgentGate，执行安全命令和危险命令，验证：

• 危险命令 rm -rf / → Deny (critical)
• 危险命令 curl | bash → Deny (high)
• Session 历史自动累积
• 系统上下文完整采集

3.2 手动 curl 测试

# 安全命令
curl -s -X POST http://127.0.0.1:9100/api/v1/analyze \
  -H "Content-Type: application/json" \
  -d '{"agent_type":"test","session_id":"sess-001",
       "tool_name":"bash","command":"ls -la","cwd":"/tmp"}' | python3 -m json.tool

# 危险命令
curl -s -X POST http://127.0.0.1:9100/api/v1/analyze \
  -H "Content-Type: application/json" \
  -d '{"agent_type":"test","session_id":"sess-001",
       "tool_name":"bash","command":"rm -rf /","cwd":"/root"}' | python3 -m json.tool

---

4. 三方 Agent 如何调用 AgentGate

4.1 Python Agent（SDK）

from agent_gate.sdk import GateClient

client = GateClient(base_url="http://127.0.0.1:9100")

# 安全分析
response = client.analyze(
    agent_type="my-agent",
    session_id="sess-001",
    tool_name="bash",
    command="ls -la",
    cwd="/home/user/project",
)

if response["decision"] == "Allow":
    # 执行命令
    print("允许执行")
else:
    print(f"被阻止: {response['reason']}")

4.2 TypeScript / Electron Agent（SDK）

import { GateClient } from './gate_client';

const client = new GateClient({ baseUrl: 'http://127.0.0.1:9100' });

const result = await client.analyze({
    agentType: 'opendes',
    sessionId: 'sess-001',
    toolName: 'executeBash',
    command: 'ls -la',
    cwd: '/home/user/project',
});

if (result.decision === 'Allow') {
    // 执行命令
} else {
    console.error(`被阻止: ${result.reason}`);
}

4.3 任意 Agent（HTTP / curl）

curl -s -X POST http://127.0.0.1:9100/api/v1/analyze \
  -H "Content-Type: application/json" \
  -d '{
    "agent_type": "my-agent",
    "agent_version": "1.0",
    "session_id": "sess-abc",
    "tool_name": "bash",
    "command": "rm -rf /tmp/cache",
    "cwd": "/home/user/project",
    "description": "清理缓存",
    "user_prompt": "清理项目临时文件"
  }'

4.4 Shell 包装器（无代码接入）

# 通过环境变量配置
export AGENT_GATE_URL=http://127.0.0.1:9100
export AGENT_SESSION_ID=shell-session-001

# 用包装器执行命令
agent-gate-wrap ls -la           # Allow → 正常执行
agent-gate-wrap rm -rf /         # Deny  → 阻断并退出 126

4.5 OpenDesk 集成

参考 examples/opendesk/：

• agent_gate_plugin.ts — 插件代码（放到 OpenDesk 源码中）
• README.md — 3 步集成指南
• TESTING.md — 分层测试指南
• test_e2e.py — 端到端自动测试

核心：在 OpenDesk 的 afterParseToolcall 钩子中调用 AgentGate API，返回 reject 即可阻断工具执行。

---

5. API 接口

方法	路径	说明
POST	/api/v1/analyze	安全分析（核心接口）
GET	/api/v1/health	健康检查
GET	/api/v1/sessions/{id}/traces	查询 session 完整调用轨迹
GET	/api/v1/agents/{id}/sessions	查询某 agent 所有 session
GET	/api/v1/tool_calls	条件查询工具调用记录

请求格式

{
  "agent_type": "opendes",
  "session_id": "session-abc",
  "tool_name": "bash",
  "command": "rm -rf /tmp/cache/*",
  "cwd": "/home/user/project",
  "description": "清理缓存",
  "user_prompt": "清理项目临时文件",
  "action_history": [],
  "metadata": {}
}

action_history 可选 — AgentGate 从 SQLite 自动累积同一 session 的历史。

响应格式

{
  "decision": "Deny",
  "reason": "检测到递归强制删除操作",
  "risk_level": "high",
  "risk_type": "dangerous_command",
  "violated_layers": ["1.2"],
  "confidence": 97,
  "analysis_duration_ms": 12.5,
  "collected_context": {
    "os": { "system": "linux", "distribution": "Ubuntu 24.04", ... },
    "resources": { "cpu_count": 8, "memory_total_gb": 31.2, ... },
    "process": { "pid": 12345, "name": "opendesk", ... },
    "user": { "uid": 1000, "username": "developer", ... }
  },
  "source": "agent_moss"
}

---

6. 关键设计

特性	说明
Fail-Closed	AgentMoss 不可用时 AgentGate 返回 Deny（安全优先）
决策缓存	LRU 缓存 Allow 决策（同 session+同 cwd+同 command 可复用）
Deny 不缓存	每次 Deny 都重新分析，确保安全
Session 持久化	SQLite 存储全量 tool_call 记录，Agent 无需自维护 history
action_history 自动累积	同一 session_id 的调用序列自动拼接，传给 AgentMoss
系统上下文采集	5 模块并行（OS/进程/网络/用户/环境变量），< 50ms
环境变量安全过滤	白名单机制，自动排除 KEY/TOKEN/SECRET 等敏感变量

---

7. 测试

# 单元测试（42 项）
python3 -m pytest tests/test_server.py -v

# 端到端测试（AgentMoss + AgentGate 联调）
python3 examples/opendesk/test_e2e.py

---

8. 高权限需求

采集模块	所需权限	读取路径
system_info	普通用户	/proc/version, /etc/os-release
process_info	CAP_SYS_PTRACE	/proc/*/status, capabilities
env_info	普通用户	自身进程环境变量
network_info	CAP_NET_ADMIN	/proc/net/*
user_info	普通用户	/etc/passwd, /etc/group

推荐以专用 agent-gate 用户 + Linux capabilities 运行，不使用 root。

---

9. Dashboard 可视化面板

AgentGate 内置实时 Dashboard（http://localhost:9100/dashboard/），提供：

• 实时统计卡片（Agents / Sessions / Tool Calls / Allow / Deny / Avg Latency）
• Risk Distribution 风险分布条
• Recent Tool Calls 操作记录表格
• Agent 类型分布
• Timeline 折线图（过去 60 分钟 Allow/Deny 趋势）

在 OpenDesk 中查看

AgentGate 已集成到 OpenDesk 桌面端作为侧边栏应用，点击图标即可在应用内查看 Dashboard。

---

10. 项目结构

AgentGate/
├── agent_gate/                    # 主包
│   ├── cli.py                     # 命令行入口 (agent-gate server / clean)
│   ├── config.py                  # YAML 配置 + XDG 路径自动适配
│   ├── normalizer.py              # AgentGateInput → AgentMoss AnalyzeRequest
│   ├── sdk.py                     # Python SDK
│   ├── server/                    # FastAPI 服务
│   │   ├── app.py                 # HTTP / Unix Socket 双模式
│   │   ├── routes.py              # /analyze /traces /tool_calls /health
│   │   ├── models.py              # Pydantic 数据模型
│   │   └── middleware.py          # 请求日志
│   ├── collector/                 # 系统数据采集（5 模块并行）
│   │   ├── system_info.py         # OS/内核/CPU/内存/磁盘
│   │   ├── process_info.py        # PID/capabilities/cgroup
│   │   ├── env_info.py            # 环境变量白名单过滤
│   │   ├── network_info.py        # 网络接口/连接数
│   │   └── user_info.py           # UID/GID/用户组
│   ├── client/                    # AgentMoss 客户端
│   │   ├── moss_client.py         # 异步 HTTP + Fail-Closed
│   │   └── cache.py               # LRU 决策缓存
│   └── storage/                   # SQLite 持久化
│       ├── db.py                  # 建表/迁移/WAL 模式
│       └── repository.py          # CRUD (agents/sessions/tool_calls)
├── sdk/                           # 客户端 SDK
│   ├── python/gate_client.py      # Python SDK
│   └── typescript/                # TypeScript SDK (package.json + tsconfig)
├── examples/opendesk/             # OpenDesk 集成
│   ├── agent_gate_plugin.ts       # 插件（afterParseToolcall 钩子）
│   ├── test_integration.py        # 接口测试（不改 OpenDesk）
│   ├── test_e2e.py                # 端到端测试（自动启双服务）
│   ├── README.md                  # 集成指南
│   └── TESTING.md                 # 分层测试指南
├── config/agent_gate.yaml         # YAML 配置模板
├── scripts/agent-gate-wrap.sh     # Shell 包装器
├── tests/test_server.py           # 单元 + 集成测试（42 项）
└── docs/agent_collector_design_v1.md  # 设计文档

许可证

MIT