cl/sentinel: reject invalid ping and status requests

cl/sentinel/communication/ssz_snappy/encoding.go

@@ -32,6 +32,27 @@ import (
 	"github.com/erigontech/erigon/common/ssz"
 )
 
+var errCompressedPayloadLimit = errors.New("compressed payload exceeds maximum size")
+
+type compressedPayloadReader struct {
+	r         io.Reader
+	remaining uint64
+	read      uint64
+}
+
+func (r *compressedPayloadReader) Read(p []byte) (int, error) {
+	if r.remaining == 0 {
+		return 0, errCompressedPayloadLimit
+	}
+	if uint64(len(p)) > r.remaining {
+		p = p[:r.remaining]
+	}
+	n, err := r.r.Read(p)
+	r.remaining -= uint64(n)
+	r.read += uint64(n)
+	return n, err
+}
+
 func EncodeAndWrite(w io.Writer, val ssz.Marshaler, prefix ...byte) error {
 	enc := make([]byte, 0, val.EncodingSizeSSZ())
 	var err error
@@ -82,22 +103,59 @@ func DecodeAndRead(r io.Reader, val ssz.EncodableSSZ, b *clparams.BeaconChainCon
 }
 
 func DecodeAndReadNoForkDigest(r io.Reader, val ssz.EncodableSSZ, version clparams.StateVersion) error {
+	return decodeAndReadNoForkDigest(r, val, version, nil)
+}
+
+// DecodeAndReadNoForkDigestExact decodes a payload with an exact uncompressed size and no trailing data.
+func DecodeAndReadNoForkDigestExact(r io.Reader, val ssz.EncodableSSZ, version clparams.StateVersion, expectedSize uint64) error {
+	return decodeAndReadNoForkDigest(r, val, version, &expectedSize)
+}
+
+func decodeAndReadNoForkDigest(r io.Reader, val ssz.EncodableSSZ, version clparams.StateVersion, expectedSize *uint64) error {
 	// Read varint for length of message.
 	encodedLn, _, err := ReadUvarint(r)
 	if err != nil {
 		return fmt.Errorf("unable to read varint from message prefix: %w", err)
 	}
+	if expectedSize != nil && encodedLn != *expectedSize {
+		return fmt.Errorf("unexpected payload size: got %d, want %d", encodedLn, *expectedSize)
+	}
 	if encodedLn > uint64(16*datasize.MB) {
 		return errors.New("payload too big")
 	}
 
-	sr := snappypool.Reader(r)
+	compressedInput := r
+	var compressedReader *compressedPayloadReader
+	var maxCompressedSize uint64
+	if expectedSize != nil {
+		maxCompressedSize = 32 + encodedLn + encodedLn/6
+		compressedReader = &compressedPayloadReader{r: r, remaining: maxCompressedSize}
+		compressedInput = compressedReader
+	}
+	sr := snappypool.Reader(compressedInput)
 	defer snappypool.PutReader(sr)
 	raw := make([]byte, encodedLn)
 	if _, err := io.ReadFull(sr, raw); err != nil {
 		// fetch struct name of val
 		return fmt.Errorf("unable to readPacket: %w", err)
 	}
+	if expectedSize != nil {
+		if compressedReader.read >= maxCompressedSize {
+			return errCompressedPayloadLimit
+		}
+		compressedBytes := compressedReader.read
+		var extra [1]byte
+		_, err := io.ReadFull(sr, extra[:])
+		if compressedReader.read >= maxCompressedSize {
+			return errCompressedPayloadLimit
+		}
+		if err != nil && err != io.EOF { //nolint:errorlint // Only bare EOF proves clean stream termination.
+			return fmt.Errorf("unable to verify payload end: %w", err)
+		}
+		if err == nil || compressedReader.read != compressedBytes {
+			return errors.New("payload contains trailing bytes")
+		}
+	}
 
 	err = val.DecodeSSZ(raw, int(version))
 	if err != nil {

cl/sentinel/communication/ssz_snappy/encoding_test.go

@@ -0,0 +1,136 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package ssz_snappy
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/erigontech/erigon/cl/clparams"
+	"github.com/erigontech/erigon/cl/cltypes"
+)
+
+var snappyStreamIdentifier = []byte{0xff, 0x06, 0x00, 0x00, 's', 'N', 'a', 'P', 'p', 'Y'}
+
+type countingReader struct {
+	r     *bytes.Reader
+	bytes int
+}
+
+type terminalErrorReader struct {
+	r   *bytes.Reader
+	err error
+}
+
+func (r *terminalErrorReader) Read(p []byte) (int, error) {
+	n, err := r.r.Read(p)
+	if n == 0 {
+		return 0, r.err
+	}
+	return n, err
+}
+
+func (r *countingReader) Read(p []byte) (int, error) {
+	n, err := r.r.Read(p)
+	r.bytes += n
+	return n, err
+}
+
+func TestDecodeAndReadNoForkDigestExactRejectsTrailingFrames(t *testing.T) {
+	for _, test := range []struct {
+		name  string
+		frame []byte
+	}{
+		{name: "stream identifier", frame: snappyStreamIdentifier},
+		{name: "skippable frame", frame: []byte{0x80, 0x01, 0x00, 0x00, 0x00}},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			var encoded bytes.Buffer
+			require.NoError(t, EncodeAndWrite(&encoded, &cltypes.Ping{Id: 1}))
+			encoded.Write(test.frame)
+
+			err := DecodeAndReadNoForkDigestExact(bytes.NewReader(encoded.Bytes()), &cltypes.Ping{}, clparams.Phase0Version, 8)
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestDecodeAndReadNoForkDigestExactBoundsCompressedInput(t *testing.T) {
+	var encoded bytes.Buffer
+	require.NoError(t, EncodeAndWrite(&encoded, &cltypes.Ping{Id: 1}))
+	for range 10 {
+		encoded.Write(snappyStreamIdentifier)
+	}
+
+	reader := &countingReader{r: bytes.NewReader(encoded.Bytes())}
+	err := DecodeAndReadNoForkDigestExact(reader, &cltypes.Ping{}, clparams.Phase0Version, 8)
+	require.Error(t, err)
+	require.LessOrEqual(t, reader.bytes, 1+32+8+8/6)
+}
+
+func TestDecodeAndReadNoForkDigestExactFailsClosedAtCompressedLimit(t *testing.T) {
+	const payloadSize = 8
+	const maxCompressedSize = 32 + payloadSize + payloadSize/6
+
+	var encoded bytes.Buffer
+	require.NoError(t, EncodeAndWrite(&encoded, &cltypes.Ping{Id: 1}))
+	prefix, body := encoded.Bytes()[:1], encoded.Bytes()[1:]
+	require.Less(t, len(body), maxCompressedSize)
+	var ping cltypes.Ping
+	require.NoError(t, DecodeAndReadNoForkDigestExact(bytes.NewReader(encoded.Bytes()), &ping, clparams.Phase0Version, payloadSize))
+	require.Equal(t, uint64(1), ping.Id)
+
+	padding := make([]byte, maxCompressedSize-len(body))
+	padding[0] = 0x80
+	padding[1] = byte(len(padding) - 4)
+	exactMax := append(append(append(append([]byte{}, prefix...), body[:len(snappyStreamIdentifier)]...), padding...), body[len(snappyStreamIdentifier):]...)
+	require.Len(t, exactMax, 1+maxCompressedSize)
+
+	exactMaxReader := &countingReader{r: bytes.NewReader(exactMax)}
+	require.ErrorIs(t, DecodeAndReadNoForkDigestExact(exactMaxReader, &cltypes.Ping{}, clparams.Phase0Version, payloadSize), errCompressedPayloadLimit)
+	require.LessOrEqual(t, exactMaxReader.bytes, 1+maxCompressedSize)
+
+	overMax := append(append([]byte{}, exactMax...), 0)
+	reader := &countingReader{r: bytes.NewReader(overMax)}
+	require.ErrorIs(t, DecodeAndReadNoForkDigestExact(reader, &cltypes.Ping{}, clparams.Phase0Version, payloadSize), errCompressedPayloadLimit)
+	require.LessOrEqual(t, reader.bytes, 1+maxCompressedSize)
+}
+
+func TestDecodeAndReadNoForkDigestExactPreservesTerminalReadError(t *testing.T) {
+	var encoded bytes.Buffer
+	require.NoError(t, EncodeAndWrite(&encoded, &cltypes.Ping{Id: 1}))
+
+	terminalErr := errors.New("terminal read error")
+	reader := &terminalErrorReader{r: bytes.NewReader(encoded.Bytes()), err: terminalErr}
+	err := DecodeAndReadNoForkDigestExact(reader, &cltypes.Ping{}, clparams.Phase0Version, 8)
+	require.ErrorIs(t, err, terminalErr)
+}
+
+func TestDecodeAndReadNoForkDigestExactRejectsWrappedEOF(t *testing.T) {
+	var encoded bytes.Buffer
+	require.NoError(t, EncodeAndWrite(&encoded, &cltypes.Ping{Id: 1}))
+
+	terminalErr := fmt.Errorf("transport failed: %w", io.EOF)
+	reader := &terminalErrorReader{r: bytes.NewReader(encoded.Bytes()), err: terminalErr}
+	err := DecodeAndReadNoForkDigestExact(reader, &cltypes.Ping{}, clparams.Phase0Version, 8)
+	require.ErrorIs(t, err, terminalErr)
+}

cl/sentinel/handlers/handlers.go

@@ -192,6 +192,9 @@ func (c *ConsensusHandlers) wrapStreamHandler(name string, fn func(s network.Str
 		// SetDeadline covers both directions.
 		if err := s.SetDeadline(time.Now().Add(5 * time.Second)); err != nil {
 			log.Trace("failed to set stream deadline", "err", err)
+			_ = s.Reset()
+			_ = s.Close()
+			return
 		}
 
 		if err := fn(s); err != nil {

cl/sentinel/handlers/handlers_test.go

@@ -0,0 +1,78 @@
+// Copyright 2026 The Erigon Authors
+// This file is part of Erigon.
+//
+// Erigon is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// Erigon is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with Erigon. If not, see <http://www.gnu.org/licenses/>.
+
+package handlers
+
+import (
+	"errors"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/stretchr/testify/require"
+
+	"github.com/erigontech/erigon/cl/sentinel/communication"
+)
+
+type remotePeerConn struct {
+	network.Conn
+	peerID peer.ID
+}
+
+func (c *remotePeerConn) RemotePeer() peer.ID { return c.peerID }
+
+type deadlineFailingStream struct {
+	network.Stream
+	conn   network.Conn
+	err    error
+	reset  bool
+	closed bool
+}
+
+func (s *deadlineFailingStream) Conn() network.Conn          { return s.conn }
+func (s *deadlineFailingStream) SetDeadline(time.Time) error { return s.err }
+func (s *deadlineFailingStream) Reset() error                { s.reset = true; return nil }
+func (s *deadlineFailingStream) Close() error                { s.closed = true; return nil }
+
+func TestStreamHandlerStopsWhenDeadlineCannotBeSet(t *testing.T) {
+	h, err := libp2p.New(libp2p.NoListenAddrs)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, h.Close()) })
+
+	peerID := peer.ID("deadline-failure-peer")
+	c := &ConsensusHandlers{host: h, rateLimiter: newPeerRateLimiter()}
+	stream := &deadlineFailingStream{
+		conn: &remotePeerConn{peerID: peerID},
+		err:  errors.New("deadline unavailable"),
+	}
+	handlerCalled := false
+	handler := c.wrapStreamHandler(communication.PingProtocolV1, func(network.Stream) error {
+		handlerCalled = true
+		return nil
+	})
+
+	handler(stream)
+
+	require.False(t, handlerCalled)
+	require.True(t, stream.reset)
+	require.True(t, stream.closed)
+	counter, ok := c.rateLimiter.concurrency.Load(peerID.String())
+	require.True(t, ok)
+	require.Zero(t, counter.(*atomic.Int32).Load())
+}

cl/sentinel/handlers/heartbeats.go

@@ -18,7 +18,6 @@ package handlers
 
 import (
 	"encoding/hex"
-	"io"
 	"strings"
 
 	"github.com/libp2p/go-libp2p/core/network"
@@ -34,6 +33,10 @@ import (
 // Since packets are just structs, they can be resent with no issue
 
 func (c *ConsensusHandlers) pingHandler(s network.Stream) error {
+	request := &cltypes.Ping{}
+	if err := ssz_snappy.DecodeAndReadNoForkDigestExact(s, request, clparams.Phase0Version, uint64(request.EncodingSizeSSZ())); err != nil {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
+	}
 	return ssz_snappy.EncodeAndWrite(s, &cltypes.Ping{
 		Id: c.me.Seq(),
 	}, SuccessfulResponsePrefix)
@@ -122,23 +125,19 @@ func (c *ConsensusHandlers) metadataV3Handler(s network.Stream) error {
 }
 
 func (c *ConsensusHandlers) statusHandler(s network.Stream) error {
-	// Per eth2 spec the responder must read the peer's Status before replying.
-	// Read and discard the incoming request body so the stream advances correctly.
 	peerStatus := &cltypes.Status{}
-	if err := ssz_snappy.DecodeAndReadNoForkDigest(s, peerStatus, clparams.Phase0Version); err != nil {
-		// If we cannot read the request, drain whatever is left and proceed.
-		_, _ = io.Copy(io.Discard, s)
+	if err := ssz_snappy.DecodeAndReadNoForkDigestExact(s, peerStatus, clparams.Phase0Version, uint64(peerStatus.EncodingSizeSSZ())); err != nil {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
 	}
 	status := c.hs.Status()
 	status.EarliestAvailableSlot = nil
 	return ssz_snappy.EncodeAndWrite(s, status, SuccessfulResponsePrefix)
 }
 
 func (c *ConsensusHandlers) statusV2Handler(s network.Stream) error {
-	// Per eth2 spec the responder must read the peer's Status before replying.
-	peerStatus := &cltypes.Status{}
-	if err := ssz_snappy.DecodeAndReadNoForkDigest(s, peerStatus, clparams.Phase0Version); err != nil {
-		_, _ = io.Copy(io.Discard, s)
+	peerStatus := &cltypes.Status{EarliestAvailableSlot: new(uint64)}
+	if err := ssz_snappy.DecodeAndReadNoForkDigestExact(s, peerStatus, clparams.FuluVersion, uint64(peerStatus.EncodingSizeSSZ())); err != nil {
+		return ssz_snappy.EncodeAndWrite(s, &emptyString{}, InvalidRequestPrefix)
 	}
 	status := c.hs.Status()
 	forkDigest, err := c.ethClock.CurrentForkDigest()