fix: derive NAT telemetry from ICE, drop leaky go-nats discovery

clientcore/consumer.go

@@ -503,7 +503,7 @@ func NewConsumerWebRTC(options *WebRTCOptions, wg *sync.WaitGroup) *WorkerFSM {
 				return 0, []interface{}{}
 			case http.StatusOK:
 				// Signaling is complete, so we can short circuit instead of awaiting the response body
-				return 4, []interface{}{peerConnection, connectionEstablished, connectionChange, connectionClosed}
+				return 4, []interface{}{peerConnection, candidates, connectionEstablished, connectionChange, connectionClosed}
 			default:
 				slog.Debug(
 					// Borked!
@@ -515,28 +515,30 @@ func NewConsumerWebRTC(options *WebRTCOptions, wg *sync.WaitGroup) *WorkerFSM {
 		FSMstate(func(ctx context.Context, com *ipcChan, input []interface{}) (int, []interface{}) {
 			// State 4
 			// input[0]: *webrtc.PeerConnection
-			// input[1]: chan *webrtc.DataChannel
-			// input[2]: chan webrtc.PeerConnectionState
-			// input[3]: chan struct{}
+			// input[1]: []webrtc.ICECandidate
+			// input[2]: chan *webrtc.DataChannel
+			// input[3]: chan webrtc.PeerConnectionState
+			// input[4]: chan struct{}
 			peerConnection := input[0].(*webrtc.PeerConnection)
-			connectionEstablished := input[1].(chan *webrtc.DataChannel)
-			connectionChange := input[2].(chan webrtc.PeerConnectionState)
-			connectionClosed := input[3].(chan struct{})
-			slog.Debug(fmt.Sprintf("Consumer state 4, signaling complete!"))
+			candidates := input[1].([]webrtc.ICECandidate)
+			connectionEstablished := input[2].(chan *webrtc.DataChannel)
+			connectionChange := input[3].(chan webrtc.PeerConnectionState)
+			connectionClosed := input[4].(chan struct{})
+			slog.Debug("Consumer state 4, signaling complete!")
 
-			// XXX: Use our current cohort of STUN servers to perform NAT behavior discovery such that we
-			// can send interesting traces revealing the outcome of our NAT traversal attempt. If the
-			// cohort fails here, we won't drop it.
-			STUNSrvs := scache.cohort()
+			// Summarize NAT behavior from the ICE candidates we just gathered so we can send
+			// interesting traces revealing the outcome of our NAT traversal attempt, without a
+			// redundant standalone STUN probe.
+			natSummary := summarizeNATFromICE(candidates)
 
 			select {
 			case d := <-connectionEstablished:
-				slog.Debug(fmt.Sprintf("A WebRTC connection has been established!"))
-				go otel.CollectAndSendNATBehaviorTelemetry(STUNSrvs, "nat_success")
+				slog.Debug("A WebRTC connection has been established!")
+				go otel.SendNATBehaviorTelemetry(natSummary, "nat_success")
 				return 5, []interface{}{peerConnection, d, connectionChange, connectionClosed}
 			case <-time.After(options.NATFailTimeout):
-				slog.Debug(fmt.Sprintf("NAT failure, aborting!"))
-				go otel.CollectAndSendNATBehaviorTelemetry(STUNSrvs, "nat_failure")
+				slog.Debug("NAT failure, aborting!")
+				go otel.SendNATBehaviorTelemetry(natSummary, "nat_failure")
 				// Borked!
 				peerConnection.Close() // TODO: there's an err we should handle here
 				return 0, []interface{}{}

clientcore/nat.go

@@ -0,0 +1,114 @@
+package clientcore
+
+import (
+	"strings"
+
+	"github.com/getlantern/broflake/otel"
+	"github.com/pion/webrtc/v4"
+)
+
+// NAT behavior vocabulary. Mapping/filtering values and the non-cone NAT types reuse the
+// strings go-nats previously emitted so dashboards keep grouping; the three go-nats cone
+// variants (full / address-restricted / port-restricted) collapse to a single "Cone NAT"
+// because filtering behavior is not observable from ICE (see otel.NATSummary).
+const (
+	natMappingIndependent = "independent"
+	natMappingDependent   = "address-port dependent"
+	natBehaviorUnknown    = "unspecified"
+
+	natTypeOpen       = "Open to the Internet"
+	natTypeCone       = "Cone NAT"
+	natTypeSymmetric  = "Symmetric NAT"
+	natTypeUDPBlocked = "UDP blocked by firewall"
+	natTypeUnknown    = "Unknown"
+)
+
+// summarizeNATFromICE derives a NAT-behavior summary from the local ICE candidates
+// gathered during connection establishment, avoiding a redundant standalone STUN
+// probe. The mapping behavior is inferred by comparing the server-reflexive (srflx)
+// mapped addresses returned by the multiple STUN servers in the cohort: identical
+// mapped endpoints across servers indicate endpoint-independent (cone) mapping, while
+// differing endpoints indicate address/port-dependent (symmetric) mapping.
+//
+// Filtering behavior is not observable from ICE — it requires the RFC 5780
+// CHANGE-REQUEST transaction, which ICE never issues — so it is always reported as
+// unspecified and the cone variants of NATType are not distinguished.
+func summarizeNATFromICE(candidates []webrtc.ICECandidate) otel.NATSummary {
+	var srflx []webrtc.ICECandidate
+	for _, c := range candidates {
+		if c.Typ == webrtc.ICECandidateTypeSrflx {
+			srflx = append(srflx, c)
+		}
+	}
+
+	// No server-reflexive candidate means the STUN servers were unreachable (only
+	// host candidates gathered), which we treat as UDP being blocked.
+	if len(srflx) == 0 {
+		return otel.NATSummary{
+			MappingBehavior:   natBehaviorUnknown,
+			FilteringBehavior: natBehaviorUnknown,
+			NATType:           natTypeUDPBlocked,
+		}
+	}
+
+	// Mapping behavior compares mapped endpoints from the same address family, so
+	// classify against whichever family gathered more srflx candidates.
+	var v4, v6 []webrtc.ICECandidate
+	for _, c := range srflx {
+		if strings.Contains(c.Address, ":") {
+			v6 = append(v6, c)
+		} else {
+			v4 = append(v4, c)
+		}
+	}
+	fam := v4
+	if len(v6) > len(v4) {
+		fam = v6
+	}
+
+	sum := otel.NATSummary{
+		ExternalIP:        fam[0].Address,
+		FilteringBehavior: natBehaviorUnknown,
+	}
+
+	for _, c := range fam {
+		if c.RelatedAddress != "" && c.Address != c.RelatedAddress {
+			sum.IsNatted = true
+		}
+	}
+
+	if fam[0].RelatedPort != 0 {
+		sum.PortPreservation = fam[0].Port == fam[0].RelatedPort
+	}
+
+	type endpoint struct {
+		addr string
+		port uint16
+	}
+	distinct := make(map[endpoint]struct{}, len(fam))
+	for _, c := range fam {
+		distinct[endpoint{c.Address, c.Port}] = struct{}{}
+	}
+	switch {
+	case len(fam) < 2:
+		// A single srflx candidate can't reveal whether the mapping varies per server.
+		sum.MappingBehavior = natBehaviorUnknown
+	case len(distinct) == 1:
+		sum.MappingBehavior = natMappingIndependent
+	default:
+		sum.MappingBehavior = natMappingDependent
+	}
+
+	switch {
+	case !sum.IsNatted:
+		sum.NATType = natTypeOpen
+	case sum.MappingBehavior == natMappingDependent:
+		sum.NATType = natTypeSymmetric
+	case sum.MappingBehavior == natMappingIndependent:
+		sum.NATType = natTypeCone
+	default:
+		sum.NATType = natTypeUnknown
+	}
+
+	return sum
+}

clientcore/nat_test.go

@@ -0,0 +1,112 @@
+package clientcore
+
+import (
+	"testing"
+
+	"github.com/pion/webrtc/v4"
+)
+
+func host(addr string, port uint16) webrtc.ICECandidate {
+	return webrtc.ICECandidate{Typ: webrtc.ICECandidateTypeHost, Address: addr, Port: port}
+}
+
+func srflx(addr string, port uint16, base string, basePort uint16) webrtc.ICECandidate {
+	return webrtc.ICECandidate{
+		Typ:            webrtc.ICECandidateTypeSrflx,
+		Address:        addr,
+		Port:           port,
+		RelatedAddress: base,
+		RelatedPort:    basePort,
+	}
+}
+
+func TestSummarizeNATFromICE(t *testing.T) {
+	tests := []struct {
+		name        string
+		candidates  []webrtc.ICECandidate
+		wantNatted  bool
+		wantMapping string
+		wantType    string
+		wantPortPsv bool
+		wantExtIP   string
+	}{
+		{
+			name:        "no srflx means UDP blocked",
+			candidates:  []webrtc.ICECandidate{host("192.168.1.5", 50000)},
+			wantMapping: natBehaviorUnknown,
+			wantType:    natTypeUDPBlocked,
+		},
+		{
+			name: "identical mapped endpoint across servers is cone",
+			candidates: []webrtc.ICECandidate{
+				host("192.168.1.5", 50000),
+				srflx("203.0.113.7", 50000, "192.168.1.5", 50000),
+				srflx("203.0.113.7", 50000, "192.168.1.5", 50000),
+			},
+			wantNatted:  true,
+			wantMapping: natMappingIndependent,
+			wantType:    natTypeCone,
+			wantPortPsv: true,
+			wantExtIP:   "203.0.113.7",
+		},
+		{
+			name: "differing mapped ports across servers is symmetric",
+			candidates: []webrtc.ICECandidate{
+				srflx("203.0.113.7", 50001, "192.168.1.5", 50000),
+				srflx("203.0.113.7", 50002, "192.168.1.5", 50000),
+			},
+			wantNatted:  true,
+			wantMapping: natMappingDependent,
+			wantType:    natTypeSymmetric,
+			wantPortPsv: false,
+			wantExtIP:   "203.0.113.7",
+		},
+		{
+			name: "single srflx can't classify mapping",
+			candidates: []webrtc.ICECandidate{
+				srflx("203.0.113.7", 50000, "192.168.1.5", 50000),
+			},
+			wantNatted:  true,
+			wantMapping: natBehaviorUnknown,
+			wantType:    natTypeUnknown,
+			wantPortPsv: true,
+			wantExtIP:   "203.0.113.7",
+		},
+		{
+			name: "mapped equals base means not natted, open",
+			candidates: []webrtc.ICECandidate{
+				srflx("203.0.113.7", 50000, "203.0.113.7", 50000),
+				srflx("203.0.113.7", 50000, "203.0.113.7", 50000),
+			},
+			wantNatted:  false,
+			wantMapping: natMappingIndependent,
+			wantType:    natTypeOpen,
+			wantPortPsv: true,
+			wantExtIP:   "203.0.113.7",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := summarizeNATFromICE(tt.candidates)
+			if got.IsNatted != tt.wantNatted {
+				t.Errorf("IsNatted = %v, want %v", got.IsNatted, tt.wantNatted)
+			}
+			if got.MappingBehavior != tt.wantMapping {
+				t.Errorf("MappingBehavior = %q, want %q", got.MappingBehavior, tt.wantMapping)
+			}
+			if got.NATType != tt.wantType {
+				t.Errorf("NATType = %q, want %q", got.NATType, tt.wantType)
+			}
+			if got.PortPreservation != tt.wantPortPsv {
+				t.Errorf("PortPreservation = %v, want %v", got.PortPreservation, tt.wantPortPsv)
+			}
+			if got.ExternalIP != tt.wantExtIP {
+				t.Errorf("ExternalIP = %q, want %q", got.ExternalIP, tt.wantExtIP)
+			}
+			if got.FilteringBehavior != natBehaviorUnknown {
+				t.Errorf("FilteringBehavior = %q, want %q", got.FilteringBehavior, natBehaviorUnknown)
+			}
+		})
+	}
+}

go.mod

@@ -2,15 +2,12 @@ module github.com/getlantern/broflake
 
 go 1.24.0
 
-replace github.com/enobufs/go-nats => github.com/noahlevenson/go-nats v0.0.0-20230720174341-49df1f749775
-
 replace github.com/quic-go/quic-go => github.com/getlantern/quic-go-unbounded-fork v0.59.0-unbounded
 
 require (
 	github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
 	github.com/coder/websocket v1.8.12
 	github.com/elazarl/goproxy v1.7.2
-	github.com/enobufs/go-nats v0.0.1
 	github.com/getlantern/geo v0.0.0-20240108161311-50692a1b69a9
 	github.com/getlantern/telemetry v0.0.0-20250606052628-8960164ec1f5
 	github.com/google/uuid v1.6.0
@@ -51,7 +48,6 @@ require (
 	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect
 	github.com/pierrec/lz4/v4 v4.1.12 // indirect
 	github.com/pion/datachannel v1.6.0 // indirect
-	github.com/pion/dtls/v2 v2.2.12 // indirect
 	github.com/pion/dtls/v3 v3.1.2 // indirect
 	github.com/pion/ice/v4 v4.2.2 // indirect
 	github.com/pion/interceptor v0.1.44 // indirect
@@ -63,13 +59,8 @@ require (
 	github.com/pion/sctp v1.9.4 // indirect
 	github.com/pion/sdp/v3 v3.0.18 // indirect
 	github.com/pion/srtp/v3 v3.0.10 // indirect
-	github.com/pion/stun v0.6.1 // indirect
 	github.com/pion/stun/v3 v3.1.1 // indirect
-	github.com/pion/transport v0.14.1 // indirect
-	github.com/pion/transport/v2 v2.2.10 // indirect
-	github.com/pion/turn v1.3.7 // indirect
 	github.com/pion/turn/v4 v4.1.4 // indirect
-	github.com/pkg/errors v0.8.1 // indirect
 	github.com/ulikunitz/xz v0.5.10 // indirect
 	github.com/wlynxg/anet v0.0.5 // indirect
 	github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect