fix(consensus/reactor): reject oversized proposals (#5324)

CHANGELOG.md

@@ -20,6 +20,9 @@
 
 ### BUG-FIXES
 
+- `[consensus]` Reject oversized proposals
+  ([\#5324](https://github.com/cometbft/cometbft/pull/5324))
+
 ### STATE-BREAKING
 
 ### API-BREAKING

consensus/byzantine_test.go

@@ -604,3 +604,86 @@
 }
 
 func (br *ByzantineReactor) InitPeer(peer p2p.Peer) p2p.Peer { return peer }
+
+// Large/oversized proposals should be rejected
+func TestRejectOversizedProposals(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	n := 2
+	css, cleanup := randConsensusNet(t, n, "consensus_reactor_test", newMockTickerFunc(false), newKVStore)
+	defer cleanup()
+
+	switches := make([]*p2p.Switch, n)
+	p2pLogger := consensusLogger().With("module", "p2p")
+	for i := 0; i < n; i++ {
+		switches[i] = p2p.MakeSwitch(
+			config.P2P,
+			i,
+			func(_ int, sw *p2p.Switch) *p2p.Switch {
+				return sw
+			})
+		switches[i].SetLogger(p2pLogger.With("validator", i))
+	}
+
+	reactors := make([]p2p.Reactor, n)
+	for i := 0; i < n; i++ {
+		conR := NewReactor(css[i], false)
+		defer func() { require.NoError(t, conR.Stop()) }()
+
+		conR.SetLogger(consensusLogger().With("validator", i))
+		reactors[i] = conR
+	}
+
+	p2p.MakeConnectedSwitches(config.P2P, n, func(i int, _ *p2p.Switch) *p2p.Switch {
+		switches[i].AddReactor("CONSENSUS", reactors[i])
+		return switches[i]
+	}, p2p.Connect2Switches)
+
+	peers := switches[0].Peers().List()
+	targetPeer := peers[0]
+
+	height := int64(1)
+	round := int32(0)
+	cs := css[0]
+
+	block, err := cs.createProposalBlock(ctx)
+	require.NoError(t, err)
+
+	blockParts, err := block.MakePartSet(types.BlockPartSizeBytes)
+	require.NoError(t, err)
+
+	// create oversized proposal
+	propBlockID := types.BlockID{Hash: block.Hash(), PartSetHeader: blockParts.Header()}
+	propBlockID.PartSetHeader.Total = 4294967295
+
+	proposal := types.NewProposal(height, round, -1, propBlockID)
+	p := proposal.ToProto()
+	if err := cs.privValidator.SignProposal(cs.state.ChainID, p); err != nil {
+		t.Error(err)
+	}
+	proposal.Signature = p.Signature
+
+	success := targetPeer.Send(p2p.Envelope{
+		ChannelID: DataChannel,
+		Message:   &cmtcons.Proposal{Proposal: *proposal.ToProto()},
+	})
+	require.True(t, success)
+
+	select {
+	case e := <-css[1].peerMsgQueue:
+		// if we receive a message here, the peer incorrectly accepted the
+		// oversized proposal
+		if _, receivedProposal := e.Msg.(*ProposalMessage); receivedProposal {
+			assert.Fail(t, "peer incorrectly accepted oversized proposal")
+			return
+		}
+		// invalid state, we received some other unexpected message type, fail
+		// the test
+		assert.Fail(t, "received unexpected message type on peer msg queue, expected *ProposalMessage")
+	case <-ctx.Done():
+	case <-time.After(500 * time.Millisecond):
+		// timeout after 500ms if nothing has happened and assume peer rejected
+		// the proposal
+	}
+}

consensus/reactor.go

@@ -322,6 +322,15 @@ func (conR *Reactor) ReceiveEnvelope(e p2p.Envelope) {
 		}
 		switch msg := msg.(type) {
 		case *ProposalMessage:
+			conR.conS.mtx.RLock()
+			maxBytes := conR.conS.state.ConsensusParams.Block.MaxBytes
+			conR.conS.mtx.RUnlock()
+			if err := msg.Proposal.ValidateBlockSize(maxBytes); err != nil {
+				conR.Logger.Error("Rejecting oversized proposal", "peer", e.Src, "height", msg.Proposal.Height)
+				conR.Switch.StopPeerForError(e.Src, ErrProposalTooManyParts)
+				return
+			}
+
 			ps.SetHasProposal(msg.Proposal)
 			conR.conS.peerMsgQueue <- msgInfo{msg, e.Src.ID()}
 		case *ProposalPOLMessage:
@@ -1625,6 +1634,12 @@ func (m *ProposalMessage) ValidateBasic() error {
 	return m.Proposal.ValidateBasic()
 }
 
+// ValidateBlockSize validates the proposals block size against a maximum. If
+// -1 is passed, types.MaxBlockSizeBytes will be used as the maximum.
+func (m *ProposalMessage) ValidateBlockSize(maxBlockSizeBytes int64) error {
+	return m.Proposal.ValidateBlockSize(maxBlockSizeBytes)
+}
+
 // String returns a string representation.
 func (m *ProposalMessage) String() string {
 	return fmt.Sprintf("[Proposal %v]", m.Proposal)

types/proposal.go

@@ -79,6 +79,22 @@ func (p *Proposal) ValidateBasic() error {
 	return nil
 }
 
+// ValidateBlockSize block size ensures that a proposal block is not larger
+// than a maximum number of bytes, based on the total amount of parts reported
+// in the PartSetHeader. If -1 is passed as the maxBlockSizeBytes,
+// types.MaxBlockSizeBytes will be used as the maximum.
+func (p *Proposal) ValidateBlockSize(maxBlockSizeBytes int64) error {
+	if maxBlockSizeBytes == -1 {
+		maxBlockSizeBytes = int64(MaxBlockSizeBytes)
+	}
+	totalParts := int64(p.BlockID.PartSetHeader.Total)
+	maxParts := (maxBlockSizeBytes-1)/int64(BlockPartSizeBytes) + 1
+	if totalParts > maxParts {
+		return fmt.Errorf("proposal has too many parts %d (max: %d)", totalParts, maxParts)
+	}
+	return nil
+}
+
 // String returns a string representation of the Proposal.
 //
 // 1. height

types/proposal_test.go

@@ -190,3 +190,30 @@ func TestProposalProtoBuf(t *testing.T) {
 		}
 	}
 }
+
+func TestProposalValidateBlockSize(t *testing.T) {
+	testCases := []struct {
+		testName     string
+		maxBlockSize int64
+		proposal     *Proposal
+		expectPass   bool
+	}{
+		{"10 chunk max, 5 chunk proposal, success", int64(10 * BlockPartSizeBytes), NewProposal(0, 0, 0, BlockID{PartSetHeader: PartSetHeader{Total: 5}}), true},
+		{"10 chunk max, 20 chunk proposal, fail", int64(10 * BlockPartSizeBytes), NewProposal(0, 0, 0, BlockID{PartSetHeader: PartSetHeader{Total: 20}}), false},
+		{"10 chunk max, max uint32 chunk proposal, fail", int64(10 * BlockPartSizeBytes), NewProposal(0, 0, 0, BlockID{PartSetHeader: PartSetHeader{Total: math.MaxUint32}}), false},
+		{"-1 chunk max, max uint32 chunk proposal, fail", -1, NewProposal(0, 0, 0, BlockID{PartSetHeader: PartSetHeader{Total: math.MaxUint32}}), false},
+		{"0 chunk max, max uint32 chunk proposal, fail", -1, NewProposal(0, 0, 0, BlockID{PartSetHeader: PartSetHeader{Total: math.MaxUint32}}), false},
+		{"total parts equals chunk max, success", -1, NewProposal(0, 0, 0, BlockID{PartSetHeader: PartSetHeader{Total: 1600}}), true},
+	}
+	for _, tc := range testCases {
+		tc := tc
+		t.Run(tc.testName, func(t *testing.T) {
+			err := tc.proposal.ValidateBlockSize(tc.maxBlockSize)
+			if tc.expectPass {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+			}
+		})
+	}
+}