From 922b8e2215d3a303a812ff65de9cebd68d92f72c Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Mon, 26 Oct 2026 22:04:20 +0000 Subject: [PATCH] fix(config): resolve arming grants before the global-config transaction PUT /api/config and PUT /api/ri-exchange/config checked execute:purchases and execute:ri-exchange inside the UpdateGlobalConfigAtomic closure. That closure runs while the transaction holds a pooled connection, and the permission lookup reads the auth store through the same pool, so as many concurrent arming saves as the pool has connections each held one and waited forever for a second. Both permission outcomes are now resolved through requirePermission before the transaction opens. The gates inside the closure still decide escalation against the advisory-locked row and only read the snapshot; a lookup error is raised only when the write escalates, as before. Closes #1517 Co-Authored-By: Claude Opus 5.5 --- internal/api/auto_purchase_grant.go | 65 +++++- .../api/global_config_arming_grants_test.go | 147 +++++++++++++ internal/api/handler.go | 26 ++- internal/api/handler_config.go | 16 +- ...ler_config_arming_pool_integration_test.go | 203 ++++++++++++++++++ internal/api/handler_ri_exchange.go | 5 +- internal/api/ladder_arming_grant.go | 8 +- 7 files changed, 445 insertions(+), 25 deletions(-) create mode 100644 internal/api/global_config_arming_grants_test.go create mode 100644 internal/api/handler_config_arming_pool_integration_test.go diff --git a/internal/api/auto_purchase_grant.go b/internal/api/auto_purchase_grant.go index 856c75f67..b55f960d2 100644 --- a/internal/api/auto_purchase_grant.go +++ b/internal/api/auto_purchase_grant.go @@ -72,10 +72,55 @@ func credentialKind(ctx context.Context, session *Session) string { // requireAutoPurchaseGrant refuses the admin API key and otherwise requires // execute:purchases, evaluated against the credential presented. func (h *Handler) requireAutoPurchaseGrant(ctx context.Context, req *events.LambdaFunctionURLRequest, session *Session) error { + if err := refuseAutoPurchaseCredential(session); err != nil { + return err + } + _, err := h.requirePermission(ctx, req, auth.ActionExecute, auth.ResourcePurchases) + return err +} + +// refuseAutoPurchaseCredential is the credential half of the auto-purchase +// grant: no session, or the stateless admin API key, can never arm. +func refuseAutoPurchaseCredential(session *Session) error { if session == nil || session.UserID == apiKeyAdminUserID { return NewClientError(403, autoPurchaseGrantDenied) } - _, err := h.requirePermission(ctx, req, auth.ActionExecute, auth.ResourcePurchases) + return nil +} + +// armingGrants holds the outcome (nil = granted) of the two requirePermission +// checks a global_config write may need to arm unattended spend: +// execute:purchases (auto-purchase and ladder execution switches) and +// execute:ri-exchange (RI exchange auto mode and its caps). +// +// They are resolved by resolveArmingGrants BEFORE UpdateGlobalConfigAtomic +// opens its transaction, and the gates inside the closure read them from here. +// The closure runs while the transaction holds a pooled connection; a +// requirePermission call there needs a second one from the same pool, so as +// many concurrent arming saves as the pool has connections deadlocked the +// instance (issue #1517). Whether a write escalates still depends on the +// advisory-locked row, so it is still decided inside the closure; only the +// permission lookup moved out. A lookup error is kept, not raised, so a write +// that escalates nothing is unaffected by it, exactly as when the lookup ran +// only on escalation. +type armingGrants struct { + purchases error + riExchange error +} + +// resolveArmingGrants evaluates both execute verbs against the credential +// presented, through requirePermission, so a user API key is held to its own +// effective permissions just as the in-transaction checks did. +func (h *Handler) resolveArmingGrants(ctx context.Context, req *events.LambdaFunctionURLRequest) armingGrants { + return armingGrants{ + purchases: h.resolveGrant(ctx, req, auth.ResourcePurchases), + riExchange: h.resolveGrant(ctx, req, auth.ResourceRIExchange), + } +} + +// resolveGrant returns the requirePermission outcome for execute on resource. +func (h *Handler) resolveGrant(ctx context.Context, req *events.LambdaFunctionURLRequest, resource string) error { + _, err := h.requirePermission(ctx, req, auth.ActionExecute, resource) return err } @@ -114,13 +159,15 @@ func formatCap(v *float64) string { // requireAutoPurchaseSwitchGrant gates and stamps a change of the global kill // switch inside the UpdateGlobalConfigAtomic closure. Only off->on needs the -// grant; any change is stamped so the last actor is always recorded. -func (h *Handler) requireAutoPurchaseSwitchGrant(ctx context.Context, req *events.LambdaFunctionURLRequest, session *Session, before bool, cfg *config.GlobalConfig) error { +// grant; any change is stamped so the last actor is always recorded. grant is +// the pre-resolved execute:purchases outcome (armingGrants.purchases): nothing +// here may read through the pool while the transaction holds a connection. +func requireAutoPurchaseSwitchGrant(ctx context.Context, session *Session, grant error, before bool, cfg *config.GlobalConfig) error { if before == cfg.AutoPurchaseEnabled { return nil } if cfg.AutoPurchaseEnabled { - if err := h.requireAutoPurchaseGrant(ctx, req, session); err != nil { + if err := prefetchedAutoPurchaseGrant(session, grant); err != nil { return err } } @@ -129,6 +176,16 @@ func (h *Handler) requireAutoPurchaseSwitchGrant(ctx context.Context, req *event return nil } +// prefetchedAutoPurchaseGrant is requireAutoPurchaseGrant with the +// execute:purchases lookup already done: the credential refusal first, then the +// pre-resolved permission outcome. +func prefetchedAutoPurchaseGrant(session *Session, grant error) error { + if err := refuseAutoPurchaseCredential(session); err != nil { + return err + } + return grant +} + // logAutoPurchaseSwitchChange records a committed change of the kill switch. func logAutoPurchaseSwitchChange(ctx context.Context, session *Session, before bool, after *config.GlobalConfig) { if after == nil || after.AutoPurchaseEnabled == before { diff --git a/internal/api/global_config_arming_grants_test.go b/internal/api/global_config_arming_grants_test.go new file mode 100644 index 000000000..5e7f93bf4 --- /dev/null +++ b/internal/api/global_config_arming_grants_test.go @@ -0,0 +1,147 @@ +package api + +import ( + "context" + "errors" + "testing" + + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// Issue #1517: the global_config write closures run inside the transaction +// UpdateGlobalConfigAtomic opens, which holds a pooled connection. Any auth +// lookup from there needs a second connection from the same pool, so enough +// concurrent arming saves deadlock the instance. These tests pin that the +// closure makes no auth call at all, while the grant outcomes stay the same. +// The real-pool reproduction is TestGlobalConfigArmingSavesDoNotExhaustPool. + +// authFreeTxStore models UpdateGlobalConfigAtomic and records how many auth +// calls the apply closure made while the "transaction" was open. +type authFreeTxStore struct { + *MockConfigStore + auth *MockAuthService + stored config.GlobalConfig + callsInTx int +} + +func (s *authFreeTxStore) UpdateGlobalConfigAtomic(_ context.Context, apply func(*config.GlobalConfig) error) (*config.GlobalConfig, error) { + cfg := s.stored + before := len(s.auth.Calls) + err := apply(&cfg) + s.callsInTx += len(s.auth.Calls) - before + if err != nil { + return nil, err + } + s.stored = cfg + return &cfg, nil +} + +func (s *authFreeTxStore) ListServiceConfigs(context.Context) ([]config.ServiceConfig, error) { + return []config.ServiceConfig{}, nil +} + +func newAuthFreeTxHandler(perms []auth.Permission, stored config.GlobalConfig) (*Handler, *authFreeTxStore, *MockAuthService) { + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "tok").Return(&Session{UserID: apgUserID}, nil).Maybe() + mockAuth.grantPermissions(perms) + store := &authFreeTxStore{MockConfigStore: new(MockConfigStore), auth: mockAuth, stored: stored} + return &Handler{config: store, auth: mockAuth}, store, mockAuth +} + +func authFreeReq(body string) *events.LambdaFunctionURLRequest { + return &events.LambdaFunctionURLRequest{Headers: map[string]string{"Authorization": "Bearer tok"}, Body: body} +} + +var ( + afAdmin = []auth.Permission{{Action: auth.ActionAdmin, Resource: auth.ResourceAll}} + afGranted = []auth.Permission{ + {Action: auth.ActionAdmin, Resource: auth.ResourceAll}, + {Action: auth.ActionExecute, Resource: auth.ResourcePurchases}, + {Action: auth.ActionExecute, Resource: auth.ResourceRIExchange}, + } +) + +const afRIExchangeArm = `{"auto_exchange_enabled":true,"mode":"auto","utilization_threshold":50,"max_payment_per_exchange_usd":100,"max_payment_daily_usd":500,"lookback_days":30}` + +func TestGlobalConfigArmingGates_NoAuthCallInsideTransaction(t *testing.T) { + tests := []struct { + name string + perms []auth.Permission + dedicated bool + body string + wantRefused bool + }{ + {"arm auto-purchase, granted", afGranted, false, `{"auto_purchase_enabled":true}`, false}, + {"arm auto-purchase, refused", afAdmin, false, `{"auto_purchase_enabled":true}`, true}, + {"arm ladder execution, granted", afGranted, false, `{"laddering_enabled":true,"ladder_execution_enabled":true}`, false}, + {"arm ladder execution, refused", afAdmin, false, `{"laddering_enabled":true,"ladder_execution_enabled":true}`, true}, + {"arm RI exchange via config, granted", afGranted, false, `{"ri_exchange_enabled":true,"ri_exchange_mode":"auto"}`, false}, + {"arm RI exchange via config, refused", afAdmin, false, `{"ri_exchange_enabled":true,"ri_exchange_mode":"auto"}`, true}, + {"arm RI exchange via dedicated route, granted", afGranted, true, afRIExchangeArm, false}, + {"arm RI exchange via dedicated route, refused", afAdmin, true, afRIExchangeArm, true}, + {"ordinary save, admin", afAdmin, false, `{"approval_required":false}`, false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + h, store, _ := newAuthFreeTxHandler(tc.perms, baseValidConfig()) + var err error + if tc.dedicated { + _, err = h.updateRIExchangeConfig(context.Background(), authFreeReq(tc.body)) + } else { + _, err = h.updateConfig(context.Background(), authFreeReq(tc.body)) + } + if tc.wantRefused { + apgRefused(t, err) + } else { + require.NoError(t, err) + } + assert.Zero(t, store.callsInTx, "the write closure must not call the auth service: it would need a second pooled connection") + }) + } +} + +// A failed execute-verb lookup is raised only when the write escalates, exactly +// as when the lookup ran inside the closure on escalation alone. Resolving it +// up front must not fail an ordinary save, nor turn a lookup error into a pass. +func TestGlobalConfigArmingGates_LookupErrorOnlyMattersOnEscalation(t *testing.T) { + lookupErr := errors.New("auth store unavailable") + newHandler := func() (*Handler, *authFreeTxStore) { + mockAuth := new(MockAuthService) + mockAuth.On("ValidateSession", mock.Anything, "tok").Return(&Session{UserID: apgUserID}, nil).Maybe() + mockAuth.On("HasPermissionAPI", mock.Anything, apgUserID, auth.ActionExecute, mock.Anything).Return(false, lookupErr) + mockAuth.grantPermissions(afAdmin) + store := &authFreeTxStore{MockConfigStore: new(MockConfigStore), auth: mockAuth, stored: baseValidConfig()} + return &Handler{config: store, auth: mockAuth}, store + } + + t.Run("ordinary save succeeds", func(t *testing.T) { + h, store := newHandler() + _, err := h.updateConfig(context.Background(), authFreeReq(`{"approval_required":false}`)) + require.NoError(t, err) + assert.False(t, store.stored.ApprovalRequired) + }) + for _, body := range []string{ + `{"auto_purchase_enabled":true}`, + `{"laddering_enabled":true,"ladder_execution_enabled":true}`, + `{"ri_exchange_enabled":true,"ri_exchange_mode":"auto"}`, + } { + t.Run("escalation fails closed: "+body, func(t *testing.T) { + h, store := newHandler() + _, err := h.updateConfig(context.Background(), authFreeReq(body)) + require.ErrorIs(t, err, lookupErr) + assert.Equal(t, baseValidConfig(), store.stored, "a refused write must not land") + }) + } + t.Run("dedicated route escalation fails closed", func(t *testing.T) { + h, store := newHandler() + _, err := h.updateRIExchangeConfig(context.Background(), authFreeReq(afRIExchangeArm)) + require.ErrorIs(t, err, lookupErr) + assert.Equal(t, baseValidConfig(), store.stored) + }) +} diff --git a/internal/api/handler.go b/internal/api/handler.go index f2dc5fd06..c0ee31a23 100644 --- a/internal/api/handler.go +++ b/internal/api/handler.go @@ -604,12 +604,14 @@ func (s riExchangeArmState) armed() bool { // threat the carve-out closed (issue #1765). // // Authorization goes through requirePermission, taking the same request the -// caller was authenticated from, so the verb is evaluated against THE CREDENTIAL -// PRESENTED rather than the owning user's group permissions. That distinction is -// load-bearing: for a user API key, requirePermission routes to authorizeAPIKey -// and checks the KEY's effective permissions. Checking the session's user id -// instead let a CI key that is correctly denied execute:ri-exchange arm the -// scheduler to execute every exchange -- the same inheritance bug +// caller was authenticated from (resolved ahead of the transaction by +// resolveGrant and passed in as grant), so the verb is evaluated against +// THE CREDENTIAL PRESENTED rather than the owning user's group permissions. +// That distinction is load-bearing: for a user API key, requirePermission +// routes to authorizeAPIKey and checks the KEY's effective permissions. +// Checking the session's user id instead let a CI key that is correctly +// denied execute:ri-exchange arm the scheduler to execute every exchange -- +// the same inheritance bug // requirePermissionConstraints already guards against one function above, and // the reason this does not call requireSessionPermission directly. It also // gives the stateless admin API key the same answer here as on the direct @@ -633,7 +635,12 @@ func (s riExchangeArmState) armed() bool { // these fields exactly as effectively as the dedicated PUT /api/ri-exchange/config // (updateRIExchangeConfig), so a gate on one alone would look complete and // would not be. -func (h *Handler) requireRIExchangeAutoModeGrant(ctx context.Context, req *events.LambdaFunctionURLRequest, before, after riExchangeArmState) error { +// +// It runs inside the UpdateGlobalConfigAtomic closure, so it must not touch +// the pool (issue #1517): grant is the outcome of that requirePermission call, +// resolved before the transaction opened, and is returned only when the write +// escalates. +func requireRIExchangeAutoModeGrant(before, after riExchangeArmState, grant error) error { if !after.armed() { return nil } @@ -641,10 +648,7 @@ func (h *Handler) requireRIExchangeAutoModeGrant(ctx context.Context, req *event if before.armed() && !raisesCap { return nil } - if _, err := h.requirePermission(ctx, req, auth.ActionExecute, auth.ResourceRIExchange); err != nil { - return err - } - return nil + return grant } // getAccountScope returns the auth.AccountScope describing which cloud diff --git a/internal/api/handler_config.go b/internal/api/handler_config.go index 182e2425a..46c8b3460 100644 --- a/internal/api/handler_config.go +++ b/internal/api/handler_config.go @@ -84,6 +84,9 @@ func (h *Handler) updateConfig(ctx context.Context, req *events.LambdaFunctionUR if err := h.requireGlobalConfigScope(ctx, session); err != nil { return nil, err } + // Resolved before the transaction: the closure below runs while it holds a + // pooled connection and must not need a second one (issue #1517). + grants := h.resolveArmingGrants(ctx, req) // Serialized read-modify-write: the store loads the stored config and // applies this closure under an advisory-locked transaction, then upserts @@ -121,10 +124,10 @@ func (h *Handler) updateConfig(ctx context.Context, req *events.LambdaFunctionUR if vErr := existing.Validate(); vErr != nil { return NewClientError(400, fmt.Sprintf("validation error: %s", vErr)) } - if gErr := h.requireRIExchangeAutoModeGrant(ctx, req, before, riExchangeArmStateOf(existing)); gErr != nil { + if gErr := requireRIExchangeAutoModeGrant(before, riExchangeArmStateOf(existing), grants.riExchange); gErr != nil { return gErr } - return h.requireSwitchGrants(ctx, req, session, switchBefore, ladderBefore, existing) + return requireSwitchGrants(ctx, session, grants.purchases, switchBefore, ladderBefore, existing) }) if err != nil { return nil, err @@ -136,12 +139,13 @@ func (h *Handler) updateConfig(ctx context.Context, req *events.LambdaFunctionUR } // requireSwitchGrants chains the arming gates of the two money-spending -// switches: auto-purchase, then ladder execution. -func (h *Handler) requireSwitchGrants(ctx context.Context, req *events.LambdaFunctionURLRequest, session *Session, autoBefore, ladderBefore bool, cfg *config.GlobalConfig) error { - if err := h.requireAutoPurchaseSwitchGrant(ctx, req, session, autoBefore, cfg); err != nil { +// switches: auto-purchase, then ladder execution. purchasesGrant is the +// pre-resolved execute:purchases outcome (armingGrants.purchases). +func requireSwitchGrants(ctx context.Context, session *Session, purchasesGrant error, autoBefore, ladderBefore bool, cfg *config.GlobalConfig) error { + if err := requireAutoPurchaseSwitchGrant(ctx, session, purchasesGrant, autoBefore, cfg); err != nil { return err } - return h.requireLadderExecutionSwitchGrant(ctx, req, session, ladderBefore, cfg) + return requireLadderExecutionSwitchGrant(ctx, session, purchasesGrant, ladderBefore, cfg) } func (h *Handler) requireGlobalConfigScope(ctx context.Context, session *Session) error { diff --git a/internal/api/handler_config_arming_pool_integration_test.go b/internal/api/handler_config_arming_pool_integration_test.go new file mode 100644 index 000000000..e2c1cb6ff --- /dev/null +++ b/internal/api/handler_config_arming_pool_integration_test.go @@ -0,0 +1,203 @@ +//go:build integration + +package api + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/auth" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/aws/aws-lambda-go/events" + "github.com/stretchr/testify/require" +) + +// armingSavePoolSize is the Fargate default pool (db_max_connections = 3). +const armingSavePoolSize = 3 + +// TestGlobalConfigArmingSavesDoNotExhaustPool is the pool deadlock of issue +// #1517. UpdateGlobalConfigAtomic holds one pooled connection for its locked +// transaction. When a save arms unattended spend, the grant check runs inside +// that transaction; if it reads the auth store through the same pool, then as +// many concurrent saves as the pool has connections each hold one (the lock +// holder, plus the rest waiting on the advisory lock) and the holder waits for +// a second that never frees. Every save then hangs until its deadline. +// +// The auth service, its store and the config store all share one pool of the +// Fargate default size, as internal/server/app.go wires them. Arming and +// disarming saves alternate so that many lock holders need the grant, not just +// the first. Before the fix each subtest fails at the deadline; after it the +// whole batch takes well under a second. +func TestGlobalConfigArmingSavesDoNotExhaustPool(t *testing.T) { + for _, tc := range []struct { + name string + arm, disarm string + put func(h *Handler) func(context.Context, *events.LambdaFunctionURLRequest) error + armed func(*config.GlobalConfig) bool + }{ + { + name: "config PUT arms auto-purchase", + arm: `{"auto_purchase_enabled":true}`, + disarm: `{"auto_purchase_enabled":false}`, + put: putGlobalConfig, + armed: func(c *config.GlobalConfig) bool { return c.AutoPurchaseEnabled }, + }, + { + name: "config PUT arms ladder execution", + arm: `{"laddering_enabled":true,"ladder_execution_enabled":true}`, + disarm: `{"ladder_execution_enabled":false}`, + put: putGlobalConfig, + armed: func(c *config.GlobalConfig) bool { return c.LadderExecutionEnabled }, + }, + { + name: "config PUT arms RI exchange auto mode", + arm: `{"ri_exchange_enabled":true,"ri_exchange_mode":"auto"}`, + disarm: `{"ri_exchange_mode":"manual"}`, + put: putGlobalConfig, + armed: func(c *config.GlobalConfig) bool { return riExchangeArmStateOf(c).armed() }, + }, + { + name: "RI exchange config PUT arms auto mode", + arm: `{"auto_exchange_enabled":true,"mode":"auto","utilization_threshold":50,"max_payment_per_exchange_usd":100,"max_payment_daily_usd":500,"lookback_days":30}`, + disarm: `{"auto_exchange_enabled":true,"mode":"manual","utilization_threshold":50,"max_payment_per_exchange_usd":100,"max_payment_daily_usd":500,"lookback_days":30}`, + put: putRIExchangeConfig, + armed: func(c *config.GlobalConfig) bool { return riExchangeArmStateOf(c).armed() }, + }, + } { + t.Run(tc.name, func(t *testing.T) { + h, store, token := newSmallPoolArmingHandler(t) + put := tc.put(h) + ctx := t.Context() + + // Prove the fixture actually reaches the grant: the arming save + // is authorized through the real auth store, and the session + // holds the execute verbs. + require.NoError(t, put(ctx, bearerRequest(token, tc.arm))) + require.NoError(t, put(ctx, bearerRequest(token, tc.disarm))) + + const saves = armingSavePoolSize * 4 + reqCtx, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + start := make(chan struct{}) + errs := make([]error, saves) + var wg sync.WaitGroup + for i := range saves { + body := tc.arm + if i%2 == 1 { + body = tc.disarm + } + wg.Add(1) + go func() { + defer wg.Done() + <-start + errs[i] = put(reqCtx, bearerRequest(token, body)) + }() + } + began := time.Now() + close(start) + wg.Wait() + t.Logf("%d concurrent arming/disarming saves on a pool of %d took %v", saves, armingSavePoolSize, time.Since(began)) + + for i, err := range errs { + require.NoError(t, err, "save %d", i) + } + // A final arming save lands, so the batch did not leave the + // switch in a state the gate never saw. + require.NoError(t, put(ctx, bearerRequest(token, tc.arm))) + got, err := store.GetGlobalConfig(ctx) + require.NoError(t, err) + require.True(t, tc.armed(got)) + }) + } +} + +// TestGlobalConfigArmingSaveGrantStaysStrict checks, on the real auth store, +// that a session holding update:config but neither execute verb is still +// refused every arming save while an ordinary save goes through. +func TestGlobalConfigArmingSaveGrantStaysStrict(t *testing.T) { + ctx := t.Context() + pg := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()) + store, authStore := config.NewPostgresStore(pg.DB), auth.NewPostgresStore(pg.DB) + csrfKey := []byte(strings.Repeat("g", 32)) + h := NewHandler(HandlerConfig{ConfigStore: store, AuthService: &marketplaceAuthFixture{service: auth.NewService(auth.ServiceConfig{Store: authStore, CSRFKey: csrfKey})}}) + _, token, _ := marketplaceSession(t, authStore, []auth.Permission{ + {Action: auth.ActionView, Resource: auth.ResourceConfig}, + {Action: auth.ActionUpdate, Resource: auth.ResourceConfig}, + }, []string{"*"}, csrfKey) + + require.NoError(t, h.updateConfigErr(ctx, bearerRequest(token, `{"laddering_enabled":true}`))) + for _, tc := range []struct { + put func(context.Context, *events.LambdaFunctionURLRequest) error + body string + }{ + {h.updateConfigErr, `{"auto_purchase_enabled":true}`}, + {h.updateConfigErr, `{"ladder_execution_enabled":true}`}, + {h.updateConfigErr, `{"ri_exchange_enabled":true,"ri_exchange_mode":"auto"}`}, + {h.updateRIExchangeConfigErr, `{"auto_exchange_enabled":true,"mode":"auto","utilization_threshold":50,"lookback_days":30}`}, + } { + err := tc.put(ctx, bearerRequest(token, tc.body)) + ce, ok := IsClientError(err) + require.True(t, ok, "%s: want a 403, got %v", tc.body, err) + require.Equal(t, 403, ce.code, tc.body) + } + got, err := store.GetGlobalConfig(ctx) + require.NoError(t, err) + require.True(t, got.LadderingEnabled) + require.False(t, got.AutoPurchaseEnabled) + require.False(t, got.LadderExecutionEnabled) + require.False(t, riExchangeArmStateOf(got).armed()) +} + +// newSmallPoolArmingHandler wires a handler whose config store and real auth +// service share one pool of armingSavePoolSize connections, and returns a +// session token whose user may arm every money-spending switch. +func newSmallPoolArmingHandler(t *testing.T) (*Handler, *config.PostgresStore, string) { + t.Helper() + ctx := t.Context() + // The limit under test is this client pool's size, so the shared server + // is fine. + pg := testhelpers.RequireMigratedDatabase(ctx, t, getMigrationsPath()) + poolCfg := *pg.Config + poolCfg.MaxConnections, poolCfg.MinConnections = armingSavePoolSize, 0 + db, err := database.NewConnection(ctx, &poolCfg, nil) + require.NoError(t, err) + t.Cleanup(db.Close) + + store, authStore := config.NewPostgresStore(db), auth.NewPostgresStore(db) + csrfKey := []byte(strings.Repeat("p", 32)) + h := NewHandler(HandlerConfig{ConfigStore: store, AuthService: &marketplaceAuthFixture{service: auth.NewService(auth.ServiceConfig{Store: authStore, CSRFKey: csrfKey})}}) + _, token, _ := marketplaceSession(t, authStore, []auth.Permission{ + {Action: auth.ActionView, Resource: auth.ResourceConfig}, + {Action: auth.ActionUpdate, Resource: auth.ResourceConfig}, + {Action: auth.ActionExecute, Resource: auth.ResourcePurchases}, + {Action: auth.ActionExecute, Resource: auth.ResourceRIExchange}, + }, []string{"*"}, csrfKey) + return h, store, token +} + +func putGlobalConfig(h *Handler) func(context.Context, *events.LambdaFunctionURLRequest) error { + return h.updateConfigErr +} + +func putRIExchangeConfig(h *Handler) func(context.Context, *events.LambdaFunctionURLRequest) error { + return h.updateRIExchangeConfigErr +} + +func (h *Handler) updateConfigErr(ctx context.Context, req *events.LambdaFunctionURLRequest) error { + _, err := h.updateConfig(ctx, req) + return err +} + +func (h *Handler) updateRIExchangeConfigErr(ctx context.Context, req *events.LambdaFunctionURLRequest) error { + _, err := h.updateRIExchangeConfig(ctx, req) + return err +} + +func bearerRequest(token, body string) *events.LambdaFunctionURLRequest { + return &events.LambdaFunctionURLRequest{Headers: map[string]string{"Authorization": "Bearer " + token}, Body: body} +} diff --git a/internal/api/handler_ri_exchange.go b/internal/api/handler_ri_exchange.go index 085c2e63a..4010ee438 100644 --- a/internal/api/handler_ri_exchange.go +++ b/internal/api/handler_ri_exchange.go @@ -2245,6 +2245,9 @@ func (h *Handler) updateRIExchangeConfig(ctx context.Context, req *events.Lambda if err := h.requireGlobalConfigScope(ctx, session); err != nil { return nil, err } + // Resolved before the transaction: the closure below runs while it holds a + // pooled connection and must not need a second one (issue #1517). + riExchangeGrant := h.resolveGrant(ctx, req, auth.ResourceRIExchange) // Route through the serialized read-modify-write so this partial update // (only the ri_exchange_* columns) shares the advisory lock with other @@ -2269,7 +2272,7 @@ func (h *Handler) updateRIExchangeConfig(ctx context.Context, req *events.Lambda existing.RIExchangeMaxDailyUSD = body.MaxPaymentDailyUSD existing.RIExchangeLookbackDays = body.LookbackDays - return h.requireRIExchangeAutoModeGrant(ctx, req, before, riExchangeArmStateOf(existing)) + return requireRIExchangeAutoModeGrant(before, riExchangeArmStateOf(existing), riExchangeGrant) }); err != nil { // A refused write is not a failed save. Surface the gate's 403 (and any // other ClientError the closure raises) unwrapped, so the response says diff --git a/internal/api/ladder_arming_grant.go b/internal/api/ladder_arming_grant.go index a3010b938..3ce2f71e0 100644 --- a/internal/api/ladder_arming_grant.go +++ b/internal/api/ladder_arming_grant.go @@ -19,13 +19,15 @@ import ( // requireLadderExecutionSwitchGrant gates and stamps a change of // GlobalConfig.LadderExecutionEnabled inside the UpdateGlobalConfigAtomic -// closure. Only off->on needs the grant; any change is stamped. -func (h *Handler) requireLadderExecutionSwitchGrant(ctx context.Context, req *events.LambdaFunctionURLRequest, session *Session, before bool, cfg *config.GlobalConfig) error { +// closure. Only off->on needs the grant; any change is stamped. grant is the +// pre-resolved execute:purchases outcome (armingGrants.purchases), so nothing +// here reads through the pool inside the transaction (issue #1517). +func requireLadderExecutionSwitchGrant(ctx context.Context, session *Session, grant error, before bool, cfg *config.GlobalConfig) error { if before == cfg.LadderExecutionEnabled { return nil } if cfg.LadderExecutionEnabled { - if err := h.requireAutoPurchaseGrant(ctx, req, session); err != nil { + if err := prefetchedAutoPurchaseGrant(session, grant); err != nil { return err } }