Allow capability DONs to include OCR attestation of the responses (#1907)

* Allow capability DONs to include OCR attestation of the responses

* Added tests for ResponseToReportData

* Move OCR attestation to response

* Set OCRAttestation to the response

* Hash based OCR feature flag

Author: dhaidashenko

keystore/corekeys/ocr2key/evm_keyring.go

@@ -87,13 +87,7 @@ func (ekr *evmKeyring) Verify3(publicKey ocrtypes.OnchainPublicKey, cd ocrtypes.
 }
 
 func (ekr *evmKeyring) VerifyBlob(pubkey types.OnchainPublicKey, b, sig []byte) bool {
-	authorPubkey, err := crypto.SigToPub(b, sig)
-	if err != nil {
-		return false
-	}
-	authorAddress := crypto.PubkeyToAddress(*authorPubkey)
-	// no need for constant time compare since neither arg is sensitive
-	return bytes.Equal(pubkey[:], authorAddress[:])
+	return EvmVerifyBlob(pubkey, b, sig)
 }
 
 func (ekr *evmKeyring) MaxSignatureLength() int {
@@ -116,3 +110,13 @@ func (ekr *evmKeyring) Unmarshal(in []byte) error {
 	ekr.privateKey = func() *ecdsa.PrivateKey { return privateKey }
 	return nil
 }
+
+func EvmVerifyBlob(pubkey types.OnchainPublicKey, b, sig []byte) bool {
+	authorPubkey, err := crypto.SigToPub(b, sig)
+	if err != nil {
+		return false
+	}
+	authorAddress := crypto.PubkeyToAddress(*authorPubkey)
+	// no need for constant time compare since neither arg is sensitive
+	return bytes.Equal(pubkey[:], authorAddress[:])
+}

pkg/capabilities/capabilities.go

@@ -2,6 +2,7 @@ package capabilities
 
 import (
 	"context"
+	"encoding/binary"
 	"errors"
 	"fmt"
 	"iter"
@@ -10,6 +11,7 @@ import (
 	"strings"
 	"time"
 
+	"golang.org/x/crypto/sha3"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 
@@ -39,6 +41,21 @@ func (e errStopExecution) Is(err error) bool {
 	return strings.Contains(err.Error(), errStopExecutionMsg)
 }
 
+// ErrResponsePayloadNotAvailable is returned when a capability's Execute method cannot provide a response payload and engine should wait for another response instead of treating it as an error.
+var ErrResponsePayloadNotAvailable = &responsePayloadNotAvailableError{}
+
+type responsePayloadNotAvailableError struct{}
+
+const errResponsePayloadNotAvailableMsg = "__response_payload_not_available"
+
+func (e responsePayloadNotAvailableError) Error() string {
+	return errResponsePayloadNotAvailableMsg
+}
+
+func (e responsePayloadNotAvailableError) Is(err error) bool {
+	return strings.Contains(err.Error(), errResponsePayloadNotAvailableMsg)
+}
+
 // CapabilityType enum values.
 const (
 	CapabilityTypeUnknown   CapabilityType = "unknown"
@@ -76,14 +93,63 @@ type CapabilityResponse struct {
 	Metadata ResponseMetadata
 
 	// Payload is used for no DAG workflows
-	Payload *anypb.Any
+	Payload        *anypb.Any
+	OCRAttestation *OCRAttestation
 }
 
 type ResponseMetadata struct {
 	Metering []MeteringNodeDetail
 	CapDON_N uint32
 }
 
+type OCRAttestation struct {
+	ConfigDigest   ocrtypes.ConfigDigest
+	SequenceNumber uint64
+	Sigs           []AttributedSignature
+}
+
+type AttributedSignature struct {
+	Signature []byte
+	Signer    uint32
+}
+
+func ExtractMeteringFromMetadata(sender p2ptypes.PeerID, metadata ResponseMetadata) (MeteringNodeDetail, error) {
+	if len(metadata.Metering) != 1 {
+		return MeteringNodeDetail{}, fmt.Errorf("unexpected number of metering records received from peer %s: got %d, want 1", sender, len(metadata.Metering))
+	}
+
+	rpt := metadata.Metering[0]
+	rpt.Peer2PeerID = sender.String()
+	return rpt, nil
+}
+
+func ResponseToReportData(workflowExecutionID, referenceID string, responsePayload []byte, metadata ResponseMetadata) ([32]byte, error) {
+	// use empty PeerID since the sender must not be included in the hash
+	metering, err := ExtractMeteringFromMetadata(p2ptypes.PeerID{}, metadata)
+	if err != nil {
+		return [32]byte{}, fmt.Errorf("failed to extract metering from metadata: %w", err)
+	}
+
+	hash := sha3.New256()
+	const domainSeparator = "CapabilityResponseReportData:v1"
+	hash.Write([]byte(domainSeparator))
+	// Helper to write a length-prefixed byte slice.
+	writeField := func(b []byte) {
+		// Use a fixed-width length prefix to make encoding unambiguous.
+		_ = binary.Write(hash, binary.BigEndian, uint64(len(b)))
+		_, _ = hash.Write(b)
+	}
+	writeField([]byte(workflowExecutionID))
+	writeField([]byte(referenceID))
+	writeField(responsePayload)
+	writeField([]byte(metering.SpendUnit))
+	writeField([]byte(metering.SpendValue))
+
+	var result [32]byte
+	copy(result[:], hash.Sum(nil))
+	return result, nil
+}
+
 type MeteringNodeDetail struct {
 	Peer2PeerID string
 	SpendUnit   string
@@ -94,6 +160,7 @@ type MeteringNodeDetail struct {
 type ResponseAndMetadata[T proto.Message] struct {
 	Response         T
 	ResponseMetadata ResponseMetadata
+	OCRAttestation   *OCRAttestation
 }
 
 type SpendLimit struct {

pkg/capabilities/capabilities_test.go

@@ -3,6 +3,7 @@ package capabilities
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"errors"
 	"maps"
 	"strings"
@@ -360,3 +361,113 @@ func TestRegistrationMetadata_ContextWithCRE(t *testing.T) {
 	ctx = md.ContextWithCRE(ctx)
 	require.Equal(t, "org-id", contexts.CREValue(ctx).Org)
 }
+
+func testResponseMetadata(spendUnit, spendValue string) ResponseMetadata {
+	return ResponseMetadata{
+		Metering: []MeteringNodeDetail{{SpendUnit: spendUnit, SpendValue: spendValue}},
+	}
+}
+
+func TestResponseToReportData(t *testing.T) {
+	t.Run("deterministic golden digest", func(t *testing.T) {
+		// SHA3-256 over domain "CapabilityResponseReportData:v1" plus length-prefixed fields.
+		want, err := hex.DecodeString("5d39c100ecf57f83b095cc9e129e0be24809e208af97a40e6447932749272a50")
+		require.NoError(t, err)
+		require.Len(t, want, 32)
+
+		got, err := ResponseToReportData(
+			"wf-exec",
+			"ref-1",
+			[]byte("payload"),
+			testResponseMetadata("unit", "42"),
+		)
+		require.NoError(t, err)
+		assert.Equal(t, want, got[:])
+	})
+
+	t.Run("same inputs yield same hash", func(t *testing.T) {
+		md := testResponseMetadata("u", "1")
+		a, err := ResponseToReportData("wf", "ref", []byte("p"), md)
+		require.NoError(t, err)
+		b, err := ResponseToReportData("wf", "ref", []byte("p"), md)
+		require.NoError(t, err)
+		assert.Equal(t, a, b)
+	})
+
+	t.Run("nil payload matches empty payload", func(t *testing.T) {
+		md := testResponseMetadata("u", "v")
+		a, err := ResponseToReportData("w", "r", nil, md)
+		require.NoError(t, err)
+		b, err := ResponseToReportData("w", "r", []byte{}, md)
+		require.NoError(t, err)
+		assert.Equal(t, a, b)
+	})
+
+	t.Run("each field affects the digest", func(t *testing.T) {
+		base, err := ResponseToReportData(
+			"workflow-exec-id",
+			"reference-id",
+			[]byte{0x01, 0x02},
+			testResponseMetadata("spend-unit", "spend-value"),
+		)
+		require.NoError(t, err)
+
+		other, err := ResponseToReportData(
+			"other-workflow",
+			"reference-id",
+			[]byte{0x01, 0x02},
+			testResponseMetadata("spend-unit", "spend-value"),
+		)
+		require.NoError(t, err)
+		assert.NotEqual(t, base, other)
+
+		other, err = ResponseToReportData(
+			"workflow-exec-id",
+			"other-ref",
+			[]byte{0x01, 0x02},
+			testResponseMetadata("spend-unit", "spend-value"),
+		)
+		require.NoError(t, err)
+		assert.NotEqual(t, base, other)
+
+		other, err = ResponseToReportData(
+			"workflow-exec-id",
+			"reference-id",
+			[]byte{0x01},
+			testResponseMetadata("spend-unit", "spend-value"),
+		)
+		require.NoError(t, err)
+		assert.NotEqual(t, base, other)
+
+		other, err = ResponseToReportData(
+			"workflow-exec-id",
+			"reference-id",
+			[]byte{0x01, 0x02},
+			testResponseMetadata("other-unit", "spend-value"),
+		)
+		require.NoError(t, err)
+		assert.NotEqual(t, base, other)
+
+		other, err = ResponseToReportData(
+			"workflow-exec-id",
+			"reference-id",
+			[]byte{0x01, 0x02},
+			testResponseMetadata("spend-unit", "other-value"),
+		)
+		require.NoError(t, err)
+		assert.NotEqual(t, base, other)
+	})
+
+	t.Run("metering must contain exactly one entry", func(t *testing.T) {
+		_, err := ResponseToReportData("w", "r", nil, ResponseMetadata{})
+		require.ErrorContains(t, err, "failed to extract metering from metadata: unexpected number of metering records received from peer 12D3KooW9pNAk8aiBuGVQtWRdbkLmo5qVL3e2h5UxbN2Nz9ttwiw: got 0, want 1")
+
+		_, err = ResponseToReportData("w", "r", nil, ResponseMetadata{
+			Metering: []MeteringNodeDetail{
+				{SpendUnit: "a", SpendValue: "1"},
+				{SpendUnit: "b", SpendValue: "2"},
+			},
+		})
+		require.ErrorContains(t, err, "failed to extract metering from metadata: unexpected number of metering records received from peer 12D3KooW9pNAk8aiBuGVQtWRdbkLmo5qVL3e2h5UxbN2Nz9ttwiw: got 2, want 1")
+	})
+}