[BFT] ReceiptValidator ensures k receipts committing to the execution result

engine/testutil/nodes.go

@@ -462,7 +462,13 @@ func ConsensusNode(t *testing.T, hub *stub.Hub, identity *flow.Identity, identit
 	assigner, err := chunks.NewChunkAssigner(flow.DefaultChunkAssignmentAlpha, node.State)
 	require.Nil(t, err)
 
-	receiptValidator := validation.NewReceiptValidator(node.State, node.Headers, node.Index, resultsDB, node.Seals)
+	receiptValidator := validation.NewReceiptValidator(
+		node.State,
+		node.Headers,
+		node.Index,
+		resultsDB,
+		node.Seals,
+	)
 
 	sealingEngine, err := sealing.NewEngine(
 		node.Log,

module/validation/receipt_validator.go

@@ -7,6 +7,7 @@ import (
 	"github.com/onflow/flow-go/crypto/hash"
 	"github.com/onflow/flow-go/engine"
 	"github.com/onflow/flow-go/model/flow"
+	"github.com/onflow/flow-go/module"
 	"github.com/onflow/flow-go/module/signature"
 	"github.com/onflow/flow-go/state"
 	"github.com/onflow/flow-go/state/fork"
@@ -15,7 +16,7 @@ import (
 )
 
 // receiptValidator holds all needed context for checking
-// receipt validity against current protocol state.
+// receipt validity against the current protocol state.
 type receiptValidator struct {
 	headers         storage.Headers
 	seals           storage.Seals
@@ -25,12 +26,14 @@ type receiptValidator struct {
 	signatureHasher hash.Hasher
 }
 
+var _ module.ReceiptValidator = (*receiptValidator)(nil)
+
 func NewReceiptValidator(state protocol.State,
 	headers storage.Headers,
 	index storage.Index,
 	results storage.ExecutionResults,
 	seals storage.Seals,
-) *receiptValidator {
+) module.ReceiptValidator {
 	rv := &receiptValidator{
 		state:           state,
 		headers:         headers,
@@ -39,7 +42,6 @@ func NewReceiptValidator(state protocol.State,
 		signatureHasher: signature.NewBLSHasher(signature.ExecutionReceiptTag),
 		seals:           seals,
 	}
-
 	return rv
 }
 
@@ -272,12 +274,24 @@ func (v *receiptValidator) ValidatePayload(candidate *flow.Block) error {
 		return fmt.Errorf("internal error while traversing the ancestor fork of unsealed blocks: %w", err)
 	}
 
+	// tracks the number of receipts committing to each result.
+	// it's ok to only index receipts at this point, because we will perform
+	// all needed checks after we have validated all results.
+	receiptsByResult := payload.Receipts.GroupByResultID()
+
 	// first validate all results that were included into payload
-	// if one of results is invalid we fail the whole check because it could be violating
-	// parent-children relationship
+	// if one of results is invalid we fail the whole check because it could be violating parent-children relationship
+	// each execution
 	for i, result := range payload.Results {
 		resultID := result.ID()
 
+		// Every included result must be accompanied by a receipt with a corresponding `ResultID`, in the same block.
+		// If a result is included without a corresponding receipt, it cannot be attributed to any executor.
+		receiptsForResult := uint(len(receiptsByResult.GetGroup(resultID)))
+		if receiptsForResult == 0 {
+			return engine.NewInvalidInputErrorf("no receipts for result %v at index %d", resultID, i)
+		}
+
 		// check for duplicated results
 		if _, isDuplicate := executionTree[resultID]; isDuplicate {
 			return engine.NewInvalidInputErrorf("duplicate result %v at index %d", resultID, i)

module/validation/receipt_validator_test.go

@@ -29,7 +29,13 @@ func (s *ReceiptValidationSuite) SetupTest() {
 	s.SetupChain()
 	s.publicKey = &fmock.PublicKey{}
 	s.Identities[s.ExeID].StakingPubKey = s.publicKey
-	s.receiptValidator = NewReceiptValidator(s.State, s.HeadersDB, s.IndexDB, s.ResultsDB, s.SealsDB)
+	s.receiptValidator = NewReceiptValidator(
+		s.State,
+		s.HeadersDB,
+		s.IndexDB,
+		s.ResultsDB,
+		s.SealsDB,
+	)
 }
 
 // TestReceiptValid try submitting valid receipt
@@ -745,3 +751,97 @@ func (s *ReceiptValidationSuite) TestValidateReceiptAfterBootstrap() {
 	err := s.receiptValidator.ValidatePayload(candidate)
 	s.Require().NoError(err)
 }
+
+// TestValidateReceiptResultWithoutReceipt tests a case when a malicious leader incorporates a made-up execution result
+// into their proposal. ReceiptValidator must ensure that for each result included in the block, there must be
+// at least one receipt included in that block as well.
+func (s *ReceiptValidationSuite) TestValidateReceiptResultWithoutReceipt() {
+	// assuming signatures are all good
+	s.publicKey.On("Verify", mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
+
+	// G <- A <- B
+	blocks, result0, seal := unittest.ChainFixture(2)
+	s.SealsIndex[blocks[0].ID()] = seal
+
+	receipts := unittest.ReceiptChainFor(blocks, result0)
+	blockA, blockB := blocks[1], blocks[2]
+	receiptA, receiptB := receipts[1], receipts[2]
+
+	blockA.Payload.Receipts = []*flow.ExecutionReceiptMeta{}
+	blockB.Payload.Receipts = []*flow.ExecutionReceiptMeta{receiptA.Meta()}
+	blockB.Payload.Results = []*flow.ExecutionResult{&receiptA.ExecutionResult}
+	// update block header so that blocks are chained together
+	unittest.ReconnectBlocksAndReceipts(blocks, receipts)
+	// assuming all receipts are executed by the correct executor
+	for _, r := range receipts {
+		r.ExecutorID = s.ExeID
+	}
+
+	for _, b := range blocks {
+		s.Extend(b)
+	}
+	s.PersistedResults[result0.ID()] = result0
+
+	candidate := unittest.BlockWithParentFixture(blockB.Header)
+	candidate.Payload = &flow.Payload{
+		Receipts: []*flow.ExecutionReceiptMeta{},
+		Results:  []*flow.ExecutionResult{&receiptB.ExecutionResult},
+	}
+
+	err := s.receiptValidator.ValidatePayload(candidate)
+	s.Require().Error(err)
+	s.Require().True(engine.IsInvalidInputError(err))
+}
+
+// TestValidateReceiptResultHasEnoughReceipts tests a case where a leader incorporates an execution result
+// into their proposal, which has multiple receipts and ReceiptValidator accepts it as valid payload.
+func (s *ReceiptValidationSuite) TestValidateReceiptResultHasEnoughReceipts() {
+	k := uint(5)
+	// assuming signatures are all good
+	s.publicKey.On("Verify", mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
+
+	// G <- A <- B
+	blocks, result0, seal := unittest.ChainFixture(2)
+	s.SealsIndex[blocks[0].ID()] = seal
+
+	receipts := unittest.ReceiptChainFor(blocks, result0)
+	blockA, blockB := blocks[1], blocks[2]
+	receiptA, receiptB := receipts[1], receipts[2]
+
+	blockA.Payload.Receipts = []*flow.ExecutionReceiptMeta{}
+	blockB.Payload.Receipts = []*flow.ExecutionReceiptMeta{receiptA.Meta()}
+	blockB.Payload.Results = []*flow.ExecutionResult{&receiptA.ExecutionResult}
+	// update block header so that blocks are chained together
+	unittest.ReconnectBlocksAndReceipts(blocks, receipts)
+	// assuming all receipts are executed by the correct executor
+	for _, r := range receipts {
+		r.ExecutorID = s.ExeID
+	}
+
+	for _, b := range blocks {
+		s.Extend(b)
+	}
+	s.PersistedResults[result0.ID()] = result0
+
+	candidateReceipts := []*flow.ExecutionReceiptMeta{receiptB.Meta()}
+	// add k-1 more receipts for the same execution result
+	for i := uint(1); i < k; i++ {
+		// use base receipt and change the executor ID, we don't care about signatures since we are not validating them
+		receipt := *receiptB.Meta()
+		// create a mock executor which submitted the receipt
+		executor := unittest.IdentityFixture(unittest.WithRole(flow.RoleExecution), unittest.WithStakingPubKey(s.publicKey))
+		receipt.ExecutorID = executor.NodeID
+		// update local identity table so the receipt is considered valid
+		s.Identities[executor.NodeID] = executor
+		candidateReceipts = append(candidateReceipts, &receipt)
+	}
+
+	candidate := unittest.BlockWithParentFixture(blockB.Header)
+	candidate.Payload = &flow.Payload{
+		Receipts: candidateReceipts,
+		Results:  []*flow.ExecutionResult{&receiptB.ExecutionResult},
+	}
+
+	err := s.receiptValidator.ValidatePayload(candidate)
+	s.Require().NoError(err)
+}