Merge branch 'main' into apply-routes-early

Because the code generated new channel for every single event, was easy to miss notification.
Use single channel.

client/anonymize/anonymize.go

@@ -69,6 +69,22 @@ func (a *Anonymizer) AnonymizeIP(ip netip.Addr) netip.Addr {
 	return a.ipAnonymizer[ip]
 }
 
+func (a *Anonymizer) AnonymizeUDPAddr(addr net.UDPAddr) net.UDPAddr {
+	// Convert IP to netip.Addr
+	ip, ok := netip.AddrFromSlice(addr.IP)
+	if !ok {
+		return addr
+	}
+
+	anonIP := a.AnonymizeIP(ip)
+
+	return net.UDPAddr{
+		IP:   anonIP.AsSlice(),
+		Port: addr.Port,
+		Zone: addr.Zone,
+	}
+}
+
 // isInAnonymizedRange checks if an IP is within the range of already assigned anonymized IPs
 func (a *Anonymizer) isInAnonymizedRange(ip netip.Addr) bool {
 	if ip.Is4() && ip.Compare(a.startAnonIPv4) >= 0 && ip.Compare(a.currentAnonIPv4) <= 0 {

client/firewall/uspfilter/forwarder/tcp.go

@@ -111,12 +111,12 @@ func (f *Forwarder) proxyTCP(id stack.TransportEndpointID, inConn *gonet.TCPConn
 
 	if errInToOut != nil {
 		if !isClosedError(errInToOut) {
-			f.logger.Error("proxyTCP: copy error (in -> out): %v", errInToOut)
+			f.logger.Error("proxyTCP: copy error (in -> out) for %s: %v", epID(id), errInToOut)
 		}
 	}
 	if errOutToIn != nil {
 		if !isClosedError(errOutToIn) {
-			f.logger.Error("proxyTCP: copy error (out -> in): %v", errOutToIn)
+			f.logger.Error("proxyTCP: copy error (out -> in) for %s: %v", epID(id), errOutToIn)
 		}
 	}
 

client/firewall/uspfilter/forwarder/udp.go

@@ -250,10 +250,10 @@ func (f *Forwarder) proxyUDP(ctx context.Context, pConn *udpPacketConn, id stack
 	wg.Wait()
 
 	if outboundErr != nil && !isClosedError(outboundErr) {
-		f.logger.Error("proxyUDP: copy error (outbound->inbound): %v", outboundErr)
+		f.logger.Error("proxyUDP: copy error (outbound->inbound) for %s: %v", epID(id), outboundErr)
 	}
 	if inboundErr != nil && !isClosedError(inboundErr) {
-		f.logger.Error("proxyUDP: copy error (inbound->outbound): %v", inboundErr)
+		f.logger.Error("proxyUDP: copy error (inbound->outbound) for %s: %v", epID(id), inboundErr)
 	}
 
 	var rxPackets, txPackets uint64

client/firewall/uspfilter/uspfilter.go

@@ -39,8 +39,12 @@ const (
 	// EnvForceUserspaceRouter forces userspace routing even if native routing is available.
 	EnvForceUserspaceRouter = "NB_FORCE_USERSPACE_ROUTER"
 
-	// EnvEnableNetstackLocalForwarding enables forwarding of local traffic to the native stack when running netstack
-	// Leaving this on by default introduces a security risk as sockets on listening on localhost only will be accessible
+	// EnvEnableLocalForwarding enables forwarding of local traffic to the native stack for internal (non-NetBird) interfaces.
+	// Default off as it might be security risk because sockets listening on localhost only will become accessible.
+	EnvEnableLocalForwarding = "NB_ENABLE_LOCAL_FORWARDING"
+
+	// EnvEnableNetstackLocalForwarding is an alias for EnvEnableLocalForwarding.
+	// In netstack mode, it enables forwarding of local traffic to the native stack for all interfaces.
 	EnvEnableNetstackLocalForwarding = "NB_ENABLE_NETSTACK_LOCAL_FORWARDING"
 )
 
@@ -147,6 +151,11 @@ func parseCreateEnv() (bool, bool) {
 		if err != nil {
 			log.Warnf("failed to parse %s: %v", EnvEnableNetstackLocalForwarding, err)
 		}
+	} else if val := os.Getenv(EnvEnableLocalForwarding); val != "" {
+		enableLocalForwarding, err = strconv.ParseBool(val)
+		if err != nil {
+			log.Warnf("failed to parse %s: %v", EnvEnableLocalForwarding, err)
+		}
 	}
 
 	return disableConntrack, enableLocalForwarding
@@ -779,9 +788,10 @@ func (m *Manager) handleLocalTraffic(d *decoder, srcIP, dstIP netip.Addr, packet
 		return true
 	}
 
-	// if running in netstack mode we need to pass this to the forwarder
-	if m.netstack && m.localForwarding {
-		return m.handleNetstackLocalTraffic(packetData)
+	// If requested we pass local traffic to internal interfaces to the forwarder.
+	// netstack doesn't have an interface to forward packets to the native stack so we always need to use the forwarder.
+	if m.localForwarding && (m.netstack || dstIP != m.wgIface.Address().IP) {
+		return m.handleForwardedLocalTraffic(packetData)
 	}
 
 	// track inbound packets to get the correct direction and session id for flows
@@ -791,8 +801,7 @@ func (m *Manager) handleLocalTraffic(d *decoder, srcIP, dstIP netip.Addr, packet
 	return false
 }
 
-func (m *Manager) handleNetstackLocalTraffic(packetData []byte) bool {
-
+func (m *Manager) handleForwardedLocalTraffic(packetData []byte) bool {
 	fwd := m.forwarder.Load()
 	if fwd == nil {
 		m.logger.Trace("Dropping local packet (forwarder not initialized)")

client/iface/configurer/kernel_unix.go

@@ -12,6 +12,8 @@ import (
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 )
 
+var zeroKey wgtypes.Key
+
 type KernelConfigurer struct {
 	deviceName string
 }
@@ -201,6 +203,47 @@ func (c *KernelConfigurer) configure(config wgtypes.Config) error {
 func (c *KernelConfigurer) Close() {
 }
 
+func (c *KernelConfigurer) FullStats() (*Stats, error) {
+	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 nil, fmt.Errorf("get device %s: %w", c.deviceName, err)
+	}
+	fullStats := &Stats{
+		DeviceName: wgDevice.Name,
+		PublicKey:  wgDevice.PublicKey.String(),
+		ListenPort: wgDevice.ListenPort,
+		FWMark:     wgDevice.FirewallMark,
+		Peers:      []Peer{},
+	}
+
+	for _, p := range wgDevice.Peers {
+		peer := Peer{
+			PublicKey:     p.PublicKey.String(),
+			AllowedIPs:    p.AllowedIPs,
+			TxBytes:       p.TransmitBytes,
+			RxBytes:       p.ReceiveBytes,
+			LastHandshake: p.LastHandshakeTime,
+			PresharedKey:  p.PresharedKey != zeroKey,
+		}
+		if p.Endpoint != nil {
+			peer.Endpoint = *p.Endpoint
+		}
+		fullStats.Peers = append(fullStats.Peers, peer)
+	}
+	return fullStats, nil
+}
+
 func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) {
 	stats := make(map[string]WGStats)
 	wg, err := wgctrl.New()

client/iface/configurer/usp.go

@@ -19,10 +19,17 @@ import (
 )
 
 const (
+	privateKey                  = "private_key"
 	ipcKeyLastHandshakeTimeSec  = "last_handshake_time_sec"
 	ipcKeyLastHandshakeTimeNsec = "last_handshake_time_nsec"
 	ipcKeyTxBytes               = "tx_bytes"
 	ipcKeyRxBytes               = "rx_bytes"
+	allowedIP                   = "allowed_ip"
+	endpoint                    = "endpoint"
+	fwmark                      = "fwmark"
+	listenPort                  = "listen_port"
+	publicKey                   = "public_key"
+	presharedKey                = "preshared_key"
 )
 
 var ErrAllowedIPNotFound = fmt.Errorf("allowed IP not found")
@@ -186,6 +193,15 @@ func (c *WGUSPConfigurer) RemoveAllowedIP(peerKey string, ip string) error {
 	return c.device.IpcSet(toWgUserspaceString(config))
 }
 
+func (c *WGUSPConfigurer) FullStats() (*Stats, error) {
+	ipcStr, err := c.device.IpcGet()
+	if err != nil {
+		return nil, fmt.Errorf("IpcGet failed: %w", err)
+	}
+
+	return parseStatus(c.deviceName, ipcStr)
+}
+
 // startUAPI starts the UAPI listener for managing the WireGuard interface via external tool
 func (t *WGUSPConfigurer) startUAPI() {
 	var err error
@@ -365,3 +381,136 @@ func getFwmark() int {
 	}
 	return 0
 }
+
+func hexToWireguardKey(hexKey string) (wgtypes.Key, error) {
+	// Decode hex string to bytes
+	keyBytes, err := hex.DecodeString(hexKey)
+	if err != nil {
+		return wgtypes.Key{}, fmt.Errorf("failed to decode hex key: %w", err)
+	}
+
+	// Check if we have the right number of bytes (WireGuard keys are 32 bytes)
+	if len(keyBytes) != 32 {
+		return wgtypes.Key{}, fmt.Errorf("invalid key length: expected 32 bytes, got %d", len(keyBytes))
+	}
+
+	// Convert to wgtypes.Key
+	var key wgtypes.Key
+	copy(key[:], keyBytes)
+
+	return key, nil
+}
+
+func parseStatus(deviceName, ipcStr string) (*Stats, error) {
+	stats := &Stats{DeviceName: deviceName}
+	var currentPeer *Peer
+	for _, line := range strings.Split(strings.TrimSpace(ipcStr), "\n") {
+		if line == "" {
+			continue
+		}
+		parts := strings.SplitN(line, "=", 2)
+		if len(parts) != 2 {
+			continue
+		}
+		key := parts[0]
+		val := parts[1]
+
+		switch key {
+		case privateKey:
+			key, err := hexToWireguardKey(val)
+			if err != nil {
+				log.Errorf("failed to parse private key: %v", err)
+				continue
+			}
+			stats.PublicKey = key.PublicKey().String()
+		case publicKey:
+			// Save previous peer
+			if currentPeer != nil {
+				stats.Peers = append(stats.Peers, *currentPeer)
+			}
+			key, err := hexToWireguardKey(val)
+			if err != nil {
+				log.Errorf("failed to parse public key: %v", err)
+				continue
+			}
+			currentPeer = &Peer{
+				PublicKey: key.String(),
+			}
+		case listenPort:
+			if port, err := strconv.Atoi(val); err == nil {
+				stats.ListenPort = port
+			}
+		case fwmark:
+			if fwmark, err := strconv.Atoi(val); err == nil {
+				stats.FWMark = fwmark
+			}
+		case endpoint:
+			if currentPeer == nil {
+				continue
+			}
+
+			host, portStr, err := net.SplitHostPort(strings.Trim(val, "[]"))
+			if err != nil {
+				log.Errorf("failed to parse endpoint: %v", err)
+				continue
+			}
+			port, err := strconv.Atoi(portStr)
+			if err != nil {
+				log.Errorf("failed to parse endpoint port: %v", err)
+				continue
+			}
+			currentPeer.Endpoint = net.UDPAddr{
+				IP:   net.ParseIP(host),
+				Port: port,
+			}
+		case allowedIP:
+			if currentPeer == nil {
+				continue
+			}
+			_, ipnet, err := net.ParseCIDR(val)
+			if err == nil {
+				currentPeer.AllowedIPs = append(currentPeer.AllowedIPs, *ipnet)
+			}
+		case ipcKeyTxBytes:
+			if currentPeer == nil {
+				continue
+			}
+			rxBytes, err := toBytes(val)
+			if err != nil {
+				continue
+			}
+			currentPeer.TxBytes = rxBytes
+		case ipcKeyRxBytes:
+			if currentPeer == nil {
+				continue
+			}
+			rxBytes, err := toBytes(val)
+			if err != nil {
+				continue
+			}
+			currentPeer.RxBytes = rxBytes
+
+		case ipcKeyLastHandshakeTimeSec:
+			if currentPeer == nil {
+				continue
+			}
+
+			ts, err := toLastHandshake(val)
+			if err != nil {
+				continue
+			}
+			currentPeer.LastHandshake = ts
+		case presharedKey:
+			if currentPeer == nil {
+				continue
+			}
+			if val != "" {
+				currentPeer.PresharedKey = true
+			}
+		}
+	}
+	if currentPeer != nil {
+		stats.Peers = append(stats.Peers, *currentPeer)
+	}
+	return stats, nil
+}

client/iface/configurer/wgshow.go

@@ -0,0 +1,24 @@
+package configurer
+
+import (
+	"net"
+	"time"
+)
+
+type Peer struct {
+	PublicKey     string
+	Endpoint      net.UDPAddr
+	AllowedIPs    []net.IPNet
+	TxBytes       int64
+	RxBytes       int64
+	LastHandshake time.Time
+	PresharedKey  bool
+}
+
+type Stats struct {
+	DeviceName string
+	PublicKey  string
+	ListenPort int
+	FWMark     int
+	Peers      []Peer
+}

client/iface/device/interface.go

@@ -17,4 +17,5 @@ type WGConfigurer interface {
 	RemoveAllowedIP(peerKey string, allowedIP string) error
 	Close()
 	GetStats() (map[string]configurer.WGStats, error)
+	FullStats() (*configurer.Stats, error)
 }

client/iface/iface.go

@@ -216,6 +216,10 @@ func (w *WGIface) GetStats() (map[string]configurer.WGStats, error) {
 	return w.configurer.GetStats()
 }
 
+func (w *WGIface) FullStats() (*configurer.Stats, error) {
+	return w.configurer.FullStats()
+}
+
 func (w *WGIface) waitUntilRemoved() error {
 	maxWaitTime := 5 * time.Second
 	timeout := time.NewTimer(maxWaitTime)

client/internal/debug/debug.go

@@ -270,11 +270,16 @@ func (g *BundleGenerator) createArchive() error {
 		log.Errorf("Failed to add corrupted state files to debug bundle: %v", err)
 	}
 
+	if err := g.addWgShow(); err != nil {
+		log.Errorf("Failed to add wg show output: %v", err)
+	}
+
 	if g.logFile != "console" {
 		if err := g.addLogfile(); err != nil {
 			return fmt.Errorf("add log file: %w", err)
 		}
 	}
+
 	return nil
 }
 

client/internal/debug/wgshow.go

@@ -0,0 +1,66 @@
+package debug
+
+import (
+	"bytes"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/netbirdio/netbird/client/iface/configurer"
+)
+
+type WGIface interface {
+	FullStats() (*configurer.Stats, error)
+}
+
+func (g *BundleGenerator) addWgShow() error {
+	result, err := g.statusRecorder.PeersStatus()
+	if err != nil {
+		return err
+	}
+
+	output := g.toWGShowFormat(result)
+	reader := bytes.NewReader([]byte(output))
+
+	if err := g.addFileToZip(reader, "wgshow.txt"); err != nil {
+		return fmt.Errorf("add wg show to zip: %w", err)
+	}
+	return nil
+}
+
+func (g *BundleGenerator) toWGShowFormat(s *configurer.Stats) string {
+	var sb strings.Builder
+
+	sb.WriteString(fmt.Sprintf("interface: %s\n", s.DeviceName))
+	sb.WriteString(fmt.Sprintf("  public key: %s\n", s.PublicKey))
+	sb.WriteString(fmt.Sprintf("  listen port: %d\n", s.ListenPort))
+	if s.FWMark != 0 {
+		sb.WriteString(fmt.Sprintf("  fwmark: %#x\n", s.FWMark))
+	}
+
+	for _, peer := range s.Peers {
+		sb.WriteString(fmt.Sprintf("\npeer: %s\n", peer.PublicKey))
+		if peer.Endpoint.IP != nil {
+			if g.anonymize {
+				anonEndpoint := g.anonymizer.AnonymizeUDPAddr(peer.Endpoint)
+				sb.WriteString(fmt.Sprintf("  endpoint: %s\n", anonEndpoint.String()))
+			} else {
+				sb.WriteString(fmt.Sprintf("  endpoint: %s\n", peer.Endpoint.String()))
+			}
+		}
+		if len(peer.AllowedIPs) > 0 {
+			var ipStrings []string
+			for _, ipnet := range peer.AllowedIPs {
+				ipStrings = append(ipStrings, ipnet.String())
+			}
+			sb.WriteString(fmt.Sprintf("  allowed ips: %s\n", strings.Join(ipStrings, ", ")))
+		}
+		sb.WriteString(fmt.Sprintf("  latest handshake: %s\n", peer.LastHandshake.Format(time.RFC1123)))
+		sb.WriteString(fmt.Sprintf("  transfer: %d B received, %d B sent\n", peer.RxBytes, peer.TxBytes))
+		if peer.PresharedKey {
+			sb.WriteString("  preshared key: (hidden)\n")
+		}
+	}
+
+	return sb.String()
+}

client/internal/engine.go

@@ -359,6 +359,7 @@ func (e *Engine) Start() error {
 		return fmt.Errorf("new wg interface: %w", err)
 	}
 	e.wgInterface = wgIface
+	e.statusRecorder.SetWgIface(wgIface)
 
 	// start flow manager right after interface creation
 	publicKey := e.config.WgPrivateKey.PublicKey()
@@ -380,7 +381,6 @@ func (e *Engine) Start() error {
 			return fmt.Errorf("run rosenpass manager: %w", err)
 		}
 	}
-
 	e.stateManager.Start()
 
 	initialRoutes, dnsServer, err := e.newDnsServer()
@@ -1453,6 +1453,7 @@ func (e *Engine) close() {
 			log.Errorf("failed closing Netbird interface %s %v", e.config.WgIfaceName, err)
 		}
 		e.wgInterface = nil
+		e.statusRecorder.SetWgIface(nil)
 	}
 
 	if !isNil(e.sshServer) {

client/internal/engine_test.go

@@ -99,6 +99,10 @@ type MockWGIface struct {
 	GetNetFunc                 func() *netstack.Net
 }
 
+func (m *MockWGIface) FullStats() (*configurer.Stats, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
 func (m *MockWGIface) GetInterfaceGUIDString() (string, error) {
 	return m.GetInterfaceGUIDStringFunc()
 }

client/internal/iface_common.go

@@ -37,4 +37,5 @@ type wgIfaceBase interface {
 	GetWGDevice() *wgdevice.Device
 	GetStats() (map[string]configurer.WGStats, error)
 	GetNet() *netstack.Net
+	FullStats() (*configurer.Stats, error)
 }

client/internal/peer/status.go

@@ -3,6 +3,7 @@ package peer
 import (
 	"context"
 	"errors"
+	"fmt"
 	"net/netip"
 	"slices"
 	"sync"
@@ -32,6 +33,10 @@ type ResolvedDomainInfo struct {
 	ParentDomain domain.Domain
 }
 
+type WGIfaceStatus interface {
+	FullStats() (*configurer.Stats, error)
+}
+
 type EventListener interface {
 	OnEvent(event *proto.SystemEvent)
 }
@@ -202,6 +207,7 @@ type Status struct {
 	ingressGwMgr *ingressgw.Manager
 
 	routeIDLookup routeIDLookup
+	wgIface       WGIfaceStatus
 }
 
 // NewRecorder returns a new Status instance
@@ -1078,6 +1084,23 @@ func (d *Status) GetEventHistory() []*proto.SystemEvent {
 	return d.eventQueue.GetAll()
 }
 
+func (d *Status) SetWgIface(wgInterface WGIfaceStatus) {
+	d.mux.Lock()
+	defer d.mux.Unlock()
+
+	d.wgIface = wgInterface
+}
+
+func (d *Status) PeersStatus() (*configurer.Stats, error) {
+	d.mux.Lock()
+	defer d.mux.Unlock()
+	if d.wgIface == nil {
+		return nil, fmt.Errorf("wgInterface is nil, cannot retrieve peers status")
+	}
+
+	return d.wgIface.FullStats()
+}
+
 type EventQueue struct {
 	maxSize int
 	events  []*proto.SystemEvent