Handle peer interface config change (#348)

Before this change, NetBird Agent wasn't handling
peer interface configuration changes dynamically.
Also, remote peer configuration changes have
not been applied (e.g. AllowedIPs changed).
Not a very common cause, but still it should be handled.
Now, Agent reacts to PeerConfig changes sent from the
management service and restarts remote connections
if AllowedIps have been changed.

client/internal/engine.go

@@ -78,7 +78,7 @@ type Engine struct {
 
 	ctx context.Context
 
-	wgInterface iface.WGIface
+	wgInterface *iface.WGIface
 
 	udpMux          ice.UDPMux
 	udpMuxSrflx     ice.UniversalUDPMux
@@ -177,7 +177,7 @@ func (e *Engine) Start() error {
 	myPrivateKey := e.config.WgPrivateKey
 	var err error
 
-	e.wgInterface, err = iface.NewWGIface(wgIfaceName, wgAddr, iface.DefaultMTU)
+	e.wgInterface, err = iface.NewWGIFace(wgIfaceName, wgAddr, iface.DefaultMTU)
 	if err != nil {
 		log.Errorf("failed creating wireguard interface instance %s: [%s]", wgIfaceName, err.Error())
 		return err
@@ -216,7 +216,39 @@ func (e *Engine) Start() error {
 	return nil
 }
 
-// removePeers finds and removes peers that do not exist anymore in the network map received from the Management Service
+// modifyPeers updates peers that have been modified (e.g. IP address has been changed).
+// It closes the existing connection, removes it from the peerConns map, and creates a new one.
+func (e *Engine) modifyPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
+
+	// first, check if peers have been modified
+	var modified []*mgmProto.RemotePeerConfig
+	for _, p := range peersUpdate {
+		if peerConn, ok := e.peerConns[p.GetWgPubKey()]; ok {
+			if peerConn.GetConf().ProxyConfig.AllowedIps != strings.Join(p.AllowedIps, ",") {
+				modified = append(modified, p)
+			}
+		}
+	}
+
+	// second, close all modified connections and remove them from the state map
+	for _, p := range modified {
+		err := e.removePeer(p.GetWgPubKey())
+		if err != nil {
+			return err
+		}
+	}
+	// third, add the peer connections again
+	for _, p := range modified {
+		err := e.addNewPeer(p)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// removePeers finds and removes peers that do not exist anymore in the network map received from the Management Service.
+// It also removes peers that have been modified (e.g. change of IP address). They will be added again in addPeers method.
 func (e *Engine) removePeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
 	currentPeers := make([]string, 0, len(e.peerConns))
 	for p := range e.peerConns {
@@ -366,6 +398,12 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
 	}
 
 	if update.GetNetworkMap() != nil {
+		if update.GetNetworkMap().GetPeerConfig() != nil {
+			err := e.updateConfig(update.GetNetworkMap().GetPeerConfig())
+			if err != nil {
+				return err
+			}
+		}
 		// only apply new changes and ignore old ones
 		err := e.updateNetworkMap(update.GetNetworkMap())
 		if err != nil {
@@ -376,6 +414,20 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
 	return nil
 }
 
+func (e *Engine) updateConfig(conf *mgmProto.PeerConfig) error {
+	if e.wgInterface.Address.String() != conf.Address {
+		oldAddr := e.wgInterface.Address.String()
+		log.Debugf("updating peer address from %s to %s", oldAddr, conf.Address)
+		err := e.wgInterface.UpdateAddr(conf.Address)
+		if err != nil {
+			return err
+		}
+		log.Infof("updated peer address from %s to %s", oldAddr, conf.Address)
+	}
+
+	return nil
+}
+
 // receiveManagementEvents connects to the Management Service event stream to receive updates from the management service
 // E.g. when a new peer has been registered and we are allowed to connect to it.
 func (e *Engine) receiveManagementEvents() {
@@ -454,6 +506,11 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {
 			return err
 		}
 
+		err = e.modifyPeers(networkMap.GetRemotePeers())
+		if err != nil {
+			return err
+		}
+
 		err = e.addNewPeers(networkMap.GetRemotePeers())
 		if err != nil {
 			return err
@@ -464,21 +521,29 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap) error {
 	return nil
 }
 
-// addNewPeers finds and adds peers that were not know before but arrived from the Management service with the update
+// addNewPeers adds peers that were not know before but arrived from the Management service with the update
 func (e *Engine) addNewPeers(peersUpdate []*mgmProto.RemotePeerConfig) error {
 	for _, p := range peersUpdate {
-		peerKey := p.GetWgPubKey()
-		peerIPs := p.GetAllowedIps()
-		if _, ok := e.peerConns[peerKey]; !ok {
-			conn, err := e.createPeerConn(peerKey, strings.Join(peerIPs, ","))
-			if err != nil {
-				return err
-			}
-			e.peerConns[peerKey] = conn
-
-			go e.connWorker(conn, peerKey)
+		err := e.addNewPeer(p)
+		if err != nil {
+			return err
 		}
+	}
+	return nil
+}
 
+// addNewPeer add peer if connection doesn't exist
+func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig) error {
+	peerKey := peerConfig.GetWgPubKey()
+	peerIPs := peerConfig.GetAllowedIps()
+	if _, ok := e.peerConns[peerKey]; !ok {
+		conn, err := e.createPeerConn(peerKey, strings.Join(peerIPs, ","))
+		if err != nil {
+			return err
+		}
+		e.peerConns[peerKey] = conn
+
+		go e.connWorker(conn, peerKey)
 	}
 	return nil
 }
@@ -505,6 +570,12 @@ func (e Engine) connWorker(conn *peer.Conn, peerKey string) {
 		err := conn.Open()
 		if err != nil {
 			log.Debugf("connection to peer %s failed: %v", peerKey, err)
+			switch err.(type) {
+			case *peer.ConnectionClosedError:
+				// conn has been forced to close, so we exit the loop
+				return
+			default:
+			}
 		}
 	}
 }

client/internal/engine_test.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -62,7 +63,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 		networkMap *mgmtProto.NetworkMap
 
 		expectedLen    int
-		expectedPeers  []string
+		expectedPeers  []*mgmtProto.RemotePeerConfig
 		expectedSerial uint64
 	}
 
@@ -81,6 +82,11 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 		AllowedIps: []string{"100.64.0.12/24"},
 	}
 
+	modifiedPeer3 := &mgmtProto.RemotePeerConfig{
+		WgPubKey:   "GGHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=",
+		AllowedIps: []string{"100.64.0.20/24"},
+	}
+
 	case1 := testCase{
 		name: "input with a new peer to add",
 		networkMap: &mgmtProto.NetworkMap{
@@ -92,7 +98,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 			RemotePeersIsEmpty: false,
 		},
 		expectedLen:    1,
-		expectedPeers:  []string{peer1.GetWgPubKey()},
+		expectedPeers:  []*mgmtProto.RemotePeerConfig{peer1},
 		expectedSerial: 1,
 	}
 
@@ -108,7 +114,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 			RemotePeersIsEmpty: false,
 		},
 		expectedLen:    2,
-		expectedPeers:  []string{peer1.GetWgPubKey(), peer2.GetWgPubKey()},
+		expectedPeers:  []*mgmtProto.RemotePeerConfig{peer1, peer2},
 		expectedSerial: 2,
 	}
 
@@ -123,7 +129,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 			RemotePeersIsEmpty: false,
 		},
 		expectedLen:    2,
-		expectedPeers:  []string{peer1.GetWgPubKey(), peer2.GetWgPubKey()},
+		expectedPeers:  []*mgmtProto.RemotePeerConfig{peer1, peer2},
 		expectedSerial: 2,
 	}
 
@@ -138,11 +144,26 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 			RemotePeersIsEmpty: false,
 		},
 		expectedLen:    2,
-		expectedPeers:  []string{peer2.GetWgPubKey(), peer3.GetWgPubKey()},
+		expectedPeers:  []*mgmtProto.RemotePeerConfig{peer2, peer3},
 		expectedSerial: 4,
 	}
 
 	case5 := testCase{
+		name: "input with one peer to modify",
+		networkMap: &mgmtProto.NetworkMap{
+			Serial:     4,
+			PeerConfig: nil,
+			RemotePeers: []*mgmtProto.RemotePeerConfig{
+				modifiedPeer3, peer2,
+			},
+			RemotePeersIsEmpty: false,
+		},
+		expectedLen:    2,
+		expectedPeers:  []*mgmtProto.RemotePeerConfig{peer2, modifiedPeer3},
+		expectedSerial: 4,
+	}
+
+	case6 := testCase{
 		name: "input with all peers to remove",
 		networkMap: &mgmtProto.NetworkMap{
 			Serial:             5,
@@ -155,7 +176,7 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 		expectedSerial: 5,
 	}
 
-	for _, c := range []testCase{case1, case2, case3, case4, case5} {
+	for _, c := range []testCase{case1, case2, case3, case4, case5, case6} {
 		t.Run(c.name, func(t *testing.T) {
 			err = engine.updateNetworkMap(c.networkMap)
 			if err != nil {
@@ -172,9 +193,15 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
 			}
 
 			for _, p := range c.expectedPeers {
-				if _, ok := engine.peerConns[p]; !ok {
+				conn, ok := engine.peerConns[p.GetWgPubKey()]
+				if !ok {
 					t.Errorf("expecting Engine.peerConns to contain peer %s", p)
 				}
+				expectedAllowedIPs := strings.Join(p.AllowedIps, ",")
+				if conn.GetConf().ProxyConfig.AllowedIps != expectedAllowedIPs {
+					t.Errorf("expecting peer %s to have AllowedIPs= %s, got %s", p.GetWgPubKey(),
+						expectedAllowedIPs, conn.GetConf().ProxyConfig.AllowedIps)
+				}
 			}
 		})
 	}

client/internal/peer/conn.go

@@ -66,6 +66,11 @@ type Conn struct {
 	proxy proxy.Proxy
 }
 
+// GetConf returns the connection config
+func (conn *Conn) GetConf() ConnConfig {
+	return conn.config
+}
+
 // NewConn creates a new not opened Conn to the remote peer.
 // To establish a connection run Conn.Open
 func NewConn(config ConnConfig) (*Conn, error) {
@@ -437,7 +442,7 @@ func (conn *Conn) Close() error {
 		// before conn.Open() another update from management arrives with peers: [1,2,3,4,5]
 		// engine adds a new Conn for 4 and 5
 		// therefore peer 4 has 2 Conn objects
-		log.Warnf("closing not started coonection %s", conn.config.Key)
+		log.Warnf("connection has been already closed or attempted closing not started coonection %s", conn.config.Key)
 		return NewConnectionAlreadyClosed(conn.config.Key)
 	}
 }

client/internal/proxy/proxy.go

@@ -21,7 +21,7 @@ const (
 type Config struct {
 	WgListenAddr string
 	RemoteKey    string
-	WgInterface  iface.WGIface
+	WgInterface  *iface.WGIface
 	AllowedIps   string
 	PreSharedKey *wgtypes.Key
 }