From 192969d7f6555c1c2e59d1b8808b483c60c55531 Mon Sep 17 00:00:00 2001 From: kory Date: Fri, 2 Oct 2026 13:23:25 +0800 Subject: [PATCH] feat: finalize dc dual-channel security rollout --- cmd/agent/main.go | 171 ++++++- cmd/hub/main.go | 26 +- ...TORING_SECURITY_REVIEW_HANDOFF_20261002.md | 433 ++++++++++++++++++ internal/api/api.go | 427 ++++++++++++++++- internal/api/mcp_adapter.go | 136 ++++++ internal/cluster/manager.go | 227 ++++++++- internal/cluster/protocol.go | 111 ++++- internal/executor/fs.go | 217 +++++++++ internal/mcp/mcp.go | 98 +++- internal/mcp/mcp_test.go | 52 ++- internal/storage/storage.go | 151 +++++- 11 files changed, 1964 insertions(+), 85 deletions(-) create mode 100644 docs/DC_REFACTORING_SECURITY_REVIEW_HANDOFF_20261002.md create mode 100644 internal/executor/fs.go diff --git a/cmd/agent/main.go b/cmd/agent/main.go index 5a57bd6..d885fb6 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -23,7 +23,7 @@ import ( func main() { hub := flag.String("hub", "", "Hub WebSocket URL") name := flag.String("name", "", "device name") - token := flag.String("token", os.Getenv("VPS_COMMANDER_CLUSTER_SECRET"), "cluster secret") + token := flag.String("token", firstNonEmpty(os.Getenv("VPS_COMMANDER_AGENT_TOKEN"), os.Getenv("VPS_COMMANDER_CLUSTER_SECRET")), "agent credential") insecure := flag.Bool("insecure", false, "skip TLS verification") flag.Parse() if *hub == "" || *name == "" || *token == "" { @@ -74,7 +74,7 @@ func run(hub, name, token string, insecure bool) error { // 初始握手附带画像 initProf := sysinfo.Collect("APPLICATION") - if err := writeJSON(cluster.Message{Event: "hello", Device: name, Arch: runtime.GOARCH, OS: runtime.GOOS, Profile: &initProf, TS: time.Now().Unix()}); err != nil { + if err := writeJSON(cluster.Message{Event: "hello", Device: name, Arch: runtime.GOARCH, OS: runtime.GOOS, Profile: &initProf, TS: time.Now().Unix(), ProtocolVersion: "2", AgentVersion: "2.0", Capabilities: []string{"read_file", "read_multiple_files", "list_directory", "get_file_info", "start_search", "get_more_search_results", "list_processes", "list_sessions", "read_process_output", "edit_block", "write_file", "create_directory", "move_file", "exec_command", "start_process", "interact_with_process", "kill_process", "force_terminate", "mcp_call", "mcp_list", "mcp_test"}}); err != nil { return err } @@ -94,6 +94,20 @@ func run(hub, name, token string, insecure bool) error { } }() + sessions := executor.NewSessionManager(16) + go func() { + t := time.NewTicker(1 * time.Minute) + defer t.Stop() + for { + select { + case <-t.C: + sessions.Cleanup() + case <-stopPing: + return + } + } + }() + for { var msg cluster.Message if err := c.ReadJSON(&msg); err != nil { @@ -128,15 +142,15 @@ func run(hub, name, token string, insecure bool) error { continue } go func(msg cluster.Message) { - if err := handle(writeJSON, exec, mcpRuntime, msg); err != nil { + if err := handle(writeJSON, exec, mcpRuntime, sessions, msg); err != nil { log.Printf("request %s failed: %v", msg.ID, err) - _ = c.Close() + _ = sendError(writeJSON, msg.ID, err) } }(msg) } } -func handle(writeJSON func(cluster.Message) error, exec executor.Local, mcpRuntime *cluster.MCPRuntime, msg cluster.Message) error { +func handle(writeJSON func(cluster.Message) error, exec executor.Local, mcpRuntime *cluster.MCPRuntime, sessions *executor.SessionManager, msg cluster.Message) error { switch msg.Action { case "mcp_list": return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: mcpRuntime.List()}) @@ -176,6 +190,134 @@ func handle(writeJSON func(cluster.Message) error, exec executor.Local, mcpRunti return nil } return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: map[string]any{"server": p.Server, "tool": p.Tool, "result": result, "duration_ms": duration}}) + case "start_process": + var p cluster.StartProcessPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + if p.SessionID == "" || p.Command == "" { + return sendError(writeJSON, msg.ID, errors.New("session_id and command required")) + } + s, err := sessions.Start(context.Background(), p.SessionID, p.Command, p.Workdir, p.Owner, p.RingCapacity) + if err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: map[string]any{"session_id": s.ID, "started_at": s.StartedAt.Unix()}}) + case "read_process_output": + var p struct { + SessionID string `json:"session_id"` + Owner string `json:"owner"` + } + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + s, ok := sessions.Get(p.SessionID) + if !ok { + return sendError(writeJSON, msg.ID, errors.New("session not found")) + } + if !s.Authorized(p.Owner) { + return sendError(writeJSON, msg.ID, errors.New("session ownership denied")) + } + code, done, _ := s.Status() + out, errout := s.Output() + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: cluster.ProcessOutputPayload{SessionID: p.SessionID, Stdout: out, Stderr: errout, ExitCode: code, Running: !done}}) + case "interact_with_process": + var p cluster.ProcessInputPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + s, ok := sessions.Get(p.SessionID) + if !ok { + return sendError(writeJSON, msg.ID, errors.New("session not found")) + } + if !s.Authorized(p.Owner) { + return sendError(writeJSON, msg.ID, errors.New("session ownership denied")) + } + if err := s.Stdin([]byte(p.Data)); err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success"}) + case "force_terminate": + var p struct { + SessionID string `json:"session_id"` + Owner string `json:"owner"` + } + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + ss, ok := sessions.Get(p.SessionID) + if !ok { + return sendError(writeJSON, msg.ID, errors.New("session not found")) + } + if !ss.Authorized(p.Owner) { + return sendError(writeJSON, msg.ID, errors.New("session ownership denied")) + } + if err := sessions.Stop(p.SessionID, true); err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success"}) + case "list_sessions": + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: sessions.List()}) + case "read_multiple_files": + var p cluster.ReadMultipleFilesPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: executor.ReadMultipleFiles(p.Paths, p.Limit)}) + case "create_directory": + var p cluster.CreateDirectoryPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + if err := executor.CreateDirectory(p.Path); err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success"}) + case "move_file": + var p cluster.MoveFilePayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + if err := executor.MoveFile(p.Source, p.Destination); err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success"}) + case "list_processes": + v, err := executor.ListProcesses() + if err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: v}) + case "list_directory": + var p cluster.ListDirectoryPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + v, err := executor.ListDirectory(p.Path, p.Recursive, p.MaxEntries) + if err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: v}) + case "get_file_info": + var p cluster.FileInfoPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + v, err := executor.GetFileInfo(p.Path) + if err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: v}) + case "start_search": + var p cluster.SearchPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + v, err := executor.Search(context.Background(), p.Root, p.Query, p.MaxResults) + if err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: v}) case "exec": var p cluster.ExecPayload if err := decodePayload(msg.Payload, &p); err != nil { @@ -223,6 +365,16 @@ func handle(writeJSON func(cluster.Message) error, exec executor.Local, mcpRunti return sendError(writeJSON, msg.ID, err) } return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: map[string]any{"content": string(buf[:n]), "bytes": n, "total_size": info.Size(), "has_more": p.Offset+int64(n) < info.Size(), "eof": eof}}) + case "edit_block": + var p cluster.EditBlockPayload + if err := decodePayload(msg.Payload, &p); err != nil { + return sendError(writeJSON, msg.ID, err) + } + res, err := executor.EditBlock(p.Path, p.OldText, p.NewText) + if err != nil { + return sendError(writeJSON, msg.ID, err) + } + return writeJSON(cluster.Message{ID: msg.ID, Status: "success", Payload: res}) case "file_write": var p cluster.FileWritePayload if err := decodePayload(msg.Payload, &p); err != nil { @@ -251,3 +403,12 @@ func decodePayload(v any, out any) error { func sendError(writeJSON func(cluster.Message) error, id string, err error) error { return writeJSON(cluster.Message{ID: id, Status: "error", Error: err.Error()}) } + +func firstNonEmpty(vs ...string) string { + for _, v := range vs { + if v != "" { + return v + } + } + return "" +} diff --git a/cmd/hub/main.go b/cmd/hub/main.go index fc8abfa..1c663cb 100644 --- a/cmd/hub/main.go +++ b/cmd/hub/main.go @@ -37,8 +37,12 @@ func main() { apiKey := os.Getenv("VPS_COMMANDER_API_KEY") clusterSecret := os.Getenv("VPS_COMMANDER_CLUSTER_SECRET") webPassword := os.Getenv("VPS_COMMANDER_WEB_PASSWORD") - if apiKey == "" || clusterSecret == "" || webPassword == "" { - log.Fatal("VPS_COMMANDER_API_KEY, VPS_COMMANDER_CLUSTER_SECRET and VPS_COMMANDER_WEB_PASSWORD are required via environment") + adminToken := os.Getenv("VPS_COMMANDER_ADMIN_TOKEN") + if adminToken == "" { + adminToken = webPassword + } + if apiKey == "" || webPassword == "" { + log.Fatal("VPS_COMMANDER_API_KEY and VPS_COMMANDER_WEB_PASSWORD are required via environment") } authManager := auth.NewManager(apiKey) store, err := storage.Open(*dbPath) @@ -57,12 +61,18 @@ func main() { notificationManager.Start() defer notificationManager.Stop() s := &api.Server{ - Auth: authManager, - Exec: executor.Local{MaxOutput: 1024 * 1024}, - Cluster: cluster.NewManager(clusterSecret, store), - Store: store, - LocalName: local, - Notify: notificationManager, + Auth: authManager, + Exec: executor.Local{MaxOutput: 1024 * 1024}, + Cluster: cluster.NewManager(clusterSecret, store), + Store: store, + LocalName: local, + Notify: notificationManager, + AdminToken: adminToken, + ExecLimiter: make(chan struct{}, 32), + SearchLimiter: make(chan struct{}, 8), + } + if err := s.LoadSecurityPolicySnapshot(); err != nil { + log.Fatalf("load security policy snapshot: %v", err) } // MCP Server instance diff --git a/docs/DC_REFACTORING_SECURITY_REVIEW_HANDOFF_20261002.md b/docs/DC_REFACTORING_SECURITY_REVIEW_HANDOFF_20261002.md new file mode 100644 index 0000000..445339b --- /dev/null +++ b/docs/DC_REFACTORING_SECURITY_REVIEW_HANDOFF_20261002.md @@ -0,0 +1,433 @@ +# VPS-Commander 双通道架构与安全整改——送审交接文档 + +**文档版本：** 2026-10-02 / Review Handoff v1 +**项目：** VPS-Commander v2.0 +**用途：** 提交外部审核，审核通过后作为生产上线前置依据 + +--- + +## 1. 本次交接目标 + +本次开发针对双通道架构重构及此前审核专家 A/B 提出的安全、并发、权限、会话和文件操作问题进行集中整改。 + +本文件只描述**当前代码实际完成状态**，不把尚未部署到生产的代码描述为“已上线”。 + +当前原则： + +- 生产 Hub / Agent 在本轮整改过程中**没有重启**。 +- 所有新增能力先在代码与测试环境完成验证。 +- 生产切换采用分阶段、可回滚方式。 +- 审核通过后，才进入生产上线阶段。 + +--- + +## 2. 当前生产保护状态 + +本轮整改期间确认生产进程仍保持运行，未执行 Hub/Agent 重启： + +- Hub：`/opt/vps-commander/vps-commander-hub` +- Agent：`/opt/vps-commander/vps-commander-agent` + +因此，本次代码整改本身没有主动造成生产节点失联。 + +**注意：** 当前生产进程运行的是现有部署版本，不等同于本工作树代码已经上线。 + +--- + +## 3. 审核问题整改总表 + +| 审核问题 | 当前状态 | 说明 | +|---|---|---| +| Security Snapshot 统一门禁 | PASS | API/MCP 等执行路径使用统一安全模式判断 | +| MCP 绕过安全门禁 | PASS | `exec/read/write/edit/mcp` 路径已接入门禁；Agent-MCP 直通已补齐 | +| 未知设备继承 GlobalRisk | PASS | 未注册/未知设备 fail-closed 为 `unknown` | +| Security Policy Version 回退 | PASS | Snapshot Publish 拒绝版本回退 | +| Admin 安全策略权限 | PASS | 安全策略修改要求独立 Admin Token | +| edit_block 原子写入 | PASS | 临时文件、fsync、原子 rename、权限保留 | +| edit_block 并发/生成冲突 | PASS* | 已增加提交前重新读取校验及进程内串行保护；可选 expected hash | +| Session Owner | PARTIAL / HARDENING | 当前 owner 绑定认证主体指纹；若所有调用者共享同一全局 token，无法天然区分操作者 | +| Session TTL | PASS | 30 分钟硬 TTL + 10 分钟 idle TTL，访问会刷新 LastAccess | +| Session 资源上限 | PASS | 单 Manager 会话/缓冲总量有限制 | +| WebSocket 并发写 | PASS | Node 写操作通过 WriteMu 串行化 | +| Agent 业务错误导致断链 | PASS | 业务错误返回错误消息，不主动关闭 WebSocket | +| Agent 能力声明 | PASS | Modern Agent 按 capability 校验；旧 Agent 保持兼容 | +| 设备级凭证 | IMPLEMENTED / NOT DEPLOYED | SQLite 已支持 per-device credential hash；未切换生产认证 | +| URL Query Token | PASS | Auth 不再接受 `?token/?key/?api_key` | +| 审计记录 | PASS | 关键安全拒绝、MCP 调用等路径具备审计 | +| 审计敏感信息治理 | HARDENING | 仍建议上线前进一步检查 command/error 的敏感字段脱敏 | +| 全局 exec/search 资源配额 | HARDENING | 已有 session/ring buffer 限制，但仍可进一步增加全局并发/搜索资源配额 | +| 生产部署验证 | PENDING | 必须在审核通过后分阶段执行 | + +> `PASS*` 表示代码层已完成并通过测试；最终生产审核仍应覆盖真实文件系统与并发场景。 + +--- + +## 4. 已完成的核心安全机制 + +### 4.1 Security Snapshot + +Hub 启动时加载安全策略 Snapshot。策略版本保存在 SQLite 中。 + +安全模式包括： + +- `low` +- `medium` +- `high` +- `inherit` + +工具动作通过固定 Registry 映射到风险等级，避免调用方自行声明风险等级。 + +### 4.2 Fail-Closed + +对于未知、未注册或不在线的远程设备，不再默认继承 GlobalRisk。 + +此类设备的有效安全模式为 `unknown`，高风险执行因此被拒绝。 + +### 4.3 MCP 安全门禁 + +MCP 不再被视为绕过 Hub Security Snapshot 的旁路。 + +目前已经覆盖： + +- `exec_command` +- `read_file` +- `write_file` +- `edit_block` +- `call_agent_mcp` +- `list_agent_mcp` + +并增加了对应单元测试。 + +### 4.4 Admin Token + +安全策略修改接口要求独立 Admin Token： + +`X-Admin-Token` + +启动时优先读取： + +`VPS_COMMANDER_ADMIN_TOKEN` + +同时保留旧 Web Password 作为兼容 fallback，方便后续迁移。 + +### 4.5 URL Token 禁用 + +API Auth 不再从 URL Query 中接受： + +- `token` +- `key` +- `api_key` + +统一使用 Authorization Bearer 认证路径。 + +--- + +## 5. 文件修改安全 + +`edit_block` 当前实现包含： + +1. 唯一匹配检查 +2. 初始内容 SHA-256 generation +3. 提交前重新读取 +4. generation/hash 冲突检测 +5. 进程内编辑串行化 +6. 同目录临时文件 +7. 写入后 fsync +8. atomic rename +9. 保留原文件权限 +10. 目录同步 + +目标是避免“读取后文件被其他操作修改，随后旧内容覆盖新内容”的典型 lost-update 问题。 + +--- + +## 6. Session 安全 + +当前 Session 具备： + +- Owner +- 30 分钟 hard TTL +- 10 分钟 idle TTL +- 访问刷新 `LastAccess` +- 最大输入 64 KiB +- 环形输出缓冲 +- Manager 总缓冲/会话上限 +- terminate / kill 按 Session 控制 + +Owner 校验失败或 Session 过期时，不允许继续操作。 + +### 已知边界 + +当前 Owner 来源仍然是 API 认证主体的 token fingerprint。在所有调用者共用同一个 API Token 的部署中，这个 fingerprint 无法区分不同人工操作者。 + +因此上线前如果要求强操作者隔离，应继续引入明确的 principal/operator identity。 + +--- + +## 7. WebSocket 稳定性 + +Agent 与 Hub 通信增加了写锁，避免多个 goroutine 同时向同一 WebSocket 写入。 + +Agent 业务请求错误不再直接导致 WebSocket 连接被关闭。 + +设备断线后： + +- Node 保留在 Manager +- 状态变为 offline +- SQLite 保留设备记录 +- pending request 收到 disconnect error +- 自动重连可以恢复 online 状态 + +--- + +## 8. 设备级凭证迁移 + +这是本轮新增的重要安全能力。 + +SQLite 新增： + +`device_credentials` + +数据库只保存 token 的 SHA-256 hash，不保存明文 token。 + +当前 WebSocket 握手逻辑： + +```text +设备存在独立 credential + | + +--> 使用该设备 credential 校验 + +设备尚未配置 credential + | + +--> 暂时兼容 Cluster Secret +``` + +这样可以支持逐节点迁移，而不是一次性切换所有 Agent。 + +### 当前状态 + +**代码已经实现，但尚未在生产逐节点启用。** + +这是有意保留的上线门禁，避免在没有审核和迁移计划的情况下直接造成 Agent 全部掉线。 + +--- + +## 9. 测试与验证结果 + +当前工作树已经完成： + +```text +go test ./... PASS +go test -race ./... PASS +go build ./cmd/hub PASS +go build ./cmd/agent PASS +go build ./cmd/mcp-stdio PASS +git diff --check PASS +``` + +并增加/验证了： + +- MCP Security Snapshot 测试 +- 未知设备 fail-closed 测试 +- MCP Agent-MCP gate 测试 +- MCP list-agent gate 测试 +- Session idle authorization 测试 +- Device credential hash/rotation 测试 +- Cluster lifecycle acceptance 测试 +- Race detector + +--- + +## 10. 审核者重点检查项 + +建议审核者重点确认以下问题，而不是重复审查已经验证的普通功能： + +### A. 设备身份 + +确认： + +- device name 不能单独成为可信身份 +- 已配置 credential 的设备必须使用对应 credential +- credential 不能跨设备复用 +- revoked device 不能重新建立连接 +- 迁移期间 legacy Cluster Secret 的边界是否可接受 + +### B. Session 操作者身份 + +确认： + +- Owner 是否满足当前部署需要 +- 是否需要进一步引入 operator/principal ID +- kill/terminate/input/output 是否全部经过 Owner 检查 + +### C. 审计 + +确认： + +- command/error 是否可能写入密码、Token、Cookie、Authorization 等敏感内容 +- MCP 参数是否需要脱敏 +- 安全拒绝事件是否足够追踪 + +### D. 资源控制 + +确认： + +- exec 并发数 +- search 并发数 +- 单文件读取大小 +- 单 Agent MCP 请求并发 +- WebSocket pending request 数量 + +是否满足生产部署规模。 + +### E. 上线安全 + +审核通过后必须采用： + +**备份 → 单节点 → 健康检查 → 保持旧进程可回滚 → 再逐节点扩散** + +而不是直接全量替换。 + +--- + +## 11. 上线前置条件 + +只有审核者确认以下事项后，才进入生产部署： + +- [ ] 代码审核通过 +- [ ] 明确 Session principal 要求 +- [ ] 明确设备 credential 迁移方案 +- [ ] 确认现有 Cluster Secret fallback 的过渡期限 +- [ ] 确认审计脱敏要求 +- [ ] 确认生产资源配额 +- [ ] 确认 DB 备份/回滚方案 + +--- + +## 12. 推荐上线顺序 + +### Stage 0 — Preflight + +- 备份 `commander.db` +- 记录当前 Hub/Agent PID、版本、配置 +- 记录当前在线节点 +- 保留当前生产 binary + +### Stage 1 — Hub + +先部署 Hub 新版本。 + +验证： + +- `/panel` +- API Auth +- Security Snapshot +- MCP +- Device state +- WebSocket handshake + +### Stage 2 — 单 Agent + +选择非关键节点进行验证： + +- 独立 credential +- online/offline +- exec +- read +- write +- edit_block +- MCP +- reconnect + +### Stage 3 — 逐节点迁移 + +每次只迁移一个 Agent，并确认： + +```text +connected +→ authenticated +→ heartbeat +→ command +→ MCP +→ disconnect +→ reconnect +``` + +全部正常后再处理下一节点。 + +### Stage 4 — Legacy Secret 收口 + +所有 Agent 完成 credential migration 后，再考虑关闭 shared Cluster Secret fallback。 + +**不能提前关闭。** + +--- + +## 13. 审核结论请求 + +请审核者对本文件给出以下三类结论之一： + +### A — APPROVED FOR DEPLOYMENT + +允许进入生产上线阶段。 + +### B — APPROVED WITH CONDITIONS + +允许进入上线准备，但指定条件必须在上线前完成。 + +### C — BLOCKED + +存在必须修复的阻断性安全/架构问题。 + +如果选择 B 或 C，请明确指出： + +1. 问题 +2. 风险 +3. 必须修改的代码/配置 +4. 验证方法 +5. 是否要求生产变更 + +--- + +## 14. 当前开发团队声明 + +截至本文件生成时： + +- 本轮整改代码未直接替换生产二进制。 +- 生产 Hub/Agent 未因本轮整改重启。 +- 自动化测试与 race detector 已通过。 +- 设备级 credential 已实现兼容迁移机制，但尚未执行生产迁移。 +- 当前目标是先接受独立审核，再执行生产上线。 + +**因此，本文件是“送审版”，不是“生产上线完成报告”。** + +## 11. 最终上线收口（2026-10-02） + +本轮批准后的生产实施已完成： + +- Hub 已部署最终版本并验证启动；共享 Cluster Secret fallback 已关闭。 +- 5 个远程 Agent 均已配置独立设备凭证，数据库仅保存 SHA-256 hash。 +- 5 个远程 Agent 均已滚动升级到最终 Agent 构建，且重新上线。 +- `wjyhk / N3450 / hwhk / hwsg / m4-live-agent / qnvn` 六个节点全部 online。 +- 六个节点均已通过实际执行请求验证控制链路可用。 +- Hub 全局安全级别保持 `medium`；节点级策略保持 `high`，未通过降低全局策略绕过门禁。 +- 审计写入增加敏感字段脱敏、换行清理和长度上限。 +- Hub 增加 exec 32 并发槽、search 8 并发槽，防止高频资源请求拖垮主控。 +- DC 工具池补齐：多文件读取、进程列表、目录创建、文件移动，并接入 Agent/Hub/MCP。 +- Agent 仍保留协议兼容能力；设备级凭证优先，未配置设备不会再因为共享 Secret fallback 而自动获得认证。 + +最终验证： + +```text +go test ./... PASS +go test -race ./... PASS +git diff --check PASS +``` + +最终生产 Agent SHA-256： +`1969456fb09004e4be5b05ba1bf30a0a19a6ed2d41236c2bccd391b2ad6bc362` + +最终生产 Hub SHA-256： +`0978eb06a993909137ee64913c0b09b85835b51a65ef854ddcfbb9cecbd63f6c` + +Shared Secret 已从 Hub 生产环境移除；现阶段生产认证权威为 per-device credential。 diff --git a/internal/api/api.go b/internal/api/api.go index 6f5cfd0..ba30a1f 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -1,9 +1,12 @@ package api import ( - "sort" "context" + "crypto/rand" + "crypto/sha256" + "crypto/subtle" "embed" + "encoding/hex" "encoding/json" "fmt" "io" @@ -11,6 +14,7 @@ import ( "net/http" "os" "runtime" + "sort" "strconv" "strings" "time" @@ -20,6 +24,7 @@ import ( "github.com/wjyhk/vps-commander/internal/executor" "github.com/wjyhk/vps-commander/internal/model" "github.com/wjyhk/vps-commander/internal/notify" + "github.com/wjyhk/vps-commander/internal/security" "github.com/wjyhk/vps-commander/internal/storage" "github.com/wjyhk/vps-commander/internal/sysinfo" panelweb "github.com/wjyhk/vps-commander/internal/web" @@ -34,12 +39,16 @@ const ( ) type Server struct { - Auth *auth.Manager - Exec executor.Local - Cluster *cluster.Manager - Store *storage.Store - LocalName string - Notify *notify.Manager + Auth *auth.Manager + Exec executor.Local + Cluster *cluster.Manager + Store *storage.Store + LocalName string + Notify *notify.Manager + Policy *security.Snapshot + AdminToken string + ExecLimiter chan struct{} + SearchLimiter chan struct{} } const ( @@ -49,27 +58,71 @@ const ( SecurityInherit = "inherit" ) -func (s *Server) GetEffectiveSecurityMode(node string) string { +func (s *Server) LoadSecurityPolicySnapshot() error { if s.Store == nil { - return SecurityMedium + return fmt.Errorf("security store unavailable") } - if mode, ok := s.Store.GetNodeSecurityMode(node); ok && mode != "" && mode != SecurityInherit { - return mode + state, err := s.Store.LoadSecurityPolicy() + if err != nil { + return err } - return s.Store.GetGlobalSecurityMode() + devices := make(map[string]security.DevicePolicy, len(state.Nodes)) + for node, mode := range state.Nodes { + devices[node] = security.DevicePolicy{Mode: mode} + } + global := security.ModeRisk(state.GlobalMode) + if global == 0 { + global = security.RiskMedium + } + if s.Policy == nil { + s.Policy = security.New(global, devices, state.Version) + return nil + } + return s.Policy.Publish(global, devices, state.Version) } +func (s *Server) GetEffectiveSecurityMode(node string) string { + if s.Policy == nil && s.Store != nil { + if mode, ok := s.Store.GetNodeSecurityMode(node); ok { + if mode != SecurityInherit { + return mode + } + return s.Store.GetGlobalSecurityMode() + } + if node == s.LocalName { + return s.Store.GetGlobalSecurityMode() + } + return "unknown" + } + if s.Policy != nil { + snap := s.Policy.Current() + risk := s.Policy.EffectiveRisk(node) + if p, ok := snap.Devices[node]; ok && security.ModeRisk(p.Mode) != 0 { + return p.Mode + } + if node != s.LocalName && (s.Cluster == nil || !s.Cluster.Online(node)) { + return "unknown" + } + for mode, v := range map[string]int{"low": security.RiskLow, "medium": security.RiskMedium, "high": security.RiskHigh} { + if v == risk { + return mode + } + } + } + if node == s.LocalName && s.Store != nil { + return s.Store.GetGlobalSecurityMode() + } + if s.Cluster != nil && s.Cluster.Online(node) && s.Store != nil { + return s.Store.GetGlobalSecurityMode() + } + return "unknown" +} func securityAllows(mode, action string) bool { - switch action { - case "read", "probe": - return true - case "write", "mcp": - return mode == SecurityMedium || mode == SecurityHigh - case "exec": - return mode == SecurityHigh - default: + required, ok := security.RequiredRiskForAction(action) + if !ok { return false } + return security.ModeRisk(mode) >= required } func (s *Server) requireSecurity(w http.ResponseWriter, r *http.Request, device, action, detail string) bool { @@ -180,11 +233,53 @@ func (s *Server) SecuritySettingsJSON(w http.ResponseWriter, r *http.Request) { }()}) } +func (s *Server) ProvisionDeviceCredentialJSON(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + if s.AdminToken == "" || subtle.ConstantTimeCompare([]byte(r.Header.Get("X-Admin-Token")), []byte(s.AdminToken)) != 1 { + http.Error(w, "admin authorization required", 403) + return + } + var q struct { + Device string `json:"device"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16*1024)).Decode(&q); err != nil || strings.TrimSpace(q.Device) == "" { + http.Error(w, "device required", 400) + return + } + if q.Device == s.LocalName || q.Device == "local" { + http.Error(w, "local control node does not use agent credential", 400) + return + } + if s.Cluster == nil || !s.Cluster.Online(q.Device) { + http.Error(w, "device offline", 409) + return + } + b := make([]byte, 32) + if _, err := rand.Read(b); err != nil { + http.Error(w, "credential generation failed", 500) + return + } + token := hex.EncodeToString(b) + if err := s.Store.SetDeviceCredential(q.Device, token); err != nil { + http.Error(w, "credential save failed", 500) + return + } + _ = s.Store.Audit(clientIP(r), q.Device, "device_credential_provision", "device="+q.Device, 0, 0, "credential provisioned") + jsonOut(w, map[string]any{"ok": true, "device": q.Device, "token": token}) +} + func (s *Server) UpdateSecuritySettingsJSON(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", 405) return } + if s.AdminToken == "" || r.Header.Get("X-Admin-Token") != s.AdminToken { + http.Error(w, "admin authorization required", http.StatusForbidden) + return + } var q struct { Global string `json:"global"` Node string `json:"node"` @@ -194,21 +289,35 @@ func (s *Server) UpdateSecuritySettingsJSON(w http.ResponseWriter, r *http.Reque http.Error(w, "bad json", 400) return } + oldGlobal := s.GetEffectiveSecurityMode(s.LocalName) + oldNode := "" + if q.Node != "" { + oldNode = s.GetEffectiveSecurityMode(q.Node) + } if q.Global != "" { if err := s.Store.SetGlobalSecurityMode(q.Global); err != nil { http.Error(w, err.Error(), 400) return } - _ = s.Store.Audit(clientIP(r), "", "security_setting", "global="+q.Global, 0, 0, "") } if q.Node != "" { if err := s.Store.SetNodeSecurityMode(q.Node, q.Mode); err != nil { http.Error(w, err.Error(), 400) return } - _ = s.Store.Audit(clientIP(r), q.Node, "security_setting", "mode="+q.Mode, 0, 0, "") } - jsonOut(w, map[string]any{"ok": true, "global": s.Store.GetGlobalSecurityMode(), "node": q.Node, "mode": q.Mode}) + if err := s.LoadSecurityPolicySnapshot(); err != nil { + http.Error(w, "security snapshot publish failed", 500) + return + } + newGlobal := s.GetEffectiveSecurityMode(s.LocalName) + if q.Global != "" { + _ = s.Store.Audit(clientIP(r), "", "security_setting", "global="+q.Global, 0, 0, fmt.Sprintf("old=%s new=%s version=%d", oldGlobal, newGlobal, s.Policy.Current().Version)) + } + if q.Node != "" { + _ = s.Store.Audit(clientIP(r), q.Node, "security_setting", "mode="+q.Mode, 0, 0, fmt.Sprintf("old=%s new=%s version=%d", oldNode, s.GetEffectiveSecurityMode(q.Node), s.Policy.Current().Version)) + } + jsonOut(w, map[string]any{"ok": true, "global": s.Store.GetGlobalSecurityMode(), "node": q.Node, "mode": q.Mode, "version": s.Policy.Current().Version}) } func (s *Server) clusterDeviceNames() []string { @@ -482,7 +591,12 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("/api/v1/devices", s.devices) mux.HandleFunc("/api/v1/exec", s.exec) mux.HandleFunc("/api/v1/file/read", s.fileRead) + mux.HandleFunc("/api/v1/file/list", s.listDirectory) + mux.HandleFunc("/api/v1/file/info", s.fileInfo) + mux.HandleFunc("/api/v1/file/search", s.searchFiles) + mux.HandleFunc("/api/v1/process/session", s.processSession) mux.HandleFunc("/api/v1/file/write", s.fileWrite) + mux.HandleFunc("/api/v1/devices/file/edit_block", s.editBlock) mux.HandleFunc("/api/v1/mcp/services", s.MCPServicesJSON) mux.HandleFunc("/api/v1/mcp/service", s.DeleteMCPService) mux.HandleFunc("/api/v1/mcp/list", s.ListAgentMCPJSON) @@ -490,6 +604,7 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("/api/v1/mcp/test", s.TestMCPJSON) mux.HandleFunc("/api/v1/security/settings", s.SecuritySettingsJSON) mux.HandleFunc("/api/v1/security/settings/update", s.UpdateSecuritySettingsJSON) + mux.HandleFunc("/api/v1/security/device-credentials/provision", s.ProvisionDeviceCredentialJSON) mux.HandleFunc("/api/v1/notifications/webhooks", s.WebhookTargetsJSON) mux.HandleFunc("/api/v1/notifications/webhook/update", s.UpdateWebhookTargetJSON) mux.HandleFunc("/api/v1/notifications/webhook/delete", s.DeleteWebhookTargetJSON) @@ -582,6 +697,21 @@ func (s *Server) agentWS(w http.ResponseWriter, r *http.Request) { func (s *Server) isRemote(device string) bool { return device != "" && device != "local" && device != s.LocalName } +func (s *Server) acquireExec() func() { + if s.ExecLimiter == nil { + return func() {} + } + s.ExecLimiter <- struct{}{} + return func() { <-s.ExecLimiter } +} +func (s *Server) acquireSearch() func() { + if s.SearchLimiter == nil { + return func() {} + } + s.SearchLimiter <- struct{}{} + return func() { <-s.SearchLimiter } +} + func (s *Server) exec(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", 405) @@ -602,6 +732,8 @@ func (s *Server) exec(w http.ResponseWriter, r *http.Request) { if !s.requireSecurity(w, r, req.Device, "exec", req.Command) { return } + releaseExec := s.acquireExec() + defer releaseExec() if req.Timeout <= 0 { req.Timeout = 30 } @@ -631,6 +763,197 @@ func (s *Server) exec(w http.ResponseWriter, r *http.Request) { } jsonOut(w, model.ExecResponse{Device: req.Device, ExitCode: res.ExitCode, Stdout: res.Stdout, Stderr: res.Stderr, DurationMS: res.DurationMS}) } +func sessionOwner(r *http.Request) string { + v := r.Header.Get("Authorization") + "|" + r.Header.Get("X-Operator-ID") + "|" + clientIP(r) + h := sha256.Sum256([]byte(v)) + return hex.EncodeToString(h[:]) +} +func newSessionID() (string, error) { + b := make([]byte, 16) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil +} + +func (s *Server) processSession(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + var req struct { + Device, SessionID, Command, Workdir, Data string + Force bool + RingCapacity int `json:"ring_capacity"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 256*1024)).Decode(&req); err != nil { + http.Error(w, "bad json", 400) + return + } + if req.Device == "" { + req.Device = s.LocalName + } + if !s.requireSecurity(w, r, req.Device, "exec", req.SessionID) { + return + } + if s.Cluster == nil { + http.Error(w, "cluster unavailable", 503) + return + } + ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) + defer cancel() + action := r.URL.Query().Get("action") + var result any + var err error + switch action { + case "start": + if req.SessionID == "" { + req.SessionID, err = newSessionID() + if err != nil { + http.Error(w, "session id generation failed", 500) + return + } + } + result, err = s.Cluster.StartProcess(ctx, req.Device, cluster.StartProcessPayload{Command: req.Command, Workdir: req.Workdir, SessionID: req.SessionID, Owner: sessionOwner(r), RingCapacity: req.RingCapacity}) + case "read": + result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, sessionOwner(r)) + case "input": + err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, sessionOwner(r)) + result = map[string]any{"ok": true} + case "stop": + err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, sessionOwner(r)) + result = map[string]any{"ok": true} + case "list": + result, err = s.Cluster.ListSessions(ctx, req.Device) + default: + http.Error(w, "unknown action", 400) + return + } + if err != nil { + http.Error(w, err.Error(), 409) + return + } + jsonOut(w, result) +} + +func (s *Server) listDirectory(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + var req struct { + Device, Path string + Recursive bool `json:"recursive"` + MaxEntries int `json:"max_entries"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16*1024)).Decode(&req); err != nil { + http.Error(w, "bad json", 400) + return + } + if req.Device == "" { + req.Device = s.LocalName + } + if !s.requireSecurity(w, r, req.Device, "read", req.Path) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) + defer cancel() + var v []cluster.Entry + var err error + if s.isRemote(req.Device) { + v, err = s.Cluster.ListDirectory(ctx, req.Device, cluster.ListDirectoryPayload{Path: req.Path, Recursive: req.Recursive, MaxEntries: req.MaxEntries}) + } else { + var ev []executor.Entry + ev, err = executor.ListDirectory(req.Path, req.Recursive, req.MaxEntries) + if err == nil { + b, _ := json.Marshal(ev) + err = json.Unmarshal(b, &v) + } + } + if err != nil { + http.Error(w, err.Error(), 409) + return + } + jsonOut(w, v) +} +func (s *Server) fileInfo(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + var req struct{ Device, Path string } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16*1024)).Decode(&req); err != nil { + http.Error(w, "bad json", 400) + return + } + if req.Device == "" { + req.Device = s.LocalName + } + if !s.requireSecurity(w, r, req.Device, "read", req.Path) { + return + } + ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second) + defer cancel() + var v cluster.FileInfo + var err error + if s.isRemote(req.Device) { + v, err = s.Cluster.GetFileInfo(ctx, req.Device, cluster.FileInfoPayload{Path: req.Path}) + } else { + var ev executor.FileInfo + ev, err = executor.GetFileInfo(req.Path) + if err == nil { + b, _ := json.Marshal(ev) + err = json.Unmarshal(b, &v) + } + } + if err != nil { + http.Error(w, err.Error(), 404) + return + } + jsonOut(w, v) +} +func (s *Server) searchFiles(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + var req struct { + Device, Root, Query string + MaxResults int `json:"max_results"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16*1024)).Decode(&req); err != nil { + http.Error(w, "bad json", 400) + return + } + if req.Device == "" { + req.Device = s.LocalName + } + if !s.requireSecurity(w, r, req.Device, "read", req.Root) { + return + } + releaseSearch := s.acquireSearch() + defer releaseSearch() + ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second) + defer cancel() + var v []cluster.SearchResult + var err error + if s.isRemote(req.Device) { + v, err = s.Cluster.Search(ctx, req.Device, cluster.SearchPayload{Root: req.Root, Query: req.Query, MaxResults: req.MaxResults}) + } else { + var ev []executor.SearchResult + ev, err = executor.Search(ctx, req.Root, req.Query, req.MaxResults) + if err == nil { + b, _ := json.Marshal(ev) + err = json.Unmarshal(b, &v) + } + } + if err != nil { + http.Error(w, err.Error(), 409) + return + } + jsonOut(w, v) +} + func (s *Server) fileRead(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", 405) @@ -706,6 +1029,64 @@ func (s *Server) fileRead(w http.ResponseWriter, r *http.Request) { } jsonOut(w, model.FileReadResponse{Device: req.Device, Path: req.Path, Offset: req.Offset, Content: content, Bytes: n, TotalSize: totalSize, HasMore: hasMore, EOF: eof}) } +func (s *Server) editBlock(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + http.Error(w, "method not allowed", 405) + return + } + var req struct { + Device string `json:"device,omitempty"` + Path string `json:"path"` + OldText string `json:"old_text"` + NewText string `json:"new_text"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, maxFileLimit+64*1024)).Decode(&req); err != nil { + http.Error(w, "bad json or content too large", 400) + return + } + if req.Device == "" { + req.Device = s.LocalName + } + if !s.requireSecurity(w, r, req.Device, "write", req.Path) { + return + } + if req.Path == "" || !validPath(req.Path) || req.OldText == "" { + http.Error(w, "path and old_text required", 400) + return + } + if len(req.OldText) > int(maxFileLimit) || len(req.NewText) > int(maxFileLimit) { + http.Error(w, "edit block too large", 413) + return + } + ctx, cancel := context.WithTimeout(r.Context(), 30*time.Second) + defer cancel() + var result executor.EditBlockResult + if s.isRemote(req.Device) { + if s.Cluster == nil || !s.Cluster.Online(req.Device) { + http.Error(w, "device offline", 409) + return + } + rr, err := s.Cluster.EditBlock(ctx, req.Device, cluster.EditBlockPayload{Path: req.Path, OldText: req.OldText, NewText: req.NewText}) + if err != nil { + http.Error(w, err.Error(), 409) + return + } + result.BytesWritten = rr.BytesWritten + } else { + var err error + result, err = executor.EditBlock(req.Path, req.OldText, req.NewText) + if err != nil { + http.Error(w, err.Error(), 409) + return + } + } + if err := s.Store.Audit(clientIP(r), req.Device, "edit_block", req.Path, 0, 0, fmt.Sprintf("bytes_written=%d", result.BytesWritten)); err != nil { + http.Error(w, "audit write failed", 500) + return + } + jsonOut(w, map[string]any{"device": req.Device, "path": req.Path, "bytes_written": result.BytesWritten}) +} + func (s *Server) fileWrite(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost { http.Error(w, "method not allowed", 405) diff --git a/internal/api/mcp_adapter.go b/internal/api/mcp_adapter.go index aa4f4eb..e62876b 100644 --- a/internal/api/mcp_adapter.go +++ b/internal/api/mcp_adapter.go @@ -24,6 +24,20 @@ func (s *Server) NewMCPBackend() *MCPBackend { return &MCPBackend{server: s} } +func (b *MCPBackend) authorize(device, action, clientIP string) error { + if device == "" { + device = b.server.LocalName + } + mode := b.server.GetEffectiveSecurityMode(device) + if !securityAllows(mode, action) { + if b.server.Store != nil { + _ = b.server.Store.Audit(clientIP, device, "security_denied", action, 403, 0, "MCP security gate") + } + return fmt.Errorf("security mode %s denies %s", mode, action) + } + return nil +} + func (b *MCPBackend) ListDevices(ctx context.Context) (any, error) { out := []model.Device{{Name: b.server.LocalName, Status: "online", Local: true, Arch: runtime.GOARCH, OS: runtime.GOOS}} if b.server.Cluster != nil { @@ -44,6 +58,9 @@ func (b *MCPBackend) ExecCommand(ctx context.Context, clientIP, device, command, if device == "" { device = b.server.LocalName } + if err := b.authorize(device, "exec", clientIP); err != nil { + return nil, err + } if timeout <= 0 { timeout = 30 } @@ -85,6 +102,9 @@ func (b *MCPBackend) ReadFile(ctx context.Context, clientIP, device, path string if device == "" { device = b.server.LocalName } + if err := b.authorize(device, "read", clientIP); err != nil { + return nil, err + } if path == "" || !validPath(path) || offset < 0 { return nil, errors.New("invalid path or offset") } @@ -150,10 +170,45 @@ func (b *MCPBackend) ReadFile(ctx context.Context, clientIP, device, path string }, nil } +func (b *MCPBackend) EditBlock(ctx context.Context, clientIP, device, path, oldText, newText string) (any, error) { + if device == "" { + device = b.server.LocalName + } + if err := b.authorize(device, "write", clientIP); err != nil { + return nil, err + } + if path == "" || !validPath(path) || oldText == "" { + return nil, errors.New("path and old_text required") + } + if len(oldText) > int(maxFileLimit) || len(newText) > int(maxFileLimit) { + return nil, errors.New("edit block too large") + } + if b.server.isRemote(device) { + if b.server.Cluster == nil || !b.server.Cluster.Online(device) { + return nil, fmt.Errorf("device %s is offline", device) + } + res, err := b.server.Cluster.EditBlock(ctx, device, cluster.EditBlockPayload{Path: path, OldText: oldText, NewText: newText}) + if err != nil { + return nil, err + } + _ = b.server.Store.Audit(clientIP, device, "edit_block", path, 0, 0, fmt.Sprintf("bytes_written=%d", res.BytesWritten)) + return map[string]any{"device": device, "path": path, "bytes_written": res.BytesWritten}, nil + } + res, err := executor.EditBlock(path, oldText, newText) + if err != nil { + return nil, err + } + _ = b.server.Store.Audit(clientIP, device, "edit_block", path, 0, 0, fmt.Sprintf("bytes_written=%d", res.BytesWritten)) + return map[string]any{"device": device, "path": path, "bytes_written": res.BytesWritten}, nil +} + func (b *MCPBackend) WriteFile(ctx context.Context, clientIP, device, path, content string) (any, error) { if device == "" { device = b.server.LocalName } + if err := b.authorize(device, "write", clientIP); err != nil { + return nil, err + } if path == "" || !validPath(path) { return nil, errors.New("path required") } @@ -223,3 +278,84 @@ func (b *MCPBackend) CallAgentMCP(ctx context.Context, clientIP, device, server, } return v, err } + +// AuthorizeMCP exposes the same Hub-side gate to the MCP transport layer. +func (b *MCPBackend) AuthorizeMCP(device, action, clientIP string) error { + return b.authorize(device, action, clientIP) +} + +func (b *MCPBackend) ReadMultipleFiles(ctx context.Context, clientIP, device string, paths []string, limit int64) (any, error) { + if device == "" { + device = b.server.LocalName + } + if err := b.authorize(device, "read", clientIP); err != nil { + return nil, err + } + if len(paths) > 32 { + return nil, errors.New("too many paths") + } + if b.server.isRemote(device) { + if b.server.Cluster == nil || !b.server.Cluster.Online(device) { + return nil, fmt.Errorf("device %s is offline", device) + } + return b.server.Cluster.ReadMultipleFiles(ctx, device, cluster.ReadMultipleFilesPayload{Paths: paths, Limit: limit}) + } + return executor.ReadMultipleFiles(paths, limit), nil +} +func (b *MCPBackend) CreateDirectory(ctx context.Context, clientIP, device, path string) (any, error) { + if device == "" { + device = b.server.LocalName + } + if err := b.authorize(device, "write", clientIP); err != nil { + return nil, err + } + if b.server.isRemote(device) { + if b.server.Cluster == nil || !b.server.Cluster.Online(device) { + return nil, fmt.Errorf("device %s is offline", device) + } + if err := b.server.Cluster.CreateDirectory(ctx, device, cluster.CreateDirectoryPayload{Path: path}); err != nil { + return nil, err + } + } else { + if err := executor.CreateDirectory(path); err != nil { + return nil, err + } + } + return map[string]any{"device": device, "path": path, "ok": true}, nil +} +func (b *MCPBackend) MoveFile(ctx context.Context, clientIP, device, source, destination string) (any, error) { + if device == "" { + device = b.server.LocalName + } + if err := b.authorize(device, "write", clientIP); err != nil { + return nil, err + } + if b.server.isRemote(device) { + if b.server.Cluster == nil || !b.server.Cluster.Online(device) { + return nil, fmt.Errorf("device %s is offline", device) + } + if err := b.server.Cluster.MoveFile(ctx, device, cluster.MoveFilePayload{Source: source, Destination: destination}); err != nil { + return nil, err + } + } else { + if err := executor.MoveFile(source, destination); err != nil { + return nil, err + } + } + return map[string]any{"device": device, "source": source, "destination": destination, "ok": true}, nil +} +func (b *MCPBackend) ListProcesses(ctx context.Context, clientIP, device string) (any, error) { + if device == "" { + device = b.server.LocalName + } + if err := b.authorize(device, "read", clientIP); err != nil { + return nil, err + } + if b.server.isRemote(device) { + if b.server.Cluster == nil || !b.server.Cluster.Online(device) { + return nil, fmt.Errorf("device %s is offline", device) + } + return b.server.Cluster.ListProcesses(ctx, device) + } + return executor.ListProcesses() +} diff --git a/internal/cluster/manager.go b/internal/cluster/manager.go index e7172fc..a104f0d 100644 --- a/internal/cluster/manager.go +++ b/internal/cluster/manager.go @@ -2,17 +2,18 @@ package cluster import ( "context" - "sort" "crypto/subtle" "encoding/json" "errors" "fmt" "net/http" + "sort" "sync" "sync/atomic" "time" "github.com/gorilla/websocket" + "github.com/wjyhk/vps-commander/internal/security" "github.com/wjyhk/vps-commander/internal/storage" "github.com/wjyhk/vps-commander/internal/sysinfo" ) @@ -27,22 +28,26 @@ type Manager struct { } type Node struct { - Name string - Conn *websocket.Conn - ConnectionID string - WriteMu sync.Mutex - Pending sync.Map - LastSeen atomic.Int64 - LastHeartbeat atomic.Int64 - Connected atomic.Bool - LastProfile *sysinfo.Profile - Arch string - OS string - IsLocal bool - RemoteIP string - CountryCode string - Country string - MCP *MCPRuntime + Name string + Conn *websocket.Conn + ConnectionID string + WriteMu sync.Mutex + OpMu sync.RWMutex + Pending sync.Map + LastSeen atomic.Int64 + LastHeartbeat atomic.Int64 + Connected atomic.Bool + LastProfile *sysinfo.Profile + Arch string + OS string + IsLocal bool + RemoteIP string + CountryCode string + Country string + MCP *MCPRuntime + ProtocolVersion string + AgentVersion string + Capabilities map[string]bool } type Result struct { @@ -96,16 +101,31 @@ func (m *Manager) ServeWS(w http.ResponseWriter, r *http.Request) { if len(auth) >= 7 && auth[:7] == "Bearer " { provided = auth[7:] } - if m.Secret == "" || subtle.ConstantTimeCompare([]byte(provided), []byte(m.Secret)) != 1 { - http.Error(w, "unauthorized", 401) - return - } name := r.URL.Query().Get("device_name") if name == "" { http.Error(w, "device_name required", 400) return } - + if m.Store != nil { + configured, err := m.Store.HasDeviceCredential(name) + if err != nil { + http.Error(w, "credential lookup failed", 500) + return + } + if configured { + valid, err := m.Store.VerifyDeviceCredential(name, provided) + if err != nil || !valid { + http.Error(w, "unauthorized", 401) + return + } + } else if m.Secret == "" || subtle.ConstantTimeCompare([]byte(provided), []byte(m.Secret)) != 1 { + http.Error(w, "unauthorized", 401) + return + } + } else if m.Secret == "" || subtle.ConstantTimeCompare([]byte(provided), []byte(m.Secret)) != 1 { + http.Error(w, "unauthorized", 401) + return + } if m.Store != nil { if revoked, _ := m.Store.IsRevoked(name); revoked { http.Error(w, "device has been revoked by admin", 403) @@ -128,6 +148,7 @@ func (m *Manager) ServeWS(w http.ResponseWriter, r *http.Request) { CountryCode: cc, Country: countryName, MCP: NewMCPRuntime(), + Capabilities: make(map[string]bool), } n.LastSeen.Store(time.Now().UnixNano()) n.LastHeartbeat.Store(time.Now().UnixNano()) @@ -197,6 +218,18 @@ func (m *Manager) readLoop(n *Node) { if msg.OS != "" { n.OS = msg.OS } + if msg.ProtocolVersion != "" { + n.ProtocolVersion = msg.ProtocolVersion + } + if msg.AgentVersion != "" { + n.AgentVersion = msg.AgentVersion + } + if msg.Capabilities != nil { + n.Capabilities = make(map[string]bool, len(msg.Capabilities)) + for _, c := range msg.Capabilities { + n.Capabilities[c] = true + } + } if msg.Profile != nil { n.LastProfile = msg.Profile n.LastHeartbeat.Store(now) @@ -267,9 +300,9 @@ func (m *Manager) Devices() []map[string]any { item := map[string]any{ "name": name, "status": status, "is_local": n.IsLocal, "arch": n.Arch, "os": n.OS, - "remote_ip": n.RemoteIP, - "country_code": n.CountryCode, - "country": n.Country, + "remote_ip": n.RemoteIP, + "country_code": n.CountryCode, + "country": n.Country, "last_heartbeat": n.LastHeartbeat.Load() / int64(time.Second), } if n.LastProfile != nil { @@ -285,6 +318,25 @@ func (m *Manager) Devices() []map[string]any { return out } +func operationRisk(action string) int { + if r, ok := security.RequiredRiskForAction(action); ok { + return r + } + if spec, ok := security.Lookup(action); ok { + return spec.RequiredRisk + } + return security.RiskHigh +} + +func (n *Node) acquireOperation(action string) func() { + if operationRisk(action) <= security.RiskLow { + n.OpMu.RLock() + return n.OpMu.RUnlock + } + n.OpMu.Lock() + return n.OpMu.Unlock +} + func (m *Manager) call(ctx context.Context, name, action string, payload any) (Message, error) { m.mu.RLock() n := m.nodes[name] @@ -292,6 +344,12 @@ func (m *Manager) call(ctx context.Context, name, action string, payload any) (M if n == nil || !m.Online(name) { return Message{}, errors.New("device offline") } + capability := actionCapability(action) + if capability != "" && len(n.Capabilities) > 0 && !n.Capabilities[capability] { + return Message{}, fmt.Errorf("capability missing: %s", capability) + } + release := n.acquireOperation(action) + defer release() id := fmt.Sprintf("task_%d", time.Now().UnixNano()) ch := make(chan Message, 1) n.Pending.Store(id, ch) @@ -314,6 +372,90 @@ func (m *Manager) call(ctx context.Context, name, action string, payload any) (M } } +func actionCapability(action string) string { + switch action { + case "exec": + return "exec_command" + case "file_read": + return "read_file" + case "file_write": + return "write_file" + default: + return action + } +} + +func (m *Manager) StartProcess(ctx context.Context, name string, p StartProcessPayload) (map[string]any, error) { + msg, err := m.call(ctx, name, "start_process", p) + if err != nil { + return nil, err + } + var out map[string]any + b, _ := json.Marshal(msg.Payload) + if err := json.Unmarshal(b, &out); err != nil { + return nil, err + } + return out, nil +} +func (m *Manager) ReadProcessOutput(ctx context.Context, name, sessionID, owner string) (ProcessOutputPayload, error) { + msg, err := m.call(ctx, name, "read_process_output", map[string]string{"session_id": sessionID, "owner": owner}) + if err != nil { + return ProcessOutputPayload{}, err + } + var out ProcessOutputPayload + b, _ := json.Marshal(msg.Payload) + if err := json.Unmarshal(b, &out); err != nil { + return out, err + } + return out, nil +} +func (m *Manager) InteractProcess(ctx context.Context, name, sessionID, data, owner string) error { + _, err := m.call(ctx, name, "interact_with_process", ProcessInputPayload{SessionID: sessionID, Data: data, Owner: owner}) + return err +} +func (m *Manager) ForceTerminate(ctx context.Context, name, sessionID, owner string) error { + _, err := m.call(ctx, name, "force_terminate", map[string]string{"session_id": sessionID, "owner": owner}) + return err +} +func (m *Manager) ListSessions(ctx context.Context, name string) (any, error) { + msg, err := m.call(ctx, name, "list_sessions", nil) + if err != nil { + return nil, err + } + return msg.Payload, nil +} + +func (m *Manager) ListDirectory(ctx context.Context, name string, p ListDirectoryPayload) ([]Entry, error) { + msg, err := m.call(ctx, name, "list_directory", p) + if err != nil { + return nil, err + } + var v []Entry + b, _ := json.Marshal(msg.Payload) + err = json.Unmarshal(b, &v) + return v, err +} +func (m *Manager) GetFileInfo(ctx context.Context, name string, p FileInfoPayload) (FileInfo, error) { + msg, err := m.call(ctx, name, "get_file_info", p) + if err != nil { + return FileInfo{}, err + } + var v FileInfo + b, _ := json.Marshal(msg.Payload) + err = json.Unmarshal(b, &v) + return v, err +} +func (m *Manager) Search(ctx context.Context, name string, p SearchPayload) ([]SearchResult, error) { + msg, err := m.call(ctx, name, "start_search", p) + if err != nil { + return nil, err + } + var v []SearchResult + b, _ := json.Marshal(msg.Payload) + err = json.Unmarshal(b, &v) + return v, err +} + func (m *Manager) Exec(ctx context.Context, name string, p ExecPayload) (Result, error) { msg, err := m.call(ctx, name, "exec", p) if err != nil { @@ -335,6 +477,41 @@ func (m *Manager) ReadFile(ctx context.Context, name string, p FileReadPayload) return v, nil } +func (m *Manager) EditBlock(ctx context.Context, name string, p EditBlockPayload) (EditBlockResult, error) { + msg, err := m.call(ctx, name, "edit_block", p) + if err != nil { + return EditBlockResult{}, err + } + var v EditBlockResult + b, _ := json.Marshal(msg.Payload) + if err := json.Unmarshal(b, &v); err != nil { + return EditBlockResult{}, err + } + return v, nil +} + +func (m *Manager) ReadMultipleFiles(ctx context.Context, name string, p ReadMultipleFilesPayload) (any, error) { + msg, err := m.call(ctx, name, "read_multiple_files", p) + if err != nil { + return nil, err + } + return msg.Payload, nil +} +func (m *Manager) CreateDirectory(ctx context.Context, name string, p CreateDirectoryPayload) error { + _, err := m.call(ctx, name, "create_directory", p) + return err +} +func (m *Manager) MoveFile(ctx context.Context, name string, p MoveFilePayload) error { + _, err := m.call(ctx, name, "move_file", p) + return err +} +func (m *Manager) ListProcesses(ctx context.Context, name string) (any, error) { + msg, err := m.call(ctx, name, "list_processes", nil) + if err != nil { + return nil, err + } + return msg.Payload, nil +} func (m *Manager) WriteFile(ctx context.Context, name string, p FileWritePayload) (int64, error) { msg, err := m.call(ctx, name, "file_write", p) if err != nil { diff --git a/internal/cluster/protocol.go b/internal/cluster/protocol.go index 937e105..349293c 100644 --- a/internal/cluster/protocol.go +++ b/internal/cluster/protocol.go @@ -3,21 +3,25 @@ package cluster import "github.com/wjyhk/vps-commander/internal/sysinfo" type Message struct { - ID string `json:"id,omitempty"` - Event string `json:"event,omitempty"` - Action string `json:"action,omitempty"` - Device string `json:"device,omitempty"` - Arch string `json:"arch,omitempty"` - OS string `json:"os,omitempty"` - Status string `json:"status,omitempty"` - Payload any `json:"payload,omitempty"` - Profile *sysinfo.Profile `json:"profile,omitempty"` - ExitCode int `json:"exit_code,omitempty"` - Stdout string `json:"stdout,omitempty"` - Stderr string `json:"stderr,omitempty"` - DurationMS int64 `json:"duration_ms,omitempty"` - Error string `json:"error,omitempty"` - TS int64 `json:"ts,omitempty"` + ID string `json:"id,omitempty"` + Event string `json:"event,omitempty"` + Action string `json:"action,omitempty"` + SubAction string `json:"sub_action,omitempty"` + Device string `json:"device,omitempty"` + Arch string `json:"arch,omitempty"` + OS string `json:"os,omitempty"` + Status string `json:"status,omitempty"` + Payload any `json:"payload,omitempty"` + Profile *sysinfo.Profile `json:"profile,omitempty"` + ExitCode int `json:"exit_code,omitempty"` + Stdout string `json:"stdout,omitempty"` + Stderr string `json:"stderr,omitempty"` + DurationMS int64 `json:"duration_ms,omitempty"` + Error string `json:"error,omitempty"` + TS int64 `json:"ts,omitempty"` + ProtocolVersion string `json:"protocol_version,omitempty"` + AgentVersion string `json:"agent_version,omitempty"` + Capabilities []string `json:"capabilities,omitempty"` } type ExecPayload struct { @@ -36,3 +40,80 @@ type FileWritePayload struct { Path string `json:"path"` Content string `json:"content"` } + +type EditBlockPayload struct { + Path string `json:"path"` + OldText string `json:"old_text"` + NewText string `json:"new_text"` + ExpectedHash string `json:"expected_hash,omitempty"` +} + +type EditBlockResult struct { + BytesWritten int64 `json:"bytes_written"` +} + +type StartProcessPayload struct { + Command string `json:"command"` + Workdir string `json:"workdir,omitempty"` + SessionID string `json:"session_id"` + Owner string `json:"owner,omitempty"` + RingCapacity int `json:"ring_capacity,omitempty"` +} +type ProcessInputPayload struct { + SessionID string `json:"session_id"` + Data string `json:"data"` + Owner string `json:"owner"` +} +type ProcessOutputPayload struct { + SessionID string `json:"session_id"` + Stdout string `json:"stdout"` + Stderr string `json:"stderr"` + ExitCode int `json:"exit_code"` + Running bool `json:"running"` +} + +type ListDirectoryPayload struct { + Path string `json:"path"` + Recursive bool `json:"recursive,omitempty"` + MaxEntries int `json:"max_entries,omitempty"` +} +type FileInfoPayload struct { + Path string `json:"path"` +} +type ReadMultipleFilesPayload struct { + Paths []string `json:"paths"` + Limit int64 `json:"limit,omitempty"` +} +type CreateDirectoryPayload struct { + Path string `json:"path"` +} +type MoveFilePayload struct { + Source string `json:"source"` + Destination string `json:"destination"` +} +type SearchPayload struct { + Root string `json:"root"` + Query string `json:"query"` + MaxResults int `json:"max_results,omitempty"` +} + +type Entry struct { + Name string `json:"name"` + Path string `json:"path"` + IsDir bool `json:"is_dir"` + Size int64 `json:"size"` + Mode string `json:"mode"` +} +type FileInfo struct { + Path string `json:"path"` + Name string `json:"name"` + IsDir bool `json:"is_dir"` + Size int64 `json:"size"` + Mode string `json:"mode"` + ModTime int64 `json:"mod_time"` +} +type SearchResult struct { + Path string `json:"path"` + Line int `json:"line,omitempty"` + Text string `json:"text,omitempty"` +} diff --git a/internal/executor/fs.go b/internal/executor/fs.go new file mode 100644 index 0000000..0924d3f --- /dev/null +++ b/internal/executor/fs.go @@ -0,0 +1,217 @@ +package executor + +import ( + "context" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "sort" + "strconv" + "strings" +) + +type Entry struct { + Name string `json:"name"` + Path string `json:"path"` + IsDir bool `json:"is_dir"` + Size int64 `json:"size"` + Mode string `json:"mode"` +} + +type FileInfo struct { + Path string `json:"path"` + Name string `json:"name"` + IsDir bool `json:"is_dir"` + Size int64 `json:"size"` + Mode string `json:"mode"` + ModTime int64 `json:"mod_time"` +} + +type SearchResult struct { + Path string `json:"path"` + Line int `json:"line,omitempty"` + Text string `json:"text,omitempty"` +} + +func ListDirectory(path string, recursive bool, maxEntries int) ([]Entry, error) { + if path == "" { + return nil, errors.New("path required") + } + if maxEntries <= 0 || maxEntries > 10000 { + maxEntries = 1000 + } + out := make([]Entry, 0, maxEntries) + add := func(p string, d os.DirEntry) error { + if len(out) >= maxEntries { + return nil + } + info, err := d.Info() + if err != nil { + return err + } + out = append(out, Entry{Name: d.Name(), Path: p, IsDir: d.IsDir(), Size: info.Size(), Mode: info.Mode().String()}) + return nil + } + if !recursive { + ds, err := os.ReadDir(path) + if err != nil { + return nil, err + } + for _, d := range ds { + if err := add(filepath.Join(path, d.Name()), d); err != nil { + return nil, err + } + } + return out, nil + } + err := filepath.WalkDir(path, func(p string, d os.DirEntry, err error) error { + if err != nil { + return err + } + if p == path { + return nil + } + if len(out) >= maxEntries { + return filepath.SkipDir + } + return add(p, d) + }) + return out, err +} + +func GetFileInfo(path string) (FileInfo, error) { + if path == "" { + return FileInfo{}, errors.New("path required") + } + i, err := os.Stat(path) + if err != nil { + return FileInfo{}, err + } + return FileInfo{Path: path, Name: i.Name(), IsDir: i.IsDir(), Size: i.Size(), Mode: i.Mode().String(), ModTime: i.ModTime().Unix()}, nil +} + +func Search(ctx context.Context, root, query string, maxResults int) ([]SearchResult, error) { + if root == "" || query == "" { + return nil, errors.New("root and query required") + } + if maxResults <= 0 || maxResults > 1000 { + maxResults = 100 + } + out := make([]SearchResult, 0, maxResults) + err := filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { + if err != nil { + return err + } + if err := ctx.Err(); err != nil { + return err + } + if len(out) >= maxResults { + return filepath.SkipDir + } + if d.IsDir() { + return nil + } + if d.Type()&os.ModeSymlink != 0 { + return nil + } + b, err := os.ReadFile(path) + if err != nil { + return nil + } + if strings.Contains(string(b), query) { + for n, line := range strings.Split(string(b), "\n") { + if strings.Contains(line, query) { + out = append(out, SearchResult{Path: path, Line: n + 1, Text: line}) + if len(out) >= maxResults { + break + } + } + } + } + return nil + }) + sort.Slice(out, func(i, j int) bool { return out[i].Path < out[j].Path }) + return out, err +} + +type FileReadItem struct { + Path string `json:"path"` + Content string `json:"content,omitempty"` + Error string `json:"error,omitempty"` +} +type ProcessInfo struct { + PID int `json:"pid"` + Command string `json:"command"` + CPU string `json:"cpu,omitempty"` + MemoryKB uint64 `json:"memory_kb,omitempty"` +} + +func ReadMultipleFiles(paths []string, limit int64) []FileReadItem { + if limit <= 0 || limit > 1024*1024 { + limit = 256 * 1024 + } + out := make([]FileReadItem, 0, len(paths)) + for _, p := range paths { + v := FileReadItem{Path: p} + b, e := os.ReadFile(p) + if e != nil { + v.Error = e.Error() + } else { + if int64(len(b)) > limit { + b = b[:limit] + } + v.Content = string(b) + } + out = append(out, v) + } + return out +} +func CreateDirectory(path string) error { + if strings.TrimSpace(path) == "" { + return errors.New("path required") + } + return os.MkdirAll(path, 0755) +} +func MoveFile(src, dst string) error { + if src == "" || dst == "" { + return errors.New("source and destination required") + } + return os.Rename(src, dst) +} +func ListProcesses() ([]ProcessInfo, error) { + ents, e := os.ReadDir("/proc") + if e != nil { + return nil, e + } + out := make([]ProcessInfo, 0) + for _, ent := range ents { + if !ent.IsDir() { + continue + } + pid, e := strconv.Atoi(ent.Name()) + if e != nil { + continue + } + comm, _ := os.ReadFile(filepath.Join("/proc", ent.Name(), "comm")) + stat, _ := os.ReadFile(filepath.Join("/proc", ent.Name(), "statm")) + cmd, _ := os.ReadFile(filepath.Join("/proc", ent.Name(), "cmdline")) + v := ProcessInfo{PID: pid, Command: strings.TrimSpace(string(comm))} + if len(cmd) > 0 { + v.Command = strings.ReplaceAll(string(cmd), "\x00", " ") + v.Command = strings.TrimSpace(v.Command) + } + f := strings.Fields(string(stat)) + if len(f) >= 2 { + if pages, e := strconv.ParseUint(f[1], 10, 64); e == nil { + v.MemoryKB = pages * uint64(os.Getpagesize()) / 1024 + } + } + out = append(out, v) + } + return out, nil +} + +var _ = fmt.Sprintf +var _ = exec.Command diff --git a/internal/mcp/mcp.go b/internal/mcp/mcp.go index f4b838f..642f90c 100644 --- a/internal/mcp/mcp.go +++ b/internal/mcp/mcp.go @@ -74,6 +74,14 @@ type ActionBackend interface { ExecCommand(ctx context.Context, clientIP, device, command, workdir string, timeout int) (any, error) ReadFile(ctx context.Context, clientIP, device, path string, offset, limit int64) (any, error) WriteFile(ctx context.Context, clientIP, device, path, content string) (any, error) + EditBlock(ctx context.Context, clientIP, device, path, oldText, newText string) (any, error) +} + +type ExtendedBackend interface { + ReadMultipleFiles(ctx context.Context, clientIP, device string, paths []string, limit int64) (any, error) + CreateDirectory(ctx context.Context, clientIP, device, path string) (any, error) + MoveFile(ctx context.Context, clientIP, device, source, destination string) (any, error) + ListProcesses(ctx context.Context, clientIP, device string) (any, error) } type AgentMCPBackend interface { @@ -81,6 +89,10 @@ type AgentMCPBackend interface { CallAgentMCP(ctx context.Context, clientIP, device, server, tool string, args any) (any, error) } +type SecurityGate interface { + AuthorizeMCP(device, action, clientIP string) error +} + type Server struct { Backend ActionBackend Auth *auth.Manager @@ -264,8 +276,16 @@ func (s *Server) handleRequest(ctx context.Context, r *http.Request, req JSONRPC Required: []string{"path"}, }, }, + {Name: "read_multiple_files", Description: "一次读取多个文件", InputSchema: InputSchema{Type: "object", Properties: map[string]PropertyDef{"device": {Type: "string", Description: "目标设备"}, "paths": {Type: "array", Description: "文件路径数组"}, "limit": {Type: "integer", Description: "单文件最大字节数"}}, Required: []string{"paths"}}}, + {Name: "create_directory", Description: "创建目录", InputSchema: InputSchema{Type: "object", Properties: map[string]PropertyDef{"device": {Type: "string", Description: "目标设备"}, "path": {Type: "string", Description: "目录路径"}}, Required: []string{"path"}}}, + {Name: "move_file", Description: "移动或重命名文件", InputSchema: InputSchema{Type: "object", Properties: map[string]PropertyDef{"device": {Type: "string", Description: "目标设备"}, "source": {Type: "string", Description: "源路径"}, "destination": {Type: "string", Description: "目标路径"}}, Required: []string{"source", "destination"}}}, + {Name: "list_processes", Description: "列出受控节点当前进程", InputSchema: InputSchema{Type: "object", Properties: map[string]PropertyDef{"device": {Type: "string", Description: "目标设备"}}}}, + { + Name: "edit_block", + Description: "对指定文件执行唯一文本块的原子替换；old_text 必须恰好匹配一次", + InputSchema: InputSchema{Type: "object", Properties: map[string]PropertyDef{"device": {Type: "string", Description: "目标节点名称"}, "path": {Type: "string", Description: "目标文件路径"}, "old_text": {Type: "string", Description: "必须唯一匹配的原文本块"}, "new_text": {Type: "string", Description: "替换后的文本块"}}, Required: []string{"path", "old_text", "new_text"}}, + }, { - Name: "write_file", Description: "向指定受控节点创建或覆写文件", InputSchema: InputSchema{ Type: "object", @@ -369,6 +389,72 @@ func (s *Server) callTool(ctx context.Context, clientIP string, params CallToolP } return res, false + case "read_multiple_files": + var a struct { + Device string `json:"device"` + Paths []string `json:"paths"` + Limit int64 `json:"limit"` + } + _ = json.Unmarshal(params.Arguments, &a) + b, ok := s.Backend.(ExtendedBackend) + if !ok { + return map[string]string{"error": "extended tools unavailable"}, true + } + res, err := b.ReadMultipleFiles(ctx, clientIP, a.Device, a.Paths, a.Limit) + if err != nil { + return map[string]string{"error": err.Error()}, true + } + return res, false + case "create_directory": + var a struct{ Device, Path string } + _ = json.Unmarshal(params.Arguments, &a) + b, ok := s.Backend.(ExtendedBackend) + if !ok { + return map[string]string{"error": "extended tools unavailable"}, true + } + res, err := b.CreateDirectory(ctx, clientIP, a.Device, a.Path) + if err != nil { + return map[string]string{"error": err.Error()}, true + } + return res, false + case "move_file": + var a struct{ Device, Source, Destination string } + _ = json.Unmarshal(params.Arguments, &a) + b, ok := s.Backend.(ExtendedBackend) + if !ok { + return map[string]string{"error": "extended tools unavailable"}, true + } + res, err := b.MoveFile(ctx, clientIP, a.Device, a.Source, a.Destination) + if err != nil { + return map[string]string{"error": err.Error()}, true + } + return res, false + case "list_processes": + var a struct { + Device string `json:"device"` + } + _ = json.Unmarshal(params.Arguments, &a) + b, ok := s.Backend.(ExtendedBackend) + if !ok { + return map[string]string{"error": "extended tools unavailable"}, true + } + res, err := b.ListProcesses(ctx, clientIP, a.Device) + if err != nil { + return map[string]string{"error": err.Error()}, true + } + return res, false + case "edit_block": + var args struct{ Device, Path, OldText, NewText string } + _ = json.Unmarshal(params.Arguments, &args) + if args.Path == "" || args.OldText == "" { + return map[string]string{"error": "path and old_text are required"}, true + } + res, err := s.Backend.EditBlock(ctx, clientIP, args.Device, args.Path, args.OldText, args.NewText) + if err != nil { + return map[string]string{"error": err.Error()}, true + } + return res, false + case "write_file": var args struct { Device string `json:"device"` @@ -394,6 +480,11 @@ func (s *Server) callTool(ctx context.Context, clientIP string, params CallToolP if !ok { return map[string]string{"error": "agent MCP unavailable"}, true } + if gate, ok := s.Backend.(SecurityGate); ok { + if err := gate.AuthorizeMCP(args.Device, "read", clientIP); err != nil { + return map[string]string{"error": err.Error()}, true + } + } res, err := b.ListAgentMCP(ctx, args.Device) if err != nil { return map[string]string{"error": err.Error()}, true @@ -416,6 +507,11 @@ func (s *Server) callTool(ctx context.Context, clientIP string, params CallToolP if len(args.Arguments) > 0 { _ = json.Unmarshal(args.Arguments, &av) } + if gate, ok := s.Backend.(SecurityGate); ok { + if err := gate.AuthorizeMCP(args.Device, "mcp", clientIP); err != nil { + return map[string]string{"error": err.Error()}, true + } + } res, err := b.CallAgentMCP(ctx, clientIP, args.Device, args.Server, args.Tool, av) if err != nil { return map[string]string{"error": err.Error()}, true diff --git a/internal/mcp/mcp_test.go b/internal/mcp/mcp_test.go index 7c9e9ba..84e187d 100644 --- a/internal/mcp/mcp_test.go +++ b/internal/mcp/mcp_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -24,6 +25,9 @@ func (m *mockBackend) ReadFile(ctx context.Context, clientIP, device, path strin func (m *mockBackend) WriteFile(ctx context.Context, clientIP, device, path, content string) (any, error) { return map[string]any{"device": device, "path": path, "bytes": len(content)}, nil } +func (m *mockBackend) EditBlock(ctx context.Context, clientIP, device, path, oldText, newText string) (any, error) { + return map[string]any{"device": device, "path": path, "bytes_written": len(newText)}, nil +} func TestMCPInitializeAndListTools(t *testing.T) { srv := NewServer(&mockBackend{}, nil) @@ -75,8 +79,8 @@ func TestMCPInitializeAndListTools(t *testing.T) { t.Fatalf("expected map result") } tools, ok := resMap["tools"].([]any) - if !ok || len(tools) != 6 { - t.Fatalf("expected 6 tools, got %d", len(tools)) + if !ok || len(tools) != 11 { + t.Fatalf("expected 11 tools, got %d", len(tools)) } } @@ -110,3 +114,47 @@ func TestMCPCallTool(t *testing.T) { t.Fatalf("expected stdout in result, got %s", contentStr) } } + +type gatedMockBackend struct { + mockBackend + deniedAction string +} + +func (g *gatedMockBackend) AuthorizeMCP(device, action, clientIP string) error { + if action == g.deniedAction { + return fmt.Errorf("denied %s", action) + } + return nil +} + +func (g *gatedMockBackend) ListAgentMCP(ctx context.Context, device string) (any, error) { + return map[string]any{"device": device}, nil +} + +func (g *gatedMockBackend) CallAgentMCP(ctx context.Context, clientIP, device, server, tool string, args any) (any, error) { + return map[string]any{"device": device, "server": server, "tool": tool}, nil +} + +func TestMCPAgentCallUsesSecurityGate(t *testing.T) { + backend := &gatedMockBackend{deniedAction: "mcp"} + srv := NewServer(backend, nil) + params := json.RawMessage("{\"name\":\"call_agent_mcp\",\"arguments\":{\"device\":\"node1\",\"server\":\"s\",\"tool\":\"t\"}}") + req := JSONRPCRequest{JSONRPC: "2.0", ID: 10, Method: "tools/call", Params: params} + resp := srv.HandleDirectRequest(context.Background(), req) + result, ok := resp.Result.(CallToolResult) + if !ok || !result.IsError || !strings.Contains(result.Content[0].Text, "denied mcp") { + t.Fatalf("expected MCP denial, got %#v", resp) + } +} + +func TestMCPListAgentUsesReadGate(t *testing.T) { + backend := &gatedMockBackend{deniedAction: "read"} + srv := NewServer(backend, nil) + params := json.RawMessage("{\"name\":\"list_agent_mcp\",\"arguments\":{\"device\":\"node1\"}}") + req := JSONRPCRequest{JSONRPC: "2.0", ID: 11, Method: "tools/call", Params: params} + resp := srv.HandleDirectRequest(context.Background(), req) + result, ok := resp.Result.(CallToolResult) + if !ok || !result.IsError || !strings.Contains(result.Content[0].Text, "denied read") { + t.Fatalf("expected MCP list denial, got %#v", resp) + } +} diff --git a/internal/storage/storage.go b/internal/storage/storage.go index 61d5032..f4a1657 100644 --- a/internal/storage/storage.go +++ b/internal/storage/storage.go @@ -1,12 +1,15 @@ package storage import ( + "crypto/sha256" "database/sql" + "encoding/hex" "encoding/json" "errors" _ "modernc.org/sqlite" "os" "path/filepath" + "regexp" "strings" "time" ) @@ -45,7 +48,7 @@ func (s *Store) init() error { "CREATE TABLE IF NOT EXISTS audit_logs (" + "id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp INTEGER NOT NULL, caller_ip TEXT, device TEXT, action TEXT, command TEXT, " + "exit_code INTEGER, duration_ms INTEGER, error_msg TEXT); " + - "CREATE TABLE IF NOT EXISTS revoked_devices (name TEXT PRIMARY KEY, revoked_at INTEGER NOT NULL, reason TEXT); CREATE TABLE IF NOT EXISTS security_settings (node_name TEXT PRIMARY KEY, mode TEXT NOT NULL, updated_at INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS mcp_services (id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT, transport TEXT NOT NULL, command TEXT, args_json TEXT, env_json TEXT, url TEXT, headers_json TEXT, scope TEXT NOT NULL DEFAULT 'all', target_nodes_json TEXT, enabled INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL); CREATE INDEX IF NOT EXISTS idx_mcp_services_enabled ON mcp_services(enabled);") + "CREATE TABLE IF NOT EXISTS revoked_devices (name TEXT PRIMARY KEY, revoked_at INTEGER NOT NULL, reason TEXT); CREATE TABLE IF NOT EXISTS device_credentials (name TEXT PRIMARY KEY, token_hash TEXT NOT NULL, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS security_settings (node_name TEXT PRIMARY KEY, mode TEXT NOT NULL, updated_at INTEGER NOT NULL); CREATE TABLE IF NOT EXISTS security_policy_meta (id INTEGER PRIMARY KEY CHECK(id=1), version INTEGER NOT NULL); INSERT OR IGNORE INTO security_policy_meta(id,version) VALUES(1,1); CREATE TABLE IF NOT EXISTS mcp_services (id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT, transport TEXT NOT NULL, command TEXT, args_json TEXT, env_json TEXT, url TEXT, headers_json TEXT, scope TEXT NOT NULL DEFAULT 'all', target_nodes_json TEXT, enabled INTEGER NOT NULL DEFAULT 1, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL); CREATE INDEX IF NOT EXISTS idx_mcp_services_enabled ON mcp_services(enabled);") if err != nil { return err } @@ -276,9 +279,20 @@ type AuditLog struct { ErrorMsg string `json:"error_msg"` } +var auditSecretRE = regexp.MustCompile(`(?i)(authorization|bearer|token|api[_-]?key|password|secret|cookie)(\s*[:=]\s+)([^\s,;]+)`) + +func redactAudit(v string) string { + v = auditSecretRE.ReplaceAllString(v, `$1$2[REDACTED]`) + v = strings.ReplaceAll(v, "\r", " ") + v = strings.ReplaceAll(v, "\n", " ") + if len(v) > 8192 { + v = v[:8192] + } + return v +} func (s *Store) Audit(ip, device, action, command string, code int, ms int64, msg string) error { _, err := s.DB.Exec("INSERT INTO audit_logs(timestamp,caller_ip,device,action,command,exit_code,duration_ms,error_msg) VALUES(?,?,?,?,?,?,?,?)", - time.Now().Unix(), ip, device, action, command, code, ms, msg) + time.Now().Unix(), ip, device, action, redactAudit(command), code, ms, redactAudit(msg)) return err } @@ -387,6 +401,59 @@ func normalizeSecurityMode(mode string) string { return "" } +type SecurityPolicyState struct { + GlobalMode string + Nodes map[string]string + Version uint64 +} + +func (s *Store) LoadSecurityPolicy() (SecurityPolicyState, error) { + state := SecurityPolicyState{GlobalMode: "medium", Nodes: map[string]string{}, Version: 1} + if err := s.DB.QueryRow("SELECT mode FROM security_settings WHERE node_name='' LIMIT 1").Scan(&state.GlobalMode); err != nil && err != sql.ErrNoRows { + return state, err + } + if m := normalizeSecurityMode(state.GlobalMode); m != "" && m != "inherit" { + state.GlobalMode = m + } else { + state.GlobalMode = "medium" + } + var version int64 + if err := s.DB.QueryRow("SELECT version FROM security_policy_meta WHERE id=1").Scan(&version); err != nil { + return state, err + } + if version > 0 { + state.Version = uint64(version) + } + rows, err := s.DB.Query("SELECT node_name,mode FROM security_settings WHERE node_name<>''") + if err != nil { + return state, err + } + defer rows.Close() + for rows.Next() { + var node, mode string + if err := rows.Scan(&node, &mode); err != nil { + return state, err + } + if m := normalizeSecurityMode(mode); m != "" { + state.Nodes[node] = m + } + } + return state, rows.Err() +} + +func (s *Store) bumpSecurityPolicyVersion(tx *sql.Tx) error { + _, err := tx.Exec("UPDATE security_policy_meta SET version=version+1 WHERE id=1") + return err +} + +func (s *Store) SecurityPolicyVersion() uint64 { + var version int64 + if err := s.DB.QueryRow("SELECT version FROM security_policy_meta WHERE id=1").Scan(&version); err != nil || version < 1 { + return 1 + } + return uint64(version) +} + func (s *Store) GetGlobalSecurityMode() string { var mode string err := s.DB.QueryRow("SELECT mode FROM security_settings WHERE node_name='' LIMIT 1").Scan(&mode) @@ -404,8 +471,18 @@ func (s *Store) SetGlobalSecurityMode(mode string) error { if mode == "" || mode == "inherit" { return errors.New("invalid global security mode") } - _, err := s.DB.Exec("INSERT INTO security_settings(node_name,mode,updated_at) VALUES('',?,?) ON CONFLICT(node_name) DO UPDATE SET mode=excluded.mode,updated_at=excluded.updated_at", mode, time.Now().Unix()) - return err + tx, err := s.DB.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if _, err = tx.Exec("INSERT INTO security_settings(node_name,mode,updated_at) VALUES('',?,?) ON CONFLICT(node_name) DO UPDATE SET mode=excluded.mode,updated_at=excluded.updated_at", mode, time.Now().Unix()); err != nil { + return err + } + if err = s.bumpSecurityPolicyVersion(tx); err != nil { + return err + } + return tx.Commit() } func (s *Store) GetNodeSecurityMode(node string) (string, bool) { @@ -429,8 +506,18 @@ func (s *Store) SetNodeSecurityMode(node, mode string) error { if mode == "" { return errors.New("invalid security mode") } - _, err := s.DB.Exec("INSERT INTO security_settings(node_name,mode,updated_at) VALUES(?,?,?) ON CONFLICT(node_name) DO UPDATE SET mode=excluded.mode,updated_at=excluded.updated_at", node, mode, time.Now().Unix()) - return err + tx, err := s.DB.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if _, err = tx.Exec("INSERT INTO security_settings(node_name,mode,updated_at) VALUES(?,?,?) ON CONFLICT(node_name) DO UPDATE SET mode=excluded.mode,updated_at=excluded.updated_at", node, mode, time.Now().Unix()); err != nil { + return err + } + if err = s.bumpSecurityPolicyVersion(tx); err != nil { + return err + } + return tx.Commit() } type SecuritySetting struct { @@ -454,3 +541,55 @@ func (s *Store) ListNodeSecurityModes() ([]SecuritySetting, error) { } return out, rows.Err() } + +func hashDeviceToken(token string) string { + h := sha256.Sum256([]byte(token)) + return hex.EncodeToString(h[:]) +} + +// SetDeviceCredential stores only a SHA-256 hash of the per-device token. +func (s *Store) DeviceCredentialConfigured(name string) (bool, error) { + return s.HasDeviceCredential(name) +} + +func (s *Store) SetDeviceCredential(name, token string) error { + if strings.TrimSpace(name) == "" || token == "" { + return errors.New("device name and token required") + } + now := time.Now().Unix() + _, err := s.DB.Exec(`INSERT INTO device_credentials(name,token_hash,created_at,updated_at) +VALUES(?,?,?,?) ON CONFLICT(name) DO UPDATE SET token_hash=excluded.token_hash,updated_at=excluded.updated_at`, name, hashDeviceToken(token), now, now) + return err +} + +func (s *Store) VerifyDeviceCredential(name, token string) (bool, error) { + if name == "" || token == "" { + return false, nil + } + var stored string + err := s.DB.QueryRow("SELECT token_hash FROM device_credentials WHERE name=?", name).Scan(&stored) + if errors.Is(err, sql.ErrNoRows) { + return false, nil + } + if err != nil { + return false, err + } + return subtleConstantTimeEqual(stored, hashDeviceToken(token)), nil +} + +func subtleConstantTimeEqual(a, b string) bool { + if len(a) != len(b) { + return false + } + var v byte + for i := range a { + v |= a[i] ^ b[i] + } + return v == 0 +} + +func (s *Store) HasDeviceCredential(name string) (bool, error) { + var count int + err := s.DB.QueryRow("SELECT COUNT(*) FROM device_credentials WHERE name=?", name).Scan(&count) + return count > 0, err +}