[CRE] [5/5] Wire confidential workflow execution into CRE

.changeset/confidential-workflow-execution.md

@@ -0,0 +1,5 @@
+---
+"chainlink": minor
+---
+
+Add confidential workflow execution: ConfidentialModule, relay handler, gateway wiring, single-DON capability fix #added #db_update

core/scripts/cre/environment/environment/workflow.go

@@ -475,7 +475,7 @@ func deployWorkflow(
 
 	fmt.Printf("\n⚙️ Registering workflow '%s' with the workflow registry\n\n", workflowNameFlag)
 
-	workflowID, registerErr := creworkflow.RegisterWithContract(ctx, sethClient, common.HexToAddress(workflowRegistryAddress), workflowRegistryVersion, uint64(donIDFlag), workflowNameFlag, "file://"+wasmWorkflowFilePathFlag, configPath, secretsPath, &containerTargetDirFlag)
+	workflowID, registerErr := creworkflow.RegisterWithContract(ctx, sethClient, common.HexToAddress(workflowRegistryAddress), workflowRegistryVersion, uint64(donIDFlag), workflowNameFlag, "file://"+wasmWorkflowFilePathFlag, configPath, secretsPath, nil, &containerTargetDirFlag)
 	if registerErr != nil {
 		return errors.Wrapf(registerErr, "❌ failed to register workflow %s", workflowNameFlag)
 	}

core/services/cre/cre.go

@@ -32,6 +32,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/compute"
+	"github.com/smartcontractkit/chainlink/v2/core/capabilities/confidentialrelay"
 	gatewayconnector "github.com/smartcontractkit/chainlink/v2/core/capabilities/gateway_connector"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/localcapmgr"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/remote"
@@ -169,6 +170,13 @@ func (s *Services) newSubservices(
 		}
 		s.GatewayConnectorWrapper = gatewayConnectorWrapper
 		srvs = append(srvs, gatewayConnectorWrapper)
+
+		relayService := confidentialrelay.NewService(
+			gatewayConnectorWrapper,
+			opts.CapabilitiesRegistry,
+			lggr,
+		)
+		srvs = append(srvs, relayService)
 	}
 
 	if cfg.CRE().Linking().URL() != "" {

core/services/gateway/handler_factory.go

@@ -18,18 +18,20 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
 	v2 "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities/v2"
+	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/confidentialrelay"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/functions"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/vault"
 	"github.com/smartcontractkit/chainlink/v2/core/services/gateway/network"
 	workflowsyncerv2 "github.com/smartcontractkit/chainlink/v2/core/services/workflows/syncer/v2"
 )
 
 const (
-	FunctionsHandlerType   HandlerType = "functions"
-	DummyHandlerType       HandlerType = "dummy"
-	WebAPICapabilitiesType HandlerType = "web-api-capabilities" //  Handler for v0.1 HTTP capabilities for DAG workflows
-	HTTPCapabilityType     HandlerType = "http-capabilities"    // Handler for v1.0 HTTP capabilities for NoDAG workflows
-	VaultHandlerType       HandlerType = "vault"
+	FunctionsHandlerType         HandlerType = "functions"
+	DummyHandlerType             HandlerType = "dummy"
+	WebAPICapabilitiesType       HandlerType = "web-api-capabilities" //  Handler for v0.1 HTTP capabilities for DAG workflows
+	HTTPCapabilityType           HandlerType = "http-capabilities"    // Handler for v1.0 HTTP capabilities for NoDAG workflows
+	VaultHandlerType             HandlerType = "vault"
+	ConfidentialRelayHandlerType HandlerType = "confidential-compute-relay"
 )
 
 type handlerFactory struct {
@@ -87,6 +89,8 @@ func (hf *handlerFactory) NewHandler(
 	case VaultHandlerType:
 		requestAuthorizer := vaultcap.NewRequestAuthorizer(hf.lggr, hf.workflowRegistrySyncer)
 		return vault.NewHandler(handlerConfig, donConfig, don, hf.capabilitiesRegistry, requestAuthorizer, hf.lggr, clockwork.NewRealClock(), hf.lf)
+	case ConfidentialRelayHandlerType:
+		return confidentialrelay.NewHandler(handlerConfig, donConfig, don, hf.lggr, clockwork.NewRealClock(), hf.lf)
 	default:
 		return nil, fmt.Errorf("unsupported handler type %s", handlerType)
 	}

core/services/gateway/handlers/confidentialrelay/aggregator.go

@@ -0,0 +1,56 @@
+package confidentialrelay
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	jsonrpc "github.com/smartcontractkit/chainlink-common/pkg/jsonrpc2"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+var (
+	errInsufficientResponsesForQuorum = errors.New("insufficient valid responses to reach quorum")
+	errQuorumUnobtainable             = errors.New("quorum unobtainable")
+)
+
+type aggregator struct{}
+
+func (a *aggregator) Aggregate(resps map[string]jsonrpc.Response[json.RawMessage], donF int, donMembersCount int, l logger.Logger) (*jsonrpc.Response[json.RawMessage], error) {
+	// F+1 (QuorumFPlusOne) is sufficient because each relay node calls the
+	// target DON (Vault or capability) through CRE's standard capability
+	// dispatch, which includes DON-level consensus. Every honest relay node
+	// receives the same consensus-aggregated response and performs deterministic
+	// translation, producing byte-identical outputs. F+1 matching responses
+	// therefore guarantees at least one honest node vouched for the result.
+	requiredQuorum := donF + 1
+
+	if len(resps) < requiredQuorum {
+		return nil, errInsufficientResponsesForQuorum
+	}
+
+	shaToCount := map[string]int{}
+	maxShaToCount := 0
+	for _, r := range resps {
+		sha, err := r.Digest()
+		if err != nil {
+			l.Errorw("failed to compute digest of response during quorum validation, skipping...", "error", err)
+			continue
+		}
+		shaToCount[sha]++
+		if shaToCount[sha] > maxShaToCount {
+			maxShaToCount = shaToCount[sha]
+		}
+		if shaToCount[sha] >= requiredQuorum {
+			return &r, nil
+		}
+	}
+
+	remainingResponses := donMembersCount - len(resps)
+	if maxShaToCount+remainingResponses < requiredQuorum {
+		l.Warnw("quorum unattainable for request", "requiredQuorum", requiredQuorum, "remainingResponses", remainingResponses, "maxShaToCount", maxShaToCount)
+		return nil, fmt.Errorf("%w: requiredQuorum=%d, maxShaToCount=%d, remainingResponses=%d", errQuorumUnobtainable, requiredQuorum, maxShaToCount, remainingResponses)
+	}
+
+	return nil, errInsufficientResponsesForQuorum
+}