Don't hole punch if either peer is behind a Symmetric NAT

p2p/host/basic/basic_host.go

@@ -219,6 +219,8 @@ func NewHost(ctx context.Context, n network.Network, opts *HostOpts) (*BasicHost
 		h.mux = opts.MultistreamMuxer
 	}
 
+	h.Peerstore().Put(h.ID(), identify.UDPNATDeviceTypeKey, network.NATDeviceTypeUnknown)
+	h.Peerstore().Put(h.ID(), identify.TCPNATDeviceTypeKey, network.NATDeviceTypeUnknown)
 	// we can't set this as a default above because it depends on the *BasicHost.
 	if h.disableSignedPeerRecord {
 		h.ids, err = identify.NewIDService(h, identify.UserAgent(opts.UserAgent), identify.DisableSignedPeerRecord())
@@ -229,6 +231,13 @@ func NewHost(ctx context.Context, n network.Network, opts *HostOpts) (*BasicHost
 		return nil, fmt.Errorf("failed to create Identify service: %s", err)
 	}
 
+	if opts.EnableHolePunching {
+		h.hps, err = holepunch.NewHolePunchService(h, h.ids)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create hole punch service: %w", err)
+		}
+	}
+
 	if uint64(opts.NegotiationTimeout) != 0 {
 		h.negtimeout = opts.NegotiationTimeout
 	}

p2p/protocol/holepunch/coordination.go

@@ -7,6 +7,7 @@ import (
 
 	logging "github.com/ipfs/go-log"
 	"github.com/jpillora/backoff"
+	"github.com/libp2p/go-libp2p-core/event"
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/network"
 	"github.com/libp2p/go-libp2p-core/peer"
@@ -49,11 +50,55 @@ func NewHolePunchService(h host.Host, ids *identify.IDService) (*HolePunchServic
 	ctx, cancel := context.WithCancel(context.Background())
 	hs := &HolePunchService{ctx: ctx, ctxCancel: cancel, host: h, ids: ids}
 
+	sub, err := h.EventBus().Subscribe(new(event.EvtNATDeviceTypeChanged))
+	if err != nil {
+		return nil, err
+	}
+
 	h.SetStreamHandler(protocol, hs.handleNewStream)
 	h.Network().Notify((*netNotifiee)(hs))
+
+	hs.refCount.Add(1)
+	go hs.loop(sub)
+
 	return hs, nil
 }
 
+func (hs *HolePunchService) loop(sub event.Subscription) {
+	defer hs.refCount.Done()
+	defer sub.Close()
+
+	for {
+		select {
+		// Our local NAT device types are intialized in the peerstore when the Host is created
+		//	and updated in the peerstore by the Observed Address Manager.
+		case _, ok := <-sub.Out():
+			if !ok {
+				return
+			}
+
+			if hs.peerSupportsHolePunching(hs.host.ID(), hs.host.Addrs()) {
+				hs.host.SetStreamHandler(protocol, hs.handleNewStream)
+			} else {
+				hs.host.RemoveStreamHandler(protocol)
+			}
+
+		case <-hs.ctx.Done():
+			return
+		}
+	}
+}
+
+func hasProtoAddr(protocCode int, addrs []ma.Multiaddr) bool {
+	for _, a := range addrs {
+		if _, err := a.ValueForProtocol(protocCode); err == nil {
+			return true
+		}
+	}
+
+	return false
+}
+
 // Close closes the Hole Punch Service.
 func (hs *HolePunchService) Close() error {
 	hs.closeSync.Do(func() {
@@ -84,6 +129,12 @@ func (hs *HolePunchService) holePunch(relayConn network.Conn) {
 		}
 	}
 
+	// return if either peer does NOT support hole punching
+	if !hs.peerSupportsHolePunching(rp, hs.host.Peerstore().Addrs(rp)) ||
+		!hs.peerSupportsHolePunching(hs.host.ID(), hs.host.Addrs()) {
+		return
+	}
+
 	// hole punch
 	hpCtx := network.WithUseTransient(hs.ctx, "hole-punch")
 	sCtx := network.WithNoDial(hpCtx, "hole-punch")
@@ -150,6 +201,30 @@ func (hs *HolePunchService) holePunch(relayConn network.Conn) {
 	}
 }
 
+// We can hole punch with a peer ONLY if it is NOT behind a symmetric NAT for all the transport protocol it supports.
+func (hs *HolePunchService) peerSupportsHolePunching(p peer.ID, addrs []ma.Multiaddr) bool {
+	udpSupported := hasProtoAddr(ma.P_UDP, addrs)
+	tcpSupported := hasProtoAddr(ma.P_TCP, addrs)
+	udpNAT, _ := hs.host.Peerstore().Get(p, identify.UDPNATDeviceTypeKey)
+	tcpNAT, _ := hs.host.Peerstore().Get(p, identify.TCPNATDeviceTypeKey)
+	udpNatType := udpNAT.(network.NATDeviceType)
+	tcpNATType := tcpNAT.(network.NATDeviceType)
+
+	if udpSupported {
+		if udpNatType == network.NATDeviceTypeCone || udpNatType == network.NATDeviceTypeUnknown {
+			return true
+		}
+	}
+
+	if tcpSupported {
+		if tcpNATType == network.NATDeviceTypeCone || tcpNATType == network.NATDeviceTypeUnknown {
+			return true
+		}
+	}
+
+	return false
+}
+
 func (hs *HolePunchService) handleNewStream(s network.Stream) {
 	log.Infof("got hole punch request from peer %s", s.Conn().RemotePeer().Pretty())
 	_ = s.SetDeadline(time.Now().Add(holePunchTimeout))

p2p/protocol/identify/id.go

@@ -41,6 +41,13 @@ const ID = "/ipfs/id/1.0.0"
 // 0.4.17 which asserted an exact version match.
 const LibP2PVersion = "ipfs/0.1.0"
 
+const (
+	// UDPNATDeviceTypeKey is the key with which we will persist a peer's UDP NAT Device Type to the peerstore.
+	UDPNATDeviceTypeKey = "UDPNATDeviceType"
+	// TCPNATDeviceTypeKey is the key with which we will persist a peer's TCP NAT Device Type to the peerstore.
+	TCPNATDeviceTypeKey = "TCPNATDeviceType"
+)
+
 // ClientVersion is the default user agent.
 //
 // Deprecated: Set this with the UserAgent option.
@@ -542,10 +549,37 @@ func (ids *IDService) createBaseIdentifyResponse(
 	av := ids.UserAgent
 	mes.ProtocolVersion = &pv
 	mes.AgentVersion = &av
+	udpNAT, tcpNAT := ids.observedAddrs.getNATDeviceTypes()
+	udpNATPb := toPbNATDeviceType(udpNAT)
+	tcpNATPb := toPbNATDeviceType(tcpNAT)
+	mes.UdpNATDeviceType = &udpNATPb
+	mes.TcpNATDeviceType = &tcpNATPb
 
 	return mes
 }
 
+func toPbNATDeviceType(t network.NATDeviceType) pb.Identify_NATDeviceType {
+	switch t {
+	case network.NATDeviceTypeCone:
+		return pb.Identify_CONE
+	case network.NATDeviceTypeSymmetric:
+		return pb.Identify_SYMMETRIC
+	default:
+		return pb.Identify_UNKNOWN
+	}
+}
+
+func fromPbNATDeviceType(typ pb.Identify_NATDeviceType) network.NATDeviceType {
+	switch typ {
+	case pb.Identify_CONE:
+		return network.NATDeviceTypeCone
+	case pb.Identify_SYMMETRIC:
+		return network.NATDeviceTypeSymmetric
+	default:
+		return network.NATDeviceTypeUnknown
+	}
+}
+
 func (ids *IDService) getSignedRecord(snapshot *identifySnapshot) []byte {
 	if ids.disableSignedPeerRecord || snapshot.record == nil {
 		return nil
@@ -634,9 +668,13 @@ func (ids *IDService) consumeMessage(mes *pb.Identify, c network.Conn) {
 	// get protocol versions
 	pv := mes.GetProtocolVersion()
 	av := mes.GetAgentVersion()
+	udpNATType := mes.GetUdpNATDeviceType()
+	tcpNATType := mes.GetTcpNATDeviceType()
 
 	ids.Host.Peerstore().Put(p, "ProtocolVersion", pv)
 	ids.Host.Peerstore().Put(p, "AgentVersion", av)
+	ids.Host.Peerstore().Put(p, UDPNATDeviceTypeKey, fromPbNATDeviceType(udpNATType))
+	ids.Host.Peerstore().Put(p, TCPNATDeviceTypeKey, fromPbNATDeviceType(tcpNATType))
 
 	// get the key from the other side. we may not have it (no-auth transport)
 	ids.consumeReceivedPubKey(c, mes.PublicKey)

p2p/protocol/identify/id_test.go

@@ -126,6 +126,21 @@ func subtestIDService(t *testing.T) {
 	testHasCertifiedAddrs(t, h1, h2p, []ma.Multiaddr{})
 	testHasCertifiedAddrs(t, h2, h1p, []ma.Multiaddr{})
 
+	// test we have the TCP & UDP NAT Device Types on both sides
+	udp, err := h1.Peerstore().Get(h2p, identify.UDPNATDeviceTypeKey)
+	require.NoError(t, err)
+	require.Equal(t, network.NATDeviceTypeUnknown, udp.(network.NATDeviceType))
+	tcp, err := h1.Peerstore().Get(h2p, identify.TCPNATDeviceTypeKey)
+	require.NoError(t, err)
+	require.Equal(t, network.NATDeviceTypeUnknown, tcp.(network.NATDeviceType))
+
+	udp, err = h2.Peerstore().Get(h1p, identify.UDPNATDeviceTypeKey)
+	require.NoError(t, err)
+	require.Equal(t, network.NATDeviceTypeUnknown, udp.(network.NATDeviceType))
+	tcp, err = h2.Peerstore().Get(h1p, identify.TCPNATDeviceTypeKey)
+	require.NoError(t, err)
+	require.Equal(t, network.NATDeviceTypeUnknown, tcp.(network.NATDeviceType))
+
 	// test that we received the "identify completed" event.
 	select {
 	case <-sub.Out():

p2p/protocol/identify/obsaddr.go

@@ -151,6 +151,13 @@ func NewObservedAddrManager(ctx context.Context, host host.Host) (*ObservedAddrM
 	return oas, nil
 }
 
+func (oas *ObservedAddrManager) getNATDeviceTypes() (udp, tcp network.NATDeviceType) {
+	oas.mu.RLock()
+	defer oas.mu.RUnlock()
+
+	return oas.currentUDPNATDeviceType, oas.currentTCPNATDeviceType
+}
+
 // AddrsFor return all activated observed addresses associated with the given
 // (resolved) listen address.
 func (oas *ObservedAddrManager) AddrsFor(addr ma.Multiaddr) (addrs []ma.Multiaddr) {
@@ -200,6 +207,21 @@ func (oas *ObservedAddrManager) filter(observedAddrs []*observedAddr) []ma.Multi
 		}
 	}
 
+	// For certain use cases such as hole punching, it's better to advertise even unactivated observed addresses rather than none at all
+	// because we don't want to wait for a hole-punch till we make enough connections with other peers to discover our activated addresses.
+	// If we have activated addresses, we will use them, otherwise, let's use whatever observed addresses we do have.
+	if len(pmap) == 0 {
+		for i := range observedAddrs {
+			a := observedAddrs[i]
+			if now.Sub(a.lastSeen) <= oas.ttl {
+				// group addresses by their IPX/Transport Protocol(TCP or UDP) pattern.
+				pat := a.groupKey()
+				pmap[pat] = append(pmap[pat], a)
+
+			}
+		}
+	}
+
 	addrs := make([]ma.Multiaddr, 0, len(observedAddrs))
 	for pat := range pmap {
 		s := pmap[pat]
@@ -269,7 +291,6 @@ func (oas *ObservedAddrManager) worker(ctx context.Context) {
 
 		case obs := <-oas.wch:
 			oas.maybeRecordObservation(obs.conn, obs.observed)
-
 		case <-ticker.C:
 			oas.gc()
 		case <-oas.refreshTimer.C:

p2p/protocol/identify/obsaddr_test.go

@@ -2,11 +2,11 @@ package identify_test
 
 import (
 	"context"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
 
-	detectrace "github.com/ipfs/go-detect-race"
 	"github.com/libp2p/go-eventbus"
 	"github.com/libp2p/go-libp2p-core/event"
 	"github.com/libp2p/go-libp2p-core/host"
@@ -16,6 +16,7 @@ import (
 	mocknet "github.com/libp2p/go-libp2p/p2p/net/mock"
 	identify "github.com/libp2p/go-libp2p/p2p/protocol/identify"
 
+	detectrace "github.com/ipfs/go-detect-race"
 	ma "github.com/multiformats/go-multiaddr"
 	"github.com/stretchr/testify/require"
 )
@@ -171,25 +172,22 @@ func TestObsAddrSet(t *testing.T) {
 	harness.observe(a2, pa4)
 	harness.observe(a3, pa4)
 
-	// these are all different so we should not yet get them.
-	if !addrsMatch(harness.oas.Addrs(), nil) {
-		t.Error("addrs should _still_ be empty (once)")
-	}
-
-	// same observer, so should not yet get them.
+	// get the address as none has been activated yet.
 	harness.observe(a1, pa4)
 	harness.observe(a2, pa4)
 	harness.observe(a3, pa4)
-	if !addrsMatch(harness.oas.Addrs(), nil) {
-		t.Error("addrs should _still_ be empty (same obs)")
+
+	fmt.Println(harness.oas.Addrs())
+	if !addrsMatch(harness.oas.Addrs(), []ma.Multiaddr{a1, a2}) {
+		t.Error("should get expected addresses as there are no activated addresses")
 	}
 
 	// different observer, but same observer group.
 	harness.observe(a1, pa5)
 	harness.observe(a2, pa5)
 	harness.observe(a3, pa5)
-	if !addrsMatch(harness.oas.Addrs(), nil) {
-		t.Error("addrs should _still_ be empty (same obs group)")
+	if !addrsMatch(harness.oas.Addrs(), []ma.Multiaddr{a1, a2}) {
+		t.Error("should get expected addresses as there are no activated addresses")
 	}
 
 	harness.observe(a1, pb1)
@@ -383,6 +381,7 @@ func TestEmitNATDeviceTypeSymmetric(t *testing.T) {
 	defer cancel()
 	harness := newHarness(ctx, t)
 	require.Empty(t, harness.oas.Addrs())
+
 	emitter, err := harness.host.EventBus().Emitter(new(event.EvtLocalReachabilityChanged), eventbus.Stateful)
 	require.NoError(t, err)
 	require.NoError(t, emitter.Emit(event.EvtLocalReachabilityChanged{Reachability: network.ReachabilityPrivate}))
@@ -429,6 +428,7 @@ func TestEmitNATDeviceTypeCone(t *testing.T) {
 	defer cancel()
 	harness := newHarness(ctx, t)
 	require.Empty(t, harness.oas.Addrs())
+
 	emitter, err := harness.host.EventBus().Emitter(new(event.EvtLocalReachabilityChanged), eventbus.Stateful)
 	require.NoError(t, err)
 	require.NoError(t, emitter.Emit(event.EvtLocalReachabilityChanged{Reachability: network.ReachabilityPrivate}))