diff --git a/universalClient/tss/coordinator/coordinator.go b/universalClient/tss/coordinator/coordinator.go index f1cbe4a6..55627b11 100644 --- a/universalClient/tss/coordinator/coordinator.go +++ b/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() { diff --git a/universalClient/tss/coordinator/coordinator_test.go b/universalClient/tss/coordinator/coordinator_test.go index 7e2b6c69..72daca8e 100644 --- a/universalClient/tss/coordinator/coordinator_test.go +++ b/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")) + }) +} diff --git a/universalClient/tss/networking/libp2p/config.go b/universalClient/tss/networking/libp2p/config.go index 2acfa632..e45e11d9 100644 --- a/universalClient/tss/networking/libp2p/config.go +++ b/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. diff --git a/universalClient/tss/networking/libp2p/gater.go b/universalClient/tss/networking/libp2p/gater.go new file mode 100644 index 00000000..ac773e9e --- /dev/null +++ b/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 +} diff --git a/universalClient/tss/networking/libp2p/network.go b/universalClient/tss/networking/libp2p/network.go index 8710940c..dec38303 100644 --- a/universalClient/tss/networking/libp2p/network.go +++ b/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) { diff --git a/universalClient/tss/networking/libp2p/network_test.go b/universalClient/tss/networking/libp2p/network_test.go index 8c984668..f09aa1e7 100644 --- a/universalClient/tss/networking/libp2p/network_test.go +++ b/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): + } +} + + diff --git a/universalClient/tss/tss.go b/universalClient/tss/tss.go index 38d70b16..f54ca065 100644 --- a/universalClient/tss/tss.go +++ b/universalClient/tss/tss.go @@ -285,34 +285,9 @@ func (n *Node) Start(ctx context.Context) error { n.logger.Debug().Msg("starting TSS node") - // Start libp2p network - net, err := libp2pnet.New(ctx, n.networkCfg, n.logger) - if err != nil { - return fmt.Errorf("failed to start libp2p network: %w", err) - } - n.network = net - - // Register global message handler - if err := net.RegisterHandler(n.onReceive); err != nil { - net.Close() - return fmt.Errorf("failed to register message handler: %w", err) - } - - // Recover IN_PROGRESS events on startup. Two-pass: - // 1. Rows whose event_data already carries signing_data → SIGNED - // (signature was persisted but status got clobbered by a race). - // 2. Remaining IN_PROGRESS → CONFIRMED (genuine mid-session crashes). - signedRecovered, confirmedReset, err := n.eventStore.RecoverInProgressEvents() - if err != nil { - n.logger.Warn().Err(err).Msg("failed to recover IN_PROGRESS events, continuing anyway") - } else if signedRecovered > 0 || confirmedReset > 0 { - n.logger.Info(). - Int64("signed_recovered", signedRecovered). - Int64("confirmed_reset", confirmedReset). - Msg("recovered IN_PROGRESS events on node startup") - } - - // Create coordinator with send function using node's Send method + // Create coordinator with send function using node's Send method. + // Created before the network so the connection gater and message handler + // never observe a nil coordinator or session manager. if n.coordinator == nil { coord := coordinator.NewCoordinator( n.eventStore, @@ -351,6 +326,36 @@ func (n *Node) Start(ctx context.Context) error { n.sessionManager = sessionMgr } + // Only Universal Validators may connect and open TSS streams + n.networkCfg.Authorizer = n.coordinator.IsKnownPeer + + // Start libp2p network + net, err := libp2pnet.New(ctx, n.networkCfg, n.logger) + if err != nil { + return fmt.Errorf("failed to start libp2p network: %w", err) + } + n.network = net + + // Register global message handler + if err := net.RegisterHandler(n.onReceive); err != nil { + net.Close() + return fmt.Errorf("failed to register message handler: %w", err) + } + + // Recover IN_PROGRESS events on startup. Two-pass: + // 1. Rows whose event_data already carries signing_data → SIGNED + // (signature was persisted but status got clobbered by a race). + // 2. Remaining IN_PROGRESS → CONFIRMED (genuine mid-session crashes). + signedRecovered, confirmedReset, err := n.eventStore.RecoverInProgressEvents() + if err != nil { + n.logger.Warn().Err(err).Msg("failed to recover IN_PROGRESS events, continuing anyway") + } else if signedRecovered > 0 || confirmedReset > 0 { + n.logger.Info(). + Int64("signed_recovered", signedRecovered). + Int64("confirmed_reset", confirmedReset). + Msg("recovered IN_PROGRESS events on node startup") + } + // Start coordinator n.coordinator.Start(ctx)