openclaw/vps-commander
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

feat: finalize dc dual-channel security rollout

Browse Source
This commit is contained in:
kory
2026-10-02 13:23:25 +08:00
parent a5620cb9c4
commit 192969d7f6
11 changed files with 1964 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/api/api.go
+404 -23
View File
@@ -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)
internal/api/mcp_adapter.go
+136
View File
@@ -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()
}
internal/cluster/manager.go
+202 -25
View File
@@ -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 {
internal/cluster/protocol.go
+96 -15
View File
@@ -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"`
}
internal/executor/fs.go
+217
View File
@@ -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
internal/mcp/mcp.go
+97 -1
View File
@@ -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
internal/mcp/mcp_test.go
+50 -2
View File
@@ -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)
}
}
internal/storage/storage.go
+145 -6
View File
@@ -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
}