diff --git a/capabilities/framework/executor.go b/capabilities/framework/executor.go index d515b4d..e35f671 100644 --- a/capabilities/framework/executor.go +++ b/capabilities/framework/executor.go @@ -463,12 +463,18 @@ type RealExecutor struct { capConfigMu sync.RWMutex capConfig *ParsedConfig capabilityRegistry core.CapabilitiesRegistry - nodeID string proposalInFlight atomic.Bool - donMu sync.RWMutex - donMembers [][]byte - donF uint32 - enclaves []types.Enclave + + // stateMu guards the mutable shared state that the background refresh loop + // (EnsureFreshEnclaves) and getLocalNodeAndCapConfig rewrite while Execute + // reads it concurrently. Access these only through the getter/setter helpers. + // vaultDON.CryptographyThreshold is guarded too (via getVaultThreshold/ + // setVaultThreshold); the rest of vaultDON is publish-once at init. + stateMu sync.RWMutex + nodeID string + donMembers [][]byte + donF uint32 + enclaves []types.Enclave // Background enclave-refresh ticker, started once on init and stopped on Close. refreshCancel context.CancelFunc @@ -627,7 +633,7 @@ func (e *RealExecutor) Execute(ctx context.Context, protoBytes []byte, secrets [ // and org, matching the DON-mode base labels (workflowName/workflowOwner/orgID). // sdk is not available at this layer (not on RequestMetadata), so it is omitted. metrics := NewScopedEmitter(e.metrics, map[string]any{ - "node.id": e.nodeID, + "node.id": e.getNodeID(), "workflow.owner": metadata.WorkflowOwner, "workflow.id": metadata.WorkflowID, "workflow.name": metadata.WorkflowName, @@ -1011,7 +1017,7 @@ func (e *RealExecutor) initLazily(ctx context.Context) error { } e.enclaveClient = pool - e.enclaves = nodes + e.setEnclaves(nodes) e.rateLimiter = rateLimiter e.setCapConfig(parsedConfig) e.vaultDON = VaultDON{ @@ -1024,11 +1030,11 @@ func (e *RealExecutor) initLazily(ctx context.Context) error { e.startEnclaveRefreshLoop() e.lggr.Infow("executor initialized", "capabilityID", e.capabilityID, - "nodeID", e.nodeID, + "nodeID", e.getNodeID(), "maxRetries", parsedConfig.MaxRetries, "retryBackoffSeconds", parsedConfig.RetryBackoffSeconds, "enableSecretsCache", parsedConfig.EnableSecretsCache, - "vaultDONThreshold", e.vaultDON.CryptographyThreshold, + "vaultDONThreshold", e.getVaultThreshold(), "insecureSkipTLS", parsedConfig.InsecureSkipTLSVerify) return nil } @@ -1094,14 +1100,14 @@ func (e *RealExecutor) EnsureFreshEnclaves(ctx context.Context) error { "endpoint": "publicKeys", }) } else { - e.enclaves = nodes + e.setEnclaves(nodes) } vaultDONPossibleFaultyNodes, err := getVaultDONPossibleFaultyNodes(ctx, e.vaultDON.Capability, int(localNode.WorkflowDON.F)) if err != nil { return fmt.Errorf("failed to get VaultDON possible faulty nodes: %w", err) } - e.vaultDON.CryptographyThreshold = getVaultDONThreshold(vaultDONPossibleFaultyNodes) + e.setVaultThreshold(getVaultDONThreshold(vaultDONPossibleFaultyNodes)) newMembers := peerIDsToSortedBytes(localNode.WorkflowDON.Members) newF := uint32(localNode.WorkflowDON.F) @@ -1111,18 +1117,48 @@ func (e *RealExecutor) EnsureFreshEnclaves(ctx context.Context) error { } func (e *RealExecutor) getDONMembership() (members [][]byte, f uint32) { - e.donMu.RLock() - defer e.donMu.RUnlock() + e.stateMu.RLock() + defer e.stateMu.RUnlock() return e.donMembers, e.donF } func (e *RealExecutor) setDONMembership(members [][]byte, f uint32) { - e.donMu.Lock() - defer e.donMu.Unlock() + e.stateMu.Lock() + defer e.stateMu.Unlock() e.donMembers = members e.donF = f } +func (e *RealExecutor) getNodeID() string { + e.stateMu.RLock() + defer e.stateMu.RUnlock() + return e.nodeID +} + +func (e *RealExecutor) setNodeID(id string) { + e.stateMu.Lock() + defer e.stateMu.Unlock() + e.nodeID = id +} + +func (e *RealExecutor) setEnclaves(nodes []types.Enclave) { + e.stateMu.Lock() + defer e.stateMu.Unlock() + e.enclaves = nodes +} + +func (e *RealExecutor) getVaultThreshold() int { + e.stateMu.RLock() + defer e.stateMu.RUnlock() + return e.vaultDON.CryptographyThreshold +} + +func (e *RealExecutor) setVaultThreshold(t int) { + e.stateMu.Lock() + defer e.stateMu.Unlock() + e.vaultDON.CryptographyThreshold = t +} + func (e *RealExecutor) proposeConfigUpdateIfMembershipChanged(ctx context.Context, newMembers [][]byte, newF uint32) { curMembers, curF := e.getDONMembership() // With a recorded baseline, skip the enclave round-trip while membership is unchanged. @@ -1215,6 +1251,8 @@ func (e *RealExecutor) broadcastConfigUpdate(ctx context.Context, newMembers [][ } func (e *RealExecutor) GetEnclaves() []types.Enclave { + e.stateMu.RLock() + defer e.stateMu.RUnlock() return e.enclaves } @@ -1233,7 +1271,7 @@ func (e *RealExecutor) getLocalNodeAndCapConfig(ctx context.Context) (capabiliti return capabilities.Node{}, capabilities.CapabilityConfiguration{}, fmt.Errorf("local node does not have a WorkflowDON ID, cannot initialise confidential http action capability") } if localNode.PeerID != nil { - e.nodeID = localNode.PeerID.String() + e.setNodeID(localNode.PeerID.String()) } ownCapabilityConfig, err := e.capabilityRegistry.ConfigForCapability(ctx, e.capabilityID, localNode.WorkflowDON.ID) @@ -1602,7 +1640,7 @@ func (e *RealExecutor) GetEncryptedDecryptionShares( } else { return nil, nil, fmt.Errorf("no decryption shares found for secret %s: neither binary nor hex-encoded shares present", secretResp.GetId().GetKey()) } - minimumSharesRequired := e.vaultDON.CryptographyThreshold + minimumSharesRequired := e.getVaultThreshold() if len(encryptedDecryptionSharesForSecret) < minimumSharesRequired { return nil, nil, fmt.Errorf("not enough encrypted decryption key shares for secret %s, expected at least %d, got %d", secretResp.GetId().GetKey(), minimumSharesRequired, len(encryptedDecryptionSharesForSecret)) } diff --git a/capabilities/framework/executor_test.go b/capabilities/framework/executor_test.go index 139a377..e26baec 100644 --- a/capabilities/framework/executor_test.go +++ b/capabilities/framework/executor_test.go @@ -10,6 +10,8 @@ import ( "errors" "fmt" "slices" + "sync" + "sync/atomic" "testing" "time" @@ -299,6 +301,9 @@ func getMockCapabilitiesRegistry(t *testing.T, mockVaultDON framework.VaultDON) // MockMetrics is a stub implementation of types.Emitter for testing. type MockMetrics struct { + // mu guards the maps so Emit is safe under concurrent Execute calls. + mu sync.Mutex + // CallCounts tracks how many times each event was emitted CallCounts map[string]int @@ -316,6 +321,8 @@ func NewMockMetrics() *MockMetrics { // Emit implements the types.Emitter interface. func (m *MockMetrics) Emit(event string, details map[string]any) { + m.mu.Lock() + defer m.mu.Unlock() m.CallCounts[event] += 1 m.EmitRecords[event] = append(m.EmitRecords[event], details) } @@ -1635,6 +1642,114 @@ func TestExecutor_ExecuteWithSecretsCache(t *testing.T) { }) } +// TestExecutor_ConcurrentRefreshAndExecute_NoDataRace is a regression test for the +// executor data race: the background refresh loop (EnsureFreshEnclaves) and +// getLocalNodeAndCapConfig rewrite nodeID, enclaves, and vaultDON.CryptographyThreshold +// while Execute reads them concurrently. Run under -race; before the stateMu guards +// this trips the detector, and a torn read of the nodeID string / enclaves slice +// header can corrupt memory at runtime. +func TestExecutor_ConcurrentRefreshAndExecute_NoDataRace(t *testing.T) { + mockVaultDONCapability := &MockVaultDONCapability{} + mockVaultDONCapability.ExecuteFunc = func(ctx context.Context, req capabilities.CapabilityRequest) (capabilities.CapabilityResponse, error) { + respAny, _ := anypb.New(getValidGetSecretsResponse()) + return capabilities.CapabilityResponse{Payload: respAny}, nil + } + mockVaultDON := framework.VaultDON{CryptographyThreshold: 1, Capability: mockVaultDONCapability} + + mockEnclaveClient := &MockEnclaveClient{} + mockEnclaveClient.ExecuteBatchFunc = func(ctx context.Context, reqs []enclavetypes.SignedComputeRequest, enclaveIDs [][32]byte) ([]enclavetypes.ExecuteResponse, error) { + return mockEnclaveClient.commonExecuteBatchReturn(t) + } + + // The registry flips the local PeerID (source of nodeID) and the enclave config + // (source of e.enclaves) on every call, so the guarded fields actually change value + // under concurrency. WorkflowDON membership and F stay constant so + // validateEnclaveSigners keeps passing and Execute stays on the happy path. + var flip atomic.Uint64 + mockRegistry := &MockCapabilitiesRegistry{ + LocalNodeFunc: func(ctx context.Context) (capabilities.Node, error) { + pid := mockPeerID1 + if flip.Add(1)%2 == 0 { + pid = mockPeerID2 + } + return capabilities.Node{ + WorkflowDON: capabilities.DON{ID: 1, F: 0, Members: []p2ptypes.PeerID{mockPeerID1, mockPeerID2}}, + PeerID: &pid, + }, nil + }, + ConfigForCapabilityFunc: func(ctx context.Context, capabilityID string, donID uint32) (capabilities.CapabilityConfiguration, error) { + id, url := [32]byte{1}, "https://enclave-a.example.com" + if flip.Load()%2 == 0 { + id, url = [32]byte{2}, "https://enclave-b.example.com" + } + enclavesList := enclavetypes.EnclavesList{ + Enclaves: []enclavetypes.Enclave{{ + EnclaveID: id, EnclaveURL: url, EnclaveType: "nitro", + TrustedValues: [][]byte{[]byte("{}")}, Region: "us-west-2", + }}, + } + wrappedConfig, err := values.WrapMap(enclavesList) + require.NoError(t, err) + return capabilities.CapabilityConfiguration{DefaultConfig: wrappedConfig}, nil + }, + GetExecutableFunc: func(ctx context.Context, ID string) (capabilities.ExecutableCapability, error) { + if ID == vault.CapabilityID { + return mockVaultDONCapability, nil + } + return nil, fmt.Errorf("unknown capability ID: %s", ID) + }, + } + + executor := framework.NewTestExecutor( + logger.Test(t), getMockKeystore(), mockEnclaveClient, mockVaultDON, + NewMockMetrics(), getDefaultRateLimiter(), + 1, 0, "test-capability-id", false, TEST_NODE_ID, mockRegistry, + ) + + actionInput := getTestExecutorInput() + protoBytes, err := proto.Marshal(actionInput.GetInput()) + require.NoError(t, err) + + const goroutines = 8 + const iterations = 40 + var wg sync.WaitGroup + start := make(chan struct{}) + + worker := func(fn func(i int)) { + defer wg.Done() + <-start + for i := 0; i < iterations; i++ { + fn(i) + } + } + + for g := 0; g < goroutines; g++ { + g := g + wg.Add(3) + // Reader: Execute reads nodeID (scoped emitter) and the vault threshold. + go worker(func(i int) { + md := capabilities.RequestMetadata{ + WorkflowID: WORKFLOW_ID, + WorkflowExecutionID: fmt.Sprintf("exec-%d-%d", g, i), + WorkflowName: WORKFLOW_NAME, + WorkflowOwner: WORKFLOW_OWNER, + } + _, _ = executor.Execute(context.Background(), protoBytes, actionInput.GetVaultDonSecrets(), md) + }) + // Writer: refresh rewrites nodeID, enclaves, and the vault threshold. + go worker(func(i int) { + _ = executor.EnsureFreshEnclaves(context.Background()) + }) + // Reader: GetEnclaves reads the enclaves slice header. + go worker(func(i int) { + _ = executor.GetEnclaves() + }) + } + + close(start) + wg.Wait() +} + func TestEnsureFreshEnclaves_ConfigurationChange(t *testing.T) { t.Run("UpdateNodes is called with new enclaves when registry configuration changes", func(t *testing.T) { // Track what nodes are passed to UpdateNodes