From 343708eebe9e0a543800a7b3643f74b76b56b62e Mon Sep 17 00:00:00 2001 From: kory Date: Fri, 2 Oct 2026 14:10:56 +0800 Subject: [PATCH] fix: repair session API and local session fallback --- cmd/hub/main.go | 8 +++ internal/api/api.go | 129 ++++++++++++++++++++++++++++++++++---------- 2 files changed, 109 insertions(+), 28 deletions(-) diff --git a/cmd/hub/main.go b/cmd/hub/main.go index 1c663cb..e6a9ef6 100644 --- a/cmd/hub/main.go +++ b/cmd/hub/main.go @@ -70,6 +70,7 @@ func main() { AdminToken: adminToken, ExecLimiter: make(chan struct{}, 32), SearchLimiter: make(chan struct{}, 8), + LocalSessions: executor.NewSessionManager(0), } if err := s.LoadSecurityPolicySnapshot(); err != nil { log.Fatalf("load security policy snapshot: %v", err) @@ -121,6 +122,13 @@ func main() { IdleTimeout: 90 * time.Second, } log.Printf("vps-commander-hub listening on %s as %s", *addr, local) + go func() { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for range ticker.C { + s.LocalSessions.Cleanup() + } + }() go func() { if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatal(err) diff --git a/internal/api/api.go b/internal/api/api.go index c6979c6..dce950c 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -49,6 +49,7 @@ type Server struct { AdminToken string ExecLimiter chan struct{} SearchLimiter chan struct{} + LocalSessions *executor.SessionManager } const ( @@ -782,9 +783,13 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) { return } var req struct { - Device, SessionID, Command, Workdir, Data string - Force bool - RingCapacity int `json:"ring_capacity"` + Device string `json:"device"` + SessionID string `json:"session_id"` + Command string `json:"command"` + Workdir string `json:"workdir"` + Data string `json:"data"` + Force bool `json:"force"` + 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) @@ -796,38 +801,106 @@ func (s *Server) processSession(w http.ResponseWriter, r *http.Request) { 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") + owner := sessionOwner(r) 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 + if s.isRemote(req.Device) { + if s.Cluster == nil || !s.Cluster.Online(req.Device) { + http.Error(w, "device offline", 409) + return + } + 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: owner, RingCapacity: req.RingCapacity}) + case "read": + result, err = s.Cluster.ReadProcessOutput(ctx, req.Device, req.SessionID, owner) + case "input": + err = s.Cluster.InteractProcess(ctx, req.Device, req.SessionID, req.Data, owner) + result = map[string]any{"ok": true} + case "stop": + err = s.Cluster.ForceTerminate(ctx, req.Device, req.SessionID, owner) + result = map[string]any{"ok": true} + case "list": + result, err = s.Cluster.ListSessions(ctx, req.Device) + default: + http.Error(w, "unknown action", 400) + return + } + } else { + if s.LocalSessions == nil { + http.Error(w, "local session manager unavailable", 503) + return + } + switch action { + case "start": + if req.SessionID == "" { + req.SessionID, err = newSessionID() + if err != nil { + http.Error(w, "session id generation failed", 500) + return + } + } + var sess *executor.Session + sess, err = s.LocalSessions.Start(ctx, req.SessionID, req.Command, req.Workdir, owner, req.RingCapacity) + if err == nil { + result = map[string]any{"session_id": sess.ID, "started_at": sess.StartedAt.Unix()} + } + case "read": + var sess *executor.Session + sess, ok := s.LocalSessions.Get(req.SessionID) + if !ok { + err = fmt.Errorf("session not found") + break + } + if !sess.Authorized(owner) { + err = fmt.Errorf("session ownership denied") + break + } + code, done, _ := sess.Status() + stdout, stderr := sess.Output() + result = cluster.ProcessOutputPayload{SessionID: req.SessionID, Stdout: stdout, Stderr: stderr, ExitCode: code, Running: !done} + case "input": + var sess *executor.Session + sess, ok := s.LocalSessions.Get(req.SessionID) + if !ok { + err = fmt.Errorf("session not found") + break + } + if !sess.Authorized(owner) { + err = fmt.Errorf("session ownership denied") + break + } + err = sess.Stdin([]byte(req.Data)) + result = map[string]any{"ok": true} + case "stop": + var sess *executor.Session + sess, ok := s.LocalSessions.Get(req.SessionID) + if !ok { + err = fmt.Errorf("session not found") + break + } + if !sess.Authorized(owner) { + err = fmt.Errorf("session ownership denied") + break + } + err = s.LocalSessions.Stop(req.SessionID, true) + result = map[string]any{"ok": true} + case "list": + result = s.LocalSessions.List() + default: + http.Error(w, "unknown action", 400) + 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)