fix(security): scope task get, create, and delete to the owner

go/api/database/client.go

@@ -2,13 +2,18 @@ package database
 
 import (
 	"context"
+	"errors"
 	"time"
 
 	a2a "github.com/a2aproject/a2a-go/v2/a2a"
 	"github.com/kagent-dev/kagent/go/api/v1alpha2"
 	"github.com/pgvector/pgvector-go"
 )
 
+// ErrTaskOwnedByAnotherUser means a task with this id already belongs to a
+// different user.
+var ErrTaskOwnedByAnotherUser = errors.New("task id owned by another user")
+
 type QueryOptions struct {
 	Limit    int
 	After    time.Time
@@ -24,7 +29,7 @@ type Client interface {
 	StoreFeedback(ctx context.Context, feedback *Feedback) error
 	StoreSession(ctx context.Context, session *Session) error
 	StoreAgent(ctx context.Context, agent *Agent) error
-	StoreTask(ctx context.Context, task *a2a.Task) error
+	StoreTask(ctx context.Context, task *a2a.Task, userID string) error
 	StorePushNotification(ctx context.Context, config *a2a.PushConfig) error
 	StoreToolServer(ctx context.Context, toolServer *ToolServer) (*ToolServer, error)
 	StoreEvents(ctx context.Context, messages ...*Event) error
@@ -33,14 +38,14 @@ type Client interface {
 	DeleteSession(ctx context.Context, sessionID string, userID string) error
 	DeleteAgent(ctx context.Context, agentID string) error
 	DeleteToolServer(ctx context.Context, serverName string, groupKind string) error
-	DeleteTask(ctx context.Context, taskID string) error
+	DeleteTask(ctx context.Context, taskID string, userID string) error
 	DeletePushNotification(ctx context.Context, taskID string) error
 	DeleteToolsForServer(ctx context.Context, serverName string, groupKind string) error
 
 	// Get methods
 	GetSession(ctx context.Context, sessionID string, userID string) (*Session, error)
 	GetAgent(ctx context.Context, name string) (*Agent, error)
-	GetTask(ctx context.Context, id string) (*a2a.Task, error)
+	GetTask(ctx context.Context, id string, userID string) (*a2a.Task, error)
 	GetTool(ctx context.Context, name string) (*Tool, error)
 	GetToolServer(ctx context.Context, name string) (*ToolServer, error)
 	GetPushNotification(ctx context.Context, taskID string, configID string) (*a2a.PushConfig, error)

go/core/internal/database/client_postgres.go

@@ -272,7 +272,7 @@ func (c *postgresClient) ListEventsForSession(ctx context.Context, sessionID, us
 
 // TODO(0.11.0): Switch task writes to v1 storage format and remove legacy conversion from this write path.
 // NOTE: We will still need to keep the read compatibility for legacy rows in 0.11.0
-func (c *postgresClient) StoreTask(ctx context.Context, task *a2a.Task) error {
+func (c *postgresClient) StoreTask(ctx context.Context, task *a2a.Task, userID string) error {
 	legacyTask, err := trpcv0.ToLegacyTask(task)
 	if err != nil {
 		return fmt.Errorf("failed to convert task to legacy format: %w", err)
@@ -281,16 +281,39 @@ func (c *postgresClient) StoreTask(ctx context.Context, task *a2a.Task) error {
 	if err != nil {
 		return fmt.Errorf("failed to serialize task: %w", err)
 	}
-	return c.q.UpsertTask(ctx, dbgen.UpsertTaskParams{
+	// The WHERE clause on UpsertTask's ON CONFLICT is what actually stops a
+	// cross-user takeover; this check only turns that into a clean error.
+	if err := c.q.UpsertTask(ctx, dbgen.UpsertTaskParams{
 		ID:              string(task.ID),
 		Data:            string(data),
 		SessionID:       strPtrIfNotEmpty(task.ContextID),
 		ProtocolVersion: nil,
-	})
+		UserID:          &userID,
+	}); err != nil {
+		return fmt.Errorf("failed to store task %s: %w", task.ID, err)
+	}
+	return c.checkTaskOwner(ctx, string(task.ID), userID)
+}
+
+// checkTaskOwner returns ErrTaskOwnedByAnotherUser if taskID exists and
+// belongs to someone else. A missing task is not an error: the caller
+// decides what that means (nothing to own yet, or already deleted).
+func (c *postgresClient) checkTaskOwner(ctx context.Context, taskID, userID string) error {
+	owner, err := c.q.GetTaskOwner(ctx, taskID)
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return nil
+		}
+		return fmt.Errorf("failed to check task owner for %s: %w", taskID, err)
+	}
+	if owner != nil && *owner != userID {
+		return dbpkg.ErrTaskOwnedByAnotherUser
+	}
+	return nil
 }
 
-func (c *postgresClient) GetTask(ctx context.Context, taskID string) (*a2a.Task, error) {
-	row, err := c.q.GetTask(ctx, taskID)
+func (c *postgresClient) GetTask(ctx context.Context, taskID, userID string) (*a2a.Task, error) {
+	row, err := c.q.GetTask(ctx, dbgen.GetTaskParams{ID: taskID, UserID: &userID})
 	if err != nil {
 		return nil, fmt.Errorf("failed to get task %s: %w", taskID, err)
 	}
@@ -313,8 +336,13 @@ func (c *postgresClient) ListTasksForSession(ctx context.Context, sessionID stri
 	return tasks, nil
 }
 
-func (c *postgresClient) DeleteTask(ctx context.Context, taskID string) error {
-	return c.q.SoftDeleteTask(ctx, taskID)
+func (c *postgresClient) DeleteTask(ctx context.Context, taskID, userID string) error {
+	if err := c.q.SoftDeleteTask(ctx, dbgen.SoftDeleteTaskParams{ID: taskID, UserID: &userID}); err != nil {
+		return fmt.Errorf("failed to delete task %s: %w", taskID, err)
+	}
+	// The delete only takes effect if user_id matched; if the task is still
+	// there and owned by someone else, say so instead of a misleading success.
+	return c.checkTaskOwner(ctx, taskID, userID)
 }
 
 // ── Push Notifications ────────────────────────────────────────────────────────

go/core/internal/database/client_test.go

@@ -328,14 +328,38 @@ func TestStoreTaskTouchesSessionActivity(t *testing.T) {
 	err = client.StoreTask(ctx, &a2a.Task{
 		ID:        "task-1",
 		ContextID: sessionID,
-	})
+	}, userID)
 	require.NoError(t, err)
 
 	got, err := client.GetSession(ctx, sessionID, userID)
 	require.NoError(t, err)
 	assert.True(t, got.UpdatedAt.After(before.UpdatedAt), "session updated_at should advance after storing a task")
 }
 
+func TestTaskAccessIsScopedToOwner(t *testing.T) {
+	db := setupTestDB(t)
+	client := NewClient(db)
+	ctx := context.Background()
+
+	require.NoError(t, client.StoreTask(ctx, &a2a.Task{ID: "task-owned"}, "user-a"))
+
+	_, err := client.GetTask(ctx, "task-owned", "user-b")
+	require.Error(t, err, "another user must not read this task")
+
+	err = client.DeleteTask(ctx, "task-owned", "user-b")
+	require.ErrorIs(t, err, dbpkg.ErrTaskOwnedByAnotherUser, "another user must not delete this task")
+
+	task, err := client.GetTask(ctx, "task-owned", "user-a")
+	require.NoError(t, err, "task must still exist after another user's delete attempt")
+	assert.Equal(t, a2a.TaskID("task-owned"), task.ID)
+
+	err = client.StoreTask(ctx, &a2a.Task{ID: "task-owned"}, "user-b")
+	require.ErrorIs(t, err, dbpkg.ErrTaskOwnedByAnotherUser, "another user must not take over this task id")
+
+	require.NoError(t, client.DeleteTask(ctx, "task-owned", "user-a"), "the real owner can delete it")
+	require.NoError(t, client.DeleteTask(ctx, "task-owned", "user-a"), "deleting an already-gone task is not an error")
+}
+
 // TestStoreAgentIdempotence verifies that calling StoreAgent multiple times
 // with the same data is idempotent and doesn't error. This is critical for
 // the lock-free concurrency model where concurrent upserts must succeed.

go/core/internal/database/queries/tasks.sql

@@ -1,8 +1,11 @@
 -- name: GetTask :one
 SELECT * FROM task
-WHERE id = $1 AND deleted_at IS NULL
+WHERE id = $1 AND user_id = $2 AND deleted_at IS NULL
 LIMIT 1;
 
+-- name: GetTaskOwner :one
+SELECT user_id FROM task WHERE id = $1 AND deleted_at IS NULL LIMIT 1;
+
 -- name: TaskExists :one
 SELECT EXISTS (
     SELECT 1 FROM task WHERE id = $1 AND deleted_at IS NULL
@@ -15,13 +18,14 @@ ORDER BY created_at ASC;
 
 -- name: UpsertTask :exec
 WITH upserted_task AS (
-INSERT INTO task (id, data, session_id, protocol_version, created_at, updated_at)
-VALUES ($1, $2, $3, $4, NOW(), NOW())
+INSERT INTO task (id, data, session_id, protocol_version, user_id, created_at, updated_at)
+VALUES ($1, $2, $3, $4, $5, NOW(), NOW())
 ON CONFLICT (id) DO UPDATE SET
     data             = EXCLUDED.data,
     session_id       = EXCLUDED.session_id,
     protocol_version = EXCLUDED.protocol_version,
     updated_at       = NOW()
+WHERE task.user_id = EXCLUDED.user_id OR task.user_id IS NULL
 RETURNING session_id
 )
 UPDATE session
@@ -32,4 +36,4 @@ WHERE upserted_task.session_id IS NOT NULL
   AND session.deleted_at IS NULL;
 
 -- name: SoftDeleteTask :exec
-UPDATE task SET deleted_at = NOW() WHERE id = $1 AND deleted_at IS NULL;
+UPDATE task SET deleted_at = NOW() WHERE id = $1 AND user_id = $2 AND deleted_at IS NULL;

go/core/internal/httpserver/handlers/sessions_test.go

@@ -682,11 +682,11 @@ func TestSessionsHandler(t *testing.T) {
 			require.NoError(t, dbClient.StoreTask(context.Background(), &a2a.Task{
 				ID:        "task-1",
 				ContextID: sessionID,
-			}))
+			}, userID))
 			require.NoError(t, dbClient.StoreTask(context.Background(), &a2a.Task{
 				ID:        "task-2",
 				ContextID: sessionID,
-			}))
+			}, userID))
 
 			req := httptest.NewRequest("GET", "/api/sessions/"+sessionID+"/tasks", nil)
 			req = mux.SetURLVars(req, map[string]string{"session_id": sessionID})