From d11b9c177dce632fe0849e9e033354d63f29a817 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 04:02:10 +0100 Subject: [PATCH] fix(api): pin the approval token inside the delayed-approval status CAS The email-link approve with a purchase delay validated the token against the execution loaded at the start of the request, then scheduled the row with a status-only CAS. A token rotated by a re-notification (or expired) between the load and the CAS could still schedule the purchase. Add TransitionExecutionStatusTokenPinned, which checks approval_token_hash and approval_token_expires_at in the same UPDATE as the status change and reports a lost token pin as ErrApprovalTokenChanged. The delayed token path uses it and answers 403; the session path keeps the plain status CAS. Closes #590 Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 4 + internal/api/handler_purchases.go | 38 +++++++-- .../api/handler_purchases_delay_token_test.go | 53 ++++++++++-- internal/api/handler_purchases_test.go | 6 +- internal/config/errors.go | 7 ++ internal/config/interfaces.go | 8 ++ internal/config/store_postgres.go | 71 +++++++++++----- internal/config/store_postgres_db_test.go | 82 +++++++++++++++++++ .../config/store_postgres_pgxmock_test.go | 2 +- internal/mocks/stores.go | 17 ++++ internal/server/test_helpers_test.go | 4 + 11 files changed, 252 insertions(+), 40 deletions(-) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 552bbe4d5..eee0fad13 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -311,6 +311,10 @@ func (m *mockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, e return nil, nil } +func (m *mockConfigStore) TransitionExecutionStatusTokenPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, rawToken string) (*config.PurchaseExecution, error) { + return nil, nil +} + func (m *mockConfigStore) RescheduleAwaitingExecution(_ context.Context, _ string, _ time.Time) (bool, error) { return false, nil } diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go index fe5161b11..bfd40be6d 100644 --- a/internal/api/handler_purchases.go +++ b/internal/api/handler_purchases.go @@ -763,11 +763,14 @@ func (h *Handler) approveViaToken(ctx context.Context, req *events.LambdaFunctio } if globalCfg.GetPurchaseDelay() > 0 { // The delayed branch bypasses ApproveExecution, which is where the - // token is otherwise validated (issue #586). + // token is otherwise validated (issue #586). This check reads the + // snapshot loaded at the start of the request; approveWithDelay pins + // the token again inside the status CAS so a token rotated since + // cannot win the transition (issue #590). if err := purchase.ValidateApprovalToken(execution, token); err != nil { return nil, tokenActionError(err) } - return h.approveWithDelay(ctx, execution, globalCfg.GetPurchaseDelay(), actor, nil) + return h.approveWithDelay(ctx, execution, globalCfg.GetPurchaseDelay(), actor, nil, token) } // ApproveExecution now runs the purchase synchronously inside the // same call (issue #372). When it returns nil the AWS API call @@ -856,7 +859,7 @@ func (h *Handler) approvePurchaseViaSession(ctx context.Context, req *events.Lam return nil, fmt.Errorf("failed to read global config for purchase delay check: %w", cfgErr) } if globalCfg.GetPurchaseDelay() > 0 { - return h.approveWithDelay(ctx, execution, globalCfg.GetPurchaseDelay(), session.Email, actor) + return h.approveWithDelay(ctx, execution, globalCfg.GetPurchaseDelay(), session.Email, actor, "") } // revocationToken is the raw token ApproveAndExecute mints for this @@ -1051,9 +1054,17 @@ func (h *Handler) requirePlanStepCap(ctx context.Context, execution *config.Purc // scheduled_execution_at <= NOW() and fires the actual SDK call. // Revoking a status=scheduled execution (via the revoke handler or the // History "Revoke" button) transitions it to "canceled" at zero cloud cost. -func (h *Handler) approveWithDelay(ctx context.Context, execution *config.PurchaseExecution, delay time.Duration, actor string, transitionedBy *string) (any, error) { - updated, err := h.scheduleApprovedExecution(ctx, execution, delay, actor, transitionedBy) +// +// approvalToken is the raw email-link token on the token path and "" on the +// session path; see scheduleApprovedExecution. +func (h *Handler) approveWithDelay(ctx context.Context, execution *config.PurchaseExecution, delay time.Duration, actor string, transitionedBy *string, approvalToken string) (any, error) { + updated, err := h.scheduleApprovedExecution(ctx, execution, delay, actor, transitionedBy, approvalToken) if err != nil { + if errors.Is(err, config.ErrApprovalTokenChanged) { + // The token was rotated or expired after approveViaToken validated + // it against the request's snapshot (issue #590). + return nil, NewClientError(403, "invalid approval token") + } if errors.Is(err, config.ErrExecutionNotInExpectedStatus) { // A concurrent Cancel beat the approve: the CAS rejected because the row // is no longer in a schedulable status. Return 409 so the client knows to @@ -1090,8 +1101,21 @@ func (h *Handler) approveWithDelay(ctx context.Context, execution *config.Purcha // "canceled" between the caller's SELECT and this write, TransitionExecutionStatus // returns ErrExecutionNotInExpectedStatus and we surface a 409 instead of // blindly overwriting the canceled state. -func (h *Handler) scheduleApprovedExecution(ctx context.Context, execution *config.PurchaseExecution, delay time.Duration, actor string, transitionedBy *string) (*config.PurchaseExecution, error) { - updated, err := h.config.TransitionExecutionStatus(ctx, execution.ExecutionID, []string{"pending", "notified"}, "scheduled", transitionedBy) +// +// A non-empty approvalToken (the email-link path) is also pinned inside the +// CAS: the row's stored token hash must still match it and be unexpired, so +// a re-notification that rotated the token after the caller validated its +// snapshot rejects the transition with config.ErrApprovalTokenChanged +// (issue #590). The session path passes "" and is not token-gated. +func (h *Handler) scheduleApprovedExecution(ctx context.Context, execution *config.PurchaseExecution, delay time.Duration, actor string, transitionedBy *string, approvalToken string) (*config.PurchaseExecution, error) { + from := []string{"pending", "notified"} + var updated *config.PurchaseExecution + var err error + if approvalToken != "" { + updated, err = h.config.TransitionExecutionStatusTokenPinned(ctx, execution.ExecutionID, from, "scheduled", transitionedBy, approvalToken) + } else { + updated, err = h.config.TransitionExecutionStatus(ctx, execution.ExecutionID, from, "scheduled", transitionedBy) + } if err != nil { return nil, fmt.Errorf("failed to transition execution %s to scheduled: %w", execution.ExecutionID, err) } diff --git a/internal/api/handler_purchases_delay_token_test.go b/internal/api/handler_purchases_delay_token_test.go index 8946e141f..3f35efaf3 100644 --- a/internal/api/handler_purchases_delay_token_test.go +++ b/internal/api/handler_purchases_delay_token_test.go @@ -2,6 +2,7 @@ package api import ( "context" + "fmt" "testing" "time" @@ -16,7 +17,7 @@ import ( // token exactly like the immediate branch (issue #586), reusing the harness // from handler_purchases_token_errors_test.go. -func delayTokenHandler(t *testing.T, exec *config.PurchaseExecution, delayHours int) *Handler { +func delayTokenHandler(t *testing.T, exec *config.PurchaseExecution) *Handler { t.Helper() h := tokenErrHandler(t, exec, nil) store := h.config.(*MockConfigStore) @@ -29,7 +30,7 @@ func delayTokenHandler(t *testing.T, exec *config.PurchaseExecution, delayHours store.ExpectedCalls = kept approver := "admin@example.com" store.On("GetGlobalConfig", mock.Anything). - Return(&config.GlobalConfig{NotificationEmail: &approver, PurchaseDelayHours: delayHours}, nil) + Return(&config.GlobalConfig{NotificationEmail: &approver, PurchaseDelayHours: 1}, nil) return h } @@ -51,17 +52,22 @@ func TestHandleRequest_ApproveViaToken_DelayConfigured_ValidatesToken(t *testing for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { exec := tc.exec - h := delayTokenHandler(t, &exec, 1) + h := delayTokenHandler(t, &exec) store := h.config.(*MockConfigStore) // Registered so a regression answers 200 instead of panicking on an unmocked call. + scheduled := &config.PurchaseExecution{ExecutionID: tokenErrExecID, Status: "scheduled"} store.On("TransitionExecutionStatus", mock.Anything, tokenErrExecID, mock.Anything, "scheduled", mock.Anything). - Return(&config.PurchaseExecution{ExecutionID: tokenErrExecID, Status: "scheduled"}, nil).Maybe() + Return(scheduled, nil).Maybe() + store.On("TransitionExecutionStatusTokenPinned", mock.Anything, tokenErrExecID, mock.Anything, "scheduled", mock.Anything, mock.Anything). + Return(scheduled, nil).Maybe() store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil).Maybe() resp, err := h.HandleRequest(context.Background(), tokenErrRequest("approve", tc.token)) require.NoError(t, err) assert.Equal(t, tc.want, resp.StatusCode, resp.Body) store.AssertNotCalled(t, "TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) + store.AssertNotCalled(t, "TransitionExecutionStatusTokenPinned", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) }) } } @@ -69,25 +75,54 @@ func TestHandleRequest_ApproveViaToken_DelayConfigured_ValidatesToken(t *testing func TestHandleRequest_ApproveViaToken_DelayConfigured_ValidTokenSchedules(t *testing.T) { future := time.Now().Add(time.Hour) exec := &config.PurchaseExecution{Status: "pending", ApprovalToken: config.HashApprovalToken(tokenErrRawToken), ApprovalTokenExpiresAt: &future} - h := delayTokenHandler(t, exec, 1) + h := delayTokenHandler(t, exec) store := h.config.(*MockConfigStore) scheduled := &config.PurchaseExecution{ExecutionID: tokenErrExecID, Status: "scheduled"} - store.On("TransitionExecutionStatus", mock.Anything, tokenErrExecID, []string{"pending", "notified"}, "scheduled", mock.Anything). - Return(scheduled, nil) + // The CAS must carry the supplied token so the store re-checks it against + // the row at transition time, not only against the request's snapshot + // (issue #590). + store.On("TransitionExecutionStatusTokenPinned", mock.Anything, tokenErrExecID, []string{"pending", "notified"}, "scheduled", mock.Anything, tokenErrRawToken). + Return(scheduled, nil).Once() store.On("SavePurchaseExecution", mock.Anything, mock.Anything).Return(nil) resp, err := h.HandleRequest(context.Background(), tokenErrRequest("approve", tokenErrRawToken)) require.NoError(t, err) assert.Equal(t, 200, resp.StatusCode, resp.Body) assert.Contains(t, resp.Body, "scheduled") + store.AssertExpectations(t) + store.AssertNotCalled(t, "TransitionExecutionStatus", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) +} + +// TestHandleRequest_ApproveViaToken_DelayConfigured_TokenRotatedBeforeCASIs403 +// covers issue #590: the request's snapshot still holds the hash of the +// supplied token, so the pre-check passes, but a re-notification rotated the +// stored token before the status CAS. The token-pinned CAS rejects it, and the +// handler must answer 403 without scheduling or stamping the row. +func TestHandleRequest_ApproveViaToken_DelayConfigured_TokenRotatedBeforeCASIs403(t *testing.T) { + future := time.Now().Add(time.Hour) + exec := &config.PurchaseExecution{Status: "notified", ApprovalToken: config.HashApprovalToken(tokenErrRawToken), ApprovalTokenExpiresAt: &future} + h := delayTokenHandler(t, exec) + store := h.config.(*MockConfigStore) + store.On("TransitionExecutionStatusTokenPinned", mock.Anything, tokenErrExecID, []string{"pending", "notified"}, "scheduled", mock.Anything, tokenErrRawToken). + Return(nil, fmt.Errorf("%w: execution %s", config.ErrApprovalTokenChanged, tokenErrExecID)).Once() + + resp, err := h.HandleRequest(context.Background(), tokenErrRequest("approve", tokenErrRawToken)) + require.NoError(t, err) + assert.Equal(t, 403, resp.StatusCode, resp.Body) + assert.Contains(t, resp.Body, "invalid approval token") + store.AssertExpectations(t) + store.AssertNotCalled(t, "TransitionExecutionStatus", + mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) } func TestHandleRequest_ApproveViaToken_DelayConfigured_ReplayOfScheduledIs409(t *testing.T) { future := time.Now().Add(time.Hour) exec := &config.PurchaseExecution{Status: "scheduled", ApprovalToken: config.HashApprovalToken(tokenErrRawToken), ApprovalTokenExpiresAt: &future} - h := delayTokenHandler(t, exec, 1) + h := delayTokenHandler(t, exec) store := h.config.(*MockConfigStore) - store.On("TransitionExecutionStatus", mock.Anything, tokenErrExecID, []string{"pending", "notified"}, "scheduled", mock.Anything). + store.On("TransitionExecutionStatusTokenPinned", mock.Anything, tokenErrExecID, []string{"pending", "notified"}, "scheduled", mock.Anything, tokenErrRawToken). Return(nil, config.ErrExecutionNotInExpectedStatus) resp, err := h.HandleRequest(context.Background(), tokenErrRequest("approve", tokenErrRawToken)) diff --git a/internal/api/handler_purchases_test.go b/internal/api/handler_purchases_test.go index 8c093167a..9d842fba5 100644 --- a/internal/api/handler_purchases_test.go +++ b/internal/api/handler_purchases_test.go @@ -5614,7 +5614,7 @@ func TestHandler_scheduleApprovedExecution_CASGuardsConcurrentCancel(t *testing. handler := &Handler{config: mockConfig} - _, err := handler.scheduleApprovedExecution(ctx, exec, 48*time.Hour, "actor@example.com", nil) + _, err := handler.scheduleApprovedExecution(ctx, exec, 48*time.Hour, "actor@example.com", nil, "") require.Error(t, err, "concurrent cancel must surface as an error, not a silent overwrite") // SavePurchaseExecution must NEVER be called: the canceled row is untouched. mockConfig.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) @@ -5650,7 +5650,7 @@ func TestHandler_scheduleApprovedExecution_HappyPath(t *testing.T) { handler := &Handler{config: mockConfig} - result, err := handler.scheduleApprovedExecution(ctx, exec, 48*time.Hour, "actor@example.com", nil) + result, err := handler.scheduleApprovedExecution(ctx, exec, 48*time.Hour, "actor@example.com", nil, "") require.NoError(t, err) assert.Equal(t, "scheduled", result.Status) assert.NotNil(t, result.ScheduledExecutionAt, "ScheduledExecutionAt must be stamped") @@ -5674,7 +5674,7 @@ func TestApproveWithDelay_CASLostMaps409(t *testing.T) { handler := &Handler{config: mockConfig} - _, err := handler.approveWithDelay(ctx, exec, 48*time.Hour, "actor@example.com", nil) + _, err := handler.approveWithDelay(ctx, exec, 48*time.Hour, "actor@example.com", nil, "") require.Error(t, err) ce, ok := IsClientError(err) require.True(t, ok, "CAS-lost error must be a ClientError") diff --git a/internal/config/errors.go b/internal/config/errors.go index 0782f6f74..0831dcbc5 100644 --- a/internal/config/errors.go +++ b/internal/config/errors.go @@ -23,6 +23,13 @@ var ErrPurchasePlanConflict = errors.New("purchase plan changed") // distinguish this legitimate race-loss from a hard DB error. var ErrExecutionNotInExpectedStatus = errors.New("execution not in expected status") +// ErrApprovalTokenChanged is returned (wrapped) by +// TransitionExecutionStatusTokenPinned when the execution is still in an +// allowed status but its stored approval token no longer matches the one the +// caller supplied, or has expired: the token was rotated (re-notification) or +// lapsed after the caller validated it and before the transition (issue #590). +var ErrApprovalTokenChanged = errors.New("approval token no longer matches the execution") + // ErrRampStepIncomplete is returned (wrapped) by CompletePlanStep when the // ramp step the completing execution belongs to has not been bought in full: // another cloud account on that step is still outstanding, or nothing on the diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index ae891bf18..28a9378d5 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -225,6 +225,14 @@ type StoreInterface interface { // row whose recommendations were attached after the caller's gate check. A // lost pin returns ErrExecutionNotInExpectedStatus like any lost CAS. TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) + // TransitionExecutionStatusTokenPinned is TransitionExecutionStatus that + // also requires the row's stored approval token hash to match rawToken and + // to be unexpired, in the same statement as the status change, so a token + // rotated or expired after the caller validated it cannot win the + // transition (issue #590). A lost token pin on a row still in an allowed + // status returns ErrApprovalTokenChanged; a lost status CAS returns + // ErrExecutionNotInExpectedStatus as usual. + TransitionExecutionStatusTokenPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, rawToken string) (*PurchaseExecution, error) // RescheduleAwaitingExecution sets scheduled_date on an execution that is // still pending or notified and touches no other column. It returns false // (and writes nothing) when the row has left those statuses, so a cancel or diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 1edb42e29..a44e39304 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -9,6 +9,7 @@ import ( "hash/fnv" "math/big" "regexp" + "slices" "sort" "strings" "time" @@ -1191,7 +1192,7 @@ func (s *PostgresStore) RotatePendingApprovalToken(ctx context.Context, executio // and the legacy cancelled_by (issue #119), matching what the other cancel paths // record and what History reads. func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*PurchaseExecution, error) { - return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, nil) + return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, nil, nil) } // TransitionExecutionStatusPinned is TransitionExecutionStatus that also @@ -1200,7 +1201,21 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution // claim makes the pin fail with ErrExecutionNotInExpectedStatus, so a row is // never run on a snapshot different from the one that was gated. func (s *PostgresStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) { - return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, &sawEmptyRecommendations) + return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, &sawEmptyRecommendations, nil) +} + +// TransitionExecutionStatusTokenPinned is TransitionExecutionStatus that also +// requires approval_token_hash to equal the hash of rawToken and the token to +// be unexpired, checked in the same UPDATE as the status change. The caller's +// earlier token check ran against a snapshot; a re-notification that rotates +// the token between that read and this write would otherwise still let the +// old token win the transition (issue #590). An empty rawToken never matches. +func (s *PostgresStore) TransitionExecutionStatusTokenPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, rawToken string) (*PurchaseExecution, error) { + if rawToken == "" { + return nil, fmt.Errorf("%w: execution %s: no approval token supplied", ErrApprovalTokenChanged, executionID) + } + tokenHash := HashApprovalToken(rawToken) + return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, nil, &tokenHash) } // RescheduleAwaitingExecution is a conditional single-column update: the @@ -1215,7 +1230,7 @@ func (s *PostgresStore) RescheduleAwaitingExecution(ctx context.Context, executi return tag.RowsAffected() == 1, nil } -func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmpty *bool) (*PurchaseExecution, error) { +func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmpty *bool, tokenHash *string) (*PurchaseExecution, error) { var cancelActor *string if toStatus == StatusCanceled { cancelActor = actor @@ -1228,6 +1243,8 @@ func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, execution cancelled_by = COALESCE((SELECT email FROM users WHERE id = $5), cancelled_by) WHERE execution_id = $1 AND status = ANY($3) AND ($6::boolean IS NULL OR (recommendations IN ('[]'::jsonb, 'null'::jsonb)) = $6) + AND ($7::text IS NULL OR (approval_token_hash = $7 + AND (approval_token_expires_at IS NULL OR approval_token_expires_at > NOW()))) RETURNING plan_id, execution_id, status, step_number, scheduled_date, notification_sent, approval_token_hash, recommendations, total_upfront_cost, estimated_savings, completed_at, error, expires_at, @@ -1238,33 +1255,47 @@ func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, execution idempotency_key, scheduled_execution_at, no_submission ` - records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty) + records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty, tokenHash) if err != nil { return nil, err } if len(records) == 0 { - existing, existErr := s.GetExecutionByID(ctx, executionID) - if errors.Is(existErr, ErrNotFound) { - // Wrap ErrNotFound so callers (e.g. the purchase reaper) can - // use errors.Is to distinguish "row vanished mid-flight" — a - // legitimate CAS race-loss — from a hard DB error. - return nil, fmt.Errorf("%w: execution %s", ErrNotFound, executionID) - } - if existErr != nil { - // A hard DB error during the probe must NOT read as a benign - // race-loss: propagate it so callers see the outage. - return nil, fmt.Errorf("transition %s: probe after zero-row CAS failed: %w", executionID, existErr) - } - // Wrap ErrExecutionNotInExpectedStatus so callers can use - // errors.Is to recognize CAS rejection (status changed between - // SELECT and UPDATE) as race-lost rather than a real error. - return nil, fmt.Errorf("%w: execution %s cannot transition from %q to %q", ErrExecutionNotInExpectedStatus, executionID, existing.Status, toStatus) + return nil, s.transitionRejection(ctx, executionID, fromStatuses, toStatus, tokenHash != nil) } return &records[0], nil } +// transitionRejection explains a zero-row transitionExecutionStatus by probing +// the row: gone (ErrNotFound), a hard DB error, a token pin lost on a row still +// in an allowed status (ErrApprovalTokenChanged), or a status CAS lost +// (ErrExecutionNotInExpectedStatus). Extracted to keep the caller under the +// gocyclo threshold. +func (s *PostgresStore) transitionRejection(ctx context.Context, executionID string, fromStatuses []string, toStatus string, tokenPinned bool) error { + existing, existErr := s.GetExecutionByID(ctx, executionID) + if errors.Is(existErr, ErrNotFound) { + // Wrap ErrNotFound so callers (e.g. the purchase reaper) can + // use errors.Is to distinguish "row vanished mid-flight" — a + // legitimate CAS race-loss — from a hard DB error. + return fmt.Errorf("%w: execution %s", ErrNotFound, executionID) + } + if existErr != nil { + // A hard DB error during the probe must NOT read as a benign + // race-loss: propagate it so callers see the outage. + return fmt.Errorf("transition %s: probe after zero-row CAS failed: %w", executionID, existErr) + } + // The status still allowed the transition, so the token pin is what + // rejected it: the stored token was rotated or expired (issue #590). + if tokenPinned && slices.Contains(fromStatuses, existing.Status) { + return fmt.Errorf("%w: execution %s", ErrApprovalTokenChanged, executionID) + } + // Wrap ErrExecutionNotInExpectedStatus so callers can use + // errors.Is to recognize CAS rejection (status changed between + // SELECT and UPDATE) as race-lost rather than a real error. + return fmt.Errorf("%w: execution %s cannot transition from %q to %q", ErrExecutionNotInExpectedStatus, executionID, existing.Status, toStatus) +} + // SetCancelledBy stamps both the canceled_by and legacy cancelled_by columns // for a single execution without touching any other column. This avoids the // full-row overwrite that a SavePurchaseExecution follow-up would perform, diff --git a/internal/config/store_postgres_db_test.go b/internal/config/store_postgres_db_test.go index b24e15d1f..2c5929853 100644 --- a/internal/config/store_postgres_db_test.go +++ b/internal/config/store_postgres_db_test.go @@ -1783,6 +1783,88 @@ func TestPostgresStoreDB_TransitionExecutionStatusPinned(t *testing.T) { }) } +// TestPostgresStoreDB_TransitionExecutionStatusTokenPinned (issue #590): the +// delayed email-link approval re-checks the approval token inside the status +// CAS, so a token rotated (re-notification) or expired after the caller read +// its snapshot loses the transition and leaves the row untouched, while the +// current token still wins it. +func TestPostgresStoreDB_TransitionExecutionStatusTokenPinned(t *testing.T) { + conn := setupTestContainerDB(t) + if conn == nil { + return + } + store := NewPostgresStore(conn) + ctx := context.Background() + plan := &PurchasePlan{Name: "token pinned", Enabled: true, Services: map[string]ServiceConfig{}, RampSchedule: PresetRampSchedules["immediate"]} + require.NoError(t, store.CreatePurchasePlan(ctx, plan)) + from := []string{"pending", "notified"} + const oldToken, newToken = "old-raw-token", "new-raw-token" + newRow := func(t *testing.T, expiresAt time.Time) *PurchaseExecution { + t.Helper() + exec := &PurchaseExecution{PlanID: plan.ID, Status: "notified", StepNumber: 1, ScheduledDate: time.Now(), + ApprovalToken: HashApprovalToken(oldToken), ApprovalTokenExpiresAt: &expiresAt} + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + return exec + } + requireUntouched := func(t *testing.T, id string) { + t.Helper() + reread, err := store.GetExecutionByID(ctx, id) + require.NoError(t, err) + assert.Equal(t, "notified", reread.Status, "a rejected transition must leave the row as it was") + } + + t.Run("token rotated after the snapshot loses the transition", func(t *testing.T) { + exec := newRow(t, time.Now().Add(time.Hour)) + snapshot, err := store.GetExecutionByID(ctx, exec.ExecutionID) + require.NoError(t, err) + require.True(t, ApprovalTokenMatches(snapshot.ApprovalToken, oldToken), "the caller's snapshot validates the old token") + rotated, err := store.RotatePendingApprovalToken(ctx, exec.ExecutionID, HashApprovalToken(newToken), time.Now().Add(time.Hour)) + require.NoError(t, err) + require.True(t, rotated) + + _, err = store.TransitionExecutionStatusTokenPinned(ctx, exec.ExecutionID, from, "scheduled", nil, oldToken) + require.ErrorIs(t, err, ErrApprovalTokenChanged) + requireUntouched(t, exec.ExecutionID) + + got, err := store.TransitionExecutionStatusTokenPinned(ctx, exec.ExecutionID, from, "scheduled", nil, newToken) + require.NoError(t, err) + assert.Equal(t, "scheduled", got.Status, "the current token still schedules the row") + }) + + t.Run("token expired after the snapshot loses the transition", func(t *testing.T) { + exec := newRow(t, time.Now().Add(-time.Minute)) + _, err := store.TransitionExecutionStatusTokenPinned(ctx, exec.ExecutionID, from, "scheduled", nil, oldToken) + require.ErrorIs(t, err, ErrApprovalTokenChanged) + requireUntouched(t, exec.ExecutionID) + }) + + t.Run("wrong or empty token loses the transition", func(t *testing.T) { + exec := newRow(t, time.Now().Add(time.Hour)) + for _, tok := range []string{"not-the-token", ""} { + _, err := store.TransitionExecutionStatusTokenPinned(ctx, exec.ExecutionID, from, "scheduled", nil, tok) + require.ErrorIs(t, err, ErrApprovalTokenChanged, "token %q", tok) + requireUntouched(t, exec.ExecutionID) + } + }) + + t.Run("a row with no stored token never matches", func(t *testing.T) { + exec := &PurchaseExecution{PlanID: plan.ID, Status: "notified", StepNumber: 1, ScheduledDate: time.Now()} + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + _, err := store.TransitionExecutionStatusTokenPinned(ctx, exec.ExecutionID, from, "scheduled", nil, oldToken) + require.ErrorIs(t, err, ErrApprovalTokenChanged) + requireUntouched(t, exec.ExecutionID) + }) + + t.Run("matching token on a row past the allowed statuses is a status CAS loss", func(t *testing.T) { + exec := newRow(t, time.Now().Add(time.Hour)) + _, err := store.TransitionExecutionStatusTokenPinned(ctx, exec.ExecutionID, from, "scheduled", nil, oldToken) + require.NoError(t, err) + _, err = store.TransitionExecutionStatusTokenPinned(ctx, exec.ExecutionID, from, "scheduled", nil, oldToken) + require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus, "a replay of a scheduled row is a status race loss") + require.NotErrorIs(t, err, ErrApprovalTokenChanged) + }) +} + // TestPostgresStoreDB_StepCapAndKillSwitchRoundTrip proves, against a real // PostgreSQL, that max_step_commitment round-trips through every plan read and // write path (NULL stays NULL) and that auto_purchase_enabled round-trips diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index f5c579d45..6e216d75c 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -2802,7 +2802,7 @@ func TestPGXMock_TransitionExecutionStatus_ProbeHardErrorNotMappedToNotFound(t * // CAS UPDATE matches zero rows (status already transitioned or row gone). mock.ExpectQuery(`UPDATE purchase_executions`). - WithArgs(anyArgsCfg(6)...). + WithArgs(anyArgsCfg(7)...). WillReturnRows(pgxmock.NewRows(execCols)) // Probe fails with a hard DB error, not an empty result. diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 2a27e1bcd..f578c310b 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -483,6 +483,23 @@ func (m *MockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, e return v, args.Error(1) } +// TransitionExecutionStatusTokenPinned mocks the token-pinned transition +// (issue #590). It does not delegate to TransitionExecutionStatus: a caller +// that must pin the approval token is expected to call this method, so a test +// has to register it explicitly. +func (m *MockConfigStore) TransitionExecutionStatusTokenPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, rawToken string) (*config.PurchaseExecution, error) { + m.record("TransitionExecutionStatusTokenPinned", ctx, executionID, fromStatuses, toStatus, actor, rawToken) + args := m.Called(ctx, executionID, fromStatuses, toStatus, actor, rawToken) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*config.PurchaseExecution) + if !ok { + panic(fmt.Sprintf("mock: expected *config.PurchaseExecution, got %T", args.Get(0))) + } + return v, args.Error(1) +} + // RescheduleAwaitingExecution mocks the conditional scheduled_date update. func (m *MockConfigStore) RescheduleAwaitingExecution(ctx context.Context, executionID string, scheduledDate time.Time) (bool, error) { args := m.Called(ctx, executionID, scheduledDate) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 7e6280f7e..80ad2ec67 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -162,6 +162,10 @@ func (m *mockConfigStoreForHealth) TransitionExecutionStatusPinned(ctx context.C return nil, nil } +func (m *mockConfigStoreForHealth) TransitionExecutionStatusTokenPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, rawToken string) (*config.PurchaseExecution, error) { + return nil, nil +} + func (m *mockConfigStoreForHealth) RescheduleAwaitingExecution(_ context.Context, _ string, _ time.Time) (bool, error) { return false, nil }