Merge branch 'main' into ssh-rewrite

client/internal/peer/conn.go

@@ -6,12 +6,11 @@ import (
 	"math/rand"
 	"net"
 	"net/netip"
-	"os"
 	"runtime"
 	"sync"
 	"time"
 
-	"github.com/pion/ice/v3"
+	"github.com/pion/ice/v4"
 	log "github.com/sirupsen/logrus"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 
@@ -174,7 +173,7 @@ func (conn *Conn) Open(engineCtx context.Context) error {
 	conn.handshaker = NewHandshaker(conn.Log, conn.config, conn.signaler, conn.workerICE, conn.workerRelay)
 
 	conn.handshaker.AddOnNewOfferListener(conn.workerRelay.OnNewOffer)
-	if os.Getenv("NB_FORCE_RELAY") != "true" {
+	if !isForceRelayed() {
 		conn.handshaker.AddOnNewOfferListener(conn.workerICE.OnNewOffer)
 	}
 

client/internal/peer/env.go

@@ -0,0 +1,14 @@
+package peer
+
+import (
+	"os"
+	"strings"
+)
+
+const (
+	EnvKeyNBForceRelay = "NB_FORCE_RELAY"
+)
+
+func isForceRelayed() bool {
+	return strings.EqualFold(os.Getenv(EnvKeyNBForceRelay), "true")
+}

client/internal/peer/guard/ice_monitor.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pion/ice/v3"
+	"github.com/pion/ice/v4"
 	log "github.com/sirupsen/logrus"
 
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"

client/internal/peer/handshaker.go

@@ -43,13 +43,6 @@ type OfferAnswer struct {
 	SessionID *ICESessionID
 }
 
-func (oa *OfferAnswer) SessionIDString() string {
-	if oa.SessionID == nil {
-		return "unknown"
-	}
-	return oa.SessionID.String()
-}
-
 type Handshaker struct {
 	mu                  sync.Mutex
 	log                 *log.Entry
@@ -57,7 +50,7 @@ type Handshaker struct {
 	signaler            *Signaler
 	ice                 *WorkerICE
 	relay               *WorkerRelay
-	onNewOfferListeners []func(*OfferAnswer)
+	onNewOfferListeners []*OfferListener
 
 	// remoteOffersCh is a channel used to wait for remote credentials to proceed with the connection
 	remoteOffersCh chan OfferAnswer
@@ -78,7 +71,8 @@ func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *W
 }
 
 func (h *Handshaker) AddOnNewOfferListener(offer func(remoteOfferAnswer *OfferAnswer)) {
-	h.onNewOfferListeners = append(h.onNewOfferListeners, offer)
+	l := NewOfferListener(offer)
+	h.onNewOfferListeners = append(h.onNewOfferListeners, l)
 }
 
 func (h *Handshaker) Listen(ctx context.Context) {
@@ -91,13 +85,13 @@ func (h *Handshaker) Listen(ctx context.Context) {
 				continue
 			}
 			for _, listener := range h.onNewOfferListeners {
-				listener(&remoteOfferAnswer)
+				listener.Notify(&remoteOfferAnswer)
 			}
 			h.log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString())
 		case remoteOfferAnswer := <-h.remoteAnswerCh:
 			h.log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString())
 			for _, listener := range h.onNewOfferListeners {
-				listener(&remoteOfferAnswer)
+				listener.Notify(&remoteOfferAnswer)
 			}
 		case <-ctx.Done():
 			h.log.Infof("stop listening for remote offers and answers")

client/internal/peer/handshaker_listener.go

@@ -0,0 +1,62 @@
+package peer
+
+import (
+	"sync"
+)
+
+type callbackFunc func(remoteOfferAnswer *OfferAnswer)
+
+func (oa *OfferAnswer) SessionIDString() string {
+	if oa.SessionID == nil {
+		return "unknown"
+	}
+	return oa.SessionID.String()
+}
+
+type OfferListener struct {
+	fn      callbackFunc
+	running bool
+	latest  *OfferAnswer
+	mu      sync.Mutex
+}
+
+func NewOfferListener(fn callbackFunc) *OfferListener {
+	return &OfferListener{
+		fn: fn,
+	}
+}
+
+func (o *OfferListener) Notify(remoteOfferAnswer *OfferAnswer) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	// Store the latest offer
+	o.latest = remoteOfferAnswer
+
+	// If already running, the running goroutine will pick up this latest value
+	if o.running {
+		return
+	}
+
+	// Start processing
+	o.running = true
+
+	// Process in a goroutine to avoid blocking the caller
+	go func(remoteOfferAnswer *OfferAnswer) {
+		for {
+			o.fn(remoteOfferAnswer)
+
+			o.mu.Lock()
+			if o.latest == nil {
+				// No more work to do
+				o.running = false
+				o.mu.Unlock()
+				return
+			}
+			remoteOfferAnswer = o.latest
+			// Clear the latest to mark it as being processed
+			o.latest = nil
+			o.mu.Unlock()
+		}
+	}(remoteOfferAnswer)
+}

client/internal/peer/handshaker_listener_test.go

@@ -0,0 +1,39 @@
+package peer
+
+import (
+	"testing"
+	"time"
+)
+
+func Test_newOfferListener(t *testing.T) {
+	dummyOfferAnswer := &OfferAnswer{}
+	runChan := make(chan struct{}, 10)
+
+	longRunningFn := func(remoteOfferAnswer *OfferAnswer) {
+		time.Sleep(1 * time.Second)
+		runChan <- struct{}{}
+	}
+
+	hl := NewOfferListener(longRunningFn)
+
+	hl.Notify(dummyOfferAnswer)
+	hl.Notify(dummyOfferAnswer)
+	hl.Notify(dummyOfferAnswer)
+
+	// Wait for exactly 2 callbacks
+	for i := 0; i < 2; i++ {
+		select {
+		case <-runChan:
+		case <-time.After(3 * time.Second):
+			t.Fatal("Timeout waiting for callback")
+		}
+	}
+
+	// Verify no additional callbacks happen
+	select {
+	case <-runChan:
+		t.Fatal("Unexpected additional callback")
+	case <-time.After(100 * time.Millisecond):
+		t.Log("Correctly received exactly 2 callbacks")
+	}
+}

client/internal/peer/ice/StunTurn.go

@@ -3,7 +3,7 @@ package ice
 import (
 	"sync/atomic"
 
-	"github.com/pion/stun/v2"
+	"github.com/pion/stun/v3"
 )
 
 type StunTurn atomic.Value

client/internal/peer/ice/agent.go

@@ -4,7 +4,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pion/ice/v3"
+	"github.com/pion/ice/v4"
 	"github.com/pion/logging"
 	"github.com/pion/randutil"
 	log "github.com/sirupsen/logrus"

client/internal/peer/ice/config.go

@@ -1,7 +1,7 @@
 package ice
 
 import (
-	"github.com/pion/ice/v3"
+	"github.com/pion/ice/v4"
 )
 
 type Config struct {

client/internal/peer/signaler.go

@@ -1,7 +1,7 @@
 package peer
 
 import (
-	"github.com/pion/ice/v3"
+	"github.com/pion/ice/v4"
 	log "github.com/sirupsen/logrus"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 

client/internal/peer/wg_watcher.go

@@ -30,9 +30,10 @@ type WGWatcher struct {
 	peerKey       string
 	stateDump     *stateDump
 
-	ctx       context.Context
-	ctxCancel context.CancelFunc
-	ctxLock   sync.Mutex
+	ctx         context.Context
+	ctxCancel   context.CancelFunc
+	ctxLock     sync.Mutex
+	enabledTime time.Time
 }
 
 func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey string, stateDump *stateDump) *WGWatcher {
@@ -48,6 +49,7 @@ func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey strin
 func (w *WGWatcher) EnableWgWatcher(parentCtx context.Context, onDisconnectedFn func()) {
 	w.log.Debugf("enable WireGuard watcher")
 	w.ctxLock.Lock()
+	w.enabledTime = time.Now()
 
 	if w.ctx != nil && w.ctx.Err() == nil {
 		w.log.Errorf("WireGuard watcher already enabled")
@@ -101,6 +103,11 @@ func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, ctxCancel contex
 				onDisconnectedFn()
 				return
 			}
+			if lastHandshake.IsZero() {
+				elapsed := handshake.Sub(w.enabledTime).Seconds()
+				w.log.Infof("first wg handshake detected within: %.2fsec, (%s)", elapsed, handshake)
+			}
+
 			lastHandshake = *handshake
 
 			resetTime := time.Until(handshake.Add(checkPeriod))

client/internal/peer/worker_ice.go

@@ -8,12 +8,11 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pion/ice/v3"
-	"github.com/pion/stun/v2"
+	"github.com/pion/ice/v4"
 	log "github.com/sirupsen/logrus"
 
 	"github.com/netbirdio/netbird/client/iface"
-	"github.com/netbirdio/netbird/client/iface/bind"
+	"github.com/netbirdio/netbird/client/iface/udpmux"
 	"github.com/netbirdio/netbird/client/internal/peer/conntype"
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
 	"github.com/netbirdio/netbird/client/internal/stdnet"
@@ -55,10 +54,6 @@ type WorkerICE struct {
 	sessionID ICESessionID
 	muxAgent  sync.Mutex
 
-	StunTurn []*stun.URI
-
-	sentExtraSrflx bool
-
 	localUfrag string
 	localPwd   string
 
@@ -122,7 +117,6 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 			w.log.Warnf("failed to close ICE agent: %s", err)
 		}
 		w.agent = nil
-		// todo consider to switch to Relay connection while establishing a new ICE connection
 	}
 
 	var preferredCandidateTypes []ice.CandidateType
@@ -140,7 +134,6 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
 		w.muxAgent.Unlock()
 		return
 	}
-	w.sentExtraSrflx = false
 	w.agent = agent
 	w.agentDialerCancel = dialerCancel
 	w.agentConnecting = true
@@ -167,6 +160,21 @@ func (w *WorkerICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HA
 		w.log.Errorf("error while handling remote candidate")
 		return
 	}
+
+	if shouldAddExtraCandidate(candidate) {
+		// sends an extra server reflexive candidate to the remote peer with our related port (usually the wireguard port)
+		// this is useful when network has an existing port forwarding rule for the wireguard port and this peer
+		extraSrflx, err := extraSrflxCandidate(candidate)
+		if err != nil {
+			w.log.Errorf("failed creating extra server reflexive candidate %s", err)
+			return
+		}
+
+		if err := w.agent.AddRemoteCandidate(extraSrflx); err != nil {
+			w.log.Errorf("error while handling remote candidate")
+			return
+		}
+	}
 }
 
 func (w *WorkerICE) GetLocalUserCredentials() (frag string, pwd string) {
@@ -210,14 +218,12 @@ func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []
 		return nil, err
 	}
 
-	if err := agent.OnSelectedCandidatePairChange(w.onICESelectedCandidatePair); err != nil {
+	if err := agent.OnSelectedCandidatePairChange(func(c1, c2 ice.Candidate) {
+		w.onICESelectedCandidatePair(agent, c1, c2)
+	}); err != nil {
 		return nil, err
 	}
 
-	if err := agent.OnSuccessfulSelectedPairBindingResponse(w.onSuccessfulSelectedPairBindingResponse); err != nil {
-		return nil, fmt.Errorf("failed setting binding response callback: %w", err)
-	}
-
 	return agent, nil
 }
 
@@ -253,6 +259,11 @@ func (w *WorkerICE) connect(ctx context.Context, agent *icemaker.ThreadSafeAgent
 		w.closeAgent(agent, w.agentDialerCancel)
 		return
 	}
+	if pair == nil {
+		w.log.Warnf("selected candidate pair is nil, cannot proceed")
+		w.closeAgent(agent, w.agentDialerCancel)
+		return
+	}
 
 	if !isRelayCandidate(pair.Local) {
 		// dynamically set remote WireGuard port if other side specified a different one from the default one
@@ -327,7 +338,7 @@ func (w *WorkerICE) punchRemoteWGPort(pair *ice.CandidatePair, remoteWgPort int)
 		return
 	}
 
-	mux, ok := w.config.ICEConfig.UDPMuxSrflx.(*bind.UniversalUDPMuxDefault)
+	mux, ok := w.config.ICEConfig.UDPMuxSrflx.(*udpmux.UniversalUDPMuxDefault)
 	if !ok {
 		w.log.Warn("invalid udp mux conversion")
 		return
@@ -354,31 +365,23 @@ func (w *WorkerICE) onICECandidate(candidate ice.Candidate) {
 			w.log.Errorf("failed signaling candidate to the remote peer %s %s", w.config.Key, err)
 		}
 	}()
-
-	if !w.shouldSendExtraSrflxCandidate(candidate) {
-		return
-	}
-
-	// sends an extra server reflexive candidate to the remote peer with our related port (usually the wireguard port)
-	// this is useful when network has an existing port forwarding rule for the wireguard port and this peer
-	extraSrflx, err := extraSrflxCandidate(candidate)
-	if err != nil {
-		w.log.Errorf("failed creating extra server reflexive candidate %s", err)
-		return
-	}
-	w.sentExtraSrflx = true
-
-	go func() {
-		err = w.signaler.SignalICECandidate(extraSrflx, w.config.Key)
-		if err != nil {
-			w.log.Errorf("failed signaling the extra server reflexive candidate: %s", err)
-		}
-	}()
 }
 
-func (w *WorkerICE) onICESelectedCandidatePair(c1 ice.Candidate, c2 ice.Candidate) {
+func (w *WorkerICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent, c1, c2 ice.Candidate) {
 	w.log.Debugf("selected candidate pair [local <-> remote] -> [%s <-> %s], peer %s", c1.String(), c2.String(),
 		w.config.Key)
+
+	pairStat, ok := agent.GetSelectedCandidatePairStats()
+	if !ok {
+		w.log.Warnf("failed to get selected candidate pair stats")
+		return
+	}
+
+	duration := time.Duration(pairStat.CurrentRoundTripTime * float64(time.Second))
+	if err := w.statusRecorder.UpdateLatency(w.config.Key, duration); err != nil {
+		w.log.Debugf("failed to update latency for peer: %s", err)
+		return
+	}
 }
 
 func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dialerCancel context.CancelFunc) func(ice.ConnectionState) {
@@ -388,7 +391,10 @@ func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dia
 		case ice.ConnectionStateConnected:
 			w.lastKnownState = ice.ConnectionStateConnected
 			return
-		case ice.ConnectionStateFailed, ice.ConnectionStateDisconnected:
+		case ice.ConnectionStateFailed, ice.ConnectionStateDisconnected, ice.ConnectionStateClosed:
+			// ice.ConnectionStateClosed happens when we recreate the agent. For the P2P to TURN switch important to
+			// notify the conn.onICEStateDisconnected changes to update the current used priority
+
 			if w.lastKnownState == ice.ConnectionStateConnected {
 				w.lastKnownState = ice.ConnectionStateDisconnected
 				w.conn.onICEStateDisconnected()
@@ -400,32 +406,34 @@ func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dia
 	}
 }
 
-func (w *WorkerICE) onSuccessfulSelectedPairBindingResponse(pair *ice.CandidatePair) {
-	if err := w.statusRecorder.UpdateLatency(w.config.Key, pair.Latency()); err != nil {
-		w.log.Debugf("failed to update latency for peer: %s", err)
-		return
-	}
-}
-
-func (w *WorkerICE) shouldSendExtraSrflxCandidate(candidate ice.Candidate) bool {
-	if !w.sentExtraSrflx && candidate.Type() == ice.CandidateTypeServerReflexive && candidate.Port() != candidate.RelatedAddress().Port {
-		return true
-	}
-	return false
-}
-
 func (w *WorkerICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) (*ice.Conn, error) {
-	isControlling := w.config.LocalKey > w.config.Key
-	if isControlling {
-		return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
+	if isController(w.config) {
+		return w.agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
 	} else {
 		return agent.Accept(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
 	}
 }
 
+func shouldAddExtraCandidate(candidate ice.Candidate) bool {
+	if candidate.Type() != ice.CandidateTypeServerReflexive {
+		return false
+	}
+
+	if candidate.Port() == candidate.RelatedAddress().Port {
+		return false
+	}
+
+	// in the older version when we didn't set candidate ID extension the remote peer sent the extra candidates
+	// in newer version we generate locally the extra candidate
+	if _, ok := candidate.GetExtension(ice.ExtensionKeyCandidateID); !ok {
+		return false
+	}
+	return true
+}
+
 func extraSrflxCandidate(candidate ice.Candidate) (*ice.CandidateServerReflexive, error) {
 	relatedAdd := candidate.RelatedAddress()
-	return ice.NewCandidateServerReflexive(&ice.CandidateServerReflexiveConfig{
+	ec, err := ice.NewCandidateServerReflexive(&ice.CandidateServerReflexiveConfig{
 		Network:   candidate.NetworkType().String(),
 		Address:   candidate.Address(),
 		Port:      relatedAdd.Port,
@@ -433,6 +441,21 @@ func extraSrflxCandidate(candidate ice.Candidate) (*ice.CandidateServerReflexive
 		RelAddr:   relatedAdd.Address,
 		RelPort:   relatedAdd.Port,
 	})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, e := range candidate.Extensions() {
+		// overwrite the original candidate ID with the new one to avoid candidate duplication
+		if e.Key == ice.ExtensionKeyCandidateID {
+			e.Value = candidate.ID()
+		}
+		if err := ec.AddExtension(e); err != nil {
+			return nil, err
+		}
+	}
+
+	return ec, nil
 }
 
 func candidateViaRoutes(candidate ice.Candidate, clientRoutes route.HAMap) bool {