diff --git a/internal/api/auto_purchase_grant.go b/internal/api/auto_purchase_grant.go index 856c75f6..b55f960d 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 00000000..5e7f93bf --- /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 f2dc5fd0..c0ee31a2 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 182e2425..46c8b346 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 00000000..e2c1cb6f --- /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 085c2e63..4010ee43 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 a3010b93..3ce2f71e 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 } }