[CRE] [4/5] ConfidentialModule, config, DB migration, syncer routing (#21641)

* resolve go.mod conflicts

* Revert file fetcher HTTP URL handling

The filepath.Base() code for HTTP URLs in newFileFetcher was dead code.
No test or production path sends an HTTP URL to a file-based fetcher.
The enclave fetches binaries via its own BinaryFetcher, independent of
the node syncer's fetcher.

* Restore comments in startAndRegisterEngine, extract newV2EngineConfig

Restore comments that were dropped when extracting startAndRegisterEngine
from tryEngineCreate. Extract common EngineConfig construction into
newV2EngineConfig and initDone hook wiring into wireInitDoneHook,
reducing duplication between the normal and confidential engine paths.

* Unify engine creation flow for confidential and normal paths

Replace the early-return pattern with a symmetric if/else that picks
the factory, then converges on a single startAndRegisterEngine call.
Rename tryConfidentialEngineCreate to confidentialEngineFactory and
change its signature to return (services.Service, error).

# Conflicts:
#	core/services/workflows/syncer/v2/handler.go

* Restore initDone comment in tryEngineCreate

* Inline startAndRegisterEngine back into tryEngineCreate

No longer needed as a separate method now that both engine paths
converge in tryEngineCreate.

* Restore inline comments in wireInitDoneHook

* Restore original BeholderEmitter closure pattern in newV2EngineConfig

* Clean up factory signatures and newV2EngineConfig param ordering

Group string params together in newV2EngineConfig, move SdkName and
DebugMode into the constructor, drop unused wid param from
confidentialEngineFactory.

* Fix lint: errors.New, assert.Empty

* Add BinaryURLResolver to ConfidentialModule for presigned URL support

The enclave needs an authenticated URL to download WASM binaries from
the CRE storage service. BinaryURLResolver resolves the raw on-chain
URL into a presigned/ephemeral URL per execution. Nil-safe: falls
back to the raw URL when no resolver is set.

PR 5/5 (#21642) wires this to the storage service retriever.

* add EmitUserMetric to stubExecutionHelper

host.ExecutionHelper gained EmitUserMetric in chainlink-common #1924.

* fix struct field alignment in ConfidentialModule

* set org_id on WorkflowExecution from CRE context

Bump chainlink-common to pick up OrgId field on WorkflowExecution
proto. Read org from the CRE execution context and include it in
the proto sent to the enclave, matching the pattern used by the
framework executor for VaultDON requests.

* block user workflows from calling system-only capabilities

Adds a deny-list check in ExecutionHelper.CallCapability() that
prevents user workflow steps from invoking internal capabilities
like confidential-workflows. The ConfidentialModule bypasses this
gate because it calls the registry directly.

Author: nadahalli

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

@@ -50,6 +50,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 ""
 	}
@@ -71,6 +73,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/v2/handler.go

@@ -558,6 +558,7 @@ func (h *eventHandler) createWorkflowSpec(ctx context.Context, payload WorkflowR
 		SpecType:      job.WASMFile,
 		BinaryURL:     payload.BinaryURL,
 		ConfigURL:     payload.ConfigURL,
+		Attributes:    payload.Attributes,
 	}
 
 	if _, err = h.workflowArtifactsStore.UpsertWorkflowSpec(ctx, entry); err != nil {
@@ -658,60 +659,9 @@ func (h *eventHandler) engineFactoryFn(ctx context.Context, workflowID string, o
 	}
 
 	// V2 aka "NoDAG"
-	cfg := &v2.EngineConfig{
-		Lggr:                  h.lggr,
-		Module:                module,
-		WorkflowConfig:        config,
-		CapRegistry:           h.capRegistry,
-		DonSubscriber:         h.workflowDonSubscriber,
-		UseLocalTimeProvider:  h.useLocalTimeProvider,
-		DonTimeStore:          h.donTimeStore,
-		ExecutionsStore:       h.workflowStore,
-		WorkflowID:            workflowID,
-		WorkflowOwner:         owner,
-		WorkflowName:          name,
-		WorkflowTag:           tag,
-		WorkflowEncryptionKey: h.workflowEncryptionKey,
+	cfg := h.newV2EngineConfig(module, workflowID, owner, tag, sdkName, name, config)
 
-		LocalLimits:                       v2.EngineLimits{}, // all defaults
-		LocalLimiters:                     h.engineLimiters,
-		FeatureFlags:                      h.featureFlags,
-		GlobalExecutionConcurrencyLimiter: h.workflowLimits,
-
-		BeholderEmitter: func() custmsg.MessageEmitter {
-			h.emitterMu.RLock()
-			defer h.emitterMu.RUnlock()
-			return h.emitter
-		}(),
-		BillingClient: h.billingClient,
-
-		WorkflowRegistryAddress:       h.workflowRegistryAddress,
-		WorkflowRegistryChainSelector: h.workflowRegistryChainSelector,
-		OrgResolver:                   h.orgResolver,
-		DebugMode:                     h.debugMode,
-		SecretsFetcher:                h.secretsFetcher,
-		SdkName:                       sdkName,
-
-		ShardOrchestratorClient: h.shardOrchestratorClient,
-		ShardingEnabled:         h.shardingEnabled,
-		MyShardID:               h.myShardID,
-		ShardRoutingSteady:      h.shardRoutingSteady,
-	}
-
-	// Wire the initDone channel to the OnInitialized lifecycle hook.
-	// This will be called when the engine completes initialization (including trigger subscriptions).
-	// We compose with any existing hook to avoid overwriting test hooks or other user-provided hooks.
-	if initDone != nil {
-		existingHook := cfg.Hooks.OnInitialized
-		cfg.Hooks.OnInitialized = func(err error) {
-			// Signal completion to the handler first
-			initDone <- err
-			// Then call any existing hook (e.g., from tests)
-			if existingHook != nil {
-				existingHook(err)
-			}
-		}
-	}
+	h.wireInitDoneHook(cfg, initDone)
 
 	return v2.NewEngine(cfg)
 }
@@ -824,24 +774,23 @@ func (h *eventHandler) tryEngineCreate(ctx context.Context, spec *job.WorkflowSp
 		return fmt.Errorf("invalid workflow name: %w", err)
 	}
 
+	confidential, err := v2.IsConfidential(spec.Attributes)
+	if err != nil {
+		return fmt.Errorf("failed to parse workflow attributes: %w", err)
+	}
+
 	// Create a channel to receive the initialization result.
 	// This allows us to wait for the engine to complete initialization (including trigger subscriptions)
 	// before emitting the workflowActivated event, ensuring the event accurately reflects deployment status.
 	initDone := make(chan error, 1)
+	var engine services.Service
 
-	// Scope the engineFactory call so that decodedBinary goes out of scope immediately after the factory returns
-	engine, err := func() (services.Service, error) {
-		return h.engineFactory(
-			ctx,
-			spec.WorkflowID,
-			spec.WorkflowOwner,
-			workflowName,
-			spec.WorkflowTag,
-			configBytes,
-			decodedBinary,
-			initDone,
-		)
-	}()
+	if confidential {
+		h.lggr.Infow("routing workflow to confidential execution", "workflowID", spec.WorkflowID)
+		engine, err = h.confidentialEngineFactory(spec, workflowName, decodedBinary, initDone)
+	} else {
+		engine, err = h.engineFactory(ctx, spec.WorkflowID, spec.WorkflowOwner, workflowName, spec.WorkflowTag, configBytes, decodedBinary, initDone)
+	}
 	if err != nil {
 		return fmt.Errorf("failed to create workflow engine: %w", err)
 	}
@@ -898,6 +847,111 @@ func (h *eventHandler) tryEngineCreate(ctx context.Context, spec *job.WorkflowSp
 	return nil
 }
 
+// newV2EngineConfig builds the common EngineConfig shared by both the normal
+// WASM engine and the confidential engine paths. Caller supplies the module.
+func (h *eventHandler) newV2EngineConfig(
+	module host.ModuleV2,
+	workflowID, owner, tag, sdkName string,
+	name types.WorkflowName,
+	config []byte,
+) *v2.EngineConfig {
+	return &v2.EngineConfig{
+		Lggr:                  h.lggr,
+		Module:                module,
+		WorkflowConfig:        config,
+		CapRegistry:           h.capRegistry,
+		DonSubscriber:         h.workflowDonSubscriber,
+		UseLocalTimeProvider:  h.useLocalTimeProvider,
+		DonTimeStore:          h.donTimeStore,
+		ExecutionsStore:       h.workflowStore,
+		WorkflowID:            workflowID,
+		WorkflowOwner:         owner,
+		WorkflowName:          name,
+		WorkflowTag:           tag,
+		WorkflowEncryptionKey: h.workflowEncryptionKey,
+
+		LocalLimits:                       v2.EngineLimits{}, // all defaults
+		LocalLimiters:                     h.engineLimiters,
+		FeatureFlags:                      h.featureFlags,
+		GlobalExecutionConcurrencyLimiter: h.workflowLimits,
+
+		BeholderEmitter: func() custmsg.MessageEmitter {
+			h.emitterMu.RLock()
+			defer h.emitterMu.RUnlock()
+			return h.emitter
+		}(),
+		BillingClient: h.billingClient,
+
+		WorkflowRegistryAddress:       h.workflowRegistryAddress,
+		WorkflowRegistryChainSelector: h.workflowRegistryChainSelector,
+		OrgResolver:                   h.orgResolver,
+		SecretsFetcher:                h.secretsFetcher,
+		DebugMode:                     h.debugMode,
+		SdkName:                       sdkName,
+
+		ShardOrchestratorClient: h.shardOrchestratorClient,
+		ShardingEnabled:         h.shardingEnabled,
+		MyShardID:               h.myShardID,
+		ShardRoutingSteady:      h.shardRoutingSteady,
+	}
+}
+
+// wireInitDoneHook wires the initDone channel to the OnInitialized lifecycle hook.
+// This will be called when the engine completes initialization (including trigger subscriptions).
+// We compose with any existing hook to avoid overwriting test hooks or other user-provided hooks.
+func (h *eventHandler) wireInitDoneHook(cfg *v2.EngineConfig, initDone chan<- error) {
+	if initDone == nil {
+		return
+	}
+	existingHook := cfg.Hooks.OnInitialized
+	cfg.Hooks.OnInitialized = func(err error) {
+		// Signal completion to the handler first
+		initDone <- err
+		// Then call any existing hook (e.g., from tests)
+		if existingHook != nil {
+			existingHook(err)
+		}
+	}
+}
+
+// confidentialEngineFactory creates a V2 engine backed by a ConfidentialModule
+// instead of a local WASM module. The ConfidentialModule delegates execution to
+// the confidential-workflows capability which runs the WASM inside a TEE.
+func (h *eventHandler) confidentialEngineFactory(
+	spec *job.WorkflowSpec,
+	workflowName types.WorkflowName,
+	decodedBinary []byte,
+	initDone chan<- error,
+) (services.Service, error) {
+	attrs, err := v2.ParseWorkflowAttributes(spec.Attributes)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse workflow attributes: %w", err)
+	}
+
+	binaryHash := v2.ComputeBinaryHash(decodedBinary)
+
+	lggr := logger.Named(h.lggr, "WorkflowEngine.ConfidentialModule")
+	lggr = logger.With(lggr, "workflowID", spec.WorkflowID, "workflowName", spec.WorkflowName, "workflowOwner", spec.WorkflowOwner)
+
+	// nil resolver: raw binaryURL is passed to the enclave as-is.
+	// PR 5/5 (#21642) wires this to the storage service retriever
+	// so the enclave receives a presigned URL.
+	module := v2.NewConfidentialModule(
+		h.capRegistry,
+		spec.BinaryURL,
+		binaryHash,
+		spec.WorkflowID, spec.WorkflowOwner, workflowName.String(), spec.WorkflowTag,
+		attrs.VaultDonSecrets,
+		nil,
+		lggr,
+	)
+
+	cfg := h.newV2EngineConfig(module, spec.WorkflowID, spec.WorkflowOwner, spec.WorkflowTag, "", workflowName, []byte(spec.Config))
+	h.wireInitDoneHook(cfg, initDone)
+
+	return v2.NewEngine(cfg)
+}
+
 // logCustMsg emits a custom message to the external sink and logs an error if that fails.
 func logCustMsg(ctx context.Context, cma custmsg.MessageEmitter, msg string, log logger.Logger) {
 	err := cma.Emit(ctx, msg)