fix: F-2026-18136 | [Dual Defense] Unauthenticated Peers Can Exhaust TSS Stream Quota and Stall Outbound Signing

universalClient/tss/coordinator/coordinator.go

@@ -179,6 +179,28 @@ func (c *Coordinator) GetPeerIDFromPartyID(_ context.Context, partyID string) (s
 	return "", fmt.Errorf("partyID %s not found in validators", partyID)
 }
 
+// IsKnownPeer reports whether peerID belongs to a Universal Validator that can
+// participate in some TSS protocol (Active, Pending Join, or Pending Leave).
+// Fails closed when the cache is empty or stale.
+func (c *Coordinator) IsKnownPeer(peerID string) bool {
+	for _, v := range c.validatorsSnapshot() {
+		if v.NetworkInfo == nil || v.NetworkInfo.PeerId != peerID {
+			continue
+		}
+		if v.LifecycleInfo == nil {
+			return false
+		}
+		switch v.LifecycleInfo.CurrentStatus {
+		case types.UVStatus_UV_STATUS_ACTIVE,
+			types.UVStatus_UV_STATUS_PENDING_JOIN,
+			types.UVStatus_UV_STATUS_PENDING_LEAVE:
+			return true
+		}
+		return false
+	}
+	return false
+}
+
 // GetMultiAddrsFromPeerID gets the multiaddrs for a given peerID.
 func (c *Coordinator) GetMultiAddrsFromPeerID(_ context.Context, peerID string) ([]string, error) {
 	for _, v := range c.validatorsSnapshot() {

universalClient/tss/coordinator/coordinator_test.go

@@ -1277,3 +1277,84 @@ func TestValidatorsSnapshot(t *testing.T) {
 		assert.NotNil(t, coord.validatorsSnapshot())
 	})
 }
+
+func TestIsKnownPeer(t *testing.T) {
+	uv := func(peerID string, status types.UVStatus) *types.UniversalValidator {
+		return &types.UniversalValidator{
+			IdentifyInfo:  &types.IdentityInfo{CoreValidatorAddress: "addr-" + peerID},
+			NetworkInfo:   &types.NetworkInfo{PeerId: peerID, MultiAddrs: []string{"/ip4/127.0.0.1/tcp/9001"}},
+			LifecycleInfo: &types.LifecycleInfo{CurrentStatus: status},
+		}
+	}
+
+	setValidators := func(coord *Coordinator, vs []*types.UniversalValidator) {
+		coord.mu.Lock()
+		coord.allValidators = vs
+		coord.lastValidatorsRefreshAt = time.Now()
+		coord.mu.Unlock()
+	}
+
+	coord, _, _ := setupTestCoordinator(t)
+
+	t.Run("eligible statuses admitted", func(t *testing.T) {
+		setValidators(coord, []*types.UniversalValidator{
+			uv("active", types.UVStatus_UV_STATUS_ACTIVE),
+			uv("joining", types.UVStatus_UV_STATUS_PENDING_JOIN),
+			uv("leaving", types.UVStatus_UV_STATUS_PENDING_LEAVE),
+		})
+		assert.True(t, coord.IsKnownPeer("active"))
+		assert.True(t, coord.IsKnownPeer("joining"))
+		assert.True(t, coord.IsKnownPeer("leaving"))
+	})
+
+	t.Run("inactive and unspecified rejected", func(t *testing.T) {
+		setValidators(coord, []*types.UniversalValidator{
+			uv("active", types.UVStatus_UV_STATUS_ACTIVE),
+			uv("inactive", types.UVStatus_UV_STATUS_INACTIVE),
+			uv("unspecified", types.UVStatus_UV_STATUS_UNSPECIFIED),
+		})
+		assert.False(t, coord.IsKnownPeer("inactive"))
+		assert.False(t, coord.IsKnownPeer("unspecified"))
+	})
+
+	t.Run("unknown peer rejected", func(t *testing.T) {
+		setValidators(coord, []*types.UniversalValidator{
+			uv("active", types.UVStatus_UV_STATUS_ACTIVE),
+		})
+		assert.False(t, coord.IsKnownPeer("stranger"))
+	})
+
+	t.Run("nil lifecycle info rejected", func(t *testing.T) {
+		noLifecycle := uv("ghost", types.UVStatus_UV_STATUS_ACTIVE)
+		noLifecycle.LifecycleInfo = nil
+		setValidators(coord, []*types.UniversalValidator{
+			uv("active", types.UVStatus_UV_STATUS_ACTIVE),
+			noLifecycle,
+		})
+		assert.False(t, coord.IsKnownPeer("ghost"))
+	})
+
+	t.Run("bootstrap keygen peers admitted without any active validator", func(t *testing.T) {
+		// Fresh network: everyone is Pending Join. Strict filter must still
+		// admit them so keygen can start; Inactive stays rejected even here.
+		setValidators(coord, []*types.UniversalValidator{
+			uv("joining", types.UVStatus_UV_STATUS_PENDING_JOIN),
+			uv("joining2", types.UVStatus_UV_STATUS_PENDING_JOIN),
+			uv("inactive", types.UVStatus_UV_STATUS_INACTIVE),
+		})
+		assert.True(t, coord.IsKnownPeer("joining"))
+		assert.True(t, coord.IsKnownPeer("joining2"))
+		assert.False(t, coord.IsKnownPeer("inactive"))
+		assert.False(t, coord.IsKnownPeer("stranger"))
+	})
+
+	t.Run("stale cache fails closed", func(t *testing.T) {
+		setValidators(coord, []*types.UniversalValidator{
+			uv("active", types.UVStatus_UV_STATUS_ACTIVE),
+		})
+		coord.mu.Lock()
+		coord.lastValidatorsRefreshAt = time.Now().Add(-time.Hour)
+		coord.mu.Unlock()
+		assert.False(t, coord.IsKnownPeer("active"))
+	})
+}

universalClient/tss/networking/libp2p/config.go

@@ -15,6 +15,11 @@ type Config struct {
 	DialTimeout time.Duration
 	// IOTimeout bounds stream read/write operations.
 	IOTimeout time.Duration
+	// Authorizer reports whether a remote peer ID is allowed to connect and
+	// open TSS streams. When set, inbound connections from unauthorized peers
+	// are rejected at secured-connection admission and any stream that slips
+	// through is reset before reading. Nil disables gating (tests only).
+	Authorizer func(peerID string) bool
 }
 
 // setDefaults sets default values for unset fields.

universalClient/tss/networking/libp2p/gater.go

@@ -0,0 +1,30 @@
+package libp2p
+
+import (
+	"github.com/libp2p/go-libp2p/core/control"
+	"github.com/libp2p/go-libp2p/core/network"
+	"github.com/libp2p/go-libp2p/core/peer"
+	ma "github.com/multiformats/go-multiaddr"
+)
+
+// validatorGater rejects inbound connections whose authenticated peer ID is
+// not accepted by the authorizer. Outbound dials are not gated: this node only
+// dials peers resolved from the validator set.
+type validatorGater struct {
+	authorizer func(peerID string) bool
+}
+
+func (g *validatorGater) InterceptPeerDial(peer.ID) bool               { return true }
+func (g *validatorGater) InterceptAddrDial(peer.ID, ma.Multiaddr) bool { return true }
+func (g *validatorGater) InterceptAccept(network.ConnMultiaddrs) bool  { return true }
+
+func (g *validatorGater) InterceptSecured(dir network.Direction, p peer.ID, _ network.ConnMultiaddrs) bool {
+	if dir == network.DirOutbound {
+		return true
+	}
+	return g.authorizer(p.String())
+}
+
+func (g *validatorGater) InterceptUpgraded(network.Conn) (bool, control.DisconnectReason) {
+	return true, 0
+}

universalClient/tss/networking/libp2p/network.go

@@ -31,6 +31,10 @@ import (
 // observed DKLS Step() + coordinator.Message wrapping for our committee sizes.
 const MaxFrameSize = 1 * 1024 * 1024 // 1 MiB
 
+// maxConcurrentReads bounds in-flight framed reads across all inbound TSS
+// streams so slow peers cannot pin unbounded goroutines on blocking reads.
+const maxConcurrentReads = 64
+
 // Network implements networking.Network using libp2p.
 type Network struct {
 	cfg        Config
@@ -43,6 +47,8 @@ type Network struct {
 	peerMu sync.RWMutex
 	peers  map[string]peer.AddrInfo
 
+	readSem chan struct{}
+
 	logger zerolog.Logger
 }
 
@@ -58,10 +64,15 @@ func New(ctx context.Context, cfg Config, logger zerolog.Logger) (*Network, erro
 		return nil, err
 	}
 
-	host, err := libp2p.New(
+	opts := []libp2p.Option{
 		libp2p.Identity(priv),
 		libp2p.ListenAddrStrings(cfg.ListenAddrs...),
-	)
+	}
+	if cfg.Authorizer != nil {
+		opts = append(opts, libp2p.ConnectionGater(&validatorGater{authorizer: cfg.Authorizer}))
+	}
+
+	host, err := libp2p.New(opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -71,6 +82,7 @@ func New(ctx context.Context, cfg Config, logger zerolog.Logger) (*Network, erro
 		host:       host,
 		protocolID: protocol.ID(cfg.ProtocolID),
 		peers:      make(map[string]peer.AddrInfo),
+		readSem:    make(chan struct{}, maxConcurrentReads),
 		logger:     logger.With().Str("component", "networking_libp2p").Logger(),
 	}
 
@@ -194,6 +206,25 @@ func (n *Network) lookupPeer(peerID string) (peer.AddrInfo, error) {
 }
 
 func (n *Network) handleStream(stream network.Stream) {
+	remotePeer := stream.Conn().RemotePeer().String()
+	// Recheck authorization per stream: the gater only runs at connection
+	// admission, so this covers peers removed from the validator set while a
+	// connection is still open.
+	if n.cfg.Authorizer != nil && !n.cfg.Authorizer(remotePeer) {
+		n.logger.Warn().Str("peer_id", remotePeer).Msg("resetting stream from unauthorized peer")
+		_ = stream.Reset()
+		return
+	}
+
+	select {
+	case n.readSem <- struct{}{}:
+	default:
+		n.logger.Warn().Str("peer_id", remotePeer).Msg("concurrent read limit reached, resetting stream")
+		_ = stream.Reset()
+		return
+	}
+	defer func() { <-n.readSem }()
+
 	defer stream.Close()
 
 	if deadline := time.Now().Add(n.cfg.IOTimeout); true {
@@ -214,7 +245,7 @@ func (n *Network) handleStream(stream network.Stream) {
 	}
 
 	// Call handler in a goroutine to avoid blocking
-	go handler(stream.Conn().RemotePeer().String(), data)
+	go handler(remotePeer, data)
 }
 
 func loadIdentity(base64Key string) (crypto.PrivKey, error) {

universalClient/tss/networking/libp2p/network_test.go

@@ -2,10 +2,15 @@ package libp2p
 
 import (
 	"bytes"
+	"context"
 	"encoding/binary"
+	"fmt"
 	"io"
+	"sync"
 	"testing"
+	"time"
 
+	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -78,3 +83,143 @@ func TestWriteFramed_AcceptsAtMaxFrameSize(t *testing.T) {
 	assert.Equal(t, payload[len(payload)-1], got[len(got)-1])
 }
 
+// allowlist is a mutable peer-ID allowlist used as a test Authorizer.
+type allowlist struct {
+	mu    sync.RWMutex
+	peers map[string]bool
+}
+
+func newAllowlist() *allowlist {
+	return &allowlist{peers: make(map[string]bool)}
+}
+
+func (a *allowlist) allow(peerID string) {
+	a.mu.Lock()
+	a.peers[peerID] = true
+	a.mu.Unlock()
+}
+
+func (a *allowlist) revoke(peerID string) {
+	a.mu.Lock()
+	delete(a.peers, peerID)
+	a.mu.Unlock()
+}
+
+func (a *allowlist) authorized(peerID string) bool {
+	a.mu.RLock()
+	defer a.mu.RUnlock()
+	return a.peers[peerID]
+}
+
+func newTestNetwork(t *testing.T, authorizer func(string) bool) *Network {
+	t.Helper()
+	n, err := New(context.Background(), Config{
+		ListenAddrs: []string{"/ip4/127.0.0.1/tcp/0"},
+		DialTimeout: 5 * time.Second,
+		IOTimeout:   5 * time.Second,
+		Authorizer:  authorizer,
+	}, zerolog.New(io.Discard))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = n.Close() })
+	return n
+}
+
+func connectPeer(t *testing.T, from *Network, to *Network) {
+	t.Helper()
+	require.NoError(t, from.EnsurePeer(to.ID(), to.ListenAddrs()))
+}
+
+func collectMessages(t *testing.T, n *Network) <-chan string {
+	t.Helper()
+	msgs := make(chan string, 64)
+	require.NoError(t, n.RegisterHandler(func(peerID string, data []byte) {
+		msgs <- peerID + ":" + string(data)
+	}))
+	return msgs
+}
+
+func TestNetwork_RejectsUnknownPeer(t *testing.T) {
+	acl := newAllowlist()
+	receiver := newTestNetwork(t, acl.authorized)
+	rogue := newTestNetwork(t, nil)
+	msgs := collectMessages(t, receiver)
+
+	connectPeer(t, rogue, receiver)
+	err := rogue.Send(context.Background(), receiver.ID(), []byte("intrusion"))
+	require.Error(t, err, "unauthenticated peer must not reach the TSS protocol")
+
+	select {
+	case m := <-msgs:
+		t.Fatalf("handler received message from unauthorized peer: %s", m)
+	case <-time.After(500 * time.Millisecond):
+	}
+}
+
+func TestNetwork_AuthorizedPeerDeliversDuringUnauthenticatedFlood(t *testing.T) {
+	acl := newAllowlist()
+	receiver := newTestNetwork(t, acl.authorized)
+	validator := newTestNetwork(t, nil)
+	acl.allow(validator.ID())
+	msgs := collectMessages(t, receiver)
+
+	const rogues = 8
+	var wg sync.WaitGroup
+	for i := range rogues {
+		rogue := newTestNetwork(t, nil)
+		connectPeer(t, rogue, receiver)
+		wg.Add(1)
+		go func(r *Network, i int) {
+			defer wg.Done()
+			for j := range 5 {
+				_ = r.Send(context.Background(), receiver.ID(), fmt.Appendf(nil, "flood-%d-%d", i, j))
+			}
+		}(rogue, i)
+	}
+
+	connectPeer(t, validator, receiver)
+	require.NoError(t, validator.Send(context.Background(), receiver.ID(), []byte("ack")))
+	wg.Wait()
+
+	select {
+	case m := <-msgs:
+		assert.Equal(t, validator.ID()+":ack", m)
+	case <-time.After(5 * time.Second):
+		t.Fatal("validator message not delivered during unauthenticated flood")
+	}
+
+	select {
+	case m := <-msgs:
+		t.Fatalf("received unexpected message: %s", m)
+	case <-time.After(500 * time.Millisecond):
+	}
+}
+
+func TestNetwork_ResetsStreamAfterPeerRevoked(t *testing.T) {
+	acl := newAllowlist()
+	receiver := newTestNetwork(t, acl.authorized)
+	validator := newTestNetwork(t, nil)
+	acl.allow(validator.ID())
+	msgs := collectMessages(t, receiver)
+
+	connectPeer(t, validator, receiver)
+	require.NoError(t, validator.Send(context.Background(), receiver.ID(), []byte("before")))
+	select {
+	case m := <-msgs:
+		assert.Equal(t, validator.ID()+":before", m)
+	case <-time.After(5 * time.Second):
+		t.Fatal("message from authorized peer not delivered")
+	}
+
+	// Revoke: the existing connection survives the gater, but handleStream
+	// must reset new streams from the now-unauthorized peer.
+	acl.revoke(validator.ID())
+	_ = validator.Send(context.Background(), receiver.ID(), []byte("after"))
+
+	select {
+	case m := <-msgs:
+		t.Fatalf("handler received message from revoked peer: %s", m)
+	case <-time.After(500 * time.Millisecond):
+	}
+}
+
+