Merge branch 'main' into fix/pkg-loss

client/iface/bind/activity.go

@@ -0,0 +1,96 @@
+package bind
+
+import (
+	"net/netip"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/monotime"
+)
+
+const (
+	saveFrequency = int64(5 * time.Second)
+)
+
+type PeerRecord struct {
+	Address      netip.AddrPort
+	LastActivity atomic.Int64 // UnixNano timestamp
+}
+
+type ActivityRecorder struct {
+	mu         sync.RWMutex
+	peers      map[string]*PeerRecord         // publicKey to PeerRecord map
+	addrToPeer map[netip.AddrPort]*PeerRecord // address to PeerRecord map
+}
+
+func NewActivityRecorder() *ActivityRecorder {
+	return &ActivityRecorder{
+		peers:      make(map[string]*PeerRecord),
+		addrToPeer: make(map[netip.AddrPort]*PeerRecord),
+	}
+}
+
+// GetLastActivities returns a snapshot of peer last activity
+func (r *ActivityRecorder) GetLastActivities() map[string]monotime.Time {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+
+	activities := make(map[string]monotime.Time, len(r.peers))
+	for key, record := range r.peers {
+		monoTime := record.LastActivity.Load()
+		activities[key] = monotime.Time(monoTime)
+	}
+	return activities
+}
+
+// UpsertAddress adds or updates the address for a publicKey
+func (r *ActivityRecorder) UpsertAddress(publicKey string, address netip.AddrPort) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+
+	var record *PeerRecord
+	record, exists := r.peers[publicKey]
+	if exists {
+		delete(r.addrToPeer, record.Address)
+		record.Address = address
+	} else {
+		record = &PeerRecord{
+			Address: address,
+		}
+		record.LastActivity.Store(int64(monotime.Now()))
+		r.peers[publicKey] = record
+	}
+
+	r.addrToPeer[address] = record
+}
+
+func (r *ActivityRecorder) Remove(publicKey string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if record, exists := r.peers[publicKey]; exists {
+		delete(r.addrToPeer, record.Address)
+		delete(r.peers, publicKey)
+	}
+}
+
+// record updates LastActivity for the given address using atomic store
+func (r *ActivityRecorder) record(address netip.AddrPort) {
+	r.mu.RLock()
+	record, ok := r.addrToPeer[address]
+	r.mu.RUnlock()
+	if !ok {
+		log.Warnf("could not find record for address %s", address)
+		return
+	}
+
+	now := int64(monotime.Now())
+	last := record.LastActivity.Load()
+	if now-last < saveFrequency {
+		return
+	}
+
+	_ = record.LastActivity.CompareAndSwap(last, now)
+}

client/iface/bind/activity_test.go

@@ -0,0 +1,25 @@
+package bind
+
+import (
+	"net/netip"
+	"testing"
+	"time"
+
+	"github.com/netbirdio/netbird/monotime"
+)
+
+func TestActivityRecorder_GetLastActivities(t *testing.T) {
+	peer := "peer1"
+	ar := NewActivityRecorder()
+	ar.UpsertAddress("peer1", netip.MustParseAddrPort("192.168.0.5:51820"))
+	activities := ar.GetLastActivities()
+
+	p, ok := activities[peer]
+	if !ok {
+		t.Fatalf("Expected activity for peer %s, but got none", peer)
+	}
+
+	if monotime.Since(p) > 5*time.Second {
+		t.Fatalf("Expected activity for peer %s to be recent, but got %v", peer, p)
+	}
+}

client/iface/bind/control.go

@@ -0,0 +1,15 @@
+package bind
+
+import (
+	wireguard "golang.zx2c4.com/wireguard/conn"
+
+	nbnet "github.com/netbirdio/netbird/util/net"
+)
+
+// TODO: This is most likely obsolete since the control fns should be called by the wrapped udpconn (ice_bind.go)
+func init() {
+	listener := nbnet.NewListener()
+	if listener.ListenConfig.Control != nil {
+		*wireguard.ControlFns = append(*wireguard.ControlFns, listener.ListenConfig.Control)
+	}
+}

client/iface/bind/control_android.go

@@ -1,12 +0,0 @@
-package bind
-
-import (
-	wireguard "golang.zx2c4.com/wireguard/conn"
-
-	nbnet "github.com/netbirdio/netbird/util/net"
-)
-
-func init() {
-	// ControlFns is not thread safe and should only be modified during init.
-	*wireguard.ControlFns = append(*wireguard.ControlFns, nbnet.ControlProtectSocket)
-}

client/iface/bind/endpoint.go

@@ -12,5 +12,6 @@ func EndpointToUDPAddr(e Endpoint) *net.UDPAddr {
 	return &net.UDPAddr{
 		IP:   e.Addr().AsSlice(),
 		Port: int(e.Port()),
+		Zone: e.Addr().Zone(),
 	}
 }

client/iface/bind/ice_bind.go

@@ -2,19 +2,22 @@ package bind
 
 import (
 	"context"
+	"encoding/binary"
 	"fmt"
 	"net"
 	"net/netip"
 	"runtime"
-	"strings"
 	"sync"
 
-	"github.com/pion/stun/v2"
+	"github.com/pion/stun/v3"
 	"github.com/pion/transport/v3"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/ipv4"
 	"golang.org/x/net/ipv6"
 	wgConn "golang.zx2c4.com/wireguard/conn"
+
+	"github.com/netbirdio/netbird/client/iface/wgaddr"
+	nbnet "github.com/netbirdio/netbird/util/net"
 )
 
 type RecvMessage struct {
@@ -51,20 +54,26 @@ type ICEBind struct {
 	closedChanMu sync.RWMutex // protect the closeChan recreation from reading from it.
 	closed       bool
 
-	muUDPMux sync.Mutex
-	udpMux   *UniversalUDPMuxDefault
+	muUDPMux         sync.Mutex
+	udpMux           *UniversalUDPMuxDefault
+	address          wgaddr.Address
+	mtu              uint16
+	activityRecorder *ActivityRecorder
 }
 
-func NewICEBind(transportNet transport.Net, filterFn FilterFn) *ICEBind {
+func NewICEBind(transportNet transport.Net, filterFn FilterFn, address wgaddr.Address, mtu uint16) *ICEBind {
 	b, _ := wgConn.NewStdNetBind().(*wgConn.StdNetBind)
 	ib := &ICEBind{
-		StdNetBind:   b,
-		recvChan:     make(chan RecvMessage, 1),
-		transportNet: transportNet,
-		filterFn:     filterFn,
-		endpoints:    make(map[netip.Addr]net.Conn),
-		closedChan:   make(chan struct{}),
-		closed:       true,
+		StdNetBind:       b,
+		recvChan:         make(chan RecvMessage, 1),
+		transportNet:     transportNet,
+		filterFn:         filterFn,
+		endpoints:        make(map[netip.Addr]net.Conn),
+		closedChan:       make(chan struct{}),
+		closed:           true,
+		mtu:              mtu,
+		address:          address,
+		activityRecorder: NewActivityRecorder(),
 	}
 
 	rc := receiverCreator{
@@ -74,6 +83,10 @@ func NewICEBind(transportNet transport.Net, filterFn FilterFn) *ICEBind {
 	return ib
 }
 
+func (s *ICEBind) MTU() uint16 {
+	return s.mtu
+}
+
 func (s *ICEBind) Open(uport uint16) ([]wgConn.ReceiveFunc, uint16, error) {
 	s.closed = false
 	s.closedChanMu.Lock()
@@ -98,6 +111,10 @@ func (s *ICEBind) Close() error {
 	return s.StdNetBind.Close()
 }
 
+func (s *ICEBind) ActivityRecorder() *ActivityRecorder {
+	return s.activityRecorder
+}
+
 // GetICEMux returns the ICE UDPMux that was created and used by ICEBind
 func (s *ICEBind) GetICEMux() (*UniversalUDPMuxDefault, error) {
 	s.muUDPMux.Lock()
@@ -109,35 +126,17 @@ func (s *ICEBind) GetICEMux() (*UniversalUDPMuxDefault, error) {
 	return s.udpMux, nil
 }
 
-func (b *ICEBind) SetEndpoint(peerAddress *net.UDPAddr, conn net.Conn) (*net.UDPAddr, error) {
-	fakeUDPAddr, err := fakeAddress(peerAddress)
-	if err != nil {
-		return nil, err
-	}
-
-	// force IPv4
-	fakeAddr, ok := netip.AddrFromSlice(fakeUDPAddr.IP.To4())
-	if !ok {
-		return nil, fmt.Errorf("failed to convert IP to netip.Addr")
-	}
-
+func (b *ICEBind) SetEndpoint(fakeIP netip.Addr, conn net.Conn) {
 	b.endpointsMu.Lock()
-	b.endpoints[fakeAddr] = conn
+	b.endpoints[fakeIP] = conn
 	b.endpointsMu.Unlock()
-
-	return fakeUDPAddr, nil
 }
 
-func (b *ICEBind) RemoveEndpoint(fakeUDPAddr *net.UDPAddr) {
-	fakeAddr, ok := netip.AddrFromSlice(fakeUDPAddr.IP.To4())
-	if !ok {
-		log.Warnf("failed to convert IP to netip.Addr")
-		return
-	}
-
+func (b *ICEBind) RemoveEndpoint(fakeIP netip.Addr) {
 	b.endpointsMu.Lock()
 	defer b.endpointsMu.Unlock()
-	delete(b.endpoints, fakeAddr)
+
+	delete(b.endpoints, fakeIP)
 }
 
 func (b *ICEBind) Send(bufs [][]byte, ep wgConn.Endpoint) error {
@@ -170,9 +169,11 @@ func (s *ICEBind) createIPv4ReceiverFn(pc *ipv4.PacketConn, conn *net.UDPConn, r
 
 	s.udpMux = NewUniversalUDPMuxDefault(
 		UniversalUDPMuxParams{
-			UDPConn:  conn,
-			Net:      s.transportNet,
-			FilterFn: s.filterFn,
+			UDPConn:   nbnet.WrapPacketConn(conn),
+			Net:       s.transportNet,
+			FilterFn:  s.filterFn,
+			WGAddress: s.address,
+			MTU:       s.mtu,
 		},
 	)
 	return func(bufs [][]byte, sizes []int, eps []wgConn.Endpoint) (n int, err error) {
@@ -222,6 +223,11 @@ func (s *ICEBind) createIPv4ReceiverFn(pc *ipv4.PacketConn, conn *net.UDPConn, r
 				continue
 			}
 			addrPort := msg.Addr.(*net.UDPAddr).AddrPort()
+
+			if isTransportPkg(msg.Buffers, msg.N) {
+				s.activityRecorder.record(addrPort)
+			}
+
 			ep := &wgConn.StdNetEndpoint{AddrPort: addrPort} // TODO: remove allocation
 			wgConn.GetSrcFromControl(msg.OOB[:msg.NN], ep)
 			eps[i] = ep
@@ -280,25 +286,17 @@ func (c *ICEBind) receiveRelayed(buffs [][]byte, sizes []int, eps []wgConn.Endpo
 		copy(buffs[0], msg.Buffer)
 		sizes[0] = len(msg.Buffer)
 		eps[0] = wgConn.Endpoint(msg.Endpoint)
+
+		if isTransportPkg(buffs, sizes[0]) {
+			if ep, ok := eps[0].(*Endpoint); ok {
+				c.activityRecorder.record(ep.AddrPort)
+			}
+		}
+
 		return 1, nil
 	}
 }
 
-// fakeAddress returns a fake address that is used to as an identifier for the peer.
-// The fake address is in the format of 127.1.x.x where x.x is the last two octets of the peer address.
-func fakeAddress(peerAddress *net.UDPAddr) (*net.UDPAddr, error) {
-	octets := strings.Split(peerAddress.IP.String(), ".")
-	if len(octets) != 4 {
-		return nil, fmt.Errorf("invalid IP format")
-	}
-
-	newAddr := &net.UDPAddr{
-		IP:   net.ParseIP(fmt.Sprintf("127.1.%s.%s", octets[2], octets[3])),
-		Port: peerAddress.Port,
-	}
-	return newAddr, nil
-}
-
 func getMessages(msgsPool *sync.Pool) *[]ipv6.Message {
 	return msgsPool.Get().(*[]ipv6.Message)
 }
@@ -310,3 +308,19 @@ func putMessages(msgs *[]ipv6.Message, msgsPool *sync.Pool) {
 	}
 	msgsPool.Put(msgs)
 }
+
+func isTransportPkg(buffers [][]byte, n int) bool {
+	// The first buffer should contain at least 4 bytes for type
+	if len(buffers[0]) < 4 {
+		return true
+	}
+
+	// WireGuard packet type is a little-endian uint32 at start
+	packetType := binary.LittleEndian.Uint32(buffers[0][:4])
+
+	// Check if packetType matches known WireGuard message types
+	if packetType == 4 && n > 32 {
+		return true
+	}
+	return false
+}

client/iface/bind/udp_mux.go

@@ -8,9 +8,9 @@ import (
 	"strings"
 	"sync"
 
-	"github.com/pion/ice/v3"
+	"github.com/pion/ice/v4"
 	"github.com/pion/logging"
-	"github.com/pion/stun/v2"
+	"github.com/pion/stun/v3"
 	"github.com/pion/transport/v3"
 	"github.com/pion/transport/v3/stdnet"
 	log "github.com/sirupsen/logrus"
@@ -150,7 +150,7 @@ func isZeros(ip net.IP) bool {
 // NewUDPMuxDefault creates an implementation of UDPMux
 func NewUDPMuxDefault(params UDPMuxParams) *UDPMuxDefault {
 	if params.Logger == nil {
-		params.Logger = logging.NewDefaultLoggerFactory().NewLogger("ice")
+		params.Logger = getLogger()
 	}
 
 	mux := &UDPMuxDefault{
@@ -296,14 +296,20 @@ func (m *UDPMuxDefault) RemoveConnByUfrag(ufrag string) {
 		return
 	}
 
-	m.addressMapMu.Lock()
-	defer m.addressMapMu.Unlock()
-
+	var allAddresses []string
 	for _, c := range removedConns {
 		addresses := c.getAddresses()
-		for _, addr := range addresses {
-			delete(m.addressMap, addr)
-		}
+		allAddresses = append(allAddresses, addresses...)
+	}
+
+	m.addressMapMu.Lock()
+	for _, addr := range allAddresses {
+		delete(m.addressMap, addr)
+	}
+	m.addressMapMu.Unlock()
+
+	for _, addr := range allAddresses {
+		m.notifyAddressRemoval(addr)
 	}
 }
 
@@ -351,14 +357,13 @@ func (m *UDPMuxDefault) registerConnForAddress(conn *udpMuxedConn, addr string)
 	}
 
 	m.addressMapMu.Lock()
-	defer m.addressMapMu.Unlock()
-
 	existing, ok := m.addressMap[addr]
 	if !ok {
 		existing = []*udpMuxedConn{}
 	}
 	existing = append(existing, conn)
 	m.addressMap[addr] = existing
+	m.addressMapMu.Unlock()
 
 	log.Debugf("ICE: registered %s for %s", addr, conn.params.Key)
 }
@@ -386,12 +391,12 @@ func (m *UDPMuxDefault) HandleSTUNMessage(msg *stun.Message, addr net.Addr) erro
 	// If you are using the same socket for the Host and SRFLX candidates, it might be that there are more than one
 	// muxed connection - one for the SRFLX candidate and the other one for the HOST one.
 	// We will then forward STUN packets to each of these connections.
-	m.addressMapMu.Lock()
+	m.addressMapMu.RLock()
 	var destinationConnList []*udpMuxedConn
 	if storedConns, ok := m.addressMap[addr.String()]; ok {
 		destinationConnList = append(destinationConnList, storedConns...)
 	}
-	m.addressMapMu.Unlock()
+	m.addressMapMu.RUnlock()
 
 	var isIPv6 bool
 	if udpAddr, _ := addr.(*net.UDPAddr); udpAddr != nil && udpAddr.IP.To4() == nil {
@@ -455,3 +460,9 @@ func newBufferHolder(size int) *bufferHolder {
 		buf: make([]byte, size),
 	}
 }
+
+func getLogger() logging.LeveledLogger {
+	fac := logging.NewDefaultLoggerFactory()
+	//fac.Writer = log.StandardLogger().Writer()
+	return fac.NewLogger("ice")
+}

client/iface/bind/udp_mux_generic.go

@@ -0,0 +1,22 @@
+//go:build !ios
+
+package bind
+
+import (
+	nbnet "github.com/netbirdio/netbird/util/net"
+)
+
+func (m *UDPMuxDefault) notifyAddressRemoval(addr string) {
+	// Kernel mode: direct nbnet.PacketConn (SharedSocket wrapped with nbnet)
+	if conn, ok := m.params.UDPConn.(*nbnet.PacketConn); ok {
+		conn.RemoveAddress(addr)
+		return
+	}
+
+	// Userspace mode: UDPConn wrapper around nbnet.PacketConn
+	if wrapped, ok := m.params.UDPConn.(*UDPConn); ok {
+		if conn, ok := wrapped.GetPacketConn().(*nbnet.PacketConn); ok {
+			conn.RemoveAddress(addr)
+		}
+	}
+}

client/iface/bind/udp_mux_ios.go

@@ -0,0 +1,7 @@
+//go:build ios
+
+package bind
+
+func (m *UDPMuxDefault) notifyAddressRemoval(addr string) {
+	// iOS doesn't support nbnet hooks, so this is a no-op
+}

client/iface/bind/udp_mux_universal.go

@@ -15,8 +15,11 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/pion/logging"
-	"github.com/pion/stun/v2"
+	"github.com/pion/stun/v3"
 	"github.com/pion/transport/v3"
+
+	"github.com/netbirdio/netbird/client/iface/bufsize"
+	"github.com/netbirdio/netbird/client/iface/wgaddr"
 )
 
 // FilterFn is a function that filters out candidates based on the address.
@@ -41,12 +44,14 @@ type UniversalUDPMuxParams struct {
 	XORMappedAddrCacheTTL time.Duration
 	Net                   transport.Net
 	FilterFn              FilterFn
+	WGAddress             wgaddr.Address
+	MTU                   uint16
 }
 
 // NewUniversalUDPMuxDefault creates an implementation of UniversalUDPMux embedding UDPMux
 func NewUniversalUDPMuxDefault(params UniversalUDPMuxParams) *UniversalUDPMuxDefault {
 	if params.Logger == nil {
-		params.Logger = logging.NewDefaultLoggerFactory().NewLogger("ice")
+		params.Logger = getLogger()
 	}
 	if params.XORMappedAddrCacheTTL == 0 {
 		params.XORMappedAddrCacheTTL = time.Second * 25
@@ -59,14 +64,14 @@ func NewUniversalUDPMuxDefault(params UniversalUDPMuxParams) *UniversalUDPMuxDef
 
 	// wrap UDP connection, process server reflexive messages
 	// before they are passed to the UDPMux connection handler (connWorker)
-	m.params.UDPConn = &udpConn{
+	m.params.UDPConn = &UDPConn{
 		PacketConn: params.UDPConn,
 		mux:        m,
 		logger:     params.Logger,
 		filterFn:   params.FilterFn,
+		address:    params.WGAddress,
 	}
 
-	// embed UDPMux
 	udpMuxParams := UDPMuxParams{
 		Logger:  params.Logger,
 		UDPConn: m.params.UDPConn,
@@ -81,7 +86,7 @@ func NewUniversalUDPMuxDefault(params UniversalUDPMuxParams) *UniversalUDPMuxDef
 // just ignore other packets printing an warning message.
 // It is a blocking method, consider running in a go routine.
 func (m *UniversalUDPMuxDefault) ReadFromConn(ctx context.Context) {
-	buf := make([]byte, 1500)
+	buf := make([]byte, m.params.MTU+bufsize.WGBufferOverhead)
 	for {
 		select {
 		case <-ctx.Done():
@@ -110,17 +115,23 @@ func (m *UniversalUDPMuxDefault) ReadFromConn(ctx context.Context) {
 	}
 }
 
-// udpConn is a wrapper around UDPMux conn that overrides ReadFrom and handles STUN/TURN packets
-type udpConn struct {
+// UDPConn is a wrapper around UDPMux conn that overrides ReadFrom and handles STUN/TURN packets
+type UDPConn struct {
 	net.PacketConn
 	mux      *UniversalUDPMuxDefault
 	logger   logging.LeveledLogger
 	filterFn FilterFn
 	// TODO: reset cache on route changes
 	addrCache sync.Map
+	address   wgaddr.Address
 }
 
-func (u *udpConn) WriteTo(b []byte, addr net.Addr) (int, error) {
+// GetPacketConn returns the underlying PacketConn
+func (u *UDPConn) GetPacketConn() net.PacketConn {
+	return u.PacketConn
+}
+
+func (u *UDPConn) WriteTo(b []byte, addr net.Addr) (int, error) {
 	if u.filterFn == nil {
 		return u.PacketConn.WriteTo(b, addr)
 	}
@@ -132,21 +143,21 @@ func (u *udpConn) WriteTo(b []byte, addr net.Addr) (int, error) {
 	return u.handleUncachedAddress(b, addr)
 }
 
-func (u *udpConn) handleCachedAddress(isRouted bool, b []byte, addr net.Addr) (int, error) {
+func (u *UDPConn) handleCachedAddress(isRouted bool, b []byte, addr net.Addr) (int, error) {
 	if isRouted {
 		return 0, fmt.Errorf("address %s is part of a routed network, refusing to write", addr)
 	}
 	return u.PacketConn.WriteTo(b, addr)
 }
 
-func (u *udpConn) handleUncachedAddress(b []byte, addr net.Addr) (int, error) {
+func (u *UDPConn) handleUncachedAddress(b []byte, addr net.Addr) (int, error) {
 	if err := u.performFilterCheck(addr); err != nil {
 		return 0, err
 	}
 	return u.PacketConn.WriteTo(b, addr)
 }
 
-func (u *udpConn) performFilterCheck(addr net.Addr) error {
+func (u *UDPConn) performFilterCheck(addr net.Addr) error {
 	host, err := getHostFromAddr(addr)
 	if err != nil {
 		log.Errorf("Failed to get host from address %s: %v", addr, err)
@@ -159,6 +170,11 @@ func (u *udpConn) performFilterCheck(addr net.Addr) error {
 		return nil
 	}
 
+	if u.address.Network.Contains(a) {
+		log.Warnf("Address %s is part of the NetBird network %s, refusing to write", addr, u.address)
+		return fmt.Errorf("address %s is part of the NetBird network %s, refusing to write", addr, u.address)
+	}
+
 	if isRouted, prefix, err := u.filterFn(a); err != nil {
 		log.Errorf("Failed to check if address %s is routed: %v", addr, err)
 	} else {