diff --git a/pkg/execution/mimicry/client.go b/pkg/execution/mimicry/client.go index 3ecf96d..cb3c141 100644 --- a/pkg/execution/mimicry/client.go +++ b/pkg/execution/mimicry/client.go @@ -28,6 +28,12 @@ const ( logFieldETHCap = "ethCapVersion" ) +// writeTimeout bounds every write to the rlpx connection. Without it, a +// peer that stops draining its socket can stall Write forever while +// writeRLPx holds writeMu, wedging the session and every other sender. A +// var (not const) so tests can shrink it instead of waiting out 30s. +var writeTimeout = 30 * time.Second + type Client struct { log logrus.FieldLogger @@ -200,6 +206,10 @@ func (c *Client) writeRLPx(code uint64, data []byte) error { return fmt.Errorf("rlpx connection is not initialized") } + if err := c.conn.SetWriteDeadline(time.Now().Add(writeTimeout)); err != nil { + return fmt.Errorf("error setting rlpx write deadline: %w", err) + } + if _, err := c.rlpxConn.Write(code, data); err != nil { return err } diff --git a/pkg/execution/mimicry/client_write_deadline_test.go b/pkg/execution/mimicry/client_write_deadline_test.go new file mode 100644 index 0000000..d4ce6dc --- /dev/null +++ b/pkg/execution/mimicry/client_write_deadline_test.go @@ -0,0 +1,197 @@ +package mimicry + +import ( + "context" + "crypto/ecdsa" + "crypto/rand" + "encoding/hex" + "fmt" + "math/big" + "net" + "testing" + "time" + + gethcommon "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/p2p" + gethrlpx "github.com/ethereum/go-ethereum/p2p/rlpx" + "github.com/ethereum/go-ethereum/rlp" + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/require" +) + +// testHelloMessage mirrors the Hello struct in message_hello.go, kept +// separate so this test doesn't depend on package-internal wiring beyond +// the public devp2p wire format. +type testHelloMessage struct { + Version uint64 + Name string + Caps []p2p.Cap + ListenPort uint64 + ID []byte + Rest []rlp.RawValue `rlp:"tail"` +} + +// testRLPxPeer is a minimal, protocol-faithful devp2p peer used to drive +// Client against real handshake and Hello exchanges without needing a real +// execution node. It owns the accepted connection for its whole lifetime +// so nothing relies on a goroutine-local variable staying reachable. +type testRLPxPeer struct { + listener net.Listener + priv *ecdsa.PrivateKey + conn net.Conn +} + +func newTestRLPxPeer(t *testing.T) *testRLPxPeer { + t.Helper() + + priv, err := crypto.GenerateKey() + require.NoError(t, err) + + ln, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + + p := &testRLPxPeer{listener: ln, priv: priv} + + t.Cleanup(func() { + _ = ln.Close() + if p.conn != nil { + _ = p.conn.Close() + } + }) + + return p +} + +func (p *testRLPxPeer) enode() string { + pub := crypto.FromECDSAPub(&p.priv.PublicKey)[1:] + + return fmt.Sprintf("enode://%s@%s", hex.EncodeToString(pub), p.listener.Addr().String()) +} + +// acceptAndHandshake accepts one connection, completes the RLPx handshake +// as recipient, and exchanges Hello. The connection is stored on p for the +// lifetime of the peer (see the type doc comment). +func (p *testRLPxPeer) acceptAndHandshake(t *testing.T) *gethrlpx.Conn { + t.Helper() + + conn, err := p.listener.Accept() + require.NoError(t, err) + + p.conn = conn + + rc := gethrlpx.NewConn(conn, nil) + _, err = rc.Handshake(p.priv) + require.NoError(t, err) + + code, data, _, err := rc.Read() + require.NoError(t, err) + require.Equal(t, uint64(HelloCode), code) + + var clientHello testHelloMessage + require.NoError(t, rlp.DecodeBytes(data, &clientHello)) + + pub := crypto.FromECDSAPub(&p.priv.PublicKey)[1:] + ourHello := &testHelloMessage{ + Version: P2PProtocolVersion, + Name: "test-peer", + Caps: SupportedEthCaps(), + ListenPort: 0, + ID: pub, + } + + encoded, err := rlp.EncodeToBytes(ourHello) + require.NoError(t, err) + _, err = rc.Write(HelloCode, encoded) + require.NoError(t, err) + + // The client enables snappy on its own connection as soon as it + // processes our Hello (see handleHello); the wire format requires both + // sides to agree, so we must match it here. + rc.SetSnappy(true) + + return rc +} + +// largeTransactions builds a Transactions frame just under the RLPx frame +// size cap, filled with random (so snappy can't shrink it back down) data. +func largeTransactions(t *testing.T, size int) *Transactions { + t.Helper() + + data := make([]byte, size) + _, err := rand.Read(data) + require.NoError(t, err) + + to := gethcommon.HexToAddress("0x000000000000000000000000000000000000ff") + tx := types.NewTx(&types.LegacyTx{ + Nonce: 0, + GasPrice: big.NewInt(1), + Gas: 21000, + To: &to, + Value: big.NewInt(0), + Data: data, + }) + + var list rlp.RawList[*types.Transaction] + require.NoError(t, list.Append(tx)) + + return &Transactions{RawList: list} +} + +// TestWriteRLPx_UnresponsivePeerDoesNotBlockForever guards against +// EXECUTION-03: a peer that stops draining its socket after the handshake +// must not be able to park writeRLPx (and therefore every exported sender, +// since they all serialize on writeMu) forever. Before the fix, no write +// deadline was ever set on the underlying connection. +func TestWriteRLPx_UnresponsivePeerDoesNotBlockForever(t *testing.T) { + original := writeTimeout + writeTimeout = 200 * time.Millisecond + t.Cleanup(func() { writeTimeout = original }) + + peer := newTestRLPxPeer(t) + + handshakeDone := make(chan struct{}) + + go func() { + peer.acceptAndHandshake(t) + close(handshakeDone) + // Go silent from here: never read again. This is the "peer stops + // draining its socket" scenario EXECUTION-03 describes. peer.conn + // stays referenced (and open) via the testRLPxPeer struct itself, + // which the test keeps alive until its Cleanup runs. + }() + + log := logrus.New() + log.SetLevel(logrus.ErrorLevel) + + client, err := New(context.Background(), log, peer.enode(), "test-client") + require.NoError(t, err) + + require.NoError(t, client.Start(context.Background())) + t.Cleanup(func() { _ = client.Stop(context.Background()) }) + + select { + case <-handshakeDone: + case <-time.After(2 * time.Second): + t.Fatal("server never completed the handshake") + } + + time.Sleep(100 * time.Millisecond) // let the client process our Hello (enables snappy) + + txs := largeTransactions(t, 12_000_000) // under the ~16MB RLPx frame cap, large enough to fill kernel buffers + + done := make(chan error, 1) + start := time.Now() + + go func() { done <- client.Transactions(context.Background(), txs) }() + + select { + case err := <-done: + elapsed := time.Since(start) + require.Error(t, err, "Transactions must fail once the write deadline is exceeded") + require.Less(t, elapsed, 2*time.Second, "Transactions took too long to give up") + case <-time.After(2 * time.Second): + t.Fatal("Transactions did not return within 2s against an unresponsive peer; the write deadline is not being enforced") + } +}