fix(auth): scope in-session agent MCP calls to the task owner

apps/backend/AGENTS.md

@@ -168,7 +168,8 @@ and can leak ACP subprocesses.
 **Remote SSH executor platforms:** Treat supported remote OS/arch values as an end-to-end contract. Platform probe/normalization, lifecycle support checks, agentctl helper resolution, platform default shell, SSH readiness endpoints, frontend response types, and tests must stay aligned. Preserve raw unsupported platform details in user-facing errors, but use normalized values for supported-platform matching. Keep shell defaults platform-aware: Darwin defaults to `zsh`, Linux defaults to `bash`, unless an explicit shell is saved.
 
 **Opt-in authentication & per-user scoping** (`internal/auth/`): auth is OFF by default; the global middleware (`auth/httpmw`, installed after CORS in `backendapp.buildHTTPServer`) then injects a synthetic admin identity for the pre-auth single user, so behavior is unchanged. Enablement is the `features.auth` runtime flag (`KANDEV_FEATURES_AUTH`, Settings > System > Feature Toggles) — the auth service derives its mode from `cfg.Features.Auth` (`disabled` / `setup` = flag-on-no-admin / `enabled` = flag-on-admin-exists); there is no separate `auth.mode` setting. When enabled, requests authenticate via a `kandev_session` cookie (opaque token, SHA-256 at rest, DB-backed) or a `kandev_pat_*` bearer token. Scoping rules:
-- **Identity travels in the request context** (`authn.IdentityFromContext`). No identity = internal caller (pollers, event bus, MCP stream) = unscoped. Synthetic identity = auth disabled = unscoped.
+- **Identity travels in the request context** (`authn.IdentityFromContext`). No identity = internal caller (pollers, event bus, office schedulers) = unscoped. Synthetic identity = auth disabled = unscoped.
+- **In-session agent MCP is scoped to the task owner.** The MCP tools an agent gets inside its own session are relayed over the agent's WebSocket stream, which carries no credential of its own. `internal/mcp/scope` resolves the stream's task → workspace → owner and attaches that user's real identity before dispatch (`lifecycle.Manager.SetMCPIdentityScoper`), so the same `authorize*` checks apply as for the PAT-authenticated `/mcp` endpoint. The owning task comes from the `AgentExecution`, never from the agent-supplied payload — do not "improve" this by reading `session_id`/`task_id` out of the request. Tool handlers stay identity-agnostic.
 - **Workspaces are per-user** (`workspaces.owner_id`); the task service filters/denies at the service layer with `*NotFound` sentinels (no existence leak). Unowned rows (`owner_id=''`) stay visible until the setup wizard claims them for the admin.
 - **New user-facing service entry points must apply scoping** — call the `authorize*` helpers in `task/service/service_access.go` (or the same pattern) when adding routes that read or mutate workspace-scoped data.
 - **WS**: clients carry their identity; dispatched actions and subscriptions are scoped; workspace-carrying events route via `Hub.BroadcastToWorkspace`. A new `hub.Broadcast` (global) call site needs a `//ws:global` justification comment.

apps/backend/internal/agent/runtime/lifecycle/manager.go

@@ -349,6 +349,22 @@ func (m *Manager) SetMCPHandler(handler agentctl.MCPHandler) {
 	m.streamManager.mcpHandler = handler
 }
 
+// SetMCPIdentityScoper installs the per-user scoping hook for in-session MCP
+// tool calls.
+//
+// Unlike the external /mcp endpoint — where the agent presents a personal
+// access token and the auth middleware resolves the identity — MCP requests
+// relayed over an agent's own stream carry no credential. Without this hook
+// they reach the task service with no identity, which that service reads as an
+// internal caller and serves unscoped, so an agent supplying another user's
+// task_id or workflow_id would be given their data.
+//
+// Set once during startup wiring, before agents start making MCP calls. Leave
+// unset to keep dispatch unscoped (single-user instances).
+func (m *Manager) SetMCPIdentityScoper(scoper MCPIdentityScoper) {
+	m.streamManager.mcpIdentityScoper = scoper
+}
+
 // SetSessionAccessChecker installs the per-user session visibility check used
 // by GetOrEnsureExecution and EnsurePassthroughExecution. The checker must
 // return nil for contexts without a request identity (internal callers). Set

apps/backend/internal/agent/runtime/lifecycle/mcp_identity.go

@@ -0,0 +1,61 @@
+package lifecycle
+
+import (
+	"context"
+
+	"go.uber.org/zap"
+
+	agentctl "github.com/kandev/kandev/internal/agent/runtime/agentctl"
+	"github.com/kandev/kandev/internal/common/logger"
+	ws "github.com/kandev/kandev/pkg/websocket"
+)
+
+// MCPIdentityScoper attaches the identity of the user who owns taskID to ctx,
+// so the in-session MCP tools an agent gets automatically are authorized as
+// that user instead of running unscoped. Returning an error denies the
+// dispatch — see internal/mcp/scope for the production implementation and why
+// it fails closed rather than falling back to full access.
+type MCPIdentityScoper func(ctx context.Context, taskID string) (context.Context, error)
+
+// taskScopedMCPHandler scopes every MCP request on one agent stream to the
+// owner of that stream's task.
+//
+// The task ID comes from the AgentExecution this stream belongs to, never from
+// the request payload: an agent controls its own payloads, so honoring a
+// payload session_id would let it name another user's session and inherit
+// their identity — turning the scoping fix into a privilege escalation.
+type taskScopedMCPHandler struct {
+	inner  agentctl.MCPHandler
+	scope  MCPIdentityScoper
+	taskID string
+	logger *logger.Logger
+}
+
+func (h *taskScopedMCPHandler) Dispatch(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
+	scoped, err := h.scope(ctx, h.taskID)
+	if err != nil {
+		h.logger.Error("denying in-session MCP request: cannot resolve the task owner",
+			zap.String("task_id", h.taskID),
+			zap.String("action", msg.Action),
+			zap.Error(err))
+		return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError,
+			"failed to resolve the session owner", nil)
+	}
+	return h.inner.Dispatch(scoped, msg)
+}
+
+// mcpHandlerFor returns the MCP handler for one execution's stream, scoped to
+// the task's owning user when per-user scoping has been wired. Without a
+// scoper (tests, or the runtime tier used standalone) the handler is passed
+// through unchanged.
+func (sm *StreamManager) mcpHandlerFor(execution *AgentExecution) agentctl.MCPHandler {
+	if sm.mcpHandler == nil || sm.mcpIdentityScoper == nil || execution.TaskID == "" {
+		return sm.mcpHandler
+	}
+	return &taskScopedMCPHandler{
+		inner:  sm.mcpHandler,
+		scope:  sm.mcpIdentityScoper,
+		taskID: execution.TaskID,
+		logger: sm.logger,
+	}
+}

apps/backend/internal/agent/runtime/lifecycle/mcp_identity_test.go

@@ -0,0 +1,162 @@
+package lifecycle
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+
+	"github.com/kandev/kandev/internal/auth/authn"
+	"github.com/kandev/kandev/internal/common/logger"
+	ws "github.com/kandev/kandev/pkg/websocket"
+)
+
+// recordingMCPHandler captures the context its dispatch ran under so tests can
+// assert which identity (if any) reached the tool handlers.
+type recordingMCPHandler struct {
+	gotCtx context.Context
+	calls  int
+}
+
+func (h *recordingMCPHandler) Dispatch(ctx context.Context, msg *ws.Message) (*ws.Message, error) {
+	h.gotCtx = ctx
+	h.calls++
+	return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{"ok": true})
+}
+
+func newMCPStreamManager(t *testing.T, inner *recordingMCPHandler, scoper MCPIdentityScoper) *StreamManager {
+	t.Helper()
+	log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"})
+	if err != nil {
+		t.Fatalf("logger: %v", err)
+	}
+	sm := NewStreamManager(log, StreamCallbacks{}, inner, nil)
+	sm.mcpIdentityScoper = scoper
+	return sm
+}
+
+func mcpRequest(t *testing.T, payload map[string]interface{}) *ws.Message {
+	t.Helper()
+	data, err := json.Marshal(payload)
+	if err != nil {
+		t.Fatalf("marshal payload: %v", err)
+	}
+	return &ws.Message{ID: "req-1", Type: ws.MessageTypeRequest, Action: "mcp.list_tasks", Payload: data}
+}
+
+// TestMCPHandlerForScopesToExecutionTask is the core wiring assertion: the
+// identity handed to the tool handlers comes from the execution that owns the
+// stream.
+func TestMCPHandlerForScopesToExecutionTask(t *testing.T) {
+	inner := &recordingMCPHandler{}
+	var scopedTaskIDs []string
+	sm := newMCPStreamManager(t, inner, func(ctx context.Context, taskID string) (context.Context, error) {
+		scopedTaskIDs = append(scopedTaskIDs, taskID)
+		return authn.WithIdentity(ctx, authn.Identity{UserID: "owner-of-" + taskID, Role: authn.RoleMember}), nil
+	})
+
+	handler := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"})
+	resp, err := handler.Dispatch(context.Background(), mcpRequest(t, map[string]interface{}{}))
+	if err != nil {
+		t.Fatalf("Dispatch: %v", err)
+	}
+	if resp.Type != ws.MessageTypeResponse {
+		t.Fatalf("response type = %q, want response", resp.Type)
+	}
+
+	if len(scopedTaskIDs) != 1 || scopedTaskIDs[0] != "task-a" {
+		t.Fatalf("scoped task IDs = %v, want [task-a]", scopedTaskIDs)
+	}
+	identity, ok := authn.IdentityFromContext(inner.gotCtx)
+	if !ok {
+		t.Fatal("tool handlers received no identity")
+	}
+	if identity.UserID != "owner-of-task-a" {
+		t.Errorf("UserID = %q, want owner-of-task-a", identity.UserID)
+	}
+}
+
+// TestMCPHandlerForIgnoresPayloadSessionID is the privilege-escalation pin. The
+// payload is agent-controlled, so scoping must key off the execution's task
+// even when the request names a different session or task.
+func TestMCPHandlerForIgnoresPayloadSessionID(t *testing.T) {
+	inner := &recordingMCPHandler{}
+	var scopedTaskIDs []string
+	sm := newMCPStreamManager(t, inner, func(ctx context.Context, taskID string) (context.Context, error) {
+		scopedTaskIDs = append(scopedTaskIDs, taskID)
+		return ctx, nil
+	})
+
+	handler := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"})
+	_, err := handler.Dispatch(context.Background(), mcpRequest(t, map[string]interface{}{
+		"session_id": "session-of-victim",
+		"task_id":    "task-victim",
+	}))
+	if err != nil {
+		t.Fatalf("Dispatch: %v", err)
+	}
+
+	if len(scopedTaskIDs) != 1 || scopedTaskIDs[0] != "task-a" {
+		t.Errorf("scoped task IDs = %v, want [task-a] — payload IDs must not steer scoping", scopedTaskIDs)
+	}
+}
+
+// TestMCPHandlerForDeniesWhenScopingFails pins fail-closed behavior: an
+// unresolvable owner must not fall through to the unscoped handlers.
+func TestMCPHandlerForDeniesWhenScopingFails(t *testing.T) {
+	inner := &recordingMCPHandler{}
+	sm := newMCPStreamManager(t, inner, func(context.Context, string) (context.Context, error) {
+		return nil, errors.New("db unavailable")
+	})
+
+	handler := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"})
+	resp, err := handler.Dispatch(context.Background(), mcpRequest(t, map[string]interface{}{}))
+	if err != nil {
+		t.Fatalf("Dispatch: %v", err)
+	}
+
+	if resp.Type != ws.MessageTypeError {
+		t.Errorf("response type = %q, want error", resp.Type)
+	}
+	if inner.calls != 0 {
+		t.Errorf("inner handler ran %d times, want 0 — the request must be denied", inner.calls)
+	}
+}
+
+// TestMCPHandlerForPassesThroughWithoutScoper keeps single-user instances and
+// isolated tests on the original unwrapped handler.
+func TestMCPHandlerForPassesThroughWithoutScoper(t *testing.T) {
+	inner := &recordingMCPHandler{}
+	sm := newMCPStreamManager(t, inner, nil)
+
+	if got := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1", TaskID: "task-a"}); got != inner {
+		t.Errorf("handler = %T, want the unwrapped inner handler", got)
+	}
+}
+
+// TestMCPHandlerForPassesThroughWithoutTaskID covers executions with no task
+// (there is no owner to resolve, so wrapping would deny every call).
+func TestMCPHandlerForPassesThroughWithoutTaskID(t *testing.T) {
+	inner := &recordingMCPHandler{}
+	sm := newMCPStreamManager(t, inner, func(ctx context.Context, _ string) (context.Context, error) {
+		return ctx, nil
+	})
+
+	if got := sm.mcpHandlerFor(&AgentExecution{ID: "exec-1"}); got != inner {
+		t.Errorf("handler = %T, want the unwrapped inner handler", got)
+	}
+}
+
+func TestSetMCPIdentityScoperReachesStreamManager(t *testing.T) {
+	log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json"})
+	if err != nil {
+		t.Fatalf("logger: %v", err)
+	}
+	m := &Manager{streamManager: NewStreamManager(log, StreamCallbacks{}, &recordingMCPHandler{}, nil)}
+
+	m.SetMCPIdentityScoper(func(ctx context.Context, _ string) (context.Context, error) { return ctx, nil })
+
+	if m.streamManager.mcpIdentityScoper == nil {
+		t.Error("SetMCPIdentityScoper did not reach the stream manager")
+	}
+}

apps/backend/internal/agent/runtime/lifecycle/streams.go

@@ -31,6 +31,10 @@ type StreamManager struct {
 	logger     *logger.Logger
 	callbacks  StreamCallbacks
 	mcpHandler agentctl.MCPHandler
+	// mcpIdentityScoper scopes in-session MCP dispatches to the owner of the
+	// stream's task. Nil leaves dispatch unscoped (single-user instances and
+	// isolated tests); set via Manager.SetMCPIdentityScoper.
+	mcpIdentityScoper MCPIdentityScoper
 	// stopCh is the Manager-owned shutdown signal. The retry/backoff and
 	// connected `<-ws.Done() / <-stop>` select read from it so they drain on
 	// Manager.Stop. May be nil when isolated tests don't care about external
@@ -282,7 +286,7 @@ func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready c
 		if sm.callbacks.OnAgentEvent != nil {
 			sm.callbacks.OnAgentEvent(execution, event)
 		}
-	}, sm.mcpHandler, func(disconnectErr error) {
+	}, sm.mcpHandlerFor(execution), func(disconnectErr error) {
 		// WebSocket dropped — signal promptDoneCh so SendPrompt doesn't hang forever.
 		// Only signal on unexpected errors (not normal close).
 		if disconnectErr != nil {
@@ -324,7 +328,7 @@ func (sm *StreamManager) connectUpdatesStream(execution *AgentExecution, ready c
 // closing this stream is expected, not an error.
 func (sm *StreamManager) connectMCPStream(execution *AgentExecution) {
 	ctx := sm.streamContext(execution)
-	err := execution.agentctl.StreamUpdates(ctx, func(agentctl.AgentEvent) {}, sm.mcpHandler, func(disconnectErr error) {
+	err := execution.agentctl.StreamUpdates(ctx, func(agentctl.AgentEvent) {}, sm.mcpHandlerFor(execution), func(disconnectErr error) {
 		if disconnectErr != nil {
 			sm.logger.Debug("passthrough MCP stream disconnected",
 				zap.String("execution_id", execution.ID),

apps/backend/internal/auth/service_credentials.go

@@ -121,6 +121,28 @@ func (s *Service) ResolveBearer(ctx context.Context, token string) (authn.Identi
 	return authn.Identity{UserID: user.ID, Role: roleOf(user), TokenID: record.ID}, true
 }
 
+// IdentityForUser resolves a stored user ID to the identity that user carries
+// on an authenticated request, without needing one of their credentials.
+//
+// It exists for callers that already know *whose* work they are performing but
+// have no request to authenticate: in-session agent MCP calls arrive over the
+// agent's own WebSocket stream, so the owning user is derived from the stream's
+// task rather than from a cookie or PAT (see internal/mcp/scope). The returned
+// identity is real — never synthetic — so per-user scoping applies to it.
+//
+// Returns false while authentication is disabled (callers keep the pre-auth
+// unscoped behavior) and when the account is missing or no longer active.
+func (s *Service) IdentityForUser(ctx context.Context, userID string) (authn.Identity, bool) {
+	if s == nil || userID == "" || s.Mode() == ModeDisabled {
+		return authn.Identity{}, false
+	}
+	user, err := s.activeUser(ctx, userID)
+	if err != nil {
+		return authn.Identity{}, false
+	}
+	return authn.Identity{UserID: user.ID, Role: roleOf(user)}, true
+}
+
 // ListSessions returns the user's sessions for the account page.
 func (s *Service) ListSessions(ctx context.Context, userID string) ([]*store.Session, error) {
 	return s.store.ListSessionsByUser(ctx, userID)