From 2ac44ce937d5a93e588618b8a1d73e79bc4b0831 Mon Sep 17 00:00:00 2001 From: viettranx Date: Thu, 26 Feb 2026 22:47:59 +0700 Subject: [PATCH] feat: enhance filesystem security with symlink/hardlink protection, implement typing indicators, and improve scheduler queue management for stale requests. --- internal/channels/discord/discord.go | 26 +- internal/channels/telegram/channel.go | 1 + internal/channels/telegram/handlers.go | 25 +- internal/channels/telegram/send.go | 6 + internal/channels/typing/controller.go | 204 ++++++++++++++ internal/channels/typing/controller_test.go | 186 +++++++++++++ internal/sandbox/fsbridge.go | 15 +- internal/scheduler/errors.go | 12 + internal/scheduler/queue.go | 105 ++++++-- internal/scheduler/scheduler.go | 24 ++ internal/tools/boundary_test.go | 282 ++++++++++++++++++++ internal/tools/filesystem.go | 140 ++++++++-- 12 files changed, 968 insertions(+), 58 deletions(-) create mode 100644 internal/channels/typing/controller.go create mode 100644 internal/channels/typing/controller_test.go create mode 100644 internal/tools/boundary_test.go diff --git a/internal/channels/discord/discord.go b/internal/channels/discord/discord.go index 90520e51..3d4e7d02 100644 --- a/internal/channels/discord/discord.go +++ b/internal/channels/discord/discord.go @@ -5,11 +5,13 @@ import ( "fmt" "log/slog" "sync" + "time" "github.com/bwmarrin/discordgo" "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/channels/typing" "github.com/nextlevelbuilder/goclaw/internal/config" ) @@ -20,6 +22,7 @@ type Channel struct { config config.DiscordConfig botUserID string // populated on start placeholders sync.Map // channelID string → messageID string + typingCtrls sync.Map // channelID string → *typing.Controller } // New creates a new Discord channel from config. @@ -85,6 +88,11 @@ func (c *Channel) Send(_ context.Context, msg bus.OutboundMessage) error { return fmt.Errorf("empty chat ID for discord send") } + // Stop typing indicator controller + if ctrl, ok := c.typingCtrls.LoadAndDelete(channelID); ok { + ctrl.(*typing.Controller).Stop() + } + content := msg.Content // NO_REPLY cleanup: content is empty when agent suppresses reply. @@ -208,8 +216,22 @@ func (c *Channel) handleMessage(_ *discordgo.Session, m *discordgo.MessageCreate "preview", channels.Truncate(content, 50), ) - // Send typing indicator - _ = c.session.ChannelTyping(channelID) + // Send typing indicator with keepalive + TTL safety net. + // Discord typing expires after 10s, so keepalive every 9s. + // TTL auto-stops after 60s to prevent stuck indicators. + typingCtrl := typing.New(typing.Options{ + MaxDuration: 60 * time.Second, + KeepaliveInterval: 9 * time.Second, + StartFn: func() error { + return c.session.ChannelTyping(channelID) + }, + }) + // Stop previous typing controller for this channel (if any) + if prev, ok := c.typingCtrls.Load(channelID); ok { + prev.(*typing.Controller).Stop() + } + c.typingCtrls.Store(channelID, typingCtrl) + typingCtrl.Start() // Send placeholder "Thinking..." message placeholder, err := c.session.ChannelMessageSend(channelID, "Thinking...") diff --git a/internal/channels/telegram/channel.go b/internal/channels/telegram/channel.go index 4753182f..006a61be 100644 --- a/internal/channels/telegram/channel.go +++ b/internal/channels/telegram/channel.go @@ -28,6 +28,7 @@ type Channel struct { teamStore store.TeamStore // for /tasks, /task_detail commands (nil in standalone) placeholders sync.Map // localKey string → messageID int stopThinking sync.Map // localKey string → *thinkingCancel + typingCtrls sync.Map // localKey string → *typing.Controller streams sync.Map // localKey string → *DraftStream (streaming preview) reactions sync.Map // localKey string → *StatusReactionController pairingReplySent sync.Map // userID string → time.Time (debounce pairing replies) diff --git a/internal/channels/telegram/handlers.go b/internal/channels/telegram/handlers.go index 1533e60a..c21e0c1c 100644 --- a/internal/channels/telegram/handlers.go +++ b/internal/channels/telegram/handlers.go @@ -12,6 +12,7 @@ import ( "github.com/nextlevelbuilder/goclaw/internal/bus" "github.com/nextlevelbuilder/goclaw/internal/channels" + "github.com/nextlevelbuilder/goclaw/internal/channels/typing" ) // handleMessage processes an incoming Telegram update. @@ -286,13 +287,27 @@ func (c *Channel) handleMessage(ctx context.Context, update telego.Update) { } } - // Send typing indicator (TS ref: buildTypingThreadParams — General topic ID=1 is OK for typing). + // Send typing indicator with keepalive + TTL safety net. + // Telegram typing expires after 5s, so keepalive every 4s. + // TTL auto-stops after 60s to prevent stuck indicators. chatIDObj := tu.ID(chatID) - typingAction := tu.ChatAction(chatIDObj, telego.ChatActionTyping) - if messageThreadID > 0 { - typingAction.MessageThreadID = messageThreadID + typingCtrl := typing.New(typing.Options{ + MaxDuration: 60 * time.Second, + KeepaliveInterval: 4 * time.Second, + StartFn: func() error { + action := tu.ChatAction(chatIDObj, telego.ChatActionTyping) + if messageThreadID > 0 { + action.MessageThreadID = messageThreadID + } + return c.bot.SendChatAction(ctx, action) + }, + }) + // Stop previous typing controller for this chat/topic (if any) + if prev, ok := c.typingCtrls.Load(localKey); ok { + prev.(*typing.Controller).Stop() } - _ = c.bot.SendChatAction(ctx, typingAction) + c.typingCtrls.Store(localKey, typingCtrl) + typingCtrl.Start() // Stop previous thinking animation for this chat/topic if prevStop, ok := c.stopThinking.Load(localKey); ok { diff --git a/internal/channels/telegram/send.go b/internal/channels/telegram/send.go index 1a6ee182..ff5ec4bc 100644 --- a/internal/channels/telegram/send.go +++ b/internal/channels/telegram/send.go @@ -12,6 +12,7 @@ import ( tu "github.com/mymmrac/telego/telegoutil" "github.com/nextlevelbuilder/goclaw/internal/bus" + "github.com/nextlevelbuilder/goclaw/internal/channels/typing" ) // Error patterns for graceful handling (matching TS error constants in send.ts). @@ -57,6 +58,11 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { c.stopThinking.Delete(localKey) } + // Stop typing indicator controller (TTL keepalive) + if ctrl, ok := c.typingCtrls.LoadAndDelete(localKey); ok { + ctrl.(*typing.Controller).Stop() + } + // NO_REPLY cleanup: content is empty when agent suppresses reply (prompt injection, etc.). // Clean up placeholder, then return without sending any message. if msg.Content == "" && len(msg.Media) == 0 { diff --git a/internal/channels/typing/controller.go b/internal/channels/typing/controller.go new file mode 100644 index 00000000..b0ae0cf6 --- /dev/null +++ b/internal/channels/typing/controller.go @@ -0,0 +1,204 @@ +// Package typing provides a channel-agnostic typing indicator controller +// with TTL safety net, keepalive support, and post-close guards. +// +// The controller prevents stuck typing indicators by: +// - Auto-stopping after a configurable TTL (default 60s) +// - Requiring both MarkRunComplete + MarkDispatchIdle for graceful cleanup +// - Guarding against post-close keepalive invocations +package typing + +import ( + "log/slog" + "sync" + "time" +) + +// Options configures a typing indicator controller. +type Options struct { + // MaxDuration is the TTL safety net. If the indicator hasn't been + // stopped after this duration, it auto-stops. 0 disables the TTL. + // Default: 60s. + MaxDuration time.Duration + + // KeepaliveInterval is how often to re-send the typing action. + // Telegram typing expires after 5s, Discord after 10s. + // 0 disables keepalive (single fire-and-forget). + KeepaliveInterval time.Duration + + // StartFn sends the channel-specific typing indicator. + // Called on Start() and on each keepalive tick. + StartFn func() error + + // StopFn sends the channel-specific stop-typing signal. + // Optional — some channels (Telegram) auto-stop on message send. + StopFn func() error +} + +// Controller manages the lifecycle of a typing indicator. +// It is safe for concurrent use. +type Controller struct { + mu sync.Mutex + + // State flags + closed bool // post-close guard: prevents stale startFn calls + runComplete bool // signal 1: agent finished processing + dispatchIdle bool // signal 2: message delivery finished + stopSent bool // prevents duplicate stopFn calls + + // Configuration + maxDuration time.Duration + keepaliveInterval time.Duration + startFn func() error + stopFn func() error + + // Timers + ttlTimer *time.Timer + keepaliveDone chan struct{} +} + +// New creates a typing controller with the given options. +func New(opts Options) *Controller { + maxDur := opts.MaxDuration + if maxDur == 0 { + maxDur = 60 * time.Second + } + return &Controller{ + maxDuration: maxDur, + keepaliveInterval: opts.KeepaliveInterval, + startFn: opts.StartFn, + stopFn: opts.StopFn, + } +} + +// Start begins the typing indicator, TTL timer, and keepalive loop. +func (c *Controller) Start() { + c.mu.Lock() + defer c.mu.Unlock() + + if c.closed { + return + } + + // Fire initial typing action + c.fireStart() + + // Start TTL safety net + if c.maxDuration > 0 { + c.ttlTimer = time.AfterFunc(c.maxDuration, func() { + c.mu.Lock() + if !c.closed { + slog.Debug("typing: TTL exceeded, auto-stopping", "ttl", c.maxDuration) + c.forceStop() + } + c.mu.Unlock() + }) + } + + // Start keepalive loop + if c.keepaliveInterval > 0 { + c.keepaliveDone = make(chan struct{}) + go c.keepaliveLoop() + } +} + +// Stop forcefully stops the typing indicator and all timers. +// Safe to call multiple times (idempotent). +func (c *Controller) Stop() { + c.mu.Lock() + defer c.mu.Unlock() + c.forceStop() +} + +// MarkRunComplete signals that the agent has finished processing. +// Cleanup happens only when both MarkRunComplete and MarkDispatchIdle have been called. +func (c *Controller) MarkRunComplete() { + c.mu.Lock() + defer c.mu.Unlock() + c.runComplete = true + c.tryCleanup() +} + +// MarkDispatchIdle signals that message delivery has completed. +// Cleanup happens only when both MarkRunComplete and MarkDispatchIdle have been called. +func (c *Controller) MarkDispatchIdle() { + c.mu.Lock() + defer c.mu.Unlock() + c.dispatchIdle = true + c.tryCleanup() +} + +// tryCleanup runs cleanup only when both completion signals have been received. +// Must be called with c.mu held. +func (c *Controller) tryCleanup() { + if c.runComplete && c.dispatchIdle && !c.closed { + c.forceStop() + } +} + +// forceStop cancels all timers and sends stop signal. +// Must be called with c.mu held. +func (c *Controller) forceStop() { + if c.closed { + return + } + c.closed = true + + // Cancel TTL timer + if c.ttlTimer != nil { + c.ttlTimer.Stop() + c.ttlTimer = nil + } + + // Stop keepalive loop + if c.keepaliveDone != nil { + close(c.keepaliveDone) + c.keepaliveDone = nil + } + + // Send stop signal + c.fireStop() +} + +// fireStart invokes the channel-specific start function. +// Must be called with c.mu held. +func (c *Controller) fireStart() { + if c.closed || c.startFn == nil { + return + } + if err := c.startFn(); err != nil { + slog.Debug("typing: startFn error", "error", err) + } +} + +// fireStop invokes the channel-specific stop function (once). +// Must be called with c.mu held. +func (c *Controller) fireStop() { + if c.stopSent || c.stopFn == nil { + return + } + c.stopSent = true + if err := c.stopFn(); err != nil { + slog.Debug("typing: stopFn error", "error", err) + } +} + +// keepaliveLoop periodically re-sends the typing indicator. +func (c *Controller) keepaliveLoop() { + ticker := time.NewTicker(c.keepaliveInterval) + defer ticker.Stop() + + for { + select { + case <-c.keepaliveDone: + return + case <-ticker.C: + c.mu.Lock() + if c.closed { + c.mu.Unlock() + return + } + c.fireStart() + c.mu.Unlock() + } + } +} diff --git a/internal/channels/typing/controller_test.go b/internal/channels/typing/controller_test.go new file mode 100644 index 00000000..020b8887 --- /dev/null +++ b/internal/channels/typing/controller_test.go @@ -0,0 +1,186 @@ +package typing + +import ( + "sync/atomic" + "testing" + "time" +) + +func TestTTLAutoStop(t *testing.T) { + var startCount, stopCount atomic.Int32 + + ctrl := New(Options{ + MaxDuration: 100 * time.Millisecond, + KeepaliveInterval: 0, + StartFn: func() error { startCount.Add(1); return nil }, + StopFn: func() error { stopCount.Add(1); return nil }, + }) + + ctrl.Start() + + // Should have started once + if startCount.Load() != 1 { + t.Fatalf("expected 1 start call, got %d", startCount.Load()) + } + + // Wait for TTL to expire + time.Sleep(200 * time.Millisecond) + + if stopCount.Load() != 1 { + t.Fatalf("expected 1 stop call after TTL, got %d", stopCount.Load()) + } + + // Verify closed + ctrl.mu.Lock() + closed := ctrl.closed + ctrl.mu.Unlock() + if !closed { + t.Fatal("expected controller to be closed after TTL") + } +} + +func TestPostCloseGuard(t *testing.T) { + var startCount atomic.Int32 + + ctrl := New(Options{ + MaxDuration: 10 * time.Second, + StartFn: func() error { startCount.Add(1); return nil }, + }) + + ctrl.Start() + if startCount.Load() != 1 { + t.Fatalf("expected 1 start call, got %d", startCount.Load()) + } + + ctrl.Stop() + + // Start after stop should be no-op + ctrl.Start() + if startCount.Load() != 1 { + t.Fatalf("expected no additional start calls after Stop, got %d", startCount.Load()) + } +} + +func TestDualSignalsRequired(t *testing.T) { + var stopCount atomic.Int32 + + ctrl := New(Options{ + MaxDuration: 10 * time.Second, + StartFn: func() error { return nil }, + StopFn: func() error { stopCount.Add(1); return nil }, + }) + + ctrl.Start() + + // Only run complete — should NOT stop + ctrl.MarkRunComplete() + time.Sleep(20 * time.Millisecond) + if stopCount.Load() != 0 { + t.Fatal("expected no stop after only MarkRunComplete") + } + + // Now dispatch idle — should trigger cleanup + ctrl.MarkDispatchIdle() + time.Sleep(20 * time.Millisecond) + if stopCount.Load() != 1 { + t.Fatalf("expected 1 stop after both signals, got %d", stopCount.Load()) + } +} + +func TestDualSignalsReverseOrder(t *testing.T) { + var stopCount atomic.Int32 + + ctrl := New(Options{ + MaxDuration: 10 * time.Second, + StartFn: func() error { return nil }, + StopFn: func() error { stopCount.Add(1); return nil }, + }) + + ctrl.Start() + + // Dispatch idle first — should NOT stop + ctrl.MarkDispatchIdle() + time.Sleep(20 * time.Millisecond) + if stopCount.Load() != 0 { + t.Fatal("expected no stop after only MarkDispatchIdle") + } + + // Now run complete — should trigger cleanup + ctrl.MarkRunComplete() + time.Sleep(20 * time.Millisecond) + if stopCount.Load() != 1 { + t.Fatalf("expected 1 stop after both signals, got %d", stopCount.Load()) + } +} + +func TestKeepalive(t *testing.T) { + var startCount atomic.Int32 + + ctrl := New(Options{ + MaxDuration: 2 * time.Second, + KeepaliveInterval: 30 * time.Millisecond, + StartFn: func() error { startCount.Add(1); return nil }, + }) + + ctrl.Start() + time.Sleep(120 * time.Millisecond) + ctrl.Stop() + + // Initial start + at least 2 keepalive ticks + count := startCount.Load() + if count < 3 { + t.Fatalf("expected at least 3 start calls (1 initial + keepalive ticks), got %d", count) + } +} + +func TestKeepaliveStopsAfterClose(t *testing.T) { + var startCount atomic.Int32 + + ctrl := New(Options{ + MaxDuration: 2 * time.Second, + KeepaliveInterval: 20 * time.Millisecond, + StartFn: func() error { startCount.Add(1); return nil }, + }) + + ctrl.Start() + time.Sleep(60 * time.Millisecond) + ctrl.Stop() + countAtStop := startCount.Load() + + // Wait to verify no more keepalive ticks + time.Sleep(80 * time.Millisecond) + countAfter := startCount.Load() + + if countAfter != countAtStop { + t.Fatalf("expected no start calls after Stop, got %d more", countAfter-countAtStop) + } +} + +func TestStopIdempotent(t *testing.T) { + var stopCount atomic.Int32 + + ctrl := New(Options{ + MaxDuration: 10 * time.Second, + StartFn: func() error { return nil }, + StopFn: func() error { stopCount.Add(1); return nil }, + }) + + ctrl.Start() + ctrl.Stop() + ctrl.Stop() + ctrl.Stop() + + if stopCount.Load() != 1 { + t.Fatalf("expected exactly 1 stop call, got %d", stopCount.Load()) + } +} + +func TestNilFunctions(t *testing.T) { + // Should not panic with nil start/stop functions + ctrl := New(Options{ + MaxDuration: 50 * time.Millisecond, + }) + ctrl.Start() + time.Sleep(100 * time.Millisecond) + ctrl.Stop() +} diff --git a/internal/sandbox/fsbridge.go b/internal/sandbox/fsbridge.go index 59f6e59d..951f1378 100644 --- a/internal/sandbox/fsbridge.go +++ b/internal/sandbox/fsbridge.go @@ -11,6 +11,7 @@ import ( "context" "fmt" "os/exec" + "path/filepath" "strings" ) @@ -104,16 +105,22 @@ func (b *FsBridge) Stat(ctx context.Context, path string) (string, error) { } // resolvePath resolves a path relative to the container workdir. +// Validates that absolute paths stay within the workdir (defense in depth). func (b *FsBridge) resolvePath(path string) string { if path == "" || path == "." { return b.workdir } - // Absolute paths pass through if strings.HasPrefix(path, "/") { - return path + // Validate absolute paths stay within workdir (defense in depth, + // container is already sandboxed with read-only FS + cap-drop ALL). + cleaned := filepath.Clean(path) + if cleaned == b.workdir || strings.HasPrefix(cleaned, b.workdir+"/") { + return cleaned + } + return b.workdir // fallback to workdir for escapes } - // Relative paths resolve against workdir - return b.workdir + "/" + path + // Relative paths: use filepath.Join for proper normalization + return filepath.Clean(filepath.Join(b.workdir, path)) } // dockerExec runs a command inside the container and returns stdout, stderr, exit code. diff --git a/internal/scheduler/errors.go b/internal/scheduler/errors.go index d454537f..c4616ff7 100644 --- a/internal/scheduler/errors.go +++ b/internal/scheduler/errors.go @@ -8,4 +8,16 @@ var ( // ErrQueueDropped is returned when a queued message is evicted to make room (drop=old policy). ErrQueueDropped = errors.New("message dropped from queue") + + // ErrMessageStale is returned when a queued message is skipped because it was + // enqueued before an abort (/stopall) and is no longer relevant. + ErrMessageStale = errors.New("message stale: enqueued before abort") + + // ErrGatewayDraining is returned when the gateway is shutting down and + // new requests cannot be accepted. + ErrGatewayDraining = errors.New("gateway is shutting down, please retry shortly") + + // ErrLaneCleared is returned when a session queue is reset (e.g. during restart) + // and all pending requests are drained. + ErrLaneCleared = errors.New("session queue cleared during reset") ) diff --git a/internal/scheduler/queue.go b/internal/scheduler/queue.go index c65b9280..31c6ea27 100644 --- a/internal/scheduler/queue.go +++ b/internal/scheduler/queue.go @@ -2,6 +2,7 @@ package scheduler import ( "context" + "log/slog" "sync" "time" @@ -61,8 +62,9 @@ type TokenEstimateFunc func(sessionKey string) (tokens int, contextWindow int) // PendingRequest is a queued agent run awaiting execution. type PendingRequest struct { - Req agent.RunRequest - ResultCh chan RunOutcome + Req agent.RunRequest + ResultCh chan RunOutcome + EnqueuedAt time.Time // timestamp when enqueued, used for stale message detection } // RunOutcome is the result of a scheduled agent run. @@ -71,6 +73,12 @@ type RunOutcome struct { Err error } +// activeRunEntry tracks a running agent execution with its generation. +type activeRunEntry struct { + cancel context.CancelFunc + generation uint64 +} + // SessionQueue manages agent runs for a single session key. // Supports configurable concurrency: 1 (serial) or N (concurrent). type SessionQueue struct { @@ -80,13 +88,15 @@ type SessionQueue struct { laneMgr *LaneManager lane string - mu sync.Mutex - queue []*PendingRequest - activeRuns map[string]context.CancelFunc // runID → cancel - activeOrder []string // FIFO order of active runIDs - maxConcurrent int // effective limit (from config or per-session override) - timer *time.Timer // debounce timer - parentCtx context.Context // stored from first Enqueue call + mu sync.Mutex + queue []*PendingRequest + activeRuns map[string]activeRunEntry // runID → entry (with generation) + activeOrder []string // FIFO order of active runIDs + maxConcurrent int // effective limit (from config or per-session override) + timer *time.Timer // debounce timer + parentCtx context.Context // stored from first Enqueue call + abortCutoffTime time.Time // messages enqueued before this are stale + generation uint64 // bumped on Reset() to ignore stale completions tokenEstimateFn TokenEstimateFunc // optional: for adaptive throttle } @@ -103,7 +113,7 @@ func NewSessionQueue(key, lane string, cfg QueueConfig, laneMgr *LaneManager, ru runFn: runFn, laneMgr: laneMgr, lane: lane, - activeRuns: make(map[string]context.CancelFunc), + activeRuns: make(map[string]activeRunEntry), maxConcurrent: maxC, } } @@ -148,7 +158,7 @@ func (sq *SessionQueue) hasCapacity() bool { // Returns a channel that receives the result when the run completes. func (sq *SessionQueue) Enqueue(ctx context.Context, req agent.RunRequest) <-chan RunOutcome { outcome := make(chan RunOutcome, 1) - pending := &PendingRequest{Req: req, ResultCh: outcome} + pending := &PendingRequest{Req: req, ResultCh: outcome, EnqueuedAt: time.Now()} sq.mu.Lock() defer sq.mu.Unlock() @@ -161,8 +171,8 @@ func (sq *SessionQueue) Enqueue(ctx context.Context, req agent.RunRequest) <-cha switch sq.config.Mode { case QueueModeInterrupt: // Cancel all active runs - for runID, cancel := range sq.activeRuns { - cancel() + for runID, entry := range sq.activeRuns { + entry.cancel() delete(sq.activeRuns, runID) } sq.activeOrder = nil @@ -223,8 +233,28 @@ func (sq *SessionQueue) startAvailable(ctx context.Context) { } // startOne picks the first queued request and runs it in the lane. +// Skips stale messages that were enqueued before the last abort cutoff. // Must be called with sq.mu held. func (sq *SessionQueue) startOne(ctx context.Context) { + // Skip stale messages enqueued before the last /stopall abort cutoff. + for len(sq.queue) > 0 { + head := sq.queue[0] + if !sq.abortCutoffTime.IsZero() && head.EnqueuedAt.Before(sq.abortCutoffTime) { + sq.queue = sq.queue[1:] + head.ResultCh <- RunOutcome{Err: ErrMessageStale} + close(head.ResultCh) + slog.Debug("scheduler: skipped stale message", + "session", sq.key, + "enqueued", head.EnqueuedAt, + "cutoff", sq.abortCutoffTime, + ) + continue + } + // Clear cutoff once a non-stale message is found + sq.abortCutoffTime = time.Time{} + break + } + if len(sq.queue) == 0 { return } @@ -234,7 +264,7 @@ func (sq *SessionQueue) startOne(ctx context.Context) { runID := pending.Req.RunID runCtx, cancel := context.WithCancel(ctx) - sq.activeRuns[runID] = cancel + sq.activeRuns[runID] = activeRunEntry{cancel: cancel, generation: sq.generation} sq.activeOrder = append(sq.activeOrder, runID) lane := sq.laneMgr.Get(sq.lane) @@ -242,14 +272,16 @@ func (sq *SessionQueue) startOne(ctx context.Context) { lane = sq.laneMgr.Get(LaneMain) } + gen := sq.generation // capture generation under lock + if lane == nil { // No lane available — run directly - go sq.executeRun(runCtx, runID, pending) + go sq.executeRun(runCtx, runID, gen, pending) return } err := lane.Submit(ctx, func() { - sq.executeRun(runCtx, runID, pending) + sq.executeRun(runCtx, runID, gen, pending) }) if err != nil { pending.ResultCh <- RunOutcome{Err: err} @@ -261,14 +293,21 @@ func (sq *SessionQueue) startOne(ctx context.Context) { } // executeRun runs the agent and then starts the next queued message(s) if capacity allows. -func (sq *SessionQueue) executeRun(ctx context.Context, runID string, pending *PendingRequest) { +func (sq *SessionQueue) executeRun(ctx context.Context, runID string, runGeneration uint64, pending *PendingRequest) { result, err := sq.runFn(ctx, pending.Req) pending.ResultCh <- RunOutcome{Result: result, Err: err} close(pending.ResultCh) sq.mu.Lock() - delete(sq.activeRuns, runID) - sq.removeFromOrder(runID) + // Check generation: ignore stale completions from a previous generation. + if entry, ok := sq.activeRuns[runID]; ok && entry.generation == sq.generation { + delete(sq.activeRuns, runID) + sq.removeFromOrder(runID) + } else if runGeneration != sq.generation { + // Stale completion from old generation — skip cleanup. + sq.mu.Unlock() + return + } if sq.hasCapacity() && len(sq.queue) > 0 { // Use parentCtx (not the per-run ctx which may be cancelled) @@ -330,7 +369,7 @@ func (sq *SessionQueue) drainQueue(outcome RunOutcome) { } // CancelOne stops the oldest active run (FIFO). -// Does NOT drain the pending queue. Used by /stop command. +// Does NOT drain the pending queue or set abort cutoff. Used by /stop command. // Returns true if an active run was actually cancelled. func (sq *SessionQueue) CancelOne() bool { sq.mu.Lock() @@ -342,8 +381,8 @@ func (sq *SessionQueue) CancelOne() bool { // Cancel the oldest active run runID := sq.activeOrder[0] - if cancel, ok := sq.activeRuns[runID]; ok { - cancel() + if entry, ok := sq.activeRuns[runID]; ok { + entry.cancel() delete(sq.activeRuns, runID) sq.activeOrder = sq.activeOrder[1:] return true @@ -352,15 +391,18 @@ func (sq *SessionQueue) CancelOne() bool { } // CancelAll stops all active runs and drains all pending requests. +// Sets abort cutoff so stale queued messages are skipped on next schedule. // Used by /stopall command. // Returns true if any active run was actually cancelled. func (sq *SessionQueue) CancelAll() bool { sq.mu.Lock() defer sq.mu.Unlock() + sq.abortCutoffTime = time.Now() // mark cutoff for stale message skipping + cancelled := false - for runID, cancel := range sq.activeRuns { - cancel() + for runID, entry := range sq.activeRuns { + entry.cancel() delete(sq.activeRuns, runID) cancelled = true } @@ -395,4 +437,17 @@ func (sq *SessionQueue) QueueLen() int { return len(sq.queue) } - +// Reset bumps the generation counter, cancels all active runs, and drains +// the pending queue. Stale completions from the old generation are ignored. +// Used during in-process restart (e.g. SIGUSR1). +func (sq *SessionQueue) Reset() { + sq.mu.Lock() + defer sq.mu.Unlock() + sq.generation++ + for _, entry := range sq.activeRuns { + entry.cancel() + } + sq.activeRuns = make(map[string]activeRunEntry) + sq.activeOrder = nil + sq.drainQueue(RunOutcome{Err: ErrLaneCleared}) +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 3bac53fd..cf59767f 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -4,6 +4,7 @@ import ( "context" "log/slog" "sync" + "sync/atomic" "github.com/nextlevelbuilder/goclaw/internal/agent" ) @@ -22,6 +23,7 @@ type Scheduler struct { config QueueConfig runFn RunFunc mu sync.RWMutex + draining atomic.Bool // set during graceful shutdown to reject new requests tokenEstimateFn TokenEstimateFunc // optional: for adaptive throttle } @@ -45,15 +47,35 @@ func (s *Scheduler) SetTokenEstimateFunc(fn TokenEstimateFunc) { s.tokenEstimateFn = fn } +// MarkDraining signals that the gateway is shutting down. +// New Schedule/ScheduleWithOpts calls will return ErrGatewayDraining immediately. +// Active runs continue to completion. +func (s *Scheduler) MarkDraining() { + s.draining.Store(true) + slog.Info("scheduler: marked as draining, new requests will be rejected") +} + // Schedule submits a run request to the appropriate session queue and lane. // Returns a channel that receives the result when the run completes. func (s *Scheduler) Schedule(ctx context.Context, lane string, req agent.RunRequest) <-chan RunOutcome { + if s.draining.Load() { + ch := make(chan RunOutcome, 1) + ch <- RunOutcome{Err: ErrGatewayDraining} + close(ch) + return ch + } sq := s.getOrCreateSession(req.SessionKey, lane) return sq.Enqueue(ctx, req) } // ScheduleWithOpts submits a run request with per-session overrides. func (s *Scheduler) ScheduleWithOpts(ctx context.Context, lane string, req agent.RunRequest, opts ScheduleOpts) <-chan RunOutcome { + if s.draining.Load() { + ch := make(chan RunOutcome, 1) + ch <- RunOutcome{Err: ErrGatewayDraining} + close(ch) + return ch + } sq := s.getOrCreateSession(req.SessionKey, lane) if opts.MaxConcurrent > 0 { sq.SetMaxConcurrent(opts.MaxConcurrent) @@ -115,7 +137,9 @@ func (s *Scheduler) CancelOneSession(sessionKey string) bool { } // Stop shuts down all lanes and clears session queues. +// Automatically marks the scheduler as draining before stopping. func (s *Scheduler) Stop() { + s.MarkDraining() s.lanes.StopAll() } diff --git a/internal/tools/boundary_test.go b/internal/tools/boundary_test.go new file mode 100644 index 00000000..5c801537 --- /dev/null +++ b/internal/tools/boundary_test.go @@ -0,0 +1,282 @@ +package tools + +import ( + "os" + "path/filepath" + "runtime" + "testing" +) + +// helper to create a temp workspace with files +func setupWorkspace(t *testing.T) string { + t.Helper() + dir := t.TempDir() + // Create a normal file + if err := os.WriteFile(filepath.Join(dir, "hello.txt"), []byte("hello"), 0644); err != nil { + t.Fatal(err) + } + // Create a subdirectory + sub := filepath.Join(dir, "subdir") + if err := os.MkdirAll(sub, 0755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(sub, "nested.txt"), []byte("nested"), 0644); err != nil { + t.Fatal(err) + } + return dir +} + +func TestResolvePath_NormalFile(t *testing.T) { + ws := setupWorkspace(t) + resolved, err := resolvePath("hello.txt", ws, true) + if err != nil { + t.Fatalf("expected success, got: %v", err) + } + if filepath.Base(resolved) != "hello.txt" { + t.Fatalf("expected hello.txt, got: %s", resolved) + } +} + +func TestResolvePath_NestedFile(t *testing.T) { + ws := setupWorkspace(t) + resolved, err := resolvePath("subdir/nested.txt", ws, true) + if err != nil { + t.Fatalf("expected success, got: %v", err) + } + if filepath.Base(resolved) != "nested.txt" { + t.Fatalf("expected nested.txt, got: %s", resolved) + } +} + +func TestResolvePath_AbsolutePath(t *testing.T) { + ws := setupWorkspace(t) + absPath := filepath.Join(ws, "hello.txt") + resolved, err := resolvePath(absPath, ws, true) + if err != nil { + t.Fatalf("expected success, got: %v", err) + } + if resolved != absPath { + // canonical path might differ if ws has symlinks (e.g. /tmp on macOS) + realAbs, _ := filepath.EvalSymlinks(absPath) + if resolved != realAbs { + t.Fatalf("expected %s or %s, got: %s", absPath, realAbs, resolved) + } + } +} + +func TestResolvePath_TraversalBlocked(t *testing.T) { + ws := setupWorkspace(t) + _, err := resolvePath("../../etc/passwd", ws, true) + if err == nil { + t.Fatal("expected error for path traversal, got nil") + } +} + +func TestResolvePath_AbsoluteEscapeBlocked(t *testing.T) { + ws := setupWorkspace(t) + _, err := resolvePath("/etc/passwd", ws, true) + if err == nil { + t.Fatal("expected error for absolute path outside workspace, got nil") + } +} + +func TestResolvePath_SymlinkEscapeBlocked(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlinks require special privileges on Windows") + } + ws := setupWorkspace(t) + + // Create a file outside workspace + outside := t.TempDir() + secret := filepath.Join(outside, "secret.txt") + if err := os.WriteFile(secret, []byte("secret"), 0644); err != nil { + t.Fatal(err) + } + + // Create symlink inside workspace pointing outside + link := filepath.Join(ws, "evil_link") + if err := os.Symlink(secret, link); err != nil { + t.Fatal(err) + } + + _, err := resolvePath("evil_link", ws, true) + if err == nil { + t.Fatal("expected error for symlink escaping workspace, got nil") + } +} + +func TestResolvePath_SymlinkInsideWorkspaceAllowed(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlinks require special privileges on Windows") + } + ws := setupWorkspace(t) + + // Create symlink pointing to a file within workspace + target := filepath.Join(ws, "hello.txt") + link := filepath.Join(ws, "good_link") + if err := os.Symlink(target, link); err != nil { + t.Fatal(err) + } + + resolved, err := resolvePath("good_link", ws, true) + if err != nil { + t.Fatalf("expected success for symlink within workspace, got: %v", err) + } + + // Should resolve to canonical path of target + realTarget, _ := filepath.EvalSymlinks(target) + if resolved != realTarget { + t.Fatalf("expected %s, got: %s", realTarget, resolved) + } +} + +func TestResolvePath_BrokenSymlinkBlocked(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlinks require special privileges on Windows") + } + ws := setupWorkspace(t) + + // Create symlink pointing to non-existent file outside workspace + link := filepath.Join(ws, "broken_link") + if err := os.Symlink("/nonexistent/secret", link); err != nil { + t.Fatal(err) + } + + _, err := resolvePath("broken_link", ws, true) + if err == nil { + t.Fatal("expected error for broken symlink outside workspace, got nil") + } +} + +func TestResolvePath_DirSymlinkEscapeBlocked(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("symlinks require special privileges on Windows") + } + ws := setupWorkspace(t) + + // Create a directory symlink pointing outside workspace + outside := t.TempDir() + if err := os.WriteFile(filepath.Join(outside, "secret.txt"), []byte("secret"), 0644); err != nil { + t.Fatal(err) + } + link := filepath.Join(ws, "evil_dir") + if err := os.Symlink(outside, link); err != nil { + t.Fatal(err) + } + + _, err := resolvePath("evil_dir/secret.txt", ws, true) + if err == nil { + t.Fatal("expected error for directory symlink escape, got nil") + } +} + +func TestResolvePath_NonExistentFileInWorkspace(t *testing.T) { + ws := setupWorkspace(t) + resolved, err := resolvePath("new_file.txt", ws, true) + if err != nil { + t.Fatalf("expected success for non-existent file in workspace, got: %v", err) + } + if filepath.Dir(resolved) == "" { + t.Fatal("expected resolved path to have directory") + } +} + +func TestResolvePath_UnrestrictedAllowsEscape(t *testing.T) { + ws := setupWorkspace(t) + // restrict=false should allow any path + resolved, err := resolvePath("/etc/hosts", ws, false) + if err != nil { + t.Fatalf("expected success with restrict=false, got: %v", err) + } + if resolved != "/etc/hosts" { + t.Fatalf("expected /etc/hosts, got: %s", resolved) + } +} + +func TestCheckHardlink_NormalFile(t *testing.T) { + dir := t.TempDir() + f := filepath.Join(dir, "normal.txt") + if err := os.WriteFile(f, []byte("data"), 0644); err != nil { + t.Fatal(err) + } + if err := checkHardlink(f); err != nil { + t.Fatalf("expected no error for normal file, got: %v", err) + } +} + +func TestCheckHardlink_HardlinkedFileBlocked(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("hardlinks behave differently on Windows") + } + dir := t.TempDir() + original := filepath.Join(dir, "original.txt") + if err := os.WriteFile(original, []byte("data"), 0644); err != nil { + t.Fatal(err) + } + hardlink := filepath.Join(dir, "hardlink.txt") + if err := os.Link(original, hardlink); err != nil { + t.Fatal(err) + } + + // Both original and hardlink should be rejected (nlink=2) + if err := checkHardlink(original); err == nil { + t.Fatal("expected error for hardlinked file (original), got nil") + } + if err := checkHardlink(hardlink); err == nil { + t.Fatal("expected error for hardlinked file (link), got nil") + } +} + +func TestCheckHardlink_DirectoryAllowed(t *testing.T) { + dir := t.TempDir() + // Directories naturally have nlink > 1, should be exempt + if err := checkHardlink(dir); err != nil { + t.Fatalf("expected no error for directory, got: %v", err) + } +} + +func TestCheckHardlink_NonExistent(t *testing.T) { + if err := checkHardlink("/nonexistent/path"); err != nil { + t.Fatalf("expected no error for non-existent file, got: %v", err) + } +} + +func TestCheckDeniedPath(t *testing.T) { + ws := setupWorkspace(t) + wsReal, _ := filepath.EvalSymlinks(ws) + + denied := filepath.Join(wsReal, ".goclaw", "secrets") + if err := os.MkdirAll(filepath.Dir(denied), 0755); err != nil { + t.Fatal(err) + } + + err := checkDeniedPath(denied, ws, []string{".goclaw"}) + if err == nil { + t.Fatal("expected error for denied path, got nil") + } + + // Non-denied path should pass + err = checkDeniedPath(filepath.Join(wsReal, "hello.txt"), ws, []string{".goclaw"}) + if err != nil { + t.Fatalf("expected no error for non-denied path, got: %v", err) + } +} + +func TestIsPathInside(t *testing.T) { + tests := []struct { + child, parent string + want bool + }{ + {"/a/b/c", "/a/b", true}, + {"/a/b", "/a/b", true}, + {"/a/bc", "/a/b", false}, // not a child, just prefix match + {"/a", "/a/b", false}, + {"/x/y", "/a/b", false}, + } + for _, tt := range tests { + got := isPathInside(tt.child, tt.parent) + if got != tt.want { + t.Errorf("isPathInside(%q, %q) = %v, want %v", tt.child, tt.parent, got, tt.want) + } + } +} diff --git a/internal/tools/filesystem.go b/internal/tools/filesystem.go index 99409537..40ebcf71 100644 --- a/internal/tools/filesystem.go +++ b/internal/tools/filesystem.go @@ -7,6 +7,7 @@ import ( "os" "path/filepath" "strings" + "syscall" "github.com/nextlevelbuilder/goclaw/internal/sandbox" ) @@ -157,12 +158,27 @@ func resolvePathWithAllowed(path, workspace string, restrict bool, allowedPrefix return resolved, nil } // If restricted and denied, check if path falls under an allowed prefix. + // Resolve symlinks in the candidate path for safe comparison. cleaned := filepath.Clean(path) + absPath, _ := filepath.Abs(cleaned) + real, evalErr := filepath.EvalSymlinks(absPath) + if evalErr != nil { + // Try resolving parent for non-existent files + parentReal, parentErr := filepath.EvalSymlinks(filepath.Dir(absPath)) + if parentErr != nil { + return "", err + } + real = filepath.Join(parentReal, filepath.Base(absPath)) + } for _, prefix := range allowedPrefixes { absPrefix, _ := filepath.Abs(prefix) - if strings.HasPrefix(cleaned, absPrefix) { - slog.Debug("read_file: allowed by prefix", "path", cleaned, "prefix", absPrefix) - return cleaned, nil + prefixReal, prefixErr := filepath.EvalSymlinks(absPrefix) + if prefixErr != nil { + prefixReal = absPrefix + } + if isPathInside(real, prefixReal) { + slog.Debug("read_file: allowed by prefix", "path", real, "prefix", prefixReal) + return real, nil } } slog.Warn("read_file: access denied", "path", cleaned, "workspace", workspace, "allowedPrefixes", allowedPrefixes) @@ -171,15 +187,21 @@ func resolvePathWithAllowed(path, workspace string, restrict bool, allowedPrefix // checkDeniedPath returns an error if the resolved path falls under any denied prefix. // Denied prefixes are relative to the workspace (e.g. ".goclaw" denies workspace/.goclaw/). +// The resolved path should already be canonical (from resolvePath with restrict=true). func checkDeniedPath(resolved, workspace string, deniedPrefixes []string) error { if len(deniedPrefixes) == 0 { return nil } absResolved, _ := filepath.Abs(resolved) absWorkspace, _ := filepath.Abs(workspace) + // Resolve workspace to canonical form for consistent comparison. + wsReal, err := filepath.EvalSymlinks(absWorkspace) + if err != nil { + wsReal = absWorkspace + } for _, prefix := range deniedPrefixes { - denied := filepath.Join(absWorkspace, prefix) - if strings.HasPrefix(absResolved, denied) { + denied := filepath.Join(wsReal, prefix) + if isPathInside(absResolved, denied) { return fmt.Errorf("access denied: path %s is restricted", prefix) } } @@ -187,27 +209,101 @@ func checkDeniedPath(resolved, workspace string, deniedPrefixes []string) error } // resolvePath resolves a path relative to the workspace and validates it. +// When restrict=true, resolves symlinks to canonical paths and rejects +// paths that escape the workspace boundary (symlink/hardlink attacks). func resolvePath(path, workspace string, restrict bool) (string, error) { + var resolved string if filepath.IsAbs(path) { - if restrict { - absWorkspace, _ := filepath.Abs(workspace) - if !strings.HasPrefix(path, absWorkspace) { - return "", fmt.Errorf("access denied: path outside workspace") + resolved = filepath.Clean(path) + } else { + resolved = filepath.Clean(filepath.Join(workspace, path)) + } + + if !restrict { + return resolved, nil + } + + // Resolve workspace to canonical path (follow symlinks in workspace path itself). + absWorkspace, _ := filepath.Abs(workspace) + wsReal, err := filepath.EvalSymlinks(absWorkspace) + if err != nil { + wsReal = absWorkspace // workspace doesn't exist yet — use as-is + } + + // Resolve the target path to canonical form (follows all symlinks). + absResolved, _ := filepath.Abs(resolved) + real, err := filepath.EvalSymlinks(absResolved) + if err != nil { + if os.IsNotExist(err) { + // Check if the path itself is a symlink (broken/dangling). + // Lstat doesn't follow symlinks, so it succeeds even for broken ones. + if linfo, lerr := os.Lstat(absResolved); lerr == nil && linfo.Mode()&os.ModeSymlink != 0 { + // It's a broken symlink — read target and validate. + target, readErr := os.Readlink(absResolved) + if readErr != nil { + return "", fmt.Errorf("access denied: cannot resolve symlink") + } + if !filepath.IsAbs(target) { + target = filepath.Join(filepath.Dir(absResolved), target) + } + target = filepath.Clean(target) + if !isPathInside(target, wsReal) { + slog.Warn("security.broken_symlink_escape", "path", path, "target", target, "workspace", wsReal) + return "", fmt.Errorf("access denied: broken symlink target outside workspace") + } + real = target + } else { + // Truly non-existent file (not a symlink): resolve parent and re-validate. + parentReal, parentErr := filepath.EvalSymlinks(filepath.Dir(absResolved)) + if parentErr != nil { + return "", fmt.Errorf("access denied: cannot resolve path") + } + real = filepath.Join(parentReal, filepath.Base(absResolved)) } - } - return filepath.Clean(path), nil - } - - resolved := filepath.Join(workspace, path) - resolved = filepath.Clean(resolved) - - if restrict { - absWorkspace, _ := filepath.Abs(workspace) - absResolved, _ := filepath.Abs(resolved) - if !strings.HasPrefix(absResolved, absWorkspace) { - return "", fmt.Errorf("access denied: path outside workspace") + } else { + // Permission error or other — reject. + slog.Warn("security.path_resolve_failed", "path", path, "error", err) + return "", fmt.Errorf("access denied: cannot resolve path") } } - return resolved, nil + // Validate canonical path stays within canonical workspace. + if !isPathInside(real, wsReal) { + slog.Warn("security.path_escape", "path", path, "resolved", real, "workspace", wsReal) + return "", fmt.Errorf("access denied: path outside workspace") + } + + // Reject hardlinked files (nlink > 1) to prevent hardlink-based escapes. + if err := checkHardlink(real); err != nil { + return "", err + } + + return real, nil +} + +// isPathInside checks whether child is inside or equal to parent directory. +func isPathInside(child, parent string) bool { + if child == parent { + return true + } + return strings.HasPrefix(child, parent+string(filepath.Separator)) +} + +// checkHardlink rejects regular files with nlink > 1 (hardlink attack prevention). +// Directories naturally have nlink > 1 and are exempt. +func checkHardlink(path string) error { + info, err := os.Lstat(path) + if err != nil { + return nil // non-existent files are OK — will fail at read/write + } + if info.IsDir() { + return nil + } + if stat, ok := info.Sys().(*syscall.Stat_t); ok { + if stat.Nlink > 1 { + slog.Warn("security.hardlink_rejected", "path", path, "nlink", stat.Nlink) + return fmt.Errorf("access denied: hardlinked file not allowed") + } + } + return nil }