diff --git a/cl/sentinel/communication/ssz_snappy/encoding.go b/cl/sentinel/communication/ssz_snappy/encoding.go
index e9a2f409ef7..c86b302f523 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,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 {
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..d798a31b2c7
--- /dev/null
+++ b/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 .
+
+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)
+}
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())
+}
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)