From f076f1e6ed6ae16100098178a9b59f5b7228dba1 Mon Sep 17 00:00:00 2001 From: kewei Date: Wed, 26 Aug 2026 15:05:35 +0800 Subject: [PATCH 1/3] cl/sentinel: reject invalid ping and status requests --- .../communication/ssz_snappy/encoding.go | 57 ++++- .../communication/ssz_snappy/encoding_test.go | 99 +++++++++ cl/sentinel/handlers/heartbeats.go | 19 +- cl/sentinel/handlers/heartbeats_test.go | 202 ++++++++++++++---- .../handlers/rate_limiter_integration_test.go | 3 +- cl/sentinel/sentinel_requests_test.go | 1 + 6 files changed, 322 insertions(+), 59 deletions(-) create mode 100644 cl/sentinel/communication/ssz_snappy/encoding_test.go diff --git a/cl/sentinel/communication/ssz_snappy/encoding.go b/cl/sentinel/communication/ssz_snappy/encoding.go index e9a2f409ef7..a057170a2cb 100644 --- a/cl/sentinel/communication/ssz_snappy/encoding.go +++ b/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,56 @@ 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) +} + +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 errors.Is(err, errCompressedPayloadLimit) { + if compressedReader.read != compressedBytes { + return errors.New("payload contains trailing bytes") + } + } else if !errors.Is(err, io.EOF) || compressedReader.read != compressedBytes { + return errors.New("payload contains trailing bytes") + } + } err = val.DecodeSSZ(raw, int(version)) if err != nil { diff --git a/cl/sentinel/communication/ssz_snappy/encoding_test.go b/cl/sentinel/communication/ssz_snappy/encoding_test.go new file mode 100644 index 00000000000..9a462c7f515 --- /dev/null +++ b/cl/sentinel/communication/ssz_snappy/encoding_test.go @@ -0,0 +1,99 @@ +// 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 . + +package ssz_snappy + +import ( + "bytes" + "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 +} + +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 TestDecodeAndReadNoForkDigestExactCompressedLimitIsInclusive(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) + + 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)} + var ping cltypes.Ping + require.NoError(t, DecodeAndReadNoForkDigestExact(exactMaxReader, &ping, clparams.Phase0Version, payloadSize)) + require.Equal(t, uint64(1), ping.Id) + require.LessOrEqual(t, exactMaxReader.bytes, 1+maxCompressedSize) + + overMax := append(append([]byte{}, exactMax...), 0) + reader := &countingReader{r: bytes.NewReader(overMax)} + require.NoError(t, DecodeAndReadNoForkDigestExact(reader, &cltypes.Ping{}, clparams.Phase0Version, payloadSize)) + require.LessOrEqual(t, reader.bytes, 1+maxCompressedSize) +} diff --git a/cl/sentinel/handlers/heartbeats.go b/cl/sentinel/handlers/heartbeats.go index c3fae5b998d..e8d6d0b7f38 100644 --- a/cl/sentinel/handlers/heartbeats.go +++ b/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,12 +125,9 @@ 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 @@ -135,10 +135,9 @@ func (c *ConsensusHandlers) statusHandler(s network.Stream) error { } 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() diff --git a/cl/sentinel/handlers/heartbeats_test.go b/cl/sentinel/handlers/heartbeats_test.go index 7b75bbf8db4..e579089baaf 100644 --- a/cl/sentinel/handlers/heartbeats_test.go +++ b/cl/sentinel/handlers/heartbeats_test.go @@ -20,10 +20,12 @@ import ( "bytes" "context" "crypto/ecdsa" + "fmt" "net/http" "testing" "github.com/libp2p/go-libp2p" + "github.com/libp2p/go-libp2p/core/network" "github.com/libp2p/go-libp2p/core/peer" "github.com/libp2p/go-libp2p/core/protocol" "github.com/stretchr/testify/require" @@ -50,6 +52,16 @@ var ( syncnetsTestVal = [1]byte{56} ) +type rawSSZ []byte + +func (r rawSSZ) EncodeSSZ(dst []byte) ([]byte, error) { + return append(dst, r...), nil +} + +func (r rawSSZ) EncodingSizeSSZ() int { + return len(r) +} + func newkey() *ecdsa.PrivateKey { key, err := crypto.GenerateKey() if err != nil { @@ -70,8 +82,9 @@ func testLocalNode(t *testing.T) *enode.LocalNode { return ln } -func TestPing(t *testing.T) { - ctx := context.Background() +func newPingTestStream(t *testing.T) network.Stream { + t.Helper() + ctx := t.Context() host, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) require.NoError(t, err) @@ -87,19 +100,16 @@ func TestPing(t *testing.T) { }) require.NoError(t, err) - peersPool := peers.NewPool(host) beaconDB, indiciesDB := setupStore(t) - f := forkchoicemock.NewForkChoiceStorageMock(t) ethClock := getEthClock(t) - _, beaconCfg := clparams.GetConfigsByNetwork(1) c := NewConsensusHandlers( ctx, beaconDB, indiciesDB, host, - peersPool, + peers.NewPool(host), &clparams.NetworkConfig{}, testLocalNode(t), beaconCfg, @@ -110,14 +120,25 @@ func TestPing(t *testing.T) { stream, err := host1.NewStream(ctx, host.ID(), protocol.ID(communication.PingProtocolV1)) require.NoError(t, err) + return stream +} - _, err = stream.Write(nil) +func requireResponseCode(t *testing.T, stream network.Stream, expected byte) { + t.Helper() + responseCode := make([]byte, 1) + _, err := stream.Read(responseCode) require.NoError(t, err) + require.Equal(t, expected, responseCode[0]) +} - firstByte := make([]byte, 1) - _, err = stream.Read(firstByte) +func TestPing(t *testing.T) { + stream := newPingTestStream(t) + + err := ssz_snappy.EncodeAndWrite(stream, &cltypes.Ping{Id: 1}) require.NoError(t, err) - require.Equal(t, firstByte[0], byte(0)) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(SuccessfulResponsePrefix)) p := &cltypes.Ping{} @@ -125,6 +146,41 @@ func TestPing(t *testing.T) { require.NoError(t, err) } +func TestPingRejectsEmptyRequest(t *testing.T) { + stream := newPingTestStream(t) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + +func TestPingRejectsTruncatedRequest(t *testing.T) { + stream := newPingTestStream(t) + require.NoError(t, ssz_snappy.EncodeAndWrite(stream, rawSSZ(make([]byte, 7)))) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + +func TestPingRejectsOversizedRequest(t *testing.T) { + stream := newPingTestStream(t) + require.NoError(t, ssz_snappy.EncodeAndWrite(stream, rawSSZ(make([]byte, 9)))) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + +func TestPingRejectsTrailingBytes(t *testing.T) { + stream := newPingTestStream(t) + var request bytes.Buffer + require.NoError(t, ssz_snappy.EncodeAndWrite(&request, &cltypes.Ping{Id: 1})) + require.NoError(t, request.WriteByte(0)) + _, err := stream.Write(request.Bytes()) + require.NoError(t, err) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + func TestGoodbye(t *testing.T) { ctx := context.Background() @@ -301,8 +357,9 @@ func TestMetadataV1(t *testing.T) { require.Equal(t, attnetsTestVal, p.Attnets) } -func TestStatus(t *testing.T) { - ctx := context.Background() +func newStatusTestStream(t *testing.T, protocolID protocol.ID) (network.Stream, *cltypes.Status) { + t.Helper() + ctx := t.Context() host, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) require.NoError(t, err) @@ -318,44 +375,28 @@ func TestStatus(t *testing.T) { }) require.NoError(t, err) - peersPool := peers.NewPool(host) beaconDB, indiciesDB := setupStore(t) - f := forkchoicemock.NewForkChoiceStorageMock(t) - - // Create mock for PeerDasStateReader ctrl := gomock.NewController(t) - mockPeerDasStateReader := peerdasstatemock.NewMockPeerDasStateReader(ctrl) - mockPeerDasStateReader.EXPECT(). - GetEarliestAvailableSlot(). - Return(uint64(0)). - AnyTimes() - mockPeerDasStateReader.EXPECT(). - GetRealCgc(). - Return(uint64(0)). - AnyTimes() - mockPeerDasStateReader.EXPECT(). - GetAdvertisedCgc(). - Return(uint64(0)). - AnyTimes() - - // Create a simple HTTP handler for the handshake - handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - w.WriteHeader(http.StatusOK) - }) + peerDasStateReader := peerdasstatemock.NewMockPeerDasStateReader(ctrl) + peerDasStateReader.EXPECT().GetEarliestAvailableSlot().Return(uint64(0)).AnyTimes() + peerDasStateReader.EXPECT().GetRealCgc().Return(uint64(0)).AnyTimes() + peerDasStateReader.EXPECT().GetAdvertisedCgc().Return(uint64(0)).AnyTimes() ethClock := getEthClock(t) - hs := handshake.New(ctx, ethClock, &clparams.MainnetBeaconConfig, handler, mockPeerDasStateReader) + hs := handshake.New(ctx, ethClock, &clparams.MainnetBeaconConfig, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }), peerDasStateReader) forkDigest, err := ethClock.CurrentForkDigest() require.NoError(t, err) - s := &cltypes.Status{ + status := &cltypes.Status{ ForkDigest: forkDigest, FinalizedRoot: common.Hash{1, 2, 4}, HeadRoot: common.Hash{1, 2, 4}, FinalizedEpoch: 1, HeadSlot: 1, } - hs.SetStatus(s) + hs.SetStatus(status) nc := clparams.NetworkConfigs[chainspec.MainnetChainID] _, beaconCfg := clparams.GetConfigsByNetwork(1) c := NewConsensusHandlers( @@ -363,17 +404,22 @@ func TestStatus(t *testing.T) { beaconDB, indiciesDB, host, - peersPool, + peers.NewPool(host), &nc, testLocalNode(t), beaconCfg, - getEthClock(t), - hs, f, nil, nil, mockPeerDasStateReader, true, + ethClock, + hs, f, nil, nil, peerDasStateReader, true, ) c.Start() - stream, err := host1.NewStream(ctx, host.ID(), protocol.ID(communication.StatusProtocolV1)) + stream, err := host1.NewStream(ctx, host.ID(), protocolID) require.NoError(t, err) + return stream, status +} + +func TestStatus(t *testing.T) { + stream, expectedStatus := newStatusTestStream(t, protocol.ID(communication.StatusProtocolV1)) // Send a Status request body (per eth2 spec the requester sends its own Status). reqStatus := &cltypes.Status{ @@ -382,18 +428,80 @@ func TestStatus(t *testing.T) { FinalizedEpoch: 2, HeadSlot: 2, } - err = ssz_snappy.EncodeAndWrite(stream, reqStatus) + err := ssz_snappy.EncodeAndWrite(stream, reqStatus) require.NoError(t, err) + require.NoError(t, stream.CloseWrite()) - firstByte := make([]byte, 1) - _, err = stream.Read(firstByte) - require.NoError(t, err) - require.Equal(t, firstByte[0], byte(0)) + requireResponseCode(t, stream, byte(SuccessfulResponsePrefix)) p := &cltypes.Status{} err = ssz_snappy.DecodeAndReadNoForkDigest(stream, p, clparams.Phase0Version) require.NoError(t, err) - require.Equal(t, s, p) + require.Equal(t, expectedStatus, p) +} + +func TestStatusRejectsEmptyRequest(t *testing.T) { + stream, _ := newStatusTestStream(t, protocol.ID(communication.StatusProtocolV1)) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + +func TestStatusRejectsTruncatedRequest(t *testing.T) { + stream, _ := newStatusTestStream(t, protocol.ID(communication.StatusProtocolV1)) + require.NoError(t, ssz_snappy.EncodeAndWrite(stream, rawSSZ(make([]byte, 83)))) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + +func TestStatusRejectsOversizedRequest(t *testing.T) { + stream, _ := newStatusTestStream(t, protocol.ID(communication.StatusProtocolV1)) + require.NoError(t, ssz_snappy.EncodeAndWrite(stream, rawSSZ(make([]byte, 85)))) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + +func TestStatusV2(t *testing.T) { + stream, expectedStatus := newStatusTestStream(t, protocol.ID(communication.StatusProtocolV2)) + earliestAvailableSlot := uint64(9) + requestStatus := &cltypes.Status{ + FinalizedRoot: common.Hash{9, 8, 7}, + HeadRoot: common.Hash{9, 8, 7}, + FinalizedEpoch: 2, + HeadSlot: 2, + EarliestAvailableSlot: &earliestAvailableSlot, + } + require.NoError(t, ssz_snappy.EncodeAndWrite(stream, requestStatus)) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(SuccessfulResponsePrefix)) + + responseStatus := &cltypes.Status{} + require.NoError(t, ssz_snappy.DecodeAndReadNoForkDigest(stream, responseStatus, clparams.FuluVersion)) + earliestAvailableSlot = 0 + expectedStatus.EarliestAvailableSlot = &earliestAvailableSlot + require.Equal(t, expectedStatus, responseStatus) +} + +func TestStatusV2RejectsEmptyRequest(t *testing.T) { + stream, _ := newStatusTestStream(t, protocol.ID(communication.StatusProtocolV2)) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) +} + +func TestStatusV2RejectsInvalidRequestSize(t *testing.T) { + for _, size := range []int{91, 93} { + t.Run(fmt.Sprintf("size_%d", size), func(t *testing.T) { + stream, _ := newStatusTestStream(t, protocol.ID(communication.StatusProtocolV2)) + require.NoError(t, ssz_snappy.EncodeAndWrite(stream, rawSSZ(make([]byte, size)))) + require.NoError(t, stream.CloseWrite()) + + requireResponseCode(t, stream, byte(InvalidRequestPrefix)) + }) + } } diff --git a/cl/sentinel/handlers/rate_limiter_integration_test.go b/cl/sentinel/handlers/rate_limiter_integration_test.go index a4774d46f7f..8b4b3a86474 100644 --- a/cl/sentinel/handlers/rate_limiter_integration_test.go +++ b/cl/sentinel/handlers/rate_limiter_integration_test.go @@ -75,8 +75,9 @@ func TestPingRateLimit(t *testing.T) { }() require.NoError(t, stream.SetDeadline(time.Now().Add(5*time.Second))) - _, err = stream.Write(nil) + err = ssz_snappy.EncodeAndWrite(stream, &cltypes.Ping{Id: 1}) require.NoError(t, err) + require.NoError(t, stream.CloseWrite()) firstByte := make([]byte, 1) _, err = stream.Read(firstByte) diff --git a/cl/sentinel/sentinel_requests_test.go b/cl/sentinel/sentinel_requests_test.go index 5500584d5a4..b95ded31926 100644 --- a/cl/sentinel/sentinel_requests_test.go +++ b/cl/sentinel/sentinel_requests_test.go @@ -353,6 +353,7 @@ func testSentinelStatusRequest(t *testing.T) { defer stream.Close() noErr(ssz_snappy.EncodeAndWrite(stream, req)) + noErr(stream.CloseWrite()) code := make([]byte, 1) _, err = stream.Read(code) From 54e9d5f68881e243bb3ffa67a836ed0a0a8441a6 Mon Sep 17 00:00:00 2001 From: kewei Date: Wed, 26 Aug 2026 22:58:27 +0800 Subject: [PATCH 2/3] cl/sentinel: tighten exact request bounds --- .../communication/ssz_snappy/encoding.go | 15 ++-- .../communication/ssz_snappy/encoding_test.go | 30 ++++++- cl/sentinel/handlers/handlers.go | 3 + cl/sentinel/handlers/handlers_test.go | 78 +++++++++++++++++++ 4 files changed, 117 insertions(+), 9 deletions(-) create mode 100644 cl/sentinel/handlers/handlers_test.go diff --git a/cl/sentinel/communication/ssz_snappy/encoding.go b/cl/sentinel/communication/ssz_snappy/encoding.go index a057170a2cb..546f873c7a3 100644 --- a/cl/sentinel/communication/ssz_snappy/encoding.go +++ b/cl/sentinel/communication/ssz_snappy/encoding.go @@ -106,6 +106,7 @@ func DecodeAndReadNoForkDigest(r io.Reader, val ssz.EncodableSSZ, version clpara 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) } @@ -128,7 +129,7 @@ func decodeAndReadNoForkDigest(r io.Reader, val ssz.EncodableSSZ, version clpara var maxCompressedSize uint64 if expectedSize != nil { maxCompressedSize = 32 + encodedLn + encodedLn/6 - compressedReader = &compressedPayloadReader{r: r, remaining: maxCompressedSize} + compressedReader = &compressedPayloadReader{r: r, remaining: maxCompressedSize + 1} compressedInput = compressedReader } sr := snappypool.Reader(compressedInput) @@ -145,11 +146,13 @@ func decodeAndReadNoForkDigest(r io.Reader, val ssz.EncodableSSZ, version clpara compressedBytes := compressedReader.read var extra [1]byte _, err := io.ReadFull(sr, extra[:]) - if errors.Is(err, errCompressedPayloadLimit) { - if compressedReader.read != compressedBytes { - return errors.New("payload contains trailing bytes") - } - } else if !errors.Is(err, io.EOF) || compressedReader.read != compressedBytes { + if compressedReader.read > maxCompressedSize { + return errCompressedPayloadLimit + } + if err != nil && !errors.Is(err, io.EOF) { + return fmt.Errorf("unable to verify payload end: %w", err) + } + if err == nil || compressedReader.read != compressedBytes { return errors.New("payload contains trailing bytes") } } diff --git a/cl/sentinel/communication/ssz_snappy/encoding_test.go b/cl/sentinel/communication/ssz_snappy/encoding_test.go index 9a462c7f515..9015d65948d 100644 --- a/cl/sentinel/communication/ssz_snappy/encoding_test.go +++ b/cl/sentinel/communication/ssz_snappy/encoding_test.go @@ -18,6 +18,7 @@ package ssz_snappy import ( "bytes" + "errors" "testing" "github.com/stretchr/testify/require" @@ -33,6 +34,19 @@ type countingReader struct { 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 @@ -68,7 +82,7 @@ func TestDecodeAndReadNoForkDigestExactBoundsCompressedInput(t *testing.T) { 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) + require.LessOrEqual(t, reader.bytes, 1+32+8+8/6+1) } func TestDecodeAndReadNoForkDigestExactCompressedLimitIsInclusive(t *testing.T) { @@ -94,6 +108,16 @@ func TestDecodeAndReadNoForkDigestExactCompressedLimitIsInclusive(t *testing.T) overMax := append(append([]byte{}, exactMax...), 0) reader := &countingReader{r: bytes.NewReader(overMax)} - require.NoError(t, DecodeAndReadNoForkDigestExact(reader, &cltypes.Ping{}, clparams.Phase0Version, payloadSize)) - require.LessOrEqual(t, reader.bytes, 1+maxCompressedSize) + require.ErrorIs(t, DecodeAndReadNoForkDigestExact(reader, &cltypes.Ping{}, clparams.Phase0Version, payloadSize), errCompressedPayloadLimit) + require.LessOrEqual(t, reader.bytes, 1+maxCompressedSize+1) +} + +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) } diff --git a/cl/sentinel/handlers/handlers.go b/cl/sentinel/handlers/handlers.go index efb0ab7e448..ec8f05620f2 100644 --- a/cl/sentinel/handlers/handlers.go +++ b/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 { diff --git a/cl/sentinel/handlers/handlers_test.go b/cl/sentinel/handlers/handlers_test.go new file mode 100644 index 00000000000..0f6d9090a90 --- /dev/null +++ b/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 . + +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()) +} From fb46e1c470626199ad156c4bd5adae052e25ed7b Mon Sep 17 00:00:00 2001 From: kewei Date: Thu, 27 Aug 2026 14:47:28 +0800 Subject: [PATCH 3/3] cl/sentinel: enforce snappy read ceiling --- .../communication/ssz_snappy/encoding.go | 8 +++--- .../communication/ssz_snappy/encoding_test.go | 25 ++++++++++++++----- 2 files changed, 23 insertions(+), 10 deletions(-) diff --git a/cl/sentinel/communication/ssz_snappy/encoding.go b/cl/sentinel/communication/ssz_snappy/encoding.go index 546f873c7a3..c86b302f523 100644 --- a/cl/sentinel/communication/ssz_snappy/encoding.go +++ b/cl/sentinel/communication/ssz_snappy/encoding.go @@ -129,7 +129,7 @@ func decodeAndReadNoForkDigest(r io.Reader, val ssz.EncodableSSZ, version clpara var maxCompressedSize uint64 if expectedSize != nil { maxCompressedSize = 32 + encodedLn + encodedLn/6 - compressedReader = &compressedPayloadReader{r: r, remaining: maxCompressedSize + 1} + compressedReader = &compressedPayloadReader{r: r, remaining: maxCompressedSize} compressedInput = compressedReader } sr := snappypool.Reader(compressedInput) @@ -140,16 +140,16 @@ func decodeAndReadNoForkDigest(r io.Reader, val ssz.EncodableSSZ, version clpara return fmt.Errorf("unable to readPacket: %w", err) } if expectedSize != nil { - if compressedReader.read > maxCompressedSize { + if compressedReader.read >= maxCompressedSize { return errCompressedPayloadLimit } compressedBytes := compressedReader.read var extra [1]byte _, err := io.ReadFull(sr, extra[:]) - if compressedReader.read > maxCompressedSize { + if compressedReader.read >= maxCompressedSize { return errCompressedPayloadLimit } - if err != nil && !errors.Is(err, io.EOF) { + 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 { diff --git a/cl/sentinel/communication/ssz_snappy/encoding_test.go b/cl/sentinel/communication/ssz_snappy/encoding_test.go index 9015d65948d..d798a31b2c7 100644 --- a/cl/sentinel/communication/ssz_snappy/encoding_test.go +++ b/cl/sentinel/communication/ssz_snappy/encoding_test.go @@ -19,6 +19,8 @@ package ssz_snappy import ( "bytes" "errors" + "fmt" + "io" "testing" "github.com/stretchr/testify/require" @@ -82,10 +84,10 @@ func TestDecodeAndReadNoForkDigestExactBoundsCompressedInput(t *testing.T) { 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+1) + require.LessOrEqual(t, reader.bytes, 1+32+8+8/6) } -func TestDecodeAndReadNoForkDigestExactCompressedLimitIsInclusive(t *testing.T) { +func TestDecodeAndReadNoForkDigestExactFailsClosedAtCompressedLimit(t *testing.T) { const payloadSize = 8 const maxCompressedSize = 32 + payloadSize + payloadSize/6 @@ -93,6 +95,9 @@ func TestDecodeAndReadNoForkDigestExactCompressedLimitIsInclusive(t *testing.T) 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 @@ -101,15 +106,13 @@ func TestDecodeAndReadNoForkDigestExactCompressedLimitIsInclusive(t *testing.T) require.Len(t, exactMax, 1+maxCompressedSize) exactMaxReader := &countingReader{r: bytes.NewReader(exactMax)} - var ping cltypes.Ping - require.NoError(t, DecodeAndReadNoForkDigestExact(exactMaxReader, &ping, clparams.Phase0Version, payloadSize)) - require.Equal(t, uint64(1), ping.Id) + 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+1) + require.LessOrEqual(t, reader.bytes, 1+maxCompressedSize) } func TestDecodeAndReadNoForkDigestExactPreservesTerminalReadError(t *testing.T) { @@ -121,3 +124,13 @@ func TestDecodeAndReadNoForkDigestExactPreservesTerminalReadError(t *testing.T) 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) +}