From 9af4ce4f61df6545a2489238b52782e93dd90e35 Mon Sep 17 00:00:00 2001 From: mesutoezdil Date: Wed, 15 Jul 2026 20:59:04 +0200 Subject: [PATCH] fix(security): scope task get, create, and delete to the owner Any user could read, overwrite, or delete any other user's task by id. Task now has a user_id column, backfilled from its session, and all three endpoints check it like session and event already do. Signed-off-by: mesutoezdil --- go/api/database/client.go | 11 +++-- go/core/internal/database/client_postgres.go | 42 ++++++++++++++--- go/core/internal/database/client_test.go | 26 ++++++++++- go/core/internal/database/gen/models.go | 1 + go/core/internal/database/gen/querier.go | 5 +- go/core/internal/database/gen/tasks.sql.go | 46 +++++++++++++++---- go/core/internal/database/queries/tasks.sql | 12 +++-- .../httpserver/handlers/sessions_test.go | 4 +- go/core/internal/httpserver/handlers/tasks.go | 34 ++++++++++++-- .../core/000008_task_owner.down.sql | 1 + .../migrations/core/000008_task_owner.up.sql | 20 ++++++++ 11 files changed, 170 insertions(+), 32 deletions(-) create mode 100644 go/core/pkg/migrations/core/000008_task_owner.down.sql create mode 100644 go/core/pkg/migrations/core/000008_task_owner.up.sql diff --git a/go/api/database/client.go b/go/api/database/client.go index e80dc8647f..7c6fd9d89c 100644 --- a/go/api/database/client.go +++ b/go/api/database/client.go @@ -2,6 +2,7 @@ package database import ( "context" + "errors" "time" a2a "github.com/a2aproject/a2a-go/v2/a2a" @@ -9,6 +10,10 @@ import ( "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) diff --git a/go/core/internal/database/client_postgres.go b/go/core/internal/database/client_postgres.go index 4e0bdc66d9..11a779fd4b 100644 --- a/go/core/internal/database/client_postgres.go +++ b/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 ──────────────────────────────────────────────────────── diff --git a/go/core/internal/database/client_test.go b/go/core/internal/database/client_test.go index a8c263d7de..76e7316f84 100644 --- a/go/core/internal/database/client_test.go +++ b/go/core/internal/database/client_test.go @@ -328,7 +328,7 @@ 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) @@ -336,6 +336,30 @@ func TestStoreTaskTouchesSessionActivity(t *testing.T) { 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. diff --git a/go/core/internal/database/gen/models.go b/go/core/internal/database/gen/models.go index 2211eb1ef2..11e32e0da2 100644 --- a/go/core/internal/database/gen/models.go +++ b/go/core/internal/database/gen/models.go @@ -150,6 +150,7 @@ type Task struct { Data string SessionID *string ProtocolVersion *string + UserID *string } type Tool struct { diff --git a/go/core/internal/database/gen/querier.go b/go/core/internal/database/gen/querier.go index 078a065021..bdd9ad9615 100644 --- a/go/core/internal/database/gen/querier.go +++ b/go/core/internal/database/gen/querier.go @@ -21,7 +21,8 @@ type Querier interface { GetPushNotification(ctx context.Context, arg GetPushNotificationParams) (PushNotification, error) GetSession(ctx context.Context, arg GetSessionParams) (Session, error) GetSessionShareByToken(ctx context.Context, token string) (SessionShare, error) - GetTask(ctx context.Context, id string) (Task, error) + GetTask(ctx context.Context, arg GetTaskParams) (Task, error) + GetTaskOwner(ctx context.Context, id string) (*string, error) GetTool(ctx context.Context, id string) (Tool, error) GetToolServer(ctx context.Context, name string) (Toolserver, error) HardDeleteCrewAIMemory(ctx context.Context, arg HardDeleteCrewAIMemoryParams) error @@ -62,7 +63,7 @@ type Querier interface { SoftDeleteEvent(ctx context.Context, id string) error SoftDeletePushNotification(ctx context.Context, taskID string) error SoftDeleteSession(ctx context.Context, arg SoftDeleteSessionParams) error - SoftDeleteTask(ctx context.Context, id string) error + SoftDeleteTask(ctx context.Context, arg SoftDeleteTaskParams) error SoftDeleteToolServer(ctx context.Context, arg SoftDeleteToolServerParams) error SoftDeleteToolsForServer(ctx context.Context, arg SoftDeleteToolsForServerParams) error TaskExists(ctx context.Context, id string) (bool, error) diff --git a/go/core/internal/database/gen/tasks.sql.go b/go/core/internal/database/gen/tasks.sql.go index be4e93d752..97e1b25c27 100644 --- a/go/core/internal/database/gen/tasks.sql.go +++ b/go/core/internal/database/gen/tasks.sql.go @@ -10,13 +10,18 @@ import ( ) const getTask = `-- name: GetTask :one -SELECT id, created_at, updated_at, deleted_at, data, session_id, protocol_version FROM task -WHERE id = $1 AND deleted_at IS NULL +SELECT id, created_at, updated_at, deleted_at, data, session_id, protocol_version, user_id FROM task +WHERE id = $1 AND user_id = $2 AND deleted_at IS NULL LIMIT 1 ` -func (q *Queries) GetTask(ctx context.Context, id string) (Task, error) { - row := q.db.QueryRow(ctx, getTask, id) +type GetTaskParams struct { + ID string + UserID *string +} + +func (q *Queries) GetTask(ctx context.Context, arg GetTaskParams) (Task, error) { + row := q.db.QueryRow(ctx, getTask, arg.ID, arg.UserID) var i Task err := row.Scan( &i.ID, @@ -26,12 +31,24 @@ func (q *Queries) GetTask(ctx context.Context, id string) (Task, error) { &i.Data, &i.SessionID, &i.ProtocolVersion, + &i.UserID, ) return i, err } +const getTaskOwner = `-- name: GetTaskOwner :one +SELECT user_id FROM task WHERE id = $1 AND deleted_at IS NULL LIMIT 1 +` + +func (q *Queries) GetTaskOwner(ctx context.Context, id string) (*string, error) { + row := q.db.QueryRow(ctx, getTaskOwner, id) + var user_id *string + err := row.Scan(&user_id) + return user_id, err +} + const listTasksForSession = `-- name: ListTasksForSession :many -SELECT id, created_at, updated_at, deleted_at, data, session_id, protocol_version FROM task +SELECT id, created_at, updated_at, deleted_at, data, session_id, protocol_version, user_id FROM task WHERE session_id = $1 AND deleted_at IS NULL ORDER BY created_at ASC ` @@ -53,6 +70,7 @@ func (q *Queries) ListTasksForSession(ctx context.Context, sessionID *string) ([ &i.Data, &i.SessionID, &i.ProtocolVersion, + &i.UserID, ); err != nil { return nil, err } @@ -65,11 +83,16 @@ func (q *Queries) ListTasksForSession(ctx context.Context, sessionID *string) ([ } const softDeleteTask = `-- 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 ` -func (q *Queries) SoftDeleteTask(ctx context.Context, id string) error { - _, err := q.db.Exec(ctx, softDeleteTask, id) +type SoftDeleteTaskParams struct { + ID string + UserID *string +} + +func (q *Queries) SoftDeleteTask(ctx context.Context, arg SoftDeleteTaskParams) error { + _, err := q.db.Exec(ctx, softDeleteTask, arg.ID, arg.UserID) return err } @@ -88,13 +111,14 @@ func (q *Queries) TaskExists(ctx context.Context, id string) (bool, error) { const upsertTask = `-- 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 @@ -110,6 +134,7 @@ type UpsertTaskParams struct { Data string SessionID *string ProtocolVersion *string + UserID *string } func (q *Queries) UpsertTask(ctx context.Context, arg UpsertTaskParams) error { @@ -118,6 +143,7 @@ func (q *Queries) UpsertTask(ctx context.Context, arg UpsertTaskParams) error { arg.Data, arg.SessionID, arg.ProtocolVersion, + arg.UserID, ) return err } diff --git a/go/core/internal/database/queries/tasks.sql b/go/core/internal/database/queries/tasks.sql index d41fcdf928..1ffb5f081f 100644 --- a/go/core/internal/database/queries/tasks.sql +++ b/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; diff --git a/go/core/internal/httpserver/handlers/sessions_test.go b/go/core/internal/httpserver/handlers/sessions_test.go index c2f7a6aac3..f3fdd037ea 100644 --- a/go/core/internal/httpserver/handlers/sessions_test.go +++ b/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}) diff --git a/go/core/internal/httpserver/handlers/tasks.go b/go/core/internal/httpserver/handlers/tasks.go index 7993794d14..7b4ef17f85 100644 --- a/go/core/internal/httpserver/handlers/tasks.go +++ b/go/core/internal/httpserver/handlers/tasks.go @@ -1,10 +1,12 @@ package handlers import ( + stderrors "errors" "fmt" "net/http" a2a "github.com/a2aproject/a2a-go/v2/a2a" + "github.com/kagent-dev/kagent/go/api/database" api "github.com/kagent-dev/kagent/go/api/httpapi" "github.com/kagent-dev/kagent/go/core/internal/httpserver/errors" "github.com/kagent-dev/kagent/go/core/internal/utils" @@ -33,7 +35,13 @@ func (h *TasksHandler) HandleGetTask(w ErrorResponseWriter, r *http.Request) { } log = log.WithValues("task_id", taskID) - task, err := h.DatabaseService.GetTask(r.Context(), taskID) + userID, err := getUserIDOrAgentUser(r) + if err != nil { + w.RespondWithError(errors.NewBadRequestError("Failed to get user ID", err)) + return + } + + task, err := h.DatabaseService.GetTask(r.Context(), taskID, userID) if err != nil { w.RespondWithError(errors.NewNotFoundError("Task not found", err)) return @@ -106,7 +114,17 @@ func (h *TasksHandler) HandleCreateTask(w ErrorResponseWriter, r *http.Request) } log = log.WithValues("task_id", task.ID) - if err := h.DatabaseService.StoreTask(r.Context(), &task); err != nil { + userID, err := getUserIDOrAgentUser(r) + if err != nil { + w.RespondWithError(errors.NewBadRequestError("Failed to get user ID", err)) + return + } + + if err := h.DatabaseService.StoreTask(r.Context(), &task, userID); err != nil { + if stderrors.Is(err, database.ErrTaskOwnedByAnotherUser) { + w.RespondWithError(errors.NewConflictError("Task ID is already in use", err)) + return + } w.RespondWithError(errors.NewInternalServerError("Failed to create task", err)) return } @@ -141,7 +159,17 @@ func (h *TasksHandler) HandleDeleteTask(w ErrorResponseWriter, r *http.Request) } log = log.WithValues("task_id", taskID) - if err := h.DatabaseService.DeleteTask(r.Context(), taskID); err != nil { + userID, err := getUserIDOrAgentUser(r) + if err != nil { + w.RespondWithError(errors.NewBadRequestError("Failed to get user ID", err)) + return + } + + if err := h.DatabaseService.DeleteTask(r.Context(), taskID, userID); err != nil { + if stderrors.Is(err, database.ErrTaskOwnedByAnotherUser) { + w.RespondWithError(errors.NewNotFoundError("Task not found", err)) + return + } w.RespondWithError(errors.NewInternalServerError("Failed to delete task", err)) return } diff --git a/go/core/pkg/migrations/core/000008_task_owner.down.sql b/go/core/pkg/migrations/core/000008_task_owner.down.sql new file mode 100644 index 0000000000..2005da4a83 --- /dev/null +++ b/go/core/pkg/migrations/core/000008_task_owner.down.sql @@ -0,0 +1 @@ +ALTER TABLE task DROP COLUMN IF EXISTS user_id; diff --git a/go/core/pkg/migrations/core/000008_task_owner.up.sql b/go/core/pkg/migrations/core/000008_task_owner.up.sql new file mode 100644 index 0000000000..9ace6e3eb6 --- /dev/null +++ b/go/core/pkg/migrations/core/000008_task_owner.up.sql @@ -0,0 +1,20 @@ +-- Task get/delete/create had no owner check at all: any user could read, +-- overwrite, or delete any other user's task by id. user_id lets queries +-- scope by owner the same way session and event already do. +ALTER TABLE task ADD COLUMN IF NOT EXISTS user_id TEXT; + +-- session.id is not globally unique (session's key is (id, user_id)), so only +-- backfill from a session id that maps to exactly one live user. Anything +-- ambiguous or gone is left NULL, which the new owner checks then hide from +-- everyone rather than guessing wrong. +WITH unique_session AS ( + SELECT id, MIN(user_id) AS user_id + FROM session + WHERE deleted_at IS NULL + GROUP BY id + HAVING COUNT(*) = 1 +) +UPDATE task +SET user_id = unique_session.user_id +FROM unique_session +WHERE unique_session.id = task.session_id AND task.user_id IS NULL;