[CRE] [4/5] ConfidentialModule, config, DB migration, syncer routing

Adds the core abstractions for confidential workflow execution:

- ConfidentialModule: implements host.ModuleV2, dispatches workflow
  execution to TEE enclave via confidential-workflows capability
- Syncer routing: detects confidential workflows via on-chain attributes,
  routes to ConfidentialModule instead of local WASM engine
- Config: CRE.ConfidentialRelay TOML config (enabled, trustedPCRs, caRootsPEM)
- DB: adds attributes column to workflow_specs_v2
- WorkflowSpec.Attributes field for persisting on-chain attributes

Nothing is wired into CRE yet. The routing is inert until PR 5/5.

Part of #21635

Author: nadahalli

core/capabilities/launcher.go

@@ -46,7 +46,7 @@ var defaultStreamConfig = p2ptypes.StreamConfig{
 
 type launcher struct {
 	services.StateMachine
-	lggr                logger.Logger
+	lggr                logger.SugaredLogger
 	myPeerID            p2ptypes.PeerID
 	peerWrapper         p2ptypes.PeerWrapper
 	dispatcher          remotetypes.Dispatcher
@@ -106,7 +106,7 @@ func NewLauncher(
 		return nil, fmt.Errorf("failed to create launcher metrics: %w", err)
 	}
 	return &launcher{
-		lggr:        logger.Named(lggr, "CapabilitiesLauncher"),
+		lggr:        logger.Sugared(lggr).Named("CapabilitiesLauncher"),
 		peerWrapper: peerWrapper,
 		dispatcher:  dispatcher,
 		cachedShims: cachedShims{
@@ -342,9 +342,12 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 	for family := range myDONFamiliesSet {
 		myDONFamilies = append(myDONFamilies, family)
 	}
-	w.lggr.Debugw("Found my DON families", "count", len(myDONFamilies), "myDONFamilies", myDONFamilies)
-	w.lggr.Debugw("Found my workflow DONs", "count", len(myWorkflowDONs), "myWorkflowDONs", myWorkflowDONs)
-	w.lggr.Debugw("Found all remote workflow DONs", "count", len(remoteWorkflowDONs), "remoteWorkflowDONs", remoteWorkflowDONs)
+	w.lggr.Debugw("Found my DON families", "count", len(myDONFamilies))
+	w.lggr.Tracew("My DON families", "myDONFamilies", myDONFamilies)
+	w.lggr.Debugw("Found my workflow DONs", "count", len(myWorkflowDONs))
+	w.lggr.Tracew("My workflow DONs", "myWorkflowDONs", myWorkflowDONs)
+	w.lggr.Debugw("Found all remote workflow DONs", "count", len(remoteWorkflowDONs))
+	w.lggr.Tracew("All remote workflow DONs", "remoteWorkflowDONs", remoteWorkflowDONs)
 
 	// Capability DONs (with IsPublic = true) the current node is a part of.
 	// These need server-side shims to expose my own capabilities externally.
@@ -359,14 +362,18 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 			}
 		}
 	}
-	w.lggr.Debugw("Found my capability DONs", "count", len(myCapabilityDONs), "myCapabilityDONs", myCapabilityDONs)
-	w.lggr.Debugw("Found all remote capability DONs", "count", len(remoteCapabilityDONs), "remoteCapabilityDONs", remoteCapabilityDONs)
+	w.lggr.Debugw("Found my capability DONs", "count", len(myCapabilityDONs))
+	w.lggr.Tracew("My capability DONs", "myCapabilityDONs", myCapabilityDONs)
+	w.lggr.Debugw("Found all remote capability DONs", "count", len(remoteCapabilityDONs))
+	w.lggr.Tracew("All remote capability DONs", "remoteCapabilityDONs", remoteCapabilityDONs)
 
 	if len(myDONFamilies) > 0 {
 		remoteWorkflowDONs = filterDONsByFamilies(remoteWorkflowDONs, myDONFamilies)
 		remoteCapabilityDONs = filterDONsByFamilies(remoteCapabilityDONs, myDONFamilies)
-		w.lggr.Debugw("Filtered remote workflow DONs to my families", "count", len(remoteWorkflowDONs), "remoteWorkflowDONs", remoteWorkflowDONs)
-		w.lggr.Debugw("Filtered remote capability DONs to my families", "count", len(remoteCapabilityDONs), "remoteCapabilityDONs", remoteCapabilityDONs)
+		w.lggr.Debugw("Filtered remote workflow DONs to my families", "count", len(remoteWorkflowDONs))
+		w.lggr.Tracew("Filtered remote workflow DONs to my families", "remoteWorkflowDONs", remoteWorkflowDONs)
+		w.lggr.Debugw("Filtered remote capability DONs to my families", "count", len(remoteCapabilityDONs))
+		w.lggr.Tracew("Filtered remote capability DONs to my families", "remoteCapabilityDONs", remoteCapabilityDONs)
 	} else {
 		// legacy / Keystone setting
 		w.lggr.Debug("My node doesn't belong to any DON families. No filtering will be applied.")
@@ -401,23 +408,8 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 
 	belongsToACapabilityDON := len(myCapabilityDONs) > 0
 	if belongsToACapabilityDON {
-		// Include both remote workflow DONs and the node's own workflow DONs.
-		// In single-DON topologies (e.g. local CRE), the same DON is both a
-		// workflow DON and a capability DON, so remoteWorkflowDONs is empty.
-		// Without including myWorkflowDONs, capabilities fail to serve with
-		// "empty workflowDONs provided".
-		allWorkflowDONs := make([]registrysyncer.DON, 0, len(remoteWorkflowDONs)+len(myWorkflowDONs))
-		allWorkflowDONs = append(allWorkflowDONs, remoteWorkflowDONs...)
-		allWorkflowDONs = append(allWorkflowDONs, myWorkflowDONs...)
 		for _, myDON := range myCapabilityDONs {
-			w.serveCapabilities(ctx, w.myPeerID, myDON, localRegistry, allWorkflowDONs)
-
-			// Capability DONs also need remote capabilities (e.g. relay DON
-			// needs vault for secret fetching). Without this, only workflow
-			// DONs discover cross-DON capabilities.
-			for _, rcd := range remoteCapabilityDONs {
-				w.addRemoteCapabilities(ctx, myDON, rcd, localRegistry)
-			}
+			w.serveCapabilities(ctx, w.myPeerID, myDON, localRegistry, remoteWorkflowDONs)
 		}
 	}
 
@@ -433,7 +425,7 @@ func (w *launcher) OnNewRegistry(ctx context.Context, localRegistry *registrysyn
 	}
 	if w.don2donSharedPeer != nil {
 		donPairs := w.donPairsToUpdate(w.myPeerID, localRegistry)
-		err := w.don2donSharedPeer.UpdateConnectionsByDONs(ctx, donPairs, defaultStreamConfig)
+		err := w.don2donSharedPeer.UpdateConnectionsByDONs(ctx, donPairs, w.p2pStreamConfig)
 		if err != nil {
 			return fmt.Errorf("failed to update peer connections: %w", err)
 		}

core/capabilities/launcher_test.go

@@ -12,6 +12,7 @@ import (
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/durationpb"
 
+	"github.com/smartcontractkit/libocr/ragep2p"
 	ragetypes "github.com/smartcontractkit/libocr/ragep2p/types"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
@@ -233,73 +234,6 @@ func TestLauncher(t *testing.T) {
 		assert.Equal(t, 1, observedLogs.FilterMessage("failed to serve capability").Len())
 	})
 
-	t.Run("OK-single_DON_serves_capabilities", func(t *testing.T) {
-		// Regression test: in single-DON topologies (e.g. local CRE), the same
-		// DON is both a workflow DON and a capability DON. The DON appears in
-		// myWorkflowDONs (not remoteWorkflowDONs). Previously, serveCapabilities
-		// only received remoteWorkflowDONs, causing executable.Server.SetConfig
-		// to fail with "empty workflowDONs provided".
-		lggr, observedLogs := logger.TestObserved(t, zapcore.DebugLevel)
-		registry := NewRegistry(lggr)
-		dispatcher := remoteMocks.NewDispatcher(t)
-
-		nodes := newNodes(4)
-		peer := mocks.NewPeer(t)
-		peer.On("UpdateConnections", mock.Anything).Return(nil)
-		peer.On("ID").Return(nodes[0])
-		peer.On("IsBootstrap").Return(false)
-		wrapper := mocks.NewPeerWrapper(t)
-		wrapper.On("GetPeer").Return(peer)
-
-		fullTriggerCapID := "streams-trigger@1.0.0"
-		mt := newMockTrigger(capabilities.MustNewCapabilityInfo(
-			fullTriggerCapID,
-			capabilities.CapabilityTypeTrigger,
-			"streams trigger",
-		))
-		require.NoError(t, registry.Add(t.Context(), mt))
-
-		fullTargetID := "write-chain_evm_1@1.0.0"
-		mtarg := &mockCapability{
-			CapabilityInfo: capabilities.MustNewCapabilityInfo(
-				fullTargetID,
-				capabilities.CapabilityTypeTarget,
-				"write chain",
-			),
-		}
-		require.NoError(t, registry.Add(t.Context(), mtarg))
-
-		triggerCapID := RandomUTF8BytesWord()
-		targetCapID := RandomUTF8BytesWord()
-
-		// Single DON: acceptsWorkflows=true AND has capability configurations.
-		// This puts it in both myWorkflowDONs and myCapabilityDONs.
-		dID := uint32(1)
-		localRegistry := buildLocalRegistry()
-		addDON(localRegistry, dID, uint32(0), uint8(1), true, true, nodes, []string{"zone-a"}, 1, [][32]byte{triggerCapID, targetCapID})
-		addCapabilityToDON(localRegistry, dID, fullTriggerCapID, capabilities.CapabilityTypeTrigger, nil)
-		addCapabilityToDON(localRegistry, dID, fullTargetID, capabilities.CapabilityTypeTarget, nil)
-
-		launcher, err := NewLauncher(
-			lggr,
-			wrapper,
-			nil,
-			nil,
-			dispatcher,
-			registry,
-			&mockDonNotifier{},
-		)
-		require.NoError(t, err)
-		require.NoError(t, launcher.Start(t.Context()))
-		defer launcher.Close()
-
-		dispatcher.On("SetReceiver", fullTriggerCapID, dID, mock.AnythingOfType("*remote.triggerPublisher")).Return(nil)
-		dispatcher.On("SetReceiver", fullTargetID, dID, mock.AnythingOfType("*executable.server")).Return(nil)
-
-		require.NoError(t, launcher.OnNewRegistry(t.Context(), localRegistry))
-		assert.Equal(t, 0, observedLogs.FilterMessage("failed to serve capability").Len())
-	})
-
 	t.Run("start and close with nil peer wrapper", func(t *testing.T) {
 		lggr := logger.Test(t)
 		registry := NewRegistry(lggr)
@@ -959,10 +893,24 @@ func TestLauncher_V2CapabilitiesAddViaCombinedClient(t *testing.T) {
 	addCapabilityToDON(localRegistry, zoneBDonID, fullExecutableCapID, capabilities.CapabilityTypeTarget, execCfg)
 	addCapabilityToDON(localRegistry, capDonID, fullLocalCapID, capabilities.CapabilityTypeAction, localCfg) // should be skipped
 
+	customStreamConfig := p2ptypes.StreamConfig{
+		IncomingMessageBufferSize: 999,
+		OutgoingMessageBufferSize: 888,
+		MaxMessageLenBytes:        777777,
+		MessageRateLimiter: ragep2p.TokenBucketParams{
+			Rate:     50.0,
+			Capacity: 250,
+		},
+		BytesRateLimiter: ragep2p.TokenBucketParams{
+			Rate:     2500000.0,
+			Capacity: 5000000,
+		},
+	}
+
 	sharedPeer := mocks.NewSharedPeer(t)
 	sharedPeer.On("ID").Return(workflowDonNodes[0])
 	sharedPeer.On("IsBootstrap").Return(false)
-	sharedPeer.On("UpdateConnectionsByDONs", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	sharedPeer.On("UpdateConnectionsByDONs", mock.Anything, mock.Anything, customStreamConfig).Return(nil)
 
 	launcher, err := NewLauncher(
 		lggr,
@@ -974,6 +922,7 @@ func TestLauncher_V2CapabilitiesAddViaCombinedClient(t *testing.T) {
 		&mockDonNotifier{},
 	)
 	require.NoError(t, err)
+	launcher.p2pStreamConfig = customStreamConfig
 	servicetest.Run(t, launcher)
 
 	dispatcher.On("SetReceiverForMethod", fullTriggerCapID, capDonID, "StreamsTrigger", mock.AnythingOfType("*remote.triggerSubscriber")).Return(nil)

core/config/cre_config.go

@@ -13,6 +13,7 @@ type CRE interface {
 	// When enabled, additional OTel tracing and logging is performed.
 	DebugMode() bool
 	LocalSecrets() map[string]string
+	ConfidentialRelay() CREConfidentialRelay
 }
 
 // WorkflowFetcher defines configuration for fetching workflow files
@@ -21,6 +22,13 @@ type WorkflowFetcher interface {
 	URL() string
 }
 
+// CREConfidentialRelay defines configuration for the confidential relay handler.
+type CREConfidentialRelay interface {
+	Enabled() bool
+	TrustedPCRs() string
+	CARootsPEM() string
+}
+
 // CRELinking defines configuration for connecting to the CRE linking service
 type CRELinking interface {
 	URL() string

core/config/toml/types.go

@@ -1895,14 +1895,24 @@ type CreConfig struct {
 	// Requires [Tracing].Enabled = true for traces to be exported (trace export is gated by
 	// Tracing.Enabled in initGlobals; Telemetry.Enabled is optional—traces work with or without it).
 	// WARNING: This is not suitable for production use due to performance overhead.
-	DebugMode *bool `toml:",omitempty"`
+	DebugMode         *bool                    `toml:",omitempty"`
+	ConfidentialRelay *ConfidentialRelayConfig `toml:",omitempty"`
 }
 
 // WorkflowFetcherConfig holds the configuration for fetching workflow files
 type WorkflowFetcherConfig struct {
 	URL *string `toml:",omitempty"`
 }
 
+// ConfidentialRelayConfig holds the configuration for the confidential relay handler.
+// When Enabled is true, the node participates in the confidential relay DON,
+// validating enclave attestations and proxying capability requests.
+type ConfidentialRelayConfig struct {
+	Enabled     *bool   `toml:",omitempty"`
+	TrustedPCRs *string `toml:",omitempty"`
+	CARootsPEM  *string `toml:",omitempty"`
+}
+
 // LinkingConfig holds the configuration for connecting to the CRE linking service
 type LinkingConfig struct {
 	URL        *string `toml:",omitempty"`
@@ -1956,6 +1966,21 @@ func (c *CreConfig) setFrom(f *CreConfig) {
 	if f.DebugMode != nil {
 		c.DebugMode = f.DebugMode
 	}
+
+	if f.ConfidentialRelay != nil {
+		if c.ConfidentialRelay == nil {
+			c.ConfidentialRelay = &ConfidentialRelayConfig{}
+		}
+		if v := f.ConfidentialRelay.Enabled; v != nil {
+			c.ConfidentialRelay.Enabled = v
+		}
+		if v := f.ConfidentialRelay.TrustedPCRs; v != nil {
+			c.ConfidentialRelay.TrustedPCRs = v
+		}
+		if v := f.ConfidentialRelay.CARootsPEM; v != nil {
+			c.ConfidentialRelay.CARootsPEM = v
+		}
+	}
 }
 
 func (w *WorkflowFetcherConfig) ValidateConfig() error {

core/services/chainlink/config_cre.go

@@ -105,6 +105,35 @@ func (c *creConfig) Linking() config.CRELinking {
 	return &linkingConfig{url: url, tlsEnabled: tlsEnabled}
 }
 
+type confidentialRelayConfig struct {
+	enabled     bool
+	trustedPCRs string
+	caRootsPEM  string
+}
+
+func (cr *confidentialRelayConfig) Enabled() bool       { return cr.enabled }
+func (cr *confidentialRelayConfig) TrustedPCRs() string { return cr.trustedPCRs }
+func (cr *confidentialRelayConfig) CARootsPEM() string  { return cr.caRootsPEM }
+
+func (c *creConfig) ConfidentialRelay() config.CREConfidentialRelay {
+	if c.c.ConfidentialRelay == nil {
+		return &confidentialRelayConfig{}
+	}
+	enabled := false
+	if c.c.ConfidentialRelay.Enabled != nil {
+		enabled = *c.c.ConfidentialRelay.Enabled
+	}
+	trustedPCRs := ""
+	if c.c.ConfidentialRelay.TrustedPCRs != nil {
+		trustedPCRs = *c.c.ConfidentialRelay.TrustedPCRs
+	}
+	caRootsPEM := ""
+	if c.c.ConfidentialRelay.CARootsPEM != nil {
+		caRootsPEM = *c.c.ConfidentialRelay.CARootsPEM
+	}
+	return &confidentialRelayConfig{enabled: enabled, trustedPCRs: trustedPCRs, caRootsPEM: caRootsPEM}
+}
+
 func (c *creConfig) LocalSecrets() map[string]string {
 	return c.s.LocalSecrets
 }

core/services/job/models.go

@@ -930,6 +930,7 @@ type WorkflowSpec struct {
 	CreatedAt     time.Time          `toml:"-"`
 	UpdatedAt     time.Time          `toml:"-"`
 	SpecType      WorkflowSpecType   `toml:"spec_type" db:"spec_type"`
+	Attributes    []byte             `db:"attributes"`
 	sdkWorkflow   *sdk.WorkflowSpec
 	rawSpec       []byte
 	config        []byte

core/services/standardcapabilities/conversions/conversions.go

@@ -33,6 +33,8 @@ func GetCapabilityIDFromCommand(command string, config string) string {
 		return "http-trigger@1.0.0-alpha"
 	case "http_action":
 		return "http-actions@1.0.0-alpha" // plural "actions"
+	case "mock":
+		return "mock@1.0.0"
 	default:
 		return ""
 	}
@@ -52,6 +54,8 @@ func GetCommandFromCapabilityID(capabilityID string) string {
 		return "http_trigger"
 	case strings.HasPrefix(capabilityID, "http-actions"):
 		return "http_action"
+	case strings.HasPrefix(capabilityID, "mock"):
+		return "mock"
 	default:
 		return ""
 	}

core/services/workflows/artifacts/v2/orm.go

@@ -63,7 +63,8 @@ func (orm *orm) UpsertWorkflowSpec(ctx context.Context, spec *job.WorkflowSpec)
 				config_url,
 				created_at,
 				updated_at,
-				spec_type
+				spec_type,
+				attributes
 			) VALUES (
 				:workflow,
 				:config,
@@ -76,7 +77,8 @@ func (orm *orm) UpsertWorkflowSpec(ctx context.Context, spec *job.WorkflowSpec)
 				:config_url,
 				:created_at,
 				:updated_at,
-				:spec_type
+				:spec_type,
+				:attributes
 			) ON CONFLICT (workflow_id) DO UPDATE
 			SET
 				workflow = EXCLUDED.workflow,
@@ -89,7 +91,8 @@ func (orm *orm) UpsertWorkflowSpec(ctx context.Context, spec *job.WorkflowSpec)
 				config_url = EXCLUDED.config_url,
 				created_at = EXCLUDED.created_at,
 				updated_at = EXCLUDED.updated_at,
-				spec_type = EXCLUDED.spec_type
+				spec_type = EXCLUDED.spec_type,
+				attributes = EXCLUDED.attributes
 			RETURNING id
 		`
 

core/services/workflows/syncer/fetcher.go

@@ -177,12 +177,20 @@ func newFileFetcher(basePath string, lggr logger.Logger) types.FetcherFunc {
 		if err != nil {
 			return nil, fmt.Errorf("invalid URL: %w", err)
 		}
-		fullPath := filepath.Clean(u.Path)
 
-		// ensure that the incoming request URL is either relative or absolute but within the basePath
-		if !filepath.IsAbs(fullPath) {
-			// If it's not absolute, we assume it's relative to the basePath
-			fullPath = filepath.Join(basePath, fullPath)
+		var fullPath string
+		if u.Scheme == "http" || u.Scheme == "https" {
+			// For HTTP(S) URLs, extract just the filename and resolve against basePath.
+			// This supports confidential workflows where the on-chain URL must be HTTP
+			// (so the enclave can fetch the binary), but the syncer reads from the local filesystem.
+			fullPath = filepath.Join(basePath, filepath.Base(u.Path))
+		} else {
+			fullPath = filepath.Clean(u.Path)
+			// ensure that the incoming request URL is either relative or absolute but within the basePath
+			if !filepath.IsAbs(fullPath) {
+				// If it's not absolute, we assume it's relative to the basePath
+				fullPath = filepath.Join(basePath, fullPath)
+			}
 		}
 		if !strings.HasPrefix(fullPath, basePath) {
 			return nil, fmt.Errorf("request URL %s is not within the basePath %s", fullPath, basePath)