diff --git a/apps/backend/AGENTS.md b/apps/backend/AGENTS.md index 9d97611f7e..46f261d9dc 100644 --- a/apps/backend/AGENTS.md +++ b/apps/backend/AGENTS.md @@ -177,7 +177,7 @@ Standalone agentctl is launched in its own process group so terminal Ctrl+C is h - **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. This includes **session-keyed** entry points: any service method taking a caller-supplied `sessionID` (or a `taskID` used to reach sessions) authorizes via `authorizeTaskID(session.TaskID)`. Authorize off the row you already read rather than calling `AuthorizeSessionAccess`, which re-reads the session. Batch helpers that take ID *lists* (`BatchGetSessionsForTasks`, `GetPrimarySessionInfoForTasks`, …) are exempt by convention: their callers derive the IDs from an already-authorized task list, so per-ID checks would add N queries to list views for no gain — keep it that way, and don't hand them caller-supplied IDs. - **Two session paths bypass the task service and must guard themselves.** (1) Handlers that resolve an execution by a bare in-memory lookup (`GetExecutionBySessionID`, `*BySessionID`) skip the `GetOrEnsure*` chokepoint where the lifecycle check runs — call `service.AuthorizeSessionAccess` (see `ProcessHandlers.denySessionAccess`) or `lifecycleMgr.CheckSessionAccess` first. (2) The orchestrator reads sessions through its own repo handle, so it inherits nothing from the task service; session-keyed entry points there call `authorizeSession` / `authorizeTask`, wired by `SetSessionAccessChecker` / `SetTaskAccessChecker`. An entry point that accepts **both** a task and a session ID must use `authorizeTaskSessionPair`, never `authorizeSession` alone: the caller can satisfy a session-only check with one of their own sessions while pointing `taskID` at someone else's task, which the method then uses for its task-scoped work. Put the guard **first** in the method, before any repo/executor/agent-manager use — `TestSessionKeyedEntryPointsGuardBeforeDependencies` runs each entry point with nil dependencies and fails on a panic if a guard is placed too late. - **Dispatched WS actions have a gateway backstop.** `Client.authorizeAction` (`internal/gateway/websocket/dispatch_scope.go`) checks any payload carrying `task_id` or `session_id` against the client's identity *before* dispatch, so a newly added action is scoped by default rather than only when its author remembers. Handler-level scoping is still required (defense in depth, and it covers non-WS callers) — do not delete a service-layer check because the backstop exists. The backstop reads `task_id`, `session_id` and `task_environment_id`, plus `id` on **top-level `task.` actions only** (`task.state`, `task.move`, `task.get`, … — `task.` and exactly one more segment). Keep using those names for task-scoped actions, because an action that invents a new name for the same kind of resource silently opts out. (`user_shell.stop` was exactly that: it keys off `task_environment_id` with `task_id` optional, so a `task_id`-only backstop missed it. `task.state` and `task.move` were the second instance: they name the task `id`, so the backstop parsed no refs and let one user mutate any other user's task — hence the `id` rule.) Deeper `task.*` namespaces are deliberately excluded: in `task.plan.revision.get` and `task.review.finding.update` the `id` names a revision or a finding, so those actions must carry `task_id` when they are task-scoped. Also avoid permissive types (`any`, `json.RawMessage`) for those fields: a payload that fails to unmarshal into the ref struct is treated as naming nothing. -- **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. +- **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. Subscription actions (`*.subscribe`) `return` before reaching the dispatch backstop (`Client.authorizeAction`), so scoping a new subscribe topic means adding a hook to `SubscriptionAccessPolicy` (`internal/gateway/websocket/access.go`) *and* calling it explicitly inside that action's own handler — the backstop's `scopedActionRefs` cannot reach it. `run.subscribe`'s `Subscriptions.Run` hook (added for WO-02, resolving the run's workspace via `runs.GetRunWorkspaceID`) is the worked example. - **Task overview vs. session detail:** Keep cross-task state in the bounded, persisted `TaskStatusSummary` projection and `task.status_summary.updated`; keep unbounded transcript and execution data session-scoped. The [bounded task-status spec](../../docs/specs/platform/bounded-task-status-delivery.md) and [accepted ADR](../../docs/decisions/2026-08-01-separate-task-summary-session-stream-traffic.md) define the shared contract. - **Port-proxy subtree capability:** with auth enforced, `/port-proxy/:sessionId/:port/*` needs a credential on *every* request, but browser subresource fetches cannot always carry one (the `` fetch never sends cookies; sandboxed iframes and `?token=`-opened previews drop them elsewhere). `PortProxyHandler` mints a short-lived (15 min, sliding), path-scoped, HMAC-signed capability after the document authenticates and propagates it as a cookie scoped to the proxy subtree plus a `kandev_cap` query parameter appended to rewritten asset URLs. `requireConnectionAuth` (`internal/gateway/websocket/access.go`) accepts either form only for the exact session:port it was minted for and restores the issuing identity, so `CheckSessionAccess` still gates on the real owner. Synthetic identities (auth disabled) skip minting — proxied bodies stay byte-identical to the pre-auth behavior. - **Self-authenticating webhooks** (automation and office channels) and `/health`, `/api/v1/features`, `/api/v1/app-state` stay public — the allowlist lives in `auth/httpmw/middleware.go` with a pinning test. Plugin webhooks are not allowlisted: `auth/httpmw` structurally defers only GET/POST relay paths because it cannot read manifests, and `internal/plugins.Controller.webhook` enforces each manifest's `access` policy. diff --git a/apps/backend/internal/backendapp/auth.go b/apps/backend/internal/backendapp/auth.go index 266db228ae..11ee6921ac 100644 --- a/apps/backend/internal/backendapp/auth.go +++ b/apps/backend/internal/backendapp/auth.go @@ -14,6 +14,8 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/db" gateways "github.com/kandev/kandev/internal/gateway/websocket" + officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite" + "github.com/kandev/kandev/internal/task/repository/repoerrors" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" taskservice "github.com/kandev/kandev/internal/task/service" ) @@ -84,7 +86,9 @@ func provideAuthService( // gatewayAuthPolicy assembles the WS gateway scoping hooks from the auth and // task services. The workspace-owner resolver caches lookups briefly: it runs // on every workspace-scoped broadcast. -func gatewayAuthPolicy(authSvc *auth.Service, taskSvc *taskservice.Service, taskRepo *sqliterepo.Repository) gateways.AuthPolicy { +func gatewayAuthPolicy( + authSvc *auth.Service, taskSvc *taskservice.Service, taskRepo *sqliterepo.Repository, officeRepo *officesqlite.Repository, +) gateways.AuthPolicy { return gateways.AuthPolicy{ Enforced: func() bool { return authSvc.Mode() != auth.ModeDisabled }, ResolveToken: authSvc.ResolveBearer, @@ -98,6 +102,7 @@ func gatewayAuthPolicy(authSvc *auth.Service, taskSvc *taskservice.Service, task Subscriptions: gateways.SubscriptionAccessPolicy{ Task: taskSvc.AuthorizeTaskAccess, Session: taskSvc.AuthorizeSessionAccess, + Run: runSubscriptionCheck(taskSvc, officeRepo), }, WorkspaceOwner: newWorkspaceOwnerResolver(taskRepo), // The user-shell actions name a task environment and treat task_id as @@ -106,6 +111,38 @@ func gatewayAuthPolicy(authSvc *auth.Service, taskSvc *taskservice.Service, task } } +// runSubscriptionCheck resolves a run's owning workspace (via its agent +// profile) and defers to the task service's workspace visibility rule. +// Unlike WorkspaceOwner (which runs on every workspace broadcast), this +// runs once per run.subscribe — a rare control message — so it is not +// cached. +// +// An ordinary (non-Office) Kanban agent profile has workspace_id="" +// (schema default, never backfilled) — the dominant case, since the run +// processor and workflow-engine queue_run dispatch run for every backend, +// not just Office. AuthorizeWorkspaceAccess's workspaceID=="" branch means +// "no workspace scoping applies" (used for dangling references elsewhere +// in the task service), which would silently allow subscribing to these +// runs; that meaning does not apply here, so an empty resolution is denied +// outright before it ever reaches that helper. officeRepo==nil fails +// closed for the same reason: this is a security check, not a visibility +// fallback. +func runSubscriptionCheck(taskSvc *taskservice.Service, officeRepo *officesqlite.Repository) func(context.Context, string) error { + return func(ctx context.Context, runID string) error { + if officeRepo == nil { + return repoerrors.ErrWorkspaceNotFound + } + workspaceID, err := officeRepo.GetRunWorkspaceID(ctx, runID) + if err != nil { + return err + } + if workspaceID == "" { + return repoerrors.ErrWorkspaceNotFound + } + return taskSvc.AuthorizeWorkspaceAccess(ctx, workspaceID) + } +} + // ownerCacheTTL bounds staleness of broadcast routing after an ownership // change (only the setup wizard's claim changes owners in practice). const ownerCacheTTL = 30 * time.Second diff --git a/apps/backend/internal/backendapp/auth_test.go b/apps/backend/internal/backendapp/auth_test.go index 6c98680cc1..a910750052 100644 --- a/apps/backend/internal/backendapp/auth_test.go +++ b/apps/backend/internal/backendapp/auth_test.go @@ -4,15 +4,27 @@ import ( "context" "net/http" "net/http/httptest" + "path/filepath" "testing" + "time" "github.com/gin-gonic/gin" "github.com/jmoiron/sqlx" _ "github.com/mattn/go-sqlite3" + settingsstore "github.com/kandev/kandev/internal/agent/settings/store" "github.com/kandev/kandev/internal/auth" + "github.com/kandev/kandev/internal/auth/authn" authstore "github.com/kandev/kandev/internal/auth/store" "github.com/kandev/kandev/internal/common/config" + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/db" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/office/models" + officesqlite "github.com/kandev/kandev/internal/office/repository/sqlite" + "github.com/kandev/kandev/internal/task/repository" + sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" + taskservice "github.com/kandev/kandev/internal/task/service" userstore "github.com/kandev/kandev/internal/user/store" ) @@ -116,3 +128,206 @@ func TestPluginSSOBridgeDefaultPortPlainName(t *testing.T) { t.Fatalf("SSO Set-Cookie names = %v, want exactly [kandev_session]", names) } } + +// newRunSubscriptionCheckHarness builds a task service, task repository, and +// office repository over one shared SQLite database, the same pairing +// gatewayAuthPolicy wires in production. WO-02 round 2's runSubscriptionCheck +// tests seed agent profiles and runs against it; round 3 also seeds owned +// workspaces via the returned task repository to drive real ownership +// decisions instead of the unscoped-caller shortcut. +func newRunSubscriptionCheckHarness(t *testing.T) (*taskservice.Service, *sqliterepo.Repository, *officesqlite.Repository) { + t.Helper() + dbConn, err := db.OpenSQLite(filepath.Join(t.TempDir(), "run-subscribe.db")) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + database := sqlx.NewDb(dbConn, "sqlite3") + t.Cleanup(func() { _ = database.Close() }) + taskRepo, cleanup, err := repository.Provide(database, database, nil) + if err != nil { + t.Fatalf("task repository: %v", err) + } + t.Cleanup(func() { _ = cleanup() }) + // agents/agent_profiles live in the settings store; officeRepo's runs + // queue (GetRunWorkspaceID) joins against them. + if _, _, err := settingsstore.Provide(database, database, nil); err != nil { + t.Fatalf("settings store: %v", err) + } + officeRepo, err := officesqlite.NewWithDB(database, database, nil) + if err != nil { + t.Fatalf("office repository: %v", err) + } + log, err := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "json", OutputPath: "stdout"}) + if err != nil { + t.Fatalf("logger: %v", err) + } + taskSvc := taskservice.NewService(taskservice.Repos{ + Workspaces: taskRepo, + Tasks: taskRepo, + TaskRepos: taskRepo, + Workflows: taskRepo, + Messages: taskRepo, + Turns: taskRepo, + Sessions: taskRepo, + GitSnapshots: taskRepo, + RepoEntities: taskRepo, + Executors: taskRepo, + Environments: taskRepo, + TaskEnvironments: taskRepo, + Reviews: taskRepo, + ResourceCleanups: taskRepo, + }, bus.NewMemoryEventBus(log), log, taskservice.RepositoryDiscoveryConfig{}) + return taskSvc, taskRepo, officeRepo +} + +// seedOwnedWorkspace creates a real workspace row owned by ownerID and +// returns its id. Driven through taskSvc.CreateWorkspace on an identity-free +// context so the request-body OwnerID is honored (callerScope returns +// unscoped, exactly the "internal caller" path CreateWorkspace documents) -- +// this seeds a real workspaces row rather than hand-writing SQL, so it stays +// correct if the schema changes. +func seedOwnedWorkspace(t *testing.T, taskSvc *taskservice.Service, ownerID string) string { + t.Helper() + workspace, err := taskSvc.CreateWorkspace(context.Background(), &taskservice.CreateWorkspaceRequest{ + Name: "ws-" + ownerID, + OwnerID: ownerID, + }) + if err != nil { + t.Fatalf("seed workspace owned by %q: %v", ownerID, err) + } + return workspace.ID +} + +// seedRunForWorkspace inserts an agent + agent profile scoped to +// workspaceID (empty string reproduces an ordinary, non-Office Kanban agent +// profile — schema default, never backfilled) and a run against it, +// returning the run id. +func seedRunForWorkspace(t *testing.T, officeRepo *officesqlite.Repository, workspaceID string) string { + t.Helper() + ctx := context.Background() + now := time.Now().UTC() + agentID := "agent-" + workspaceID + if workspaceID == "" { + agentID = "agent-unscoped" + } + if _, err := officeRepo.ExecRaw(ctx, `INSERT INTO agents + (id, name, workspace_id, created_at, updated_at) VALUES (?, ?, ?, ?, ?)`, + agentID, agentID, workspaceID, now, now); err != nil { + t.Fatalf("seed agent: %v", err) + } + if _, err := officeRepo.ExecRaw(ctx, `INSERT INTO agent_profiles + (id, agent_id, name, agent_display_name, workspace_id, role, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, '', ?, ?)`, + agentID, agentID, "name", "display", workspaceID, now, now); err != nil { + t.Fatalf("seed agent profile: %v", err) + } + runID := "run-" + agentID + if err := officeRepo.CreateRun(ctx, &models.Run{ + ID: runID, + AgentProfileID: agentID, + Reason: "task_assigned", + Status: models.RunStatus("queued"), + CoalescedCount: 1, + }); err != nil { + t.Fatalf("seed run: %v", err) + } + return runID +} + +// TestRunSubscriptionCheckDeniesEmptyResolvedWorkspace pins WO-02 round 2's +// FINDING R1-1 fix: an ordinary (non-Office) Kanban agent profile resolves +// to workspace_id=” via GetRunWorkspaceID, which used to fall into +// AuthorizeWorkspaceAccess's unconditional-allow branch (workspaceID == "" +// => nil). The hook must deny outright instead of delegating — this is the +// dominant class of runs (workflow-engine queue_run against ordinary Kanban +// agent profiles), not an edge case. +func TestRunSubscriptionCheckDeniesEmptyResolvedWorkspace(t *testing.T) { + taskSvc, _, officeRepo := newRunSubscriptionCheckHarness(t) + runID := seedRunForWorkspace(t, officeRepo, "") + + check := runSubscriptionCheck(taskSvc, officeRepo) + if err := check(context.Background(), runID); err == nil { + t.Fatal("runSubscriptionCheck(empty workspace) = nil, want a denial error") + } +} + +// TestRunSubscriptionCheckAllowsResolvedWorkspace drives runSubscriptionCheck +// with the real WORKSPACE OWNER's scoped identity against a real owned +// workspace row, so the allow comes from AuthorizeWorkspaceAccess actually +// reading the workspace and matching OwnerID -- not from callerScope's +// unscoped short-circuit (authorizeWorkspaceID returns nil before ever +// reading the row when the context carries no identity, which made the +// original version of this test pass even with `return nil` substituted for +// AuthorizeWorkspaceAccess; see REVIEW ROUND 2 FINDING B2-1). +func TestRunSubscriptionCheckAllowsResolvedWorkspace(t *testing.T) { + taskSvc, _, officeRepo := newRunSubscriptionCheckHarness(t) + ownerID := "user-owner" + workspaceID := seedOwnedWorkspace(t, taskSvc, ownerID) + runID := seedRunForWorkspace(t, officeRepo, workspaceID) + + check := runSubscriptionCheck(taskSvc, officeRepo) + ctx := authn.WithIdentity(context.Background(), authn.Identity{UserID: ownerID, Role: authn.RoleMember}) + if err := check(ctx, runID); err != nil { + t.Fatalf("runSubscriptionCheck(owner) = %v, want nil", err) + } +} + +// TestRunSubscriptionCheckDeniesForeignWorkspaceOwner is the ownership-denial +// half of B2-1's fix and directly verifies the card's own SCOPE item 4 ("a +// subscribe for a run the caller does not own must be refused") against the +// real runSubscriptionCheck -- the only existing foreign-owner coverage +// (gateway TestSubscriptionChecksDenyForeignTopics) exercises a hand-written +// fake Run hook, not this one. +func TestRunSubscriptionCheckDeniesForeignWorkspaceOwner(t *testing.T) { + taskSvc, _, officeRepo := newRunSubscriptionCheckHarness(t) + workspaceID := seedOwnedWorkspace(t, taskSvc, "user-owner") + runID := seedRunForWorkspace(t, officeRepo, workspaceID) + + check := runSubscriptionCheck(taskSvc, officeRepo) + ctx := authn.WithIdentity(context.Background(), authn.Identity{UserID: "user-other", Role: authn.RoleMember}) + if err := check(ctx, runID); err == nil { + t.Fatal("runSubscriptionCheck(foreign owner) = nil, want a denial error") + } +} + +// TestRunSubscriptionCheckDeniesWhenOfficeRepoNil pins FINDING R1-2: the +// nil-officeRepo guard must fail closed, not open. Not reachable in the +// shipped binary (repos.Office is never nil), but the wrong default for a +// security check. +func TestRunSubscriptionCheckDeniesWhenOfficeRepoNil(t *testing.T) { + taskSvc, _, _ := newRunSubscriptionCheckHarness(t) + + check := runSubscriptionCheck(taskSvc, nil) + if err := check(context.Background(), "any-run"); err == nil { + t.Fatal("runSubscriptionCheck(nil officeRepo) = nil, want a denial error") + } +} + +// TestGatewayAuthPolicyWiresRunSubscriptionCheck pins the production wiring +// site itself (auth.go's Subscriptions.Run: runSubscriptionCheck(...)), +// which nothing previously asserted: deleting that one line left every +// runSubscriptionCheck/gateway test green (B2-1). Scoped to the Run hook +// only, per the routing message -- Task/Session have no precedent test +// either and are not part of this card. +// +// Beyond the nil check, it also invokes the wired hook against a real +// empty-workspace run and asserts a denial -- this additionally catches a +// no-op substitution (e.g. `func(context.Context, string) error { return +// nil }`) that a bare non-nil assertion would miss. +func TestGatewayAuthPolicyWiresRunSubscriptionCheck(t *testing.T) { + taskSvc, taskRepo, officeRepo := newRunSubscriptionCheckHarness(t) + cfg := &config.Config{} + cfg.Features.Auth = true + cfg.Auth.SessionTTLHours = 720 + authSvc := newEnabledAuthService(t, cfg) + + policy := gatewayAuthPolicy(authSvc, taskSvc, taskRepo, officeRepo) + if policy.Subscriptions.Run == nil { + t.Fatal("gatewayAuthPolicy(...).Subscriptions.Run = nil, want runSubscriptionCheck wired") + } + + runID := seedRunForWorkspace(t, officeRepo, "") + if err := policy.Subscriptions.Run(context.Background(), runID); err == nil { + t.Fatal("gatewayAuthPolicy(...).Subscriptions.Run(empty workspace) = nil, want a denial error") + } +} diff --git a/apps/backend/internal/backendapp/main.go b/apps/backend/internal/backendapp/main.go index 9a8a0f3916..2df6b50679 100644 --- a/apps/backend/internal/backendapp/main.go +++ b/apps/backend/internal/backendapp/main.go @@ -815,7 +815,7 @@ func startGatewayAndServe( // and proxy routes, subscription visibility checks, and workspace-owner // broadcast routing. Must be installed before SetupRoutes runs in // buildHTTPServer. - gateway.SetAuthPolicy(gatewayAuthPolicy(services.Auth, services.Task, repos.Task)) + gateway.SetAuthPolicy(gatewayAuthPolicy(services.Auth, services.Task, repos.Task, repos.Office)) waitForAgentctlControlHealthy(ctx, cfg, log) diff --git a/apps/backend/internal/gateway/websocket/access.go b/apps/backend/internal/gateway/websocket/access.go index 8e736e2515..02e36fd86b 100644 --- a/apps/backend/internal/gateway/websocket/access.go +++ b/apps/backend/internal/gateway/websocket/access.go @@ -20,6 +20,9 @@ import ( type SubscriptionAccessPolicy struct { Task func(ctx context.Context, taskID string) error Session func(ctx context.Context, sessionID string) error + // Run gates run.subscribe. Nil (the zero-AuthPolicy case) leaves every + // run subscription unchecked, matching Task/Session. + Run func(ctx context.Context, runID string) error } // WorkspaceOwnerResolver resolves a workspace's owning user ID ("" = unowned diff --git a/apps/backend/internal/gateway/websocket/access_test.go b/apps/backend/internal/gateway/websocket/access_test.go index 3b29799b88..97a5aec7d8 100644 --- a/apps/backend/internal/gateway/websocket/access_test.go +++ b/apps/backend/internal/gateway/websocket/access_test.go @@ -628,6 +628,13 @@ func TestSubscriptionChecksDenyForeignTopics(t *testing.T) { } return nil }, + Run: func(ctx context.Context, runID string) error { + identity, _ := authn.IdentityFromContext(ctx) + if runID == "run-b" && identity.UserID != "user-b" { + return denied + } + return nil + }, }, }) clientA := registerAccessClient(t, hub, "a", authn.Identity{UserID: "user-a", Role: authn.RoleMember}) @@ -649,12 +656,126 @@ func TestSubscriptionChecksDenyForeignTopics(t *testing.T) { t.Fatal("denied session subscription must not register") } + clientA.handleRunSubscribe(subscribe(ws.ActionRunSubscribe, map[string]interface{}{"run_id": "run-b"})) + assertErrorResponse(t, clientA, "run subscribe") + if len(hub.getSubscribersLocked(hub.runSubscribers, "run-b")) != 0 { + t.Fatal("denied run subscription must not register") + } + // Own topics still work. clientA.handleSubscribe(subscribe(ws.ActionTaskSubscribe, map[string]interface{}{"task_id": "task-a"})) assertSuccessResponse(t, clientA, "own task subscribe") if len(hub.getSubscribersLocked(hub.taskSubscribers, "task-a")) != 1 { t.Fatal("allowed task subscription must register") } + + clientA.handleRunSubscribe(subscribe(ws.ActionRunSubscribe, map[string]interface{}{"run_id": "run-a"})) + assertSuccessResponse(t, clientA, "own run subscribe") + if len(hub.getSubscribersLocked(hub.runSubscribers, "run-a")) != 1 { + t.Fatal("allowed run subscription must register") + } +} + +// TestRunSubscribeDeniesEmptyWorkspaceResolution pins WO-02 round 2's +// FINDING R1-1 fix at the gateway wiring level: a Run hook denial for a run +// that resolves to an empty workspace (the runSubscriptionCheck case, an +// ordinary non-Office agent profile) must refuse the subscribe request and +// leave it unregistered in hub.runSubscribers — the same contract already +// pinned for a foreign-owned run by TestSubscriptionChecksDenyForeignTopics, +// exercised here for the empty-workspace input specifically. +func TestRunSubscribeDeniesEmptyWorkspaceResolution(t *testing.T) { + hub := newAccessTestHub(t) + errEmptyWorkspace := errors.New("workspace not found") + hub.setAuthPolicy(AuthPolicy{ + Subscriptions: SubscriptionAccessPolicy{ + Run: func(_ context.Context, runID string) error { + if runID == "run-unscoped" { + return errEmptyWorkspace + } + return nil + }, + }, + }) + client := registerAccessClient(t, hub, "a", authn.Identity{UserID: "user-a", Role: authn.RoleMember}) + + subscribe := func(runID string) *ws.Message { + raw, _ := json.Marshal(map[string]interface{}{"run_id": runID}) + return &ws.Message{ID: "1", Type: ws.MessageTypeRequest, Action: ws.ActionRunSubscribe, Payload: raw} + } + + client.handleRunSubscribe(subscribe("run-unscoped")) + assertErrorResponse(t, client, "run subscribe with empty-workspace resolution") + if len(hub.getSubscribersLocked(hub.runSubscribers, "run-unscoped")) != 0 { + t.Fatal("denied run subscription (empty workspace) must not register") + } +} + +// TestRunSubscribeNoExistenceLeak pins the no-existence-leak property: an +// unknown run id and a foreign run id must be byte-identical on the wire. +// handleRunSubscribe maps every hook error to one fixed (code, message) +// pair rather than forwarding err.Error(), so a caller cannot distinguish +// "you don't own this" from "this doesn't exist" by probing. +func TestRunSubscribeNoExistenceLeak(t *testing.T) { + hub := newAccessTestHub(t) + errUnknown := errors.New("sql: no rows in result set") + errForeign := errors.New("run belongs to another workspace") + hub.setAuthPolicy(AuthPolicy{ + Subscriptions: SubscriptionAccessPolicy{ + Run: func(_ context.Context, runID string) error { + if runID == "run-unknown" { + return errUnknown + } + return errForeign + }, + }, + }) + client := registerAccessClient(t, hub, "a", authn.Identity{UserID: "user-a", Role: authn.RoleMember}) + + subscribe := func(id, runID string) *ws.Message { + raw, _ := json.Marshal(map[string]interface{}{"run_id": runID}) + return &ws.Message{ID: id, Type: ws.MessageTypeRequest, Action: ws.ActionRunSubscribe, Payload: raw} + } + + client.handleRunSubscribe(subscribe("1", "run-unknown")) + unknownFrame := readErrorFrame(t, client, "unknown run") + + client.handleRunSubscribe(subscribe("2", "run-foreign")) + foreignFrame := readErrorFrame(t, client, "foreign run") + + if unknownFrame.code != foreignFrame.code || unknownFrame.message != foreignFrame.message { + t.Fatalf("unknown run response (%q, %q) must match foreign run response (%q, %q)", + unknownFrame.code, unknownFrame.message, foreignFrame.code, foreignFrame.message) + } +} + +type errorFrame struct { + code string + message string +} + +// readErrorFrame reads one frame off the client's queues and requires it to +// be an error, decoding the (code, message) payload for comparison. +func readErrorFrame(t *testing.T, client *Client, label string) errorFrame { + t.Helper() + var raw []byte + select { + case raw = <-client.controlSend: + case raw = <-client.send: + case <-time.After(2 * time.Second): + t.Fatalf("%s: no response frame", label) + } + var msg ws.Message + if err := json.Unmarshal(raw, &msg); err != nil { + t.Fatalf("%s: bad frame: %v", label, err) + } + if msg.Type != ws.MessageTypeError { + t.Fatalf("%s: got %s frame, want error", label, msg.Type) + } + var payload ws.ErrorPayload + if err := json.Unmarshal(msg.Payload, &payload); err != nil { + t.Fatalf("%s: bad error payload: %v", label, err) + } + return errorFrame{code: payload.Code, message: payload.Message} } func TestUserSubscribeScopedToOwnUser(t *testing.T) { diff --git a/apps/backend/internal/gateway/websocket/client.go b/apps/backend/internal/gateway/websocket/client.go index b0635afe75..0641c39b1d 100644 --- a/apps/backend/internal/gateway/websocket/client.go +++ b/apps/backend/internal/gateway/websocket/client.go @@ -502,6 +502,18 @@ func (c *Client) handleRunSubscribe(msg *ws.Message) { c.sendError(msg.ID, msg.Action, ws.ErrorCodeValidation, "run_id is required", nil) return } + + // Per-user scoping: a client may only observe runs in its own + // workspaces. Every hook error maps to the same fixed (code, message) + // pair below so an unknown run id and a foreign run id are + // indistinguishable on the wire (no existence leak). + if check := c.hub.authPolicy.Subscriptions.Run; check != nil { + if err := check(c.dispatchContext(), req.RunID); err != nil { + c.sendError(msg.ID, msg.Action, ws.ErrorCodeForbidden, "cannot subscribe to this run", nil) + return + } + } + c.hub.SubscribeToRun(c, req.RunID) resp, _ := ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ "success": true, diff --git a/apps/backend/internal/runs/repository/sqlite/runs.go b/apps/backend/internal/runs/repository/sqlite/runs.go index 9c9d157de8..82e3fb64a8 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs.go +++ b/apps/backend/internal/runs/repository/sqlite/runs.go @@ -177,6 +177,25 @@ func (r *Repository) GetRunByID(ctx context.Context, id string) (*models.Run, er return &run, nil } +// GetRunWorkspaceID resolves a run's owning workspace via its agent +// profile, the same join idiom ListRuns uses. A raw join, not +// GetAgentInstance: GetAgentInstance filters deleted_at IS NULL, which +// would make a run under a soft-deleted agent profile permanently +// unresolvable (and so permanently deniable) to its own owner. Returns +// sql.ErrNoRows for an unknown run. +func (r *Repository) GetRunWorkspaceID(ctx context.Context, runID string) (string, error) { + var workspaceID string + err := r.ro.QueryRowxContext(ctx, r.ro.Rebind(` + SELECT a.workspace_id FROM runs r + JOIN agent_profiles a ON a.id = r.agent_profile_id + WHERE r.id = ? + `), runID).Scan(&workspaceID) + if err != nil { + return "", err + } + return workspaceID, nil +} + // GetClaimedRunByID returns a run only while it is still claimed. Lifecycle // events carry this immutable run identity so a delayed predecessor event // cannot finish a newer claimed run for the same task and agent. diff --git a/apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go b/apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go index 8d6402aaf5..0dc99f0300 100644 --- a/apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go +++ b/apps/backend/internal/runs/repository/sqlite/runs_lookup_test.go @@ -258,3 +258,71 @@ func checkInt64(t *testing.T, field string, got, want int64) { t.Errorf("%s = %d, want %d", field, got, want) } } + +// TestGetRunWorkspaceID_ResolvesViaAgentProfileJoin pins the WS gateway's +// run.subscribe authorization lookup (WO-02): the run's owning workspace +// comes from its agent profile, the same join idiom ListRuns uses. +func TestGetRunWorkspaceID_ResolvesViaAgentProfileJoin(t *testing.T) { + repo, writer := newTestRepoWithDB(t) + ctx := context.Background() + + seedAgentProfile(t, writer, "a1", "ws-1") + run := seedTaskRun(t, repo, "run-1", "a1", "t1", "queued") + + got, err := repo.GetRunWorkspaceID(ctx, run.ID) + if err != nil { + t.Fatalf("get run workspace id: %v", err) + } + checkString(t, "workspace_id", got, "ws-1") +} + +// TestGetRunWorkspaceID_StillResolvesWhenAgentProfileSoftDeleted pins the +// deliberate choice of a raw join over GetAgentInstance: GetAgentInstance +// filters deleted_at IS NULL, which would make a run under a soft-deleted +// agent profile permanently unresolvable (and so permanently deniable) to +// its own owner. +func TestGetRunWorkspaceID_StillResolvesWhenAgentProfileSoftDeleted(t *testing.T) { + repo, writer := newTestRepoWithDB(t) + ctx := context.Background() + + seedAgentProfile(t, writer, "a1", "ws-1") + run := seedTaskRun(t, repo, "run-1", "a1", "t1", "queued") + + if _, err := writer.Exec(`UPDATE agent_profiles SET deleted_at = ? WHERE id = 'a1'`, time.Now().UTC()); err != nil { + t.Fatalf("soft-delete agent profile: %v", err) + } + + got, err := repo.GetRunWorkspaceID(ctx, run.ID) + if err != nil { + t.Fatalf("get run workspace id after soft delete: %v", err) + } + checkString(t, "workspace_id", got, "ws-1") +} + +// TestGetRunWorkspaceID_UnknownRunReturnsNoRows pins the sentinel the +// gateway's subscribe hook denies on. +func TestGetRunWorkspaceID_UnknownRunReturnsNoRows(t *testing.T) { + repo := newTestRepo(t) + if _, err := repo.GetRunWorkspaceID(context.Background(), "nope"); !errors.Is(err, sql.ErrNoRows) { + t.Fatalf("err = %v, want sql.ErrNoRows", err) + } +} + +// TestGetRunWorkspaceID_ResolvesToEmptyStringForUnscopedAgentProfile pins the +// WO-02 round-2 fix: an ordinary (non-Office) agent profile has +// workspace_id=” (schema default, no backfill), and the join returns that +// empty string as-is rather than an error. The gateway's subscribe hook (not +// this repository) is responsible for denying on an empty result. +func TestGetRunWorkspaceID_ResolvesToEmptyStringForUnscopedAgentProfile(t *testing.T) { + repo, writer := newTestRepoWithDB(t) + ctx := context.Background() + + seedAgentProfile(t, writer, "a1", "") + run := seedTaskRun(t, repo, "run-1", "a1", "t1", "queued") + + got, err := repo.GetRunWorkspaceID(ctx, run.ID) + if err != nil { + t.Fatalf("get run workspace id: %v", err) + } + checkString(t, "workspace_id", got, "") +} diff --git a/docs/decisions/2026-07-24-opt-in-authentication.md b/docs/decisions/2026-07-24-opt-in-authentication.md index 2d30475b62..822ee2d6a8 100644 --- a/docs/decisions/2026-07-24-opt-in-authentication.md +++ b/docs/decisions/2026-07-24-opt-in-authentication.md @@ -85,7 +85,8 @@ out via `webhooks[].public: true`. `auth_identities`; no migration required. - Office workspace-scoped HTTP routes are ownership-gated via a group middleware (agent-JWT callers keep their workspace-claim scoping); office - run subscriptions remain unchecked — accepted gap, noted in the spec. + run subscriptions (`run.subscribe`) are now ownership-gated too, closed by + WO-02 — see `apps/backend/AGENTS.md` and the spec. - The session-access check is enforced at the lifecycle chokepoint AND at the reverse-proxy handlers (vscode/port) that resolve executions by bare lookup; message read/search and repository-script routes carry explicit diff --git a/docs/specs/office/live-updates.md b/docs/specs/office/live-updates.md index c824ffbc6d..b17e1dfcb2 100644 --- a/docs/specs/office/live-updates.md +++ b/docs/specs/office/live-updates.md @@ -288,7 +288,7 @@ The kandev backend runs in a single-user model (`store.DefaultUserID = "default- - **Workspace scoping** — every office WS notification carries `workspace_id` in its payload (when scoped). The current implementation broadcasts office notifications to every connected client and the frontend filters by `workspace_id === workspaces.activeId`. The observable contract is: a client viewing workspace A MUST NOT act on events originating in workspace B. Either server-side filtering or client-side filtering satisfies the contract. - **User subscription** — `user.subscribe` accepts only the caller's own user id. Subscribing to another user's id returns WS error `forbidden` (`ws.ErrorCodeForbidden`, "cannot subscribe to another user"). Symmetric rule for `user.unsubscribe`. -- **Task / session / run subscriptions** — no per-key authorization in the current implementation. Any connected client can subscribe to any `task_id`, `session_id`, or `run_id`. Information-leak risk is bounded by the single-user model. A future multi-user iteration MUST gate `task.subscribe` / `session.subscribe` / `run.subscribe` on workspace membership; the gate point is `Hub.SubscribeToTask` etc. +- **Task / session / run subscriptions** — auth-enabled connections apply resource-specific access hooks before registration. Task and session use `AuthorizeTaskAccess` and `AuthorizeSessionAccess`. Runs resolve workspace and use `AuthorizeWorkspaceAccess`, denying an empty workspace. Auth-disabled connections retain single-user behavior. See `apps/backend/AGENTS.md`. - **Agent-originated subscriptions** — agentctl-initiated MCP tool calls reach the backend over a tunnelled WS connection but do **not** carry a separate identity over this surface; agents cannot subscribe to other agents' streams because subscription frames are scoped per-socket and agents never open the user-facing office WS. - **Subscription requests are not approvals** — there is no approval / review gate on subscribing. Once authorized to connect, a client may subscribe and unsubscribe freely. diff --git a/docs/specs/spec-lint.json b/docs/specs/spec-lint.json index f746185ba5..98925aa0c5 100644 --- a/docs/specs/spec-lint.json +++ b/docs/specs/spec-lint.json @@ -25,7 +25,7 @@ "docs/specs/office/agents.md": 61654, "docs/specs/office/automations-settings.md": 33607, "docs/specs/office/costs.md": 33263, - "docs/specs/office/live-updates.md": 38602, + "docs/specs/office/live-updates.md": 38568, "docs/specs/office/overview.md": 34375, "docs/specs/office/routing.md": 40782, "docs/specs/office/scheduler.md": 50967,