[client] Feature/lazy connection (#3379)

With the lazy connection feature, the peer will connect to target peers on-demand. The trigger can be any IP traffic.

This feature can be enabled with the NB_ENABLE_EXPERIMENTAL_LAZY_CONN environment variable.

When the engine receives a network map, it binds a free UDP port for every remote peer, and the system configures WireGuard endpoints for these ports. When traffic appears on a UDP socket, the system removes this listener and starts the peer connection procedure immediately.

Key changes
Fix slow netbird status -d command
Move from engine.go file to conn_mgr.go the peer connection related code
Refactor the iface interface usage and moved interface file next to the engine code
Add new command line flag and UI option to enable feature
The peer.Conn struct is reusable after it has been closed.
Change connection states
Connection states
Idle: The peer is not attempting to establish a connection. This typically means it's in a lazy state or the remote peer is expired.

Connecting: The peer is actively trying to establish a connection. This occurs when the peer has entered an active state and is continuously attempting to reach the remote peer.

Connected: A successful peer-to-peer connection has been established and communication is active.

Author: pappz

client/cmd/root.go

@@ -26,22 +26,23 @@ import (
 )
 
 const (
-	externalIPMapFlag       = "external-ip-map"
-	dnsResolverAddress      = "dns-resolver-address"
-	enableRosenpassFlag     = "enable-rosenpass"
-	rosenpassPermissiveFlag = "rosenpass-permissive"
-	preSharedKeyFlag        = "preshared-key"
-	interfaceNameFlag       = "interface-name"
-	wireguardPortFlag       = "wireguard-port"
-	networkMonitorFlag      = "network-monitor"
-	disableAutoConnectFlag  = "disable-auto-connect"
-	serverSSHAllowedFlag    = "allow-server-ssh"
-	extraIFaceBlackListFlag = "extra-iface-blacklist"
-	dnsRouteIntervalFlag    = "dns-router-interval"
-	systemInfoFlag          = "system-info"
-	blockLANAccessFlag      = "block-lan-access"
-	uploadBundle            = "upload-bundle"
-	uploadBundleURL         = "upload-bundle-url"
+	externalIPMapFlag        = "external-ip-map"
+	dnsResolverAddress       = "dns-resolver-address"
+	enableRosenpassFlag      = "enable-rosenpass"
+	rosenpassPermissiveFlag  = "rosenpass-permissive"
+	preSharedKeyFlag         = "preshared-key"
+	interfaceNameFlag        = "interface-name"
+	wireguardPortFlag        = "wireguard-port"
+	networkMonitorFlag       = "network-monitor"
+	disableAutoConnectFlag   = "disable-auto-connect"
+	serverSSHAllowedFlag     = "allow-server-ssh"
+	extraIFaceBlackListFlag  = "extra-iface-blacklist"
+	dnsRouteIntervalFlag     = "dns-router-interval"
+	systemInfoFlag           = "system-info"
+	blockLANAccessFlag       = "block-lan-access"
+	enableLazyConnectionFlag = "enable-lazy-connection"
+	uploadBundle             = "upload-bundle"
+	uploadBundleURL          = "upload-bundle-url"
 )
 
 var (
@@ -80,6 +81,7 @@ var (
 	blockLANAccess          bool
 	debugUploadBundle       bool
 	debugUploadBundleURL    string
+	lazyConnEnabled         bool
 
 	rootCmd = &cobra.Command{
 		Use:          "netbird",
@@ -184,6 +186,7 @@ func init() {
 	upCmd.PersistentFlags().BoolVar(&rosenpassPermissive, rosenpassPermissiveFlag, false, "[Experimental] Enable Rosenpass in permissive mode to allow this peer to accept WireGuard connections without requiring Rosenpass functionality from peers that do not have Rosenpass enabled.")
 	upCmd.PersistentFlags().BoolVar(&serverSSHAllowed, serverSSHAllowedFlag, false, "Allow SSH server on peer. If enabled, the SSH server will be permitted")
 	upCmd.PersistentFlags().BoolVar(&autoConnectDisabled, disableAutoConnectFlag, false, "Disables auto-connect feature. If enabled, then the client won't connect automatically when the service starts.")
+	upCmd.PersistentFlags().BoolVar(&lazyConnEnabled, enableLazyConnectionFlag, false, "[Experimental] Enable the lazy connection feature. If enabled, the client will establish connections on-demand.")
 
 	debugCmd.PersistentFlags().BoolVarP(&debugSystemInfoFlag, systemInfoFlag, "S", true, "Adds system information to the debug bundle")
 	debugCmd.PersistentFlags().BoolVarP(&debugUploadBundle, uploadBundle, "U", false, fmt.Sprintf("Uploads the debug bundle to a server from URL defined by %s", uploadBundleURL))

client/cmd/status.go

@@ -44,7 +44,7 @@ func init() {
 	statusCmd.MarkFlagsMutuallyExclusive("detail", "json", "yaml", "ipv4")
 	statusCmd.PersistentFlags().StringSliceVar(&ipsFilter, "filter-by-ips", []string{}, "filters the detailed output by a list of one or more IPs, e.g., --filter-by-ips 100.64.0.100,100.64.0.200")
 	statusCmd.PersistentFlags().StringSliceVar(&prefixNamesFilter, "filter-by-names", []string{}, "filters the detailed output by a list of one or more peer FQDN or hostnames, e.g., --filter-by-names peer-a,peer-b.netbird.cloud")
-	statusCmd.PersistentFlags().StringVar(&statusFilter, "filter-by-status", "", "filters the detailed output by connection status(connected|disconnected), e.g., --filter-by-status connected")
+	statusCmd.PersistentFlags().StringVar(&statusFilter, "filter-by-status", "", "filters the detailed output by connection status(idle|connecting|connected), e.g., --filter-by-status connected")
 }
 
 func statusFunc(cmd *cobra.Command, args []string) error {
@@ -127,12 +127,12 @@ func getStatus(ctx context.Context) (*proto.StatusResponse, error) {
 
 func parseFilters() error {
 	switch strings.ToLower(statusFilter) {
-	case "", "disconnected", "connected":
+	case "", "idle", "connecting", "connected":
 		if strings.ToLower(statusFilter) != "" {
 			enableDetailFlagWhenFilterFlag()
 		}
 	default:
-		return fmt.Errorf("wrong status filter, should be one of connected|disconnected, got: %s", statusFilter)
+		return fmt.Errorf("wrong status filter, should be one of connected|connecting|idle, got: %s", statusFilter)
 	}
 
 	if len(ipsFilter) > 0 {

client/cmd/up.go

@@ -194,6 +194,10 @@ func runInForegroundMode(ctx context.Context, cmd *cobra.Command) error {
 		ic.BlockLANAccess = &blockLANAccess
 	}
 
+	if cmd.Flag(enableLazyConnectionFlag).Changed {
+		ic.LazyConnectionEnabled = &lazyConnEnabled
+	}
+
 	providedSetupKey, err := getSetupKey()
 	if err != nil {
 		return err
@@ -332,6 +336,10 @@ func runInDaemonMode(ctx context.Context, cmd *cobra.Command) error {
 		loginRequest.BlockLanAccess = &blockLANAccess
 	}
 
+	if cmd.Flag(enableLazyConnectionFlag).Changed {
+		loginRequest.LazyConnectionEnabled = &lazyConnEnabled
+	}
+
 	var loginErr error
 
 	var loginResp *proto.LoginResponse

client/iface/configurer/kernel_unix.go

@@ -201,14 +201,30 @@ func (c *KernelConfigurer) configure(config wgtypes.Config) error {
 func (c *KernelConfigurer) Close() {
 }
 
-func (c *KernelConfigurer) GetStats(peerKey string) (WGStats, error) {
-	peer, err := c.getPeer(c.deviceName, peerKey)
+func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) {
+	stats := make(map[string]WGStats)
+	wg, err := wgctrl.New()
+	if err != nil {
+		return nil, fmt.Errorf("wgctl: %w", err)
+	}
+	defer func() {
+		err = wg.Close()
+		if err != nil {
+			log.Errorf("Got error while closing wgctl: %v", err)
+		}
+	}()
+
+	wgDevice, err := wg.Device(c.deviceName)
 	if err != nil {
-		return WGStats{}, fmt.Errorf("get wireguard stats: %w", err)
+		return nil, fmt.Errorf("get device %s: %w", c.deviceName, err)
+	}
+
+	for _, peer := range wgDevice.Peers {
+		stats[peer.PublicKey.String()] = WGStats{
+			LastHandshake: peer.LastHandshakeTime,
+			TxBytes:       peer.TransmitBytes,
+			RxBytes:       peer.ReceiveBytes,
+		}
 	}
-	return WGStats{
-		LastHandshake: peer.LastHandshakeTime,
-		TxBytes:       peer.TransmitBytes,
-		RxBytes:       peer.ReceiveBytes,
-	}, nil
+	return stats, nil
 }

client/iface/configurer/usp.go

@@ -1,6 +1,7 @@
 package configurer
 
 import (
+	"encoding/base64"
 	"encoding/hex"
 	"fmt"
 	"net"
@@ -17,6 +18,13 @@ import (
 	nbnet "github.com/netbirdio/netbird/util/net"
 )
 
+const (
+	ipcKeyLastHandshakeTimeSec  = "last_handshake_time_sec"
+	ipcKeyLastHandshakeTimeNsec = "last_handshake_time_nsec"
+	ipcKeyTxBytes               = "tx_bytes"
+	ipcKeyRxBytes               = "rx_bytes"
+)
+
 var ErrAllowedIPNotFound = fmt.Errorf("allowed IP not found")
 
 type WGUSPConfigurer struct {
@@ -217,91 +225,75 @@ func (t *WGUSPConfigurer) Close() {
 	}
 }
 
-func (t *WGUSPConfigurer) GetStats(peerKey string) (WGStats, error) {
+func (t *WGUSPConfigurer) GetStats() (map[string]WGStats, error) {
 	ipc, err := t.device.IpcGet()
 	if err != nil {
-		return WGStats{}, fmt.Errorf("ipc get: %w", err)
-	}
-
-	stats, err := findPeerInfo(ipc, peerKey, []string{
-		"last_handshake_time_sec",
-		"last_handshake_time_nsec",
-		"tx_bytes",
-		"rx_bytes",
-	})
-	if err != nil {
-		return WGStats{}, fmt.Errorf("find peer info: %w", err)
-	}
-
-	sec, err := strconv.ParseInt(stats["last_handshake_time_sec"], 10, 64)
-	if err != nil {
-		return WGStats{}, fmt.Errorf("parse handshake sec: %w", err)
-	}
-	nsec, err := strconv.ParseInt(stats["last_handshake_time_nsec"], 10, 64)
-	if err != nil {
-		return WGStats{}, fmt.Errorf("parse handshake nsec: %w", err)
-	}
-	txBytes, err := strconv.ParseInt(stats["tx_bytes"], 10, 64)
-	if err != nil {
-		return WGStats{}, fmt.Errorf("parse tx_bytes: %w", err)
-	}
-	rxBytes, err := strconv.ParseInt(stats["rx_bytes"], 10, 64)
-	if err != nil {
-		return WGStats{}, fmt.Errorf("parse rx_bytes: %w", err)
+		return nil, fmt.Errorf("ipc get: %w", err)
 	}
 
-	return WGStats{
-		LastHandshake: time.Unix(sec, nsec),
-		TxBytes:       txBytes,
-		RxBytes:       rxBytes,
-	}, nil
+	return parseTransfers(ipc)
 }
 
-func findPeerInfo(ipcInput string, peerKey string, searchConfigKeys []string) (map[string]string, error) {
-	peerKeyParsed, err := wgtypes.ParseKey(peerKey)
-	if err != nil {
-		return nil, fmt.Errorf("parse key: %w", err)
-	}
-
-	hexKey := hex.EncodeToString(peerKeyParsed[:])
-
-	lines := strings.Split(ipcInput, "\n")
-
-	configFound := map[string]string{}
-	foundPeer := false
+func parseTransfers(ipc string) (map[string]WGStats, error) {
+	stats := make(map[string]WGStats)
+	var (
+		currentKey   string
+		currentStats WGStats
+		hasPeer      bool
+	)
+	lines := strings.Split(ipc, "\n")
 	for _, line := range lines {
 		line = strings.TrimSpace(line)
 
 		// If we're within the details of the found peer and encounter another public key,
 		// this means we're starting another peer's details. So, stop.
-		if strings.HasPrefix(line, "public_key=") && foundPeer {
-			break
+		if strings.HasPrefix(line, "public_key=") {
+			peerID := strings.TrimPrefix(line, "public_key=")
+			h, err := hex.DecodeString(peerID)
+			if err != nil {
+				return nil, fmt.Errorf("decode peerID: %w", err)
+			}
+			currentKey = base64.StdEncoding.EncodeToString(h)
+			currentStats = WGStats{} // Reset stats for the new peer
+			hasPeer = true
+			stats[currentKey] = currentStats
+			continue
 		}
 
-		// Identify the peer with the specific public key
-		if line == fmt.Sprintf("public_key=%s", hexKey) {
-			foundPeer = true
+		if !hasPeer {
+			continue
 		}
 
-		for _, key := range searchConfigKeys {
-			if foundPeer && strings.HasPrefix(line, key+"=") {
-				v := strings.SplitN(line, "=", 2)
-				configFound[v[0]] = v[1]
-			}
+		key := strings.SplitN(line, "=", 2)
+		if len(key) != 2 {
+			continue
 		}
-	}
-
-	// todo: use multierr
-	for _, key := range searchConfigKeys {
-		if _, ok := configFound[key]; !ok {
-			return configFound, fmt.Errorf("config key not found: %s", key)
+		switch key[0] {
+		case ipcKeyLastHandshakeTimeSec:
+			hs, err := toLastHandshake(key[1])
+			if err != nil {
+				return nil, err
+			}
+			currentStats.LastHandshake = hs
+			stats[currentKey] = currentStats
+		case ipcKeyRxBytes:
+			rxBytes, err := toBytes(key[1])
+			if err != nil {
+				return nil, fmt.Errorf("parse rx_bytes: %w", err)
+			}
+			currentStats.RxBytes = rxBytes
+			stats[currentKey] = currentStats
+		case ipcKeyTxBytes:
+			TxBytes, err := toBytes(key[1])
+			if err != nil {
+				return nil, fmt.Errorf("parse tx_bytes: %w", err)
+			}
+			currentStats.TxBytes = TxBytes
+			stats[currentKey] = currentStats
 		}
 	}
-	if !foundPeer {
-		return nil, fmt.Errorf("%w: %s", ErrPeerNotFound, peerKey)
-	}
 
-	return configFound, nil
+	return stats, nil
 }
 
 func toWgUserspaceString(wgCfg wgtypes.Config) string {
@@ -355,6 +347,18 @@ func toWgUserspaceString(wgCfg wgtypes.Config) string {
 	return sb.String()
 }
 
+func toLastHandshake(stringVar string) (time.Time, error) {
+	sec, err := strconv.ParseInt(stringVar, 10, 64)
+	if err != nil {
+		return time.Time{}, fmt.Errorf("parse handshake sec: %w", err)
+	}
+	return time.Unix(sec, 0), nil
+}
+
+func toBytes(s string) (int64, error) {
+	return strconv.ParseInt(s, 10, 64)
+}
+
 func getFwmark() int {
 	if nbnet.AdvancedRouting() {
 		return nbnet.ControlPlaneMark