diff --git a/app/app.go b/app/app.go index 12d1f208e..6ddd63188 100644 --- a/app/app.go +++ b/app/app.go @@ -374,10 +374,15 @@ func wireP2P(ctx context.Context, life *lifecycle.Manager, conf Config, wrappedRegisterer := prometheus.WrapRegistererWith(labels, promRegistry) swarmOpts := []swarm.Option{p2p.WithSwarmMetrics(wrappedRegisterer)} - opts := []libp2p.Option{ - bwOpt, - libp2p.ResourceManager(new(network.NullResourceManager)), + rmgr := network.ResourceManager(new(network.NullResourceManager)) + if featureset.Enabled(featureset.Libp2pResourceManager) { + rmgr, err = p2p.NewResourceManager(peerIDs) + if err != nil { + return nil, err + } } + + opts := []libp2p.Option{bwOpt, libp2p.ResourceManager(rmgr)} opts = append(opts, conf.TestConfig.LibP2POpts...) var p2pNode host.Host @@ -388,6 +393,7 @@ func wireP2P(ctx context.Context, life *lifecycle.Manager, conf Config, } if err != nil { + _ = rmgr.Close() // The host owns the resource manager only once created. return nil, err } diff --git a/app/featureset/featureset.go b/app/featureset/featureset.go index 07121b95a..c556f2fc9 100644 --- a/app/featureset/featureset.go +++ b/app/featureset/featureset.go @@ -85,6 +85,11 @@ const ( // DisableDutiesCache is a safety measure to disable duties cache. DisableDutiesCache = "disable_duties_cache" + + // Libp2pResourceManager enables the libp2p resource manager, limiting connections, + // streams and memory to protect against resource exhaustion (DoS). When disabled, + // the null resource manager is used which never rejects anything. + Libp2pResourceManager = "libp2p_resource_manager" ) var ( @@ -106,6 +111,7 @@ var ( FetchAttOnBlock: statusAlpha, FetchAttOnBlockWithDelay: statusAlpha, DisableDutiesCache: statusAlpha, + Libp2pResourceManager: statusAlpha, // Add all features and their status here. } diff --git a/cmd/relay/p2p.go b/cmd/relay/p2p.go index f04bf3f1a..40417e1f7 100644 --- a/cmd/relay/p2p.go +++ b/cmd/relay/p2p.go @@ -42,9 +42,15 @@ func startP2P(ctx context.Context, config Config, key *k1.PrivateKey, reporter m } } + rm, err := p2p.NewRelayResourceManager(config.MaxConns, config.MaxResPerPeer) + if err != nil { + return nil, nil, errors.Wrap(err, "new relay resource manager") + } + p2pNode, err := p2p.NewNode(ctx, config.P2PConfig, key, p2p.NewOpenGater(), config.FilterPrivAddrs, p2p.NodeTypeQUIC, nil, - libp2p.ResourceManager(new(network.NullResourceManager)), libp2p.BandwidthReporter(reporter)) + libp2p.ResourceManager(rm), libp2p.BandwidthReporter(reporter)) if err != nil { + _ = rm.Close() // The host owns the resource manager only once created. return nil, nil, errors.Wrap(err, "new relay node") } diff --git a/p2p/resourcemanager.go b/p2p/resourcemanager.go new file mode 100644 index 000000000..70c53f5af --- /dev/null +++ b/p2p/resourcemanager.go @@ -0,0 +1,245 @@ +// Copyright © 2022-2026 Obol Labs Inc. Licensed under the terms of a Business Source License 1.1 + +package p2p + +import ( + "net/netip" + "time" + + "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" + rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager" + circuitproto "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/proto" + relayv2 "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/relay" + "github.com/libp2p/go-libp2p/x/rate" + + "github.com/obolnetwork/charon/app/errors" +) + +// Per-peer limits are fixed instead of autoscaled for deterministic behaviour across machines. +// Cluster peers are authenticated but not trusted: elevated limits accommodate duty load +// spikes while still bounding the damage a Byzantine peer can do. Other peers (relays) +// only require reservation and identify streams, so they get the default limits. +const ( + defaultPeerStreamsInbound = 512 + defaultPeerStreamsOutbound = 1024 + defaultPeerConns = 16 + defaultPeerMemory = 64 << 20 + + clusterPeerStreamsInbound = 4096 + clusterPeerStreamsOutbound = 4096 + clusterPeerConns = 32 + clusterPeerMemory = 512 << 20 + + // clusterConnsPerIP limits connections per remote IP while still allowing + // multiple cluster peers to share one IP (NAT or local test clusters). + clusterConnsPerIP = 64 + + // System and transient stream and connection limits are fixed for the same reason + // as the per-peer limits. go-libp2p's autoscaled defaults are derived from an + // eighth of host memory and fall below the per-peer limits on typical validator + // hardware (3072 inbound streams and 64 transient inbound connections on a 16GB + // host, less below that), which makes the per-peer allowances unreachable and the + // effective limits depend on host memory. Memory and FD limits stay host derived, + // since those track resources the host actually has. + // + // transientConns bounds connections that have not yet negotiated a protocol. It + // exceeds clusterConnsPerIP so a cluster sharing one NAT IP can reconnect in a + // single burst, matching the connection rate limiter burst. + transientConns = 2 * clusterConnsPerIP + + // The system scope stays above the per-peer limits so no single peer can exhaust + // it: 16384 inbound streams is four cluster peers' worth. Connections are sized + // against the transient scope instead, because a pending connection holds a + // transient and a system reservation until it identifies, so the system scope must + // admit a full transient burst alongside established peers. + systemStreamsInbound = 4 * clusterPeerStreamsInbound + systemStreamsOutbound = 4 * clusterPeerStreamsOutbound + systemConns = 2 * transientConns + + // relayServiceMemory bounds the circuit relay scopes, covering per-circuit + // buffers at full circuit load. + relayServiceMemory = 512 << 20 +) + +// NewResourceManager returns a libp2p resource manager for charon nodes. +// Stream and connection limits are fixed, with elevated limits for the +// authenticated cluster peers, while memory and FD limits scale with the host. +func NewResourceManager(clusterPeers []peer.ID) (network.ResourceManager, error) { + rm, err := rcmgr.NewResourceManager( + rcmgr.NewFixedLimiter(clusterLimitConfig(clusterPeers)), + rcmgr.WithLimitPerSubnet( + []rcmgr.ConnLimitPerSubnet{{ConnCount: clusterConnsPerIP, PrefixLength: 32}}, + []rcmgr.ConnLimitPerSubnet{{ConnCount: clusterConnsPerIP, PrefixLength: 56}}, + ), + rcmgr.WithConnRateLimiters(newConnRateLimiter(clusterConnsPerIP)), + ) + if err != nil { + return nil, errors.Wrap(err, "new resource manager") + } + + return rm, nil +} + +// NewRelayResourceManager returns a libp2p resource manager for relay nodes. +// System connection limits are derived from the relay connection config while +// other limits scale with available memory. +func NewRelayResourceManager(maxConns, maxConnsPerIP int) (network.ResourceManager, error) { + rm, err := rcmgr.NewResourceManager( + rcmgr.NewFixedLimiter(relayLimitConfig(maxConns, maxConnsPerIP)), + rcmgr.WithLimitPerSubnet( + []rcmgr.ConnLimitPerSubnet{{ConnCount: maxConnsPerIP, PrefixLength: 32}}, + []rcmgr.ConnLimitPerSubnet{{ConnCount: maxConnsPerIP, PrefixLength: 56}}, + ), + rcmgr.WithConnRateLimiters(newConnRateLimiter(maxConnsPerIP)), + ) + if err != nil { + return nil, errors.Wrap(err, "new relay resource manager") + } + + return rm, nil +} + +// clusterLimitConfig returns charon node resource limits: fixed system, transient and +// per-peer stream and connection limits, with elevated allowances for cluster peers. +// Memory and FD limits are left at their host derived defaults. +func clusterLimitConfig(clusterPeers []peer.ID) rcmgr.ConcreteLimitConfig { + limits := rcmgr.DefaultLimits + libp2p.SetDefaultServiceLimits(&limits) + + clusterLimits := make(map[peer.ID]rcmgr.ResourceLimits, len(clusterPeers)) + for _, pID := range clusterPeers { + clusterLimits[pID] = peerLimits(clusterPeerStreamsInbound, clusterPeerStreamsOutbound, clusterPeerConns, clusterPeerMemory) + } + + cfg := rcmgr.PartialLimitConfig{ + System: rcmgr.ResourceLimits{ + StreamsInbound: systemStreamsInbound, + StreamsOutbound: systemStreamsOutbound, + Streams: systemStreamsInbound + systemStreamsOutbound, + ConnsInbound: systemConns, + ConnsOutbound: systemConns, + Conns: 2 * systemConns, + }, + Transient: rcmgr.ResourceLimits{ + ConnsInbound: transientConns, + ConnsOutbound: transientConns, + Conns: 2 * transientConns, + }, + PeerDefault: peerLimits(defaultPeerStreamsInbound, defaultPeerStreamsOutbound, defaultPeerConns, defaultPeerMemory), + Peer: clusterLimits, + // All charon protocols share the default protocol limits, so raise them to + // match the system and cluster peer limits in both directions; the peer and + // system scopes remain the effective bounds. + ProtocolDefault: streamLimits(systemStreamsInbound, systemStreamsOutbound), + ProtocolPeerDefault: streamLimits(clusterPeerStreamsInbound, clusterPeerStreamsOutbound), + } + + return cfg.Build(limits.AutoScale()) +} + +// relayLimitConfig returns relay resource limits derived from the relay connection config. +func relayLimitConfig(maxConns, maxConnsPerIP int) rcmgr.ConcreteLimitConfig { + limits := rcmgr.DefaultLimits + libp2p.SetDefaultServiceLimits(&limits) + + // Each relayed circuit costs an inbound and an outbound stream on the relay, + // plus reservation and identify streams, so allow multiple streams per connection. + maxStreams := 4 * maxConns + + systemLimits := rcmgr.ResourceLimits{ + Conns: rcmgr.LimitVal(maxConns), + ConnsInbound: rcmgr.LimitVal(maxConns), + FD: rcmgr.LimitVal(maxConns), + Streams: rcmgr.LimitVal(maxStreams), + StreamsInbound: rcmgr.LimitVal(maxStreams), + StreamsOutbound: rcmgr.LimitVal(maxStreams), + } + + // The circuit relay protocol and service scopes must also track the connection + // config: their go-libp2p defaults scale with host memory and would otherwise + // cap circuits well below the configured connection limits. + relayStreams := rcmgr.ResourceLimits{ + Streams: rcmgr.LimitVal(maxStreams), + StreamsInbound: rcmgr.LimitVal(maxStreams), + StreamsOutbound: rcmgr.LimitVal(maxStreams), + Memory: relayServiceMemory, + } + // The relay allows maxConnsPerIP circuit reservations per peer, each costing an + // inbound and an outbound stream. + relayPeerStreams := streamLimits(2*maxConnsPerIP, 2*maxConnsPerIP) + + cfg := rcmgr.PartialLimitConfig{ + System: systemLimits, + // Match transient limits to system limits so reconnect storms + // (e.g. after a relay restart) are not throttled below capacity. + Transient: systemLimits, + Protocol: map[protocol.ID]rcmgr.ResourceLimits{ + circuitproto.ProtoIDv2Hop: relayStreams, + circuitproto.ProtoIDv2Stop: relayStreams, + }, + ProtocolPeer: map[protocol.ID]rcmgr.ResourceLimits{ + circuitproto.ProtoIDv2Hop: relayPeerStreams, + circuitproto.ProtoIDv2Stop: relayPeerStreams, + }, + Service: map[string]rcmgr.ResourceLimits{ + relayv2.ServiceName: relayStreams, + }, + ServicePeer: map[string]rcmgr.ResourceLimits{ + relayv2.ServiceName: relayPeerStreams, + }, + } + + return cfg.Build(limits.AutoScale()) +} + +// newConnRateLimiter returns a connection rate limiter allowing bursts up to twice +// the per-IP connection limit. It replaces go-libp2p's default which caps bursts at +// 16 connections per IP regardless of the configured connection limits, throttling +// legitimate reconnect storms from peers sharing a NAT IP. +func newConnRateLimiter(perIPConns int) *rate.Limiter { + rps := float64(perIPConns) + + return &rate.Limiter{ + // Never rate limit loopback. + NetworkPrefixLimits: []rate.PrefixLimit{ + {Prefix: netip.MustParsePrefix("127.0.0.0/8"), Limit: rate.Limit{}}, + {Prefix: netip.MustParsePrefix("::1/128"), Limit: rate.Limit{}}, + }, + SubnetRateLimiter: rate.SubnetLimiter{ + IPv4SubnetLimits: []rate.SubnetLimit{ + {PrefixLength: 32, Limit: rate.Limit{RPS: rps, Burst: 2 * perIPConns}}, + }, + IPv6SubnetLimits: []rate.SubnetLimit{ + {PrefixLength: 56, Limit: rate.Limit{RPS: rps, Burst: 2 * perIPConns}}, + {PrefixLength: 48, Limit: rate.Limit{RPS: 4 * rps, Burst: 8 * perIPConns}}, + }, + GracePeriod: time.Minute, + }, + } +} + +// streamLimits returns fixed stream-only resource limits, leaving other resources at their defaults. +func streamLimits(inbound, outbound int) rcmgr.ResourceLimits { + return rcmgr.ResourceLimits{ + StreamsInbound: rcmgr.LimitVal(inbound), + StreamsOutbound: rcmgr.LimitVal(outbound), + Streams: rcmgr.LimitVal(inbound + outbound), + } +} + +// peerLimits returns fixed per-peer resource limits. +func peerLimits(streamsInbound, streamsOutbound, conns int, memory int64) rcmgr.ResourceLimits { + return rcmgr.ResourceLimits{ + StreamsInbound: rcmgr.LimitVal(streamsInbound), + StreamsOutbound: rcmgr.LimitVal(streamsOutbound), + Streams: rcmgr.LimitVal(streamsInbound + streamsOutbound), + ConnsInbound: rcmgr.LimitVal(conns), + ConnsOutbound: rcmgr.LimitVal(conns), + Conns: rcmgr.LimitVal(conns), + FD: rcmgr.LimitVal(conns), + Memory: rcmgr.LimitVal64(memory), + } +} diff --git a/p2p/resourcemanager_internal_test.go b/p2p/resourcemanager_internal_test.go new file mode 100644 index 000000000..27ddb9f78 --- /dev/null +++ b/p2p/resourcemanager_internal_test.go @@ -0,0 +1,211 @@ +// Copyright © 2022-2026 Obol Labs Inc. Licensed under the terms of a Business Source License 1.1 + +package p2p + +import ( + "fmt" + "testing" + + k1 "github.com/decred/dcrd/dcrec/secp256k1/v4" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" + rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager" + circuitproto "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/proto" + relayv2 "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/relay" + ma "github.com/multiformats/go-multiaddr" + "github.com/stretchr/testify/require" +) + +// openInboundStreams opens inbound streams for the given peer until the resource +// manager rejects one, returning the number of streams opened. +func openInboundStreams(t *testing.T, rm network.ResourceManager, pID peer.ID, maxAttempts int) int { + t.Helper() + + var scopes []network.StreamManagementScope + + t.Cleanup(func() { + for _, scope := range scopes { + scope.Done() + } + }) + + for i := range maxAttempts { + scope, err := rm.OpenStream(pID, network.DirInbound) + if err != nil { + return i + } + + scopes = append(scopes, scope) + + // Attach a protocol like negotiated streams do, moving the stream + // from the transient scope to the protocol and peer scopes. + require.NoError(t, scope.SetProtocol("/charon/test")) + } + + return maxAttempts +} + +func randomPeerID(t *testing.T) peer.ID { + t.Helper() + + key, err := k1.GeneratePrivateKey() + require.NoError(t, err) + + pID, err := PeerIDFromKey(key.PubKey()) + require.NoError(t, err) + + return pID +} + +func TestNewResourceManagerPeerStreamLimits(t *testing.T) { + clusterPeer := randomPeerID(t) + unknownPeer := randomPeerID(t) + + rm, err := NewResourceManager([]peer.ID{clusterPeer}) + require.NoError(t, err) + + defer rm.Close() + + // Unknown peers (e.g. relays) get the fixed default limit. + unknownStreams := openInboundStreams(t, rm, unknownPeer, defaultPeerStreamsInbound*2) + require.Equal(t, defaultPeerStreamsInbound, unknownStreams) + + // Cluster peers get elevated but still bounded limits. + clusterStreams := openInboundStreams(t, rm, clusterPeer, clusterPeerStreamsInbound*2) + require.Equal(t, clusterPeerStreamsInbound, clusterStreams) +} + +func TestNewRelayResourceManagerConnLimits(t *testing.T) { + const ( + maxConns = 64 + maxConnsPerIP = 4 + ) + + rm, err := NewRelayResourceManager(maxConns, maxConnsPerIP) + require.NoError(t, err) + + defer rm.Close() + + openConn := func(rm network.ResourceManager, addr string) (network.ConnManagementScope, error) { + return rm.OpenConnection(network.DirInbound, false, ma.StringCast(addr)) + } + + // Connections from distinct IPs are allowed up to maxConns system-wide. + var scopes []network.ConnManagementScope + + defer func() { + for _, scope := range scopes { + scope.Done() + } + }() + + for i := range maxConns { + scope, err := openConn(rm, fmt.Sprintf("/ip4/10.0.%d.%d/tcp/1234", i/256+1, i%256)) + require.NoError(t, err) + + scopes = append(scopes, scope) + } + + _, err = openConn(rm, "/ip4/10.99.99.99/tcp/1234") + require.Error(t, err) + + // Connections from a single IP are limited to maxConnsPerIP. + rm2, err := NewRelayResourceManager(maxConns, maxConnsPerIP) + require.NoError(t, err) + + defer rm2.Close() + + for range maxConnsPerIP { + scope, err := openConn(rm2, "/ip4/10.1.1.1/tcp/1234") + require.NoError(t, err) + + defer scope.Done() + } + + _, err = openConn(rm2, "/ip4/10.1.1.1/tcp/1234") + require.Error(t, err) +} + +// TestConnRateLimiterAllowsPerIPBurst ensures the connection rate limiter permits +// a rapid burst of connections up to the per-IP connection limit, e.g. when +// multiple peers behind one NAT IP reconnect after a relay restart. +func TestConnRateLimiterAllowsPerIPBurst(t *testing.T) { + const ( + maxConns = 128 + maxConnsPerIP = 32 // Above go-libp2p's default rate limiter burst of 16. + ) + + openConns := func(t *testing.T, rm network.ResourceManager, addr string, n int) { + t.Helper() + + for range n { + scope, err := rm.OpenConnection(network.DirInbound, false, ma.StringCast(addr)) + require.NoError(t, err) + + defer scope.Done() + } + } + + rm, err := NewRelayResourceManager(maxConns, maxConnsPerIP) + require.NoError(t, err) + + defer rm.Close() + + openConns(t, rm, "/ip4/10.2.2.2/tcp/1234", maxConnsPerIP) + + // Validator nodes similarly allow bursts up to the per-IP limit. + clusterRM, err := NewResourceManager(nil) + require.NoError(t, err) + + defer clusterRM.Close() + + openConns(t, clusterRM, "/ip4/10.2.2.2/tcp/1234", clusterConnsPerIP) +} + +// TestClusterLimitConfig ensures system and transient stream and connection limits +// are fixed instead of scaling with host memory, and that they leave room for the +// per-peer limits, which would otherwise be unreachable on smaller hosts. +func TestClusterLimitConfig(t *testing.T) { + cfg := clusterLimitConfig([]peer.ID{randomPeerID(t)}).ToPartialLimitConfig() + + require.Equal(t, rcmgr.LimitVal(systemStreamsInbound), cfg.System.StreamsInbound) + require.Equal(t, rcmgr.LimitVal(systemConns), cfg.System.ConnsInbound) + require.Equal(t, rcmgr.LimitVal(transientConns), cfg.Transient.ConnsInbound) + + // The system scope must exceed a single peer's allowance so no one peer can + // exhaust it, and the transient scope must admit a full per-IP connection burst. + require.Greater(t, cfg.System.StreamsInbound, rcmgr.LimitVal(clusterPeerStreamsInbound)) + require.Greater(t, cfg.System.ConnsInbound, rcmgr.LimitVal(clusterPeerConns)) + require.Greater(t, cfg.Transient.ConnsInbound, rcmgr.LimitVal(clusterConnsPerIP)) + + // Memory stays host derived, bounding aggregate load on small hosts. + require.Equal(t, rcmgr.LimitVal64(defaultPeerMemory), cfg.PeerDefault.Memory) + require.NotZero(t, cfg.System.Memory) +} + +// TestRelayLimitConfig ensures relay capacity limits derive from the relay +// connection config instead of scaling with host memory. +func TestRelayLimitConfig(t *testing.T) { + const ( + maxConns = 16384 + maxConnsPerIP = 512 + ) + + cfg := relayLimitConfig(maxConns, maxConnsPerIP).ToPartialLimitConfig() + + maxStreams := rcmgr.LimitVal(4 * maxConns) + perPeerStreams := rcmgr.LimitVal(2 * maxConnsPerIP) + + require.Equal(t, rcmgr.LimitVal(maxConns), cfg.System.ConnsInbound) + require.Equal(t, maxStreams, cfg.System.StreamsInbound) + + // Circuit relay protocol and service scopes must accommodate full circuit load. + for _, proto := range []protocol.ID{circuitproto.ProtoIDv2Hop, circuitproto.ProtoIDv2Stop} { + require.Equal(t, maxStreams, cfg.Protocol[proto].StreamsInbound, proto) + require.Equal(t, perPeerStreams, cfg.ProtocolPeer[proto].StreamsInbound, proto) + } + + require.Equal(t, maxStreams, cfg.Service[relayv2.ServiceName].StreamsInbound) + require.Equal(t, perPeerStreams, cfg.ServicePeer[relayv2.ServiceName].StreamsInbound) +}