fix(p2p): skip addressbook and reacher for unreachable addresses (#5370)

Author: gacevicljubisa

pkg/p2p/libp2p/connections_test.go

@@ -13,6 +13,7 @@ import (
 	"reflect"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -27,13 +28,15 @@ import (
 	"github.com/ethersphere/bee/v2/pkg/statestore/mock"
 	"github.com/ethersphere/bee/v2/pkg/swarm"
 	"github.com/ethersphere/bee/v2/pkg/topology/lightnode"
+	libp2pcrypto "github.com/libp2p/go-libp2p/core/crypto"
 	"github.com/libp2p/go-libp2p/p2p/host/eventbus"
 
 	libp2pm "github.com/libp2p/go-libp2p"
 	"github.com/libp2p/go-libp2p/core/event"
 	"github.com/libp2p/go-libp2p/core/host"
 	"github.com/libp2p/go-libp2p/core/network"
 	libp2ppeer "github.com/libp2p/go-libp2p/core/peer"
+	"github.com/libp2p/go-libp2p/core/peerstore"
 	bhost "github.com/libp2p/go-libp2p/p2p/host/basic"
 	swarmt "github.com/libp2p/go-libp2p/p2p/net/swarm/testing"
 	ma "github.com/multiformats/go-multiaddr"
@@ -1485,3 +1488,122 @@ var (
 	noopReachability = func(p2p.ReachabilityStatus) {}
 	noopReachable    = func(swarm.Address, p2p.ReachabilityStatus) {}
 )
+
+func TestPeerMultiaddrsNoFallback(t *testing.T) {
+	t.Parallel()
+
+	s1, _ := newService(t, 1, libp2pServiceOpts{})
+
+	privKey, _, err := libp2pcrypto.GenerateEd25519Key(rand.New(rand.NewSource(time.Now().UnixNano())))
+	if err != nil {
+		t.Fatal(err)
+	}
+	unknownPeerID, err := libp2ppeer.IDFromPrivateKey(privKey)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	addrs, err := s1.PeerMultiaddrs(ctx, unknownPeerID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(addrs) != 0 {
+		t.Fatalf("expected no addresses for unknown peer, got %v", addrs)
+	}
+}
+
+type emptyAddrsPeerstore struct {
+	peerstore.Peerstore
+	targetPeerID libp2ppeer.ID
+	mu           sync.RWMutex
+}
+
+func (p *emptyAddrsPeerstore) setTarget(id libp2ppeer.ID) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.targetPeerID = id
+}
+
+func (p *emptyAddrsPeerstore) isTarget(id libp2ppeer.ID) bool {
+	p.mu.RLock()
+	defer p.mu.RUnlock()
+	return p.targetPeerID == id
+}
+
+func (p *emptyAddrsPeerstore) Addrs(id libp2ppeer.ID) []ma.Multiaddr {
+	if p.isTarget(id) {
+		return nil
+	}
+	return p.Peerstore.Addrs(id)
+}
+
+func (p *emptyAddrsPeerstore) AddrStream(ctx context.Context, id libp2ppeer.ID) <-chan ma.Multiaddr {
+	if p.isTarget(id) {
+		ch := make(chan ma.Multiaddr)
+		go func() {
+			<-ctx.Done()
+			close(ch)
+		}()
+		return ch
+	}
+	return p.Peerstore.AddrStream(ctx, id)
+}
+
+type emptyAddrsHost struct {
+	host.Host
+	ps *emptyAddrsPeerstore
+}
+
+func (h *emptyAddrsHost) Peerstore() peerstore.Peerstore {
+	return h.ps
+}
+
+func TestConnectEmptyPeerstoreSkipsAddressbookAndReacher(t *testing.T) {
+	t.Parallel()
+
+	ab1 := addressbook.New(mock.NewStateStore())
+
+	s1, _ := newService(t, 1, libp2pServiceOpts{
+		Addressbook: ab1,
+		libp2pOpts: libp2p.Options{
+			FullNode: true,
+		},
+	})
+
+	psWrapper := &emptyAddrsPeerstore{Peerstore: s1.Host().Peerstore()}
+
+	var reachableCalled atomic.Bool
+	notifier1 := mockNotifier(noopCf, noopDf, true)
+	notifier1.(*notifiee).reachable = func(_ swarm.Address, _ p2p.ReachabilityStatus) {
+		reachableCalled.Store(true)
+	}
+	s1.SetPickyNotifier(notifier1)
+
+	s2, overlay2 := newService(t, 1, libp2pServiceOpts{
+		libp2pOpts: libp2p.Options{
+			FullNode: true,
+		},
+	})
+
+	psWrapper.setTarget(s2.Host().ID())
+	s1.SetHost(&emptyAddrsHost{Host: s1.Host(), ps: psWrapper})
+
+	_, err := s2.Connect(context.Background(), serviceUnderlayAddress(t, s1))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectPeersEventually(t, s1, overlay2)
+
+	_, err = ab1.Get(overlay2)
+	if err == nil {
+		t.Fatal("expected addressbook to have no entry for NAT peer, but found one")
+	}
+
+	if reachableCalled.Load() {
+		t.Fatal("expected reacher not to be notified for NAT peer")
+	}
+}

pkg/p2p/libp2p/export_test.go

@@ -32,6 +32,10 @@ func (s *Service) Host() host.Host {
 	return s.host
 }
 
+func (s *Service) SetHost(h host.Host) {
+	s.host = h
+}
+
 type StaticAddressResolver = staticAddressResolver
 
 var (
@@ -63,6 +67,10 @@ func (s *Service) FilterSupportedAddresses(addrs []ma.Multiaddr) []ma.Multiaddr
 	return s.filterSupportedAddresses(addrs)
 }
 
+func (s *Service) PeerMultiaddrs(ctx context.Context, peerID libp2ppeer.ID) ([]ma.Multiaddr, error) {
+	return s.peerMultiaddrs(ctx, peerID)
+}
+
 func (s *Service) SetTransportFlags(hasTCP, hasWS, hasWSS bool) {
 	s.enabledTransports = map[bzz.TransportType]bool{
 		bzz.TransportTCP: hasTCP,

pkg/p2p/libp2p/libp2p.go

@@ -584,7 +584,7 @@ func (s *Service) handleIncoming(stream network.Stream) {
 	peerID := stream.Conn().RemotePeer()
 	handshakeStream := newStream(stream, s.metrics)
 
-	peerMultiaddrs, err := s.peerMultiaddrs(s.ctx, stream.Conn().RemoteMultiaddr(), peerID)
+	peerAddrs, err := s.peerMultiaddrs(s.ctx, peerID)
 	if err != nil {
 		s.logger.Debug("stream handler: handshake: build remote multiaddrs", "peer_id", peerID, "error", err)
 		s.logger.Error(nil, "stream handler: handshake: build remote multiaddrs", "peer_id", peerID)
@@ -593,12 +593,28 @@ func (s *Service) handleIncoming(stream network.Stream) {
 		return
 	}
 
+	// For the handshake we always need an observed address (ObservedUnderlay).
+	// If the peerstore had no addresses, fall back to RemoteMultiaddr for the
+	// handshake only. This typically means the peer is behind NAT and its
+	// address is not reachable from the outside.
+	observedAddrs := peerAddrs
+	if len(observedAddrs) == 0 {
+		observedAddrs, err = buildFullMAs([]ma.Multiaddr{stream.Conn().RemoteMultiaddr()}, peerID)
+		if err != nil {
+			s.logger.Debug("stream handler: handshake: build remote multiaddrs fallback", "peer_id", peerID, "error", err)
+			s.logger.Error(nil, "stream handler: handshake: build remote multiaddrs fallback", "peer_id", peerID)
+			_ = handshakeStream.Reset()
+			_ = s.host.Network().ClosePeer(peerID)
+			return
+		}
+	}
+
 	bee260Compat := s.bee260BackwardCompatibility(peerID)
 
 	i, err := s.handshakeService.Handle(
 		s.ctx,
 		handshakeStream,
-		peerMultiaddrs,
+		observedAddrs,
 		handshake.WithBee260Compatibility(bee260Compat),
 	)
 	if err != nil {
@@ -644,7 +660,8 @@ func (s *Service) handleIncoming(stream network.Stream) {
 		return
 	}
 
-	if i.FullNode {
+	// Only persist in addressbook when we have real peerstore addresses.
+	if i.FullNode && len(peerAddrs) > 0 {
 		err = s.addressbook.Put(i.BzzAddress.Overlay, *i.BzzAddress)
 		if err != nil {
 			s.logger.Debug("stream handler: addressbook put error", "peer_id", peerID, "error", err)
@@ -727,7 +744,9 @@ func (s *Service) handleIncoming(stream network.Stream) {
 		return
 	}
 
-	s.notifyReacherConnected(overlay, peerMultiaddrs)
+	if len(peerAddrs) > 0 {
+		s.notifyReacherConnected(overlay, peerAddrs)
+	}
 
 	peerUserAgent := appendSpace(s.peerUserAgent(s.ctx, peerID))
 	s.networkStatus.Store(int32(p2p.NetworkStatusAvailable))
@@ -1029,19 +1048,29 @@ func (s *Service) Connect(ctx context.Context, addrs []ma.Multiaddr) (address *b
 
 	handshakeStream := newStream(stream, s.metrics)
 
-	peerMultiaddrs, err := s.peerMultiaddrs(ctx, stream.Conn().RemoteMultiaddr(), peerID)
+	peerAddrs, err := s.peerMultiaddrs(ctx, peerID)
 	if err != nil {
 		_ = handshakeStream.Reset()
 		_ = s.host.Network().ClosePeer(peerID)
 		return nil, fmt.Errorf("build peer multiaddrs: %w", err)
 	}
 
+	observedAddrs := peerAddrs
+	if len(observedAddrs) == 0 {
+		observedAddrs, err = buildFullMAs([]ma.Multiaddr{stream.Conn().RemoteMultiaddr()}, peerID)
+		if err != nil {
+			_ = handshakeStream.Reset()
+			_ = s.host.Network().ClosePeer(peerID)
+			return nil, fmt.Errorf("build peer multiaddrs fallback: %w", err)
+		}
+	}
+
 	bee260Compat := s.bee260BackwardCompatibility(peerID)
 
 	i, err := s.handshakeService.Handshake(
 		s.ctx,
 		handshakeStream,
-		peerMultiaddrs,
+		observedAddrs,
 		handshake.WithBee260Compatibility(bee260Compat),
 	)
 	if err != nil {
@@ -1120,7 +1149,9 @@ func (s *Service) Connect(ctx context.Context, addrs []ma.Multiaddr) (address *b
 
 	s.metrics.CreatedConnectionCount.Inc()
 
-	s.notifyReacherConnected(overlay, peerMultiaddrs)
+	if len(peerAddrs) > 0 {
+		s.notifyReacherConnected(overlay, peerAddrs)
+	}
 
 	peerUA := appendSpace(s.peerUserAgent(ctx, peerID))
 	loggerV1.Debug("successfully connected to peer (outbound)", "addresses", i.BzzAddress.ShortString(), "light", i.LightString(), "user_agent", peerUA)
@@ -1429,17 +1460,12 @@ func (s *Service) determineCurrentNetworkStatus(err error) error {
 	return err
 }
 
-// peerMultiaddrs builds full multiaddresses for a peer given information from
-// the libp2p host peerstore. If the peerstore doesn't have addresses yet,
-// it falls back to using the remote address from the active connection.
-func (s *Service) peerMultiaddrs(ctx context.Context, remoteAddr ma.Multiaddr, peerID libp2ppeer.ID) ([]ma.Multiaddr, error) {
+// peerMultiaddrs builds full multiaddresses for a peer using the peerstore.
+func (s *Service) peerMultiaddrs(ctx context.Context, peerID libp2ppeer.ID) ([]ma.Multiaddr, error) {
 	waitPeersCtx, cancel := context.WithTimeout(ctx, peerstoreWaitAddrsTimeout)
 	defer cancel()
 
 	mas := waitPeerAddrs(waitPeersCtx, s.host.Peerstore(), peerID)
-	if len(mas) == 0 && remoteAddr != nil {
-		mas = []ma.Multiaddr{remoteAddr}
-	}
 
 	return buildFullMAs(mas, peerID)
 }