Merge branch 'main' into refactor/reducate-signaling

client/internal/peer/conn.go

@@ -28,10 +28,6 @@ import (
 	semaphoregroup "github.com/netbirdio/netbird/util/semaphore-group"
 )
 
-const (
-	defaultWgKeepAlive = 25 * time.Second
-)
-
 type ServiceDependencies struct {
 	StatusRecorder     *Status
 	Signaler           *Signaler
@@ -121,6 +117,8 @@ type Conn struct {
 
 	// debug purpose
 	dumpState *stateDump
+
+	endpointUpdater *EndpointUpdater
 }
 
 // NewConn creates a new not opened Conn to the remote peer.
@@ -133,17 +131,18 @@ func NewConn(config ConnConfig, services ServiceDependencies) (*Conn, error) {
 	connLog := log.WithField("peer", config.Key)
 
 	var conn = &Conn{
-		Log:            connLog,
-		config:         config,
-		statusRecorder: services.StatusRecorder,
-		signaler:       services.Signaler,
-		iFaceDiscover:  services.IFaceDiscover,
-		relayManager:   services.RelayManager,
-		srWatcher:      services.SrWatcher,
-		semaphore:      services.Semaphore,
-		statusRelay:    worker.NewAtomicStatus(),
-		statusICE:      worker.NewAtomicStatus(),
-		dumpState:      newStateDump(config.Key, connLog, services.StatusRecorder),
+		Log:             connLog,
+		config:          config,
+		statusRecorder:  services.StatusRecorder,
+		signaler:        services.Signaler,
+		iFaceDiscover:   services.IFaceDiscover,
+		relayManager:    services.RelayManager,
+		srWatcher:       services.SrWatcher,
+		semaphore:       services.Semaphore,
+		statusRelay:     worker.NewAtomicStatus(),
+		statusICE:       worker.NewAtomicStatus(),
+		dumpState:       newStateDump(config.Key, connLog, services.StatusRecorder),
+		endpointUpdater: NewEndpointUpdater(connLog, config.WgConfig, isController(config)),
 	}
 
 	return conn, nil
@@ -273,7 +272,7 @@ func (conn *Conn) Close(signalToRemote bool) {
 		conn.wgProxyICE = nil
 	}
 
-	if err := conn.removeWgPeer(); err != nil {
+	if err := conn.endpointUpdater.RemoveWgPeer(); err != nil {
 		conn.Log.Errorf("failed to remove wg endpoint: %v", err)
 	}
 
@@ -399,12 +398,19 @@ func (conn *Conn) onICEConnectionIsReady(priority conntype.ConnPriority, iceConn
 		wgProxy.Work()
 	}
 
-	if err = conn.configureWGEndpoint(ep, iceConnInfo.RosenpassPubKey); err != nil {
+	conn.Log.Infof("configure WireGuard endpoint to: %s", ep.String())
+	presharedKey := conn.presharedKey(iceConnInfo.RosenpassPubKey)
+	if err = conn.endpointUpdater.ConfigureWGEndpoint(ep, presharedKey); err != nil {
 		conn.handleConfigurationFailure(err, wgProxy)
 		return
 	}
 	wgConfigWorkaround()
 
+	if conn.wgProxyRelay != nil {
+		conn.Log.Debugf("redirect packets from relayed conn to WireGuard")
+		conn.wgProxyRelay.RedirectAs(ep)
+	}
+
 	conn.currentConnPriority = priority
 	conn.statusICE.SetConnected()
 	conn.updateIceState(iceConnInfo)
@@ -433,7 +439,8 @@ func (conn *Conn) onICEStateDisconnected() {
 		conn.dumpState.SwitchToRelay()
 		conn.wgProxyRelay.Work()
 
-		if err := conn.configureWGEndpoint(conn.wgProxyRelay.EndpointAddr(), conn.rosenpassRemoteKey); err != nil {
+		presharedKey := conn.presharedKey(conn.rosenpassRemoteKey)
+		if err := conn.endpointUpdater.ConfigureWGEndpoint(conn.wgProxyRelay.EndpointAddr(), presharedKey); err != nil {
 			conn.Log.Errorf("failed to switch to relay conn: %v", err)
 		}
 
@@ -442,6 +449,7 @@ func (conn *Conn) onICEStateDisconnected() {
 			defer conn.wgWatcherWg.Done()
 			conn.workerRelay.EnableWgWatcher(conn.ctx)
 		}()
+		conn.wgProxyRelay.Work()
 		conn.currentConnPriority = conntype.Relay
 	} else {
 		conn.Log.Infof("ICE disconnected, do not switch to Relay. Reset priority to: %s", conntype.None.String())
@@ -501,7 +509,8 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
 	}
 
 	wgProxy.Work()
-	if err := conn.configureWGEndpoint(wgProxy.EndpointAddr(), rci.rosenpassPubKey); err != nil {
+	presharedKey := conn.presharedKey(rci.rosenpassPubKey)
+	if err := conn.endpointUpdater.ConfigureWGEndpoint(wgProxy.EndpointAddr(), presharedKey); err != nil {
 		if err := wgProxy.CloseConn(); err != nil {
 			conn.Log.Warnf("Failed to close relay connection: %v", err)
 		}
@@ -579,17 +588,6 @@ func (conn *Conn) onGuardEvent() {
 	}
 }
 
-func (conn *Conn) configureWGEndpoint(addr *net.UDPAddr, remoteRPKey []byte) error {
-	presharedKey := conn.presharedKey(remoteRPKey)
-	return conn.config.WgConfig.WgInterface.UpdatePeer(
-		conn.config.WgConfig.RemoteKey,
-		conn.config.WgConfig.AllowedIps,
-		defaultWgKeepAlive,
-		addr,
-		presharedKey,
-	)
-}
-
 func (conn *Conn) updateRelayStatus(relayServerAddr string, rosenpassPubKey []byte) {
 	peerState := State{
 		PubKey:             conn.config.Key,
@@ -732,10 +730,6 @@ func (conn *Conn) isICEActive() bool {
 	return (conn.currentConnPriority == conntype.ICEP2P || conn.currentConnPriority == conntype.ICETurn) && conn.statusICE.Get() == worker.StatusConnected
 }
 
-func (conn *Conn) removeWgPeer() error {
-	return conn.config.WgConfig.WgInterface.RemovePeer(conn.config.WgConfig.RemoteKey)
-}
-
 func (conn *Conn) handleConfigurationFailure(err error, wgProxy wgproxy.Proxy) {
 	conn.Log.Warnf("Failed to update wg peer configuration: %v", err)
 	if wgProxy != nil {