Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion apps/backend/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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. Under enforced auth every dispatch is scoped to *somebody*: a resolvable active owner gets their identity; an unowned workspace gets a sentinel user ID that reaches unowned rows only; anything unresolvable (missing task/workspace row, or an owner whose account was deleted or disabled) is **denied**. Never return an identity-free context from this path — the task service reads that as an internal caller and grants everything.
- **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.
Expand Down
16 changes: 16 additions & 0 deletions apps/backend/internal/agent/runtime/lifecycle/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
61 changes: 61 additions & 0 deletions apps/backend/internal/agent/runtime/lifecycle/mcp_identity.go
Original file line number Diff line number Diff line change
@@ -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,
}
}
162 changes: 162 additions & 0 deletions apps/backend/internal/agent/runtime/lifecycle/mcp_identity_test.go
Original file line number Diff line number Diff line change
@@ -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")
}
}
8 changes: 6 additions & 2 deletions apps/backend/internal/agent/runtime/lifecycle/streams.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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),
Expand Down
22 changes: 22 additions & 0 deletions apps/backend/internal/auth/service_credentials.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading
Loading