sync mock

client/internal/dnsfwd/forwarder.go

@@ -33,8 +33,6 @@ type DNSForwarder struct {
 
 	dnsServer *dns.Server
 	mux       *dns.ServeMux
-	tcpServer *dns.Server
-	tcpMux    *dns.ServeMux
 
 	mutex      sync.RWMutex
 	fwdEntries []*ForwarderEntry
@@ -52,41 +50,22 @@ func NewDNSForwarder(listenAddress string, ttl uint32, firewall firewall.Manager
 }
 
 func (f *DNSForwarder) Listen(entries []*ForwarderEntry) error {
-	log.Infof("starting DNS forwarder on address=%s", f.listenAddress)
-
-	// UDP server
+	log.Infof("listen DNS forwarder on address=%s", f.listenAddress)
 	mux := dns.NewServeMux()
-	f.mux = mux
-	f.dnsServer = &dns.Server{
+
+	dnsServer := &dns.Server{
 		Addr:    f.listenAddress,
 		Net:     "udp",
 		Handler: mux,
 	}
-	// TCP server
-	tcpMux := dns.NewServeMux()
-	f.tcpMux = tcpMux
-	f.tcpServer = &dns.Server{
-		Addr:    f.listenAddress,
-		Net:     "tcp",
-		Handler: tcpMux,
-	}
+	f.dnsServer = dnsServer
+	f.mux = mux
 
 	f.UpdateDomains(entries)
 
-	errCh := make(chan error, 2)
-
-	go func() {
-		log.Infof("DNS UDP listener running on %s", f.listenAddress)
-		errCh <- f.dnsServer.ListenAndServe()
-	}()
-	go func() {
-		log.Infof("DNS TCP listener running on %s", f.listenAddress)
-		errCh <- f.tcpServer.ListenAndServe()
-	}()
-
-	// return the first error we get (e.g. bind failure or shutdown)
-	return <-errCh
+	return dnsServer.ListenAndServe()
 }
+
 func (f *DNSForwarder) UpdateDomains(entries []*ForwarderEntry) {
 	f.mutex.Lock()
 	defer f.mutex.Unlock()
@@ -98,41 +77,31 @@ func (f *DNSForwarder) UpdateDomains(entries []*ForwarderEntry) {
 	}
 
 	oldDomains := filterDomains(f.fwdEntries)
+
 	for _, d := range oldDomains {
 		f.mux.HandleRemove(d.PunycodeString())
-		f.tcpMux.HandleRemove(d.PunycodeString())
 	}
 
 	newDomains := filterDomains(entries)
 	for _, d := range newDomains {
-		f.mux.HandleFunc(d.PunycodeString(), f.handleDNSQueryUDP)
-		f.tcpMux.HandleFunc(d.PunycodeString(), f.handleDNSQueryTCP)
+		f.mux.HandleFunc(d.PunycodeString(), f.handleDNSQuery)
 	}
 
 	f.fwdEntries = entries
+
 	log.Debugf("Updated domains from %v to %v", oldDomains, newDomains)
 }
 
 func (f *DNSForwarder) Close(ctx context.Context) error {
-	var result *multierror.Error
-
-	if f.dnsServer != nil {
-		if err := f.dnsServer.ShutdownContext(ctx); err != nil {
-			result = multierror.Append(result, fmt.Errorf("UDP shutdown: %w", err))
-		}
+	if f.dnsServer == nil {
+		return nil
 	}
-	if f.tcpServer != nil {
-		if err := f.tcpServer.ShutdownContext(ctx); err != nil {
-			result = multierror.Append(result, fmt.Errorf("TCP shutdown: %w", err))
-		}
-	}
-
-	return nberrors.FormatErrorOrNil(result)
+	return f.dnsServer.ShutdownContext(ctx)
 }
 
-func (f *DNSForwarder) handleDNSQuery(w dns.ResponseWriter, query *dns.Msg) *dns.Msg {
+func (f *DNSForwarder) handleDNSQuery(w dns.ResponseWriter, query *dns.Msg) {
 	if len(query.Question) == 0 {
-		return nil
+		return
 	}
 	question := query.Question[0]
 	log.Tracef("received DNS request for DNS forwarder: domain=%v type=%v class=%v",
@@ -154,53 +123,20 @@ func (f *DNSForwarder) handleDNSQuery(w dns.ResponseWriter, query *dns.Msg) *dns
 		if err := w.WriteMsg(resp); err != nil {
 			log.Errorf("failed to write DNS response: %v", err)
 		}
-		return nil
+		return
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), upstreamTimeout)
 	defer cancel()
 	ips, err := net.DefaultResolver.LookupNetIP(ctx, network, domain)
 	if err != nil {
-		f.handleDNSError(w, query, resp, domain, err)
-		return nil
+		f.handleDNSError(w, resp, domain, err)
+		return
 	}
 
 	f.updateInternalState(domain, ips)
 	f.addIPsToResponse(resp, domain, ips)
 
-	return resp
-}
-
-func (f *DNSForwarder) handleDNSQueryUDP(w dns.ResponseWriter, query *dns.Msg) {
-
-	resp := f.handleDNSQuery(w, query)
-	if resp == nil {
-		return
-	}
-
-	opt := query.IsEdns0()
-	maxSize := dns.MinMsgSize
-	if opt != nil {
-		// client advertised a larger EDNS0 buffer
-		maxSize = int(opt.UDPSize())
-	}
-
-	// if our response is too big, truncate and set the TC bit
-	if resp.Len() > maxSize {
-		resp.Truncate(maxSize)
-	}
-
-	if err := w.WriteMsg(resp); err != nil {
-		log.Errorf("failed to write DNS response: %v", err)
-	}
-}
-
-func (f *DNSForwarder) handleDNSQueryTCP(w dns.ResponseWriter, query *dns.Msg) {
-	resp := f.handleDNSQuery(w, query)
-	if resp == nil {
-		return
-	}
-
 	if err := w.WriteMsg(resp); err != nil {
 		log.Errorf("failed to write DNS response: %v", err)
 	}
@@ -243,7 +179,7 @@ func (f *DNSForwarder) updateFirewall(matchingEntries []*ForwarderEntry, prefixe
 }
 
 // handleDNSError processes DNS lookup errors and sends an appropriate error response
-func (f *DNSForwarder) handleDNSError(w dns.ResponseWriter, query, resp *dns.Msg, domain string, err error) {
+func (f *DNSForwarder) handleDNSError(w dns.ResponseWriter, resp *dns.Msg, domain string, err error) {
 	var dnsErr *net.DNSError
 
 	switch {
@@ -255,7 +191,7 @@ func (f *DNSForwarder) handleDNSError(w dns.ResponseWriter, query, resp *dns.Msg
 		}
 
 		if dnsErr.Server != "" {
-			log.Warnf("failed to resolve query for type=%s domain=%s server=%s: %v", dns.TypeToString[query.Question[0].Qtype], domain, dnsErr.Server, err)
+			log.Warnf("failed to resolve query for domain=%s server=%s: %v", domain, dnsErr.Server, err)
 		} else {
 			log.Warnf(errResolveFailed, domain, err)
 		}

client/internal/dnsfwd/manager.go

@@ -33,7 +33,6 @@ type Manager struct {
 	statusRecorder *peer.Status
 
 	fwRules      []firewall.Rule
-	tcpRules     []firewall.Rule
 	dnsForwarder *DNSForwarder
 }
 
@@ -108,13 +107,6 @@ func (m *Manager) allowDNSFirewall() error {
 	}
 	m.fwRules = dnsRules
 
-	tcpRules, err := m.firewall.AddPeerFiltering(nil, net.IP{0, 0, 0, 0}, firewall.ProtocolTCP, nil, dport, firewall.ActionAccept, "")
-	if err != nil {
-		log.Errorf("failed to add allow DNS router rules, err: %v", err)
-		return err
-	}
-	m.tcpRules = tcpRules
-
 	return nil
 }
 
@@ -125,13 +117,7 @@ func (m *Manager) dropDNSFirewall() error {
 			mErr = multierror.Append(mErr, fmt.Errorf("failed to delete DNS router rules, err: %v", err))
 		}
 	}
-	for _, rule := range m.tcpRules {
-		if err := m.firewall.DeletePeerRule(rule); err != nil {
-			mErr = multierror.Append(mErr, fmt.Errorf("failed to delete DNS router rules, err: %v", err))
-		}
-	}
 
 	m.fwRules = nil
-	m.tcpRules = nil
 	return nberrors.FormatErrorOrNil(mErr)
 }

client/internal/networkmonitor/check_change_bsd.go

@@ -19,7 +19,7 @@ import (
 func checkChange(ctx context.Context, nexthopv4, nexthopv6 systemops.Nexthop) error {
 	fd, err := unix.Socket(syscall.AF_ROUTE, syscall.SOCK_RAW, syscall.AF_UNSPEC)
 	if err != nil {
-		return fmt.Errorf("open routing socket: %v", err)
+		return fmt.Errorf("failed to open routing socket: %v", err)
 	}
 	defer func() {
 		err := unix.Close(fd)

client/internal/networkmonitor/check_change_windows.go

@@ -13,7 +13,7 @@ import (
 func checkChange(ctx context.Context, nexthopv4, nexthopv6 systemops.Nexthop) error {
 	routeMonitor, err := systemops.NewRouteMonitor(ctx)
 	if err != nil {
-		return fmt.Errorf("create route monitor: %w", err)
+		return fmt.Errorf("failed to create route monitor: %w", err)
 	}
 	defer func() {
 		if err := routeMonitor.Stop(); err != nil {
@@ -38,46 +38,32 @@ func checkChange(ctx context.Context, nexthopv4, nexthopv6 systemops.Nexthop) er
 }
 
 func routeChanged(route systemops.RouteUpdate, nexthopv4, nexthopv6 systemops.Nexthop) bool {
-	if intf := route.NextHop.Intf; intf != nil && isSoftInterface(intf.Name) {
-		log.Debugf("Network monitor: ignoring default route change for next hop with soft interface %s", route.NextHop)
-		return false
-	}
-
-	// TODO: for the empty nexthop ip (on-link), determine the family differently
-	nexthop := nexthopv4
-	if route.NextHop.IP.Is6() {
-		nexthop = nexthopv6
+	intf := "<nil>"
+	if route.Interface != nil {
+		intf = route.Interface.Name
+		if isSoftInterface(intf) {
+			log.Debugf("Network monitor: ignoring default route change for soft interface %s", intf)
+			return false
+		}
 	}
 
 	switch route.Type {
-	case systemops.RouteModified, systemops.RouteAdded:
-		return handleRouteAddedOrModified(route, nexthop)
-	case systemops.RouteDeleted:
-		return handleRouteDeleted(route, nexthop)
-	}
-
-	return false
-}
-
-func handleRouteAddedOrModified(route systemops.RouteUpdate, nexthop systemops.Nexthop) bool {
-	// For added/modified routes, we care about different next hops
-	if !nexthop.Equal(route.NextHop) {
-		action := "changed"
-		if route.Type == systemops.RouteAdded {
-			action = "added"
+	case systemops.RouteModified:
+		// TODO: get routing table to figure out if our route is affected for modified routes
+		log.Infof("Network monitor: default route changed: via %s, interface %s", route.NextHop, intf)
+		return true
+	case systemops.RouteAdded:
+		if route.NextHop.Is4() && route.NextHop != nexthopv4.IP || route.NextHop.Is6() && route.NextHop != nexthopv6.IP {
+			log.Infof("Network monitor: default route added: via %s, interface %s", route.NextHop, intf)
+			return true
+		}
+	case systemops.RouteDeleted:
+		if nexthopv4.Intf != nil && route.NextHop == nexthopv4.IP || nexthopv6.Intf != nil && route.NextHop == nexthopv6.IP {
+			log.Infof("Network monitor: default route removed: via %s, interface %s", route.NextHop, intf)
+			return true
 		}
-		log.Infof("Network monitor: default route %s: via %s", action, route.NextHop)
-		return true
 	}
-	return false
-}
 
-func handleRouteDeleted(route systemops.RouteUpdate, nexthop systemops.Nexthop) bool {
-	// For deleted routes, we care about our tracked next hop being deleted
-	if nexthop.Equal(route.NextHop) {
-		log.Infof("Network monitor: default route removed: via %s", route.NextHop)
-		return true
-	}
 	return false
 }
 

client/internal/networkmonitor/check_change_windows_test.go

@@ -1,404 +0,0 @@
-package networkmonitor
-
-import (
-	"net"
-	"net/netip"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-
-	"github.com/netbirdio/netbird/client/internal/routemanager/systemops"
-)
-
-func TestRouteChanged(t *testing.T) {
-	tests := []struct {
-		name      string
-		route     systemops.RouteUpdate
-		nexthopv4 systemops.Nexthop
-		nexthopv6 systemops.Nexthop
-		expected  bool
-	}{
-		{
-			name: "soft interface should be ignored",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP: netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{
-						Name: "ISATAP-Interface", // isSoftInterface checks name
-					},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP: netip.MustParseAddr("192.168.1.2"),
-			},
-			nexthopv6: systemops.Nexthop{
-				IP: netip.MustParseAddr("2001:db8::1"),
-			},
-			expected: false,
-		},
-		{
-			name: "modified route with different v4 nexthop IP should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP: netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{
-						Index: 1, Name: "eth0",
-					},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP: netip.MustParseAddr("192.168.1.2"),
-				Intf: &net.Interface{
-					Index: 1, Name: "eth0",
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP: netip.MustParseAddr("2001:db8::1"),
-			},
-			expected: true,
-		},
-		{
-			name: "modified route with same v4 nexthop (IP and Intf Index) should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP: netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{
-						Index: 1, Name: "eth0",
-					},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP: netip.MustParseAddr("192.168.1.1"),
-				Intf: &net.Interface{
-					Index: 1, Name: "eth0",
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP: netip.MustParseAddr("2001:db8::1"),
-			},
-			expected: false,
-		},
-		{
-			name: "added route with different v6 nexthop IP should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteAdded,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP: netip.MustParseAddr("2001:db8::2"),
-					Intf: &net.Interface{
-						Index: 1, Name: "eth0",
-					},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP: netip.MustParseAddr("192.168.1.1"),
-			},
-			nexthopv6: systemops.Nexthop{
-				IP: netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{
-					Index: 1, Name: "eth0",
-				},
-			},
-			expected: true,
-		},
-		{
-			name: "added route with same v6 nexthop (IP and Intf Index) should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteAdded,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP: netip.MustParseAddr("2001:db8::1"),
-					Intf: &net.Interface{
-						Index: 1, Name: "eth0",
-					},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP: netip.MustParseAddr("192.168.1.1"),
-			},
-			nexthopv6: systemops.Nexthop{
-				IP: netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{
-					Index: 1, Name: "eth0",
-				},
-			},
-			expected: false,
-		},
-		{
-			name: "deleted route matching tracked v4 nexthop (IP and Intf Index) should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteDeleted,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP: netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{
-						Index: 1, Name: "eth0",
-					},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP: netip.MustParseAddr("192.168.1.1"),
-				Intf: &net.Interface{
-					Index: 1, Name: "eth0",
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP: netip.MustParseAddr("2001:db8::1"),
-			},
-			expected: true,
-		},
-		{
-			name: "deleted route not matching tracked v4 nexthop (different IP) should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteDeleted,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP: netip.MustParseAddr("192.168.1.3"), // Different IP
-					Intf: &net.Interface{
-						Index: 1, Name: "eth0",
-					},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP: netip.MustParseAddr("192.168.1.1"),
-				Intf: &net.Interface{
-					Index: 1, Name: "eth0",
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP: netip.MustParseAddr("2001:db8::1"),
-			},
-			expected: false,
-		},
-		{
-			name: "modified v4 route with same IP, different Intf Index should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP:   netip.MustParseAddr("192.168.1.1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: true,
-		},
-		{
-			name: "modified v4 route with same IP, one Intf nil, other non-nil should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("192.168.1.1"),
-					Intf: nil, // Intf is nil
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP:   netip.MustParseAddr("192.168.1.1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"}, // Tracked Intf is not nil
-			},
-			expected: true,
-		},
-		{
-			name: "added v4 route with same IP, different Intf Index should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteAdded,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP:   netip.MustParseAddr("192.168.1.1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: true,
-		},
-		{
-			name: "deleted v4 route with same IP, different Intf Index should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteDeleted,
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{ // This is the route being deleted
-					IP:   netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{Index: 1, Name: "eth0"},
-				},
-			},
-			nexthopv4: systemops.Nexthop{ // This is our tracked nexthop
-				IP:   netip.MustParseAddr("192.168.1.1"),
-				Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
-			},
-			expected: false, // Because nexthopv4.Equal(route.NextHop) will be false
-		},
-		{
-			name: "modified v6 route with different IP, same Intf Index should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("2001:db8::3"), // Different IP
-					Intf: &net.Interface{Index: 1, Name: "eth0"},
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP:   netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: true,
-		},
-		{
-			name: "modified v6 route with same IP, different Intf Index should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("2001:db8::1"),
-					Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP:   netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: true,
-		},
-		{
-			name: "modified v6 route with same IP, same Intf Index should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteModified,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("2001:db8::1"),
-					Intf: &net.Interface{Index: 1, Name: "eth0"},
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP:   netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: false,
-		},
-		{
-			name: "deleted v6 route matching tracked nexthop (IP and Intf Index) should return true",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteDeleted,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("2001:db8::1"),
-					Intf: &net.Interface{Index: 1, Name: "eth0"},
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP:   netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: true,
-		},
-		{
-			name: "deleted v6 route not matching tracked nexthop (different IP) should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteDeleted,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("2001:db8::3"), // Different IP
-					Intf: &net.Interface{Index: 1, Name: "eth0"},
-				},
-			},
-			nexthopv6: systemops.Nexthop{
-				IP:   netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: false,
-		},
-		{
-			name: "deleted v6 route not matching tracked nexthop (same IP, different Intf Index) should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteDeleted,
-				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
-				NextHop: systemops.Nexthop{ // This is the route being deleted
-					IP:   netip.MustParseAddr("2001:db8::1"),
-					Intf: &net.Interface{Index: 1, Name: "eth0"},
-				},
-			},
-			nexthopv6: systemops.Nexthop{ // This is our tracked nexthop
-				IP:   netip.MustParseAddr("2001:db8::1"),
-				Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
-			},
-			expected: false,
-		},
-		{
-			name: "unknown route type should return false",
-			route: systemops.RouteUpdate{
-				Type:        systemops.RouteUpdateType(99), // Unknown type
-				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
-				NextHop: systemops.Nexthop{
-					IP:   netip.MustParseAddr("192.168.1.1"),
-					Intf: &net.Interface{Index: 1, Name: "eth0"},
-				},
-			},
-			nexthopv4: systemops.Nexthop{
-				IP:   netip.MustParseAddr("192.168.1.2"), // Different from route.NextHop
-				Intf: &net.Interface{Index: 1, Name: "eth0"},
-			},
-			expected: false,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result := routeChanged(tt.route, tt.nexthopv4, tt.nexthopv6)
-			assert.Equal(t, tt.expected, result)
-		})
-	}
-}
-
-func TestIsSoftInterface(t *testing.T) {
-	tests := []struct {
-		name     string
-		ifname   string
-		expected bool
-	}{
-		{
-			name:     "ISATAP interface should be detected",
-			ifname:   "ISATAP tunnel adapter",
-			expected: true,
-		},
-		{
-			name:     "lowercase soft interface should be detected",
-			ifname:   "isatap.{14A5CF17-CA72-43EC-B4EA-B4B093641B7D}",
-			expected: true,
-		},
-		{
-			name:     "Teredo interface should be detected",
-			ifname:   "Teredo Tunneling Pseudo-Interface",
-			expected: true,
-		},
-		{
-			name:     "regular interface should not be detected as soft",
-			ifname:   "eth0",
-			expected: false,
-		},
-		{
-			name:     "another regular interface should not be detected as soft",
-			ifname:   "wlan0",
-			expected: false,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			result := isSoftInterface(tt.ifname)
-			assert.Equal(t, tt.expected, result)
-		})
-	}
-}

client/internal/networkmonitor/monitor.go

@@ -118,12 +118,9 @@ func (nw *NetworkMonitor) Stop() {
 }
 
 func (nw *NetworkMonitor) checkChanges(ctx context.Context, event chan struct{}, nexthop4 systemops.Nexthop, nexthop6 systemops.Nexthop) {
-	defer close(event)
 	for {
 		if err := checkChangeFn(ctx, nexthop4, nexthop6); err != nil {
-			if !errors.Is(err, context.Canceled) {
-				log.Errorf("Network monitor: failed to check for changes: %v", err)
-			}
+			close(event)
 			return
 		}
 		// prevent blocking

client/internal/routemanager/systemops/systemops.go

@@ -1,7 +1,6 @@
 package systemops
 
 import (
-	"fmt"
 	"net"
 	"net/netip"
 	"sync"
@@ -16,20 +15,6 @@ type Nexthop struct {
 	Intf *net.Interface
 }
 
-// Equal checks if two nexthops are equal.
-func (n Nexthop) Equal(other Nexthop) bool {
-	return n.IP == other.IP && (n.Intf == nil && other.Intf == nil ||
-		n.Intf != nil && other.Intf != nil && n.Intf.Index == other.Intf.Index)
-}
-
-// String returns a string representation of the nexthop.
-func (n Nexthop) String() string {
-	if n.Intf == nil {
-		return n.IP.String()
-	}
-	return fmt.Sprintf("%s @ %d (%s)", n.IP.String(), n.Intf.Index, n.Intf.Name)
-}
-
 type ExclusionCounter = refcounter.Counter[netip.Prefix, struct{}, Nexthop]
 
 type SysOps struct {

client/internal/routemanager/systemops/systemops_windows.go

@@ -33,7 +33,8 @@ type RouteUpdateType int
 type RouteUpdate struct {
 	Type        RouteUpdateType
 	Destination netip.Prefix
-	NextHop     Nexthop
+	NextHop     netip.Addr
+	Interface   *net.Interface
 }
 
 // RouteMonitor provides a way to monitor changes in the routing table.
@@ -230,15 +231,15 @@ func (rm *RouteMonitor) parseUpdate(row *MIB_IPFORWARD_ROW2, notificationType MI
 		intf, err := net.InterfaceByIndex(idx)
 		if err != nil {
 			log.Warnf("failed to get interface name for index %d: %v", idx, err)
-			update.NextHop.Intf = &net.Interface{
+			update.Interface = &net.Interface{
 				Index: idx,
 			}
 		} else {
-			update.NextHop.Intf = intf
+			update.Interface = intf
 		}
 	}
 
-	log.Tracef("Received route update with destination %v, next hop %v, interface %v", row.DestinationPrefix, row.NextHop, update.NextHop.Intf)
+	log.Tracef("Received route update with destination %v, next hop %v, interface %v", row.DestinationPrefix, row.NextHop, update.Interface)
 	dest := parseIPPrefix(row.DestinationPrefix, idx)
 	if !dest.Addr().IsValid() {
 		return RouteUpdate{}, fmt.Errorf("invalid destination: %v", row)
@@ -261,7 +262,7 @@ func (rm *RouteMonitor) parseUpdate(row *MIB_IPFORWARD_ROW2, notificationType MI
 
 	update.Type = updateType
 	update.Destination = dest
-	update.NextHop.IP = nexthop
+	update.NextHop = nexthop
 
 	return update, nil
 }

go.mod

@@ -106,6 +106,7 @@ require (
 	golang.org/x/oauth2 v0.24.0
 	golang.org/x/sync v0.13.0
 	golang.org/x/term v0.31.0
+	golang.org/x/time v0.5.0
 	google.golang.org/api v0.177.0
 	gopkg.in/yaml.v3 v3.0.1
 	gorm.io/driver/mysql v1.5.7
@@ -240,7 +241,6 @@ require (
 	golang.org/x/image v0.18.0 // indirect
 	golang.org/x/mod v0.17.0 // indirect
 	golang.org/x/text v0.24.0 // indirect
-	golang.org/x/time v0.5.0 // indirect
 	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
 	golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240509183442-62759503f434 // indirect

management/cmd/management.go

@@ -134,6 +134,7 @@ var (
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
 			flag.Parse()
+			startPprof()
 
 			ctx, cancel := context.WithCancel(cmd.Context())
 			defer cancel()
@@ -176,7 +177,7 @@ var (
 			if disableSingleAccMode {
 				mgmtSingleAccModeDomain = ""
 			}
-			eventStore, key, err := integrations.InitEventStore(ctx, config.Datadir, config.DataStoreEncryptionKey, appMetrics)
+			eventStore, key, err := integrations.InitEventStore(ctx, config.Datadir, config.DataStoreEncryptionKey)
 			if err != nil {
 				return fmt.Errorf("failed to initialize database: %s", err)
 			}

management/cmd/root.go

@@ -2,9 +2,13 @@ package cmd
 
 import (
 	"fmt"
+	"net/http"
+	_ "net/http/pprof"
 	"os"
 	"os/signal"
+	"runtime"
 
+	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 
 	"github.com/netbirdio/netbird/management/server/types"
@@ -17,6 +21,17 @@ const (
 	idpSignKeyRefreshEnabledFlagName = "idp-sign-key-refresh-enabled"
 )
 
+func startPprof() {
+	go func() {
+		runtime.SetBlockProfileRate(1)
+		runtime.SetMutexProfileFraction(1)
+		log.Debugf("Starting pprof server on 0.0.0.0:6060")
+		if err := http.ListenAndServe("0.0.0.0:6060", nil); err != nil {
+			log.Fatalf("pprof server failed: %v", err)
+		}
+	}()
+}
+
 var (
 	dnsDomain                string
 	mgmtDataDir              string

management/server/account.go

@@ -202,7 +202,7 @@ func BuildManager(
 	if err != nil {
 		initialInterval = 1
 	} else {
-		initialInterval = int64(interval) * 10
+		initialInterval = int64(interval) * 2
 		go func() {
 			time.Sleep(30 * time.Second)
 			am.updateAccountPeersBufferInterval.Store(int64(time.Duration(interval) * time.Millisecond))

management/server/account/manager.go

@@ -117,4 +117,5 @@ type Manager interface {
 	UpdateToPrimaryAccount(ctx context.Context, accountId string) (*types.Account, error)
 	GetOwnerInfo(ctx context.Context, accountId string) (*types.UserInfo, error)
 	GetCurrentUserInfo(ctx context.Context, userAuth nbcontext.UserAuth) (*users.UserInfoWithPermissions, error)
+	BufferUpdateAccountPeers(ctx context.Context, accountID string)
 }

management/server/grpcserver.go

@@ -3,8 +3,11 @@ package server
 import (
 	"context"
 	"fmt"
+	"math/rand/v2"
 	"net"
 	"net/netip"
+	"os"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -13,6 +16,7 @@ import (
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/peer"
@@ -47,6 +51,11 @@ type GRPCServer struct {
 	ephemeralManager   *EphemeralManager
 	peerLocks          sync.Map
 	authManager        auth.Manager
+
+	syncLimiter       *rate.Limiter
+	loginLimiter      *rate.Limiter
+	loginLimiterStore sync.Map
+	loginPeerLimit    rate.Limit
 }
 
 // NewServer creates a new Management server
@@ -76,6 +85,41 @@ func NewServer(
 		}
 	}
 
+	multiplier := time.Second
+	d, e := time.ParseDuration(os.Getenv("NB_LOGIN_RATE"))
+	if e == nil {
+		multiplier = d
+	}
+
+	loginRatePerS, err := strconv.Atoi(os.Getenv("NB_LOGIN_RATE_PER_M"))
+	if loginRatePerS == 0 || err != nil {
+		loginRatePerS = 200
+	}
+
+	loginBurst, err := strconv.Atoi(os.Getenv("NB_LOGIN_BURST"))
+	if loginBurst == 0 || err != nil {
+		loginBurst = 200
+	}
+	log.WithContext(ctx).Infof("login burst limit set to %d", loginBurst)
+
+	loginPeerRatePerS, err := strconv.Atoi(os.Getenv("NB_LOGIN_PEER_RATE_PER_M"))
+	if loginPeerRatePerS == 0 || err != nil {
+		loginPeerRatePerS = 200
+	}
+	log.WithContext(ctx).Infof("login rate limit set to %d/min", loginRatePerS)
+
+	syncRatePerS, err := strconv.Atoi(os.Getenv("NB_SYNC_RATE_PER_M"))
+	if syncRatePerS == 0 || err != nil {
+		syncRatePerS = 200
+	}
+	log.WithContext(ctx).Infof("sync rate limit set to %d/min", syncRatePerS)
+
+	syncBurst, err := strconv.Atoi(os.Getenv("NB_SYNC_BURST"))
+	if syncBurst == 0 || err != nil {
+		syncBurst = 200
+	}
+	log.WithContext(ctx).Infof("sync burst limit set to %d", syncBurst)
+
 	return &GRPCServer{
 		wgKey: key,
 		// peerKey -> event channel
@@ -87,6 +131,9 @@ func NewServer(
 		authManager:        authManager,
 		appMetrics:         appMetrics,
 		ephemeralManager:   ephemeralManager,
+		syncLimiter:        rate.NewLimiter(rate.Every(time.Minute/time.Duration(syncRatePerS)), syncBurst),
+		loginLimiter:       rate.NewLimiter(rate.Every(multiplier/time.Duration(loginRatePerS)), loginBurst),
+		loginPeerLimit:     rate.Every(time.Minute / time.Duration(loginPeerRatePerS)),
 	}, nil
 }
 
@@ -128,11 +175,18 @@ func getRealIP(ctx context.Context) net.IP {
 // Sync validates the existence of a connecting peer, sends an initial state (all available for the connecting peers) and
 // notifies the connected peer of any updates (e.g. new peers under the same account)
 func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_SyncServer) error {
-	reqStart := time.Now()
 	if s.appMetrics != nil {
 		s.appMetrics.GRPCMetrics().CountSyncRequest()
 	}
 
+	if !s.syncLimiter.Allow() {
+		time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
+		log.Warnf("sync rate limit exceeded for peer %s", req.WgPubKey)
+		return status.Errorf(codes.Internal, "temp rate limit reached")
+	}
+
+	reqStart := time.Now()
+
 	ctx := srv.Context()
 
 	syncReq := &proto.SyncRequest{}
@@ -416,15 +470,58 @@ func (s *GRPCServer) parseRequest(ctx context.Context, req *proto.EncryptedMessa
 // In case it isn't, the endpoint checks whether setup key is provided within the request and tries to register a peer.
 // In case of the successful registration login is also successful
 func (s *GRPCServer) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
+	if s.appMetrics != nil {
+		s.appMetrics.GRPCMetrics().CountLoginRequest()
+	}
+
+	limiterIface, ok := s.loginLimiterStore.Load(req.WgPubKey)
+	if !ok {
+		// Check global limiter before allowing a new peer limiter
+		if !s.loginLimiter.Allow() {
+			time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
+			log.WithContext(ctx).Warnf("rate limit exceeded for peer %s", req.WgPubKey)
+			return nil, fmt.Errorf("temp rate limit reached (global limit)")
+		}
+
+		// Create new limiter for this peer
+		newLimiter := rate.NewLimiter(s.loginPeerLimit, 1000)
+		s.loginLimiterStore.Store(req.WgPubKey, newLimiter)
+
+		if !newLimiter.Allow() {
+			time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
+			log.WithContext(ctx).Warnf("rate limit exceeded for peer %s", req.WgPubKey)
+			return nil, fmt.Errorf("temp rate limit reached (new peer limit)")
+		}
+	} else {
+		// Use existing limiter for this peer
+		limiter := limiterIface.(*rate.Limiter)
+		if !limiter.Allow() {
+			time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
+			log.WithContext(ctx).Warnf("rate limit exceeded for peer %s", req.WgPubKey)
+			return nil, fmt.Errorf("temp rate limit reached (peer limit)")
+		}
+	}
+
+	// limiter, _ := s.loginLimiterStore.LoadOrStore(req.WgPubKey, rate.NewLimiter(s.loginPeerLimit, 1))
+	// if !limiter.(*rate.Limiter).Allow() {
+	//	time.Sleep(time.Millisecond * time.Duration(rand.IntN(10)*100))
+	//	log.WithContext(ctx).Warnf("rate limit exceeded for %s", req.WgPubKey)
+	//	return nil, status.Errorf(codes.Internal, "temp rate limit reached")
+	// }
+	//
+	// if os.Getenv("ENABLE_LOGIN_RATE_LIMIT") == "true" {
+	//	if !s.loginLimiter.Allow() {
+	//		return nil, status.Errorf(codes.Internal, "temp rate limit reached")
+	//	}
+	// }
+
 	reqStart := time.Now()
 	defer func() {
 		if s.appMetrics != nil {
 			s.appMetrics.GRPCMetrics().CountLoginRequestDuration(time.Since(reqStart))
 		}
 	}()
-	if s.appMetrics != nil {
-		s.appMetrics.GRPCMetrics().CountLoginRequest()
-	}
+
 	realIP := getRealIP(ctx)
 	log.WithContext(ctx).Debugf("Login request from peer [%s] [%s]", req.WgPubKey, realIP.String())
 

management/server/http/handlers/peers/peers_handler.go

@@ -4,10 +4,17 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net"
 	"net/http"
+	"os"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
 
 	"github.com/gorilla/mux"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/time/rate"
 
 	"github.com/netbirdio/netbird/management/server/account"
 	"github.com/netbirdio/netbird/management/server/activity"
@@ -23,6 +30,9 @@ import (
 // Handler is a handler that returns peers of the account
 type Handler struct {
 	accountManager account.Manager
+	rateLimiter    *rate.Limiter
+	limiterStore   sync.Map
+	reqLimit       rate.Limit
 }
 
 func AddEndpoints(accountManager account.Manager, router *mux.Router) {
@@ -35,8 +45,15 @@ func AddEndpoints(accountManager account.Manager, router *mux.Router) {
 
 // NewHandler creates a new peers Handler
 func NewHandler(accountManager account.Manager) *Handler {
+	apiRatePerM, err := strconv.Atoi(os.Getenv("NB_API_RATE_PER_M"))
+	if apiRatePerM == 0 || err != nil {
+		apiRatePerM = 60
+	}
+	log.Infof("peers API rate limit set to %d/min", apiRatePerM)
 	return &Handler{
 		accountManager: accountManager,
+		rateLimiter:    rate.NewLimiter(rate.Every(time.Minute/time.Duration(apiRatePerM)), 1),
+		reqLimit:       rate.Every(time.Minute / time.Duration(apiRatePerM)),
 	}
 }
 
@@ -54,6 +71,11 @@ func (h *Handler) checkPeerStatus(peer *nbpeer.Peer) (*nbpeer.Peer, error) {
 }
 
 func (h *Handler) getPeer(ctx context.Context, accountID, peerID, userID string, w http.ResponseWriter) {
+	if !h.rateLimiter.Allow() {
+		util.WriteError(ctx, fmt.Errorf("temp rate limit reached"), w)
+		return
+	}
+
 	peer, err := h.accountManager.GetPeer(ctx, accountID, peerID, userID)
 	if err != nil {
 		util.WriteError(ctx, err, w)
@@ -91,7 +113,7 @@ func (h *Handler) updatePeer(ctx context.Context, accountID, userID, peerID stri
 	req := &api.PeerRequest{}
 	err := json.NewDecoder(r.Body).Decode(&req)
 	if err != nil {
-		util.WriteErrorResponse("couldn't parse JSON request", http.StatusBadRequest, w)
+		util.WriteErrorResponse("couldn't parse JSON request", http.StatusPreconditionRequired, w)
 		return
 	}
 
@@ -184,9 +206,40 @@ func (h *Handler) HandlePeer(w http.ResponseWriter, r *http.Request) {
 		util.WriteError(r.Context(), status.Errorf(status.NotFound, "unknown METHOD"), w)
 	}
 }
+func getCallerIP(r *http.Request) string {
+	// Check X-Forwarded-For header first (can be a comma-separated list)
+	if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
+		// Use first IP in the list
+		parts := strings.Split(xff, ",")
+		return strings.TrimSpace(parts[0])
+	}
+
+	// Then check X-Real-IP
+	if xrip := r.Header.Get("X-Real-IP"); xrip != "" {
+		return xrip
+	}
+
+	// Fallback to RemoteAddr
+	ip, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		return r.RemoteAddr // may be raw IP
+	}
+	return ip
+}
 
 // GetAllPeers returns a list of all peers associated with a provided account
 func (h *Handler) GetAllPeers(w http.ResponseWriter, r *http.Request) {
+	ip := getCallerIP(r)
+	limiter, _ := h.limiterStore.LoadOrStore(ip, rate.NewLimiter(h.reqLimit, 1))
+	if !limiter.(*rate.Limiter).Allow() {
+		log.WithContext(r.Context()).Errorf("rate limit exceeded for IP: %s", ip)
+		util.WriteError(r.Context(), status.Errorf(status.StatusTooManyRequests, "temp rate limit reached"), w)
+		return
+	}
+	//if !h.rateLimiter.Allow() {
+	//	util.WriteError(r.Context(), status.Errorf(status.StatusTooManyRequests, "temp rate limit reached"), w)
+	//	return
+	//}
 	userAuth, err := nbcontext.GetUserAuthFromContext(r.Context())
 	if err != nil {
 		util.WriteError(r.Context(), err, w)

management/server/http/util/util.go

@@ -106,6 +106,8 @@ func WriteError(ctx context.Context, err error, w http.ResponseWriter) {
 			httpStatus = http.StatusUnauthorized
 		case status.BadRequest:
 			httpStatus = http.StatusBadRequest
+		case status.StatusTooManyRequests:
+			httpStatus = http.StatusTooManyRequests
 		default:
 		}
 		msg = strings.ToLower(err.Error())

management/server/integrations/port_forwarding/controller.go

@@ -3,12 +3,14 @@ package port_forwarding
 import (
 	"context"
 
+	"github.com/netbirdio/netbird/management/server/peer"
 	nbtypes "github.com/netbirdio/netbird/management/server/types"
 )
 
 type Controller interface {
-	SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string)
-	GetProxyNetworkMaps(ctx context.Context, accountID string) (map[string]*nbtypes.NetworkMap, error)
+	SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string, accountPeers map[string]*peer.Peer)
+	GetProxyNetworkMaps(ctx context.Context, accountID, peerID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error)
+	GetProxyNetworkMapsAll(ctx context.Context, accountID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error)
 	IsPeerInIngressPorts(ctx context.Context, accountID, peerID string) (bool, error)
 }
 
@@ -19,11 +21,15 @@ func NewControllerMock() *ControllerMock {
 	return &ControllerMock{}
 }
 
-func (c *ControllerMock) SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string) {
+func (c *ControllerMock) SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string, accountPeers map[string]*peer.Peer) {
 	// noop
 }
 
-func (c *ControllerMock) GetProxyNetworkMaps(ctx context.Context, accountID string) (map[string]*nbtypes.NetworkMap, error) {
+func (c *ControllerMock) GetProxyNetworkMaps(ctx context.Context, accountID, peerID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error) {
+	return make(map[string]*nbtypes.NetworkMap), nil
+}
+
+func (c *ControllerMock) GetProxyNetworkMapsAll(ctx context.Context, accountID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error) {
 	return make(map[string]*nbtypes.NetworkMap), nil
 }
 

management/server/mock_server/account_mock.go

@@ -889,3 +889,7 @@ func (am *MockAccountManager) GetCurrentUserInfo(ctx context.Context, userAuth n
 	}
 	return nil, status.Errorf(codes.Unimplemented, "method GetCurrentUserInfo is not implemented")
 }
+
+func (am *MockAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string) {
+	// noop
+}

management/server/peer.go

@@ -419,7 +419,7 @@ func (am *DefaultAccountManager) GetNetworkMap(ctx context.Context, peerID strin
 	}
 	customZone := account.GetPeersCustomZone(ctx, am.GetDNSDomain(account.Settings))
 
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id)
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id, peerID, account.Peers)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return nil, err
@@ -453,6 +453,7 @@ func (am *DefaultAccountManager) GetPeerNetwork(ctx context.Context, peerID stri
 // Each new Peer will be assigned a new next net.IP from the Account.Network and Account.Network.LastIP will be updated (IP's are not reused).
 // The peer property is just a placeholder for the Peer properties to pass further
 func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID string, peer *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, error) {
+	startGlobal := time.Now()
 	if setupKey == "" && userID == "" {
 		// no auth method provided => reject access
 		return nil, nil, nil, status.Errorf(status.Unauthenticated, "no peer auth method provided, please use a setup key or interactive SSO login")
@@ -505,6 +506,8 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 		var ephemeral bool
 		var groupsToAdd []string
 		var allowExtraDNSLabels bool
+
+		start := time.Now()
 		if addedByUser {
 			user, err := transaction.GetUserByUserID(ctx, store.LockingStrengthUpdate, userID)
 			if err != nil {
@@ -537,6 +540,9 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
+		log.WithContext(ctx).Debugf("AddPeer: setup key get took %v", time.Since(start))
+
+		start = time.Now()
 		if (strings.ToLower(peer.Meta.Hostname) == "iphone" || strings.ToLower(peer.Meta.Hostname) == "ipad") && userID != "" {
 			if am.idpManager != nil {
 				userdata, err := am.idpManager.GetUserDataByID(ctx, userID, idp.AppMetadata{WTAccountID: accountID})
@@ -545,16 +551,21 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 				}
 			}
 		}
+		log.WithContext(ctx).Debugf("AddPeer: idp took %v", time.Since(start))
 
+		start = time.Now()
 		freeLabel, err := am.getFreeDNSLabel(ctx, transaction, accountID, peer.Meta.Hostname)
 		if err != nil {
 			return fmt.Errorf("failed to get free DNS label: %w", err)
 		}
+		log.WithContext(ctx).Debugf("AddPeer: free label took %v", time.Since(start))
 
+		start = time.Now()
 		freeIP, err := getFreeIP(ctx, transaction, accountID)
 		if err != nil {
 			return fmt.Errorf("failed to get free IP: %w", err)
 		}
+		log.WithContext(ctx).Debugf("AddPeer: ip took %v", time.Since(start))
 
 		registrationTime := time.Now().UTC()
 		newPeer = &nbpeer.Peer{
@@ -578,17 +589,22 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			ExtraDNSLabels:              peer.ExtraDNSLabels,
 			AllowExtraDNSLabels:         allowExtraDNSLabels,
 		}
+
+		start = time.Now()
 		settings, err := transaction.GetAccountSettings(ctx, store.LockingStrengthShare, accountID)
 		if err != nil {
 			return fmt.Errorf("failed to get account settings: %w", err)
 		}
 
+		log.WithContext(ctx).Debugf("AddPeer: settings took %v", time.Since(start))
+
 		opEvent.TargetID = newPeer.ID
 		opEvent.Meta = newPeer.EventMeta(am.GetDNSDomain(settings))
 		if !addedByUser {
 			opEvent.Meta["setup_key_name"] = setupKeyName
 		}
 
+		start = time.Now()
 		if am.geo != nil && newPeer.Location.ConnectionIP != nil {
 			location, err := am.geo.Lookup(newPeer.Location.ConnectionIP)
 			if err != nil {
@@ -600,8 +616,11 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
+		log.WithContext(ctx).Debugf("AddPeer: geo took %v", time.Since(start))
+
 		newPeer = am.integratedPeerValidator.PreparePeer(ctx, accountID, newPeer, groupsToAdd, settings.Extra)
 
+		start = time.Now()
 		err = transaction.AddPeerToAllGroup(ctx, store.LockingStrengthUpdate, accountID, newPeer.ID)
 		if err != nil {
 			return fmt.Errorf("failed adding peer to All group: %w", err)
@@ -616,11 +635,16 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
+		log.WithContext(ctx).Debugf("AddPeer: add peer to group took %v", time.Since(start))
+
+		start = time.Now()
 		err = transaction.AddPeerToAccount(ctx, store.LockingStrengthUpdate, newPeer)
 		if err != nil {
 			return fmt.Errorf("failed to add peer to account: %w", err)
 		}
 
+		log.WithContext(ctx).Debugf("AddPeer: add peer to account took %v", time.Since(start))
+
 		err = transaction.IncrementNetworkSerial(ctx, store.LockingStrengthUpdate, accountID)
 		if err != nil {
 			return fmt.Errorf("failed to increment network serial: %w", err)
@@ -638,11 +662,14 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
+		start = time.Now()
 		updateAccountPeers, err = isPeerInActiveGroup(ctx, transaction, accountID, newPeer.ID)
 		if err != nil {
 			return err
 		}
 
+		log.WithContext(ctx).Debugf("AddPeer: is peer in active group took %v", time.Since(start))
+
 		log.WithContext(ctx).Debugf("Peer %s added to account %s", newPeer.ID, accountID)
 		return nil
 	})
@@ -657,8 +684,7 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 
 	am.StoreEvent(ctx, opEvent.InitiatorID, opEvent.TargetID, opEvent.AccountID, opEvent.Activity, opEvent.Meta)
 
-	unlock()
-	unlock = nil
+	log.WithContext(ctx).Debugf("AddPeer took %v", time.Since(startGlobal))
 
 	if updateAccountPeers {
 		am.BufferUpdateAccountPeers(ctx, accountID)
@@ -997,48 +1023,56 @@ func (am *DefaultAccountManager) checkIFPeerNeedsLoginWithoutLock(ctx context.Co
 }
 
 func (am *DefaultAccountManager) getValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID string, peer *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, error) {
-	start := time.Now()
+	mstart := time.Now()
 	defer func() {
-		log.WithContext(ctx).Debugf("getValidatedPeerWithMap: took %s", time.Since(start))
+		log.WithContext(ctx).Debugf("getValidatedPeerWithMap: took %s", time.Since(mstart))
 	}()
 
 	if isRequiresApproval {
+		start := time.Now()
 		network, err := am.Store.GetAccountNetwork(ctx, store.LockingStrengthShare, accountID)
 		if err != nil {
 			return nil, nil, nil, err
 		}
 
+		log.WithContext(ctx).Debugf("getValidatedPeerWithMap: took %s", time.Since(start))
+
 		emptyMap := &types.NetworkMap{
 			Network: network.Copy(),
 		}
 		return peer, emptyMap, nil, nil
 	}
-
+	start := time.Now()
 	account, err := am.requestBuffer.GetAccountWithBackpressure(ctx, accountID)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-
+	log.WithContext(ctx).Debugf("GetAccountWithBackpressure: took %s", time.Since(start))
+	start = time.Now()
 	approvedPeersMap, err := am.integratedPeerValidator.GetValidatedPeers(account.Id, maps.Values(account.Groups), maps.Values(account.Peers), account.Settings.Extra)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-
+	log.WithContext(ctx).Debugf("GetValidatedPeers: took %s", time.Since(start))
+	start = time.Now()
 	postureChecks, err := am.getPeerPostureChecks(account, peer.ID)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-
+	log.WithContext(ctx).Debugf("getPeerPostureChecks: took %s", time.Since(start))
+	start = time.Now()
 	customZone := account.GetPeersCustomZone(ctx, am.GetDNSDomain(account.Settings))
-
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id)
+	log.WithContext(ctx).Debugf("GetPeersCustomZone: took %s", time.Since(start))
+	start = time.Now()
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id, peer.ID, account.Peers)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return nil, nil, nil, err
 	}
-
+	log.WithContext(ctx).Debugf("GetProxyNetworkMaps: took %s", time.Since(start))
+	start = time.Now()
 	networkMap := account.GetPeerNetworkMap(ctx, peer.ID, customZone, approvedPeersMap, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), am.metrics.AccountManagerMetrics())
-
+	log.WithContext(ctx).Debugf("GetPeerNetworkMap: took %s", time.Since(start))
 	proxyNetworkMap, ok := proxyNetworkMaps[peer.ID]
 	if ok {
 		networkMap.Merge(proxyNetworkMap)
@@ -1166,13 +1200,16 @@ func (am *DefaultAccountManager) checkIfUserOwnsPeer(ctx context.Context, accoun
 // UpdateAccountPeers updates all peers that belong to an account.
 // Should be called when changes have to be synced to peers.
 func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, accountID string) {
+	globalStart := time.Now()
+	start := time.Now()
 	account, err := am.requestBuffer.GetAccountWithBackpressure(ctx, accountID)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to send out updates to peers. failed to get account: %v", err)
 		return
 	}
+	log.WithContext(ctx).Infof("updateAccountPeers: getAccount took %s", time.Since(start))
 
-	start := time.Now()
+	start = time.Now()
 
 	approvedPeersMap, err := am.integratedPeerValidator.GetValidatedPeers(account.Id, maps.Values(account.Groups), maps.Values(account.Peers), account.Settings.Extra)
 	if err != nil {
@@ -1180,6 +1217,8 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 		return
 	}
 
+	log.WithContext(ctx).Infof("updateAccountPeers: validatePeers took %s", time.Since(start))
+
 	var wg sync.WaitGroup
 	semaphore := make(chan struct{}, 10)
 
@@ -1189,11 +1228,21 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 	resourcePolicies := account.GetResourcePoliciesMap()
 	routers := account.GetResourceRoutersMap()
 
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, accountID)
+	start = time.Now()
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMapsAll(ctx, accountID, account.Peers)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return
 	}
+	log.WithContext(ctx).Infof("updateAccountPeers: getProxyNetworkMaps took %s", time.Since(start))
+	for _, id := range []string{"d07kd1ei389c73dq19gg", "d07kcaui389c73dq19g0", "d0e7uo6i389c73f040v0"} {
+		peerMap, ok := proxyNetworkMaps[id]
+		if !ok {
+			log.WithContext(ctx).Infof("updateAccountPeers xxx: proxy network map %s not found", id)
+			continue
+		}
+		log.WithContext(ctx).Infof("updateAccountPeers xxx: peer %s has %d peers, %d offline peers, %d, firewall rules, %d forwarding rules, %d routing rules", id, len(peerMap.Peers), len(peerMap.OfflinePeers), len(peerMap.FirewallRules), len(peerMap.ForwardingRules), len(peerMap.RoutesFirewallRules))
+	}
 
 	for _, peer := range account.Peers {
 		if !am.peersUpdateManager.HasChannel(peer.ID) {
@@ -1226,16 +1275,22 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 				return
 			}
 
+			start = time.Now()
 			update := toSyncResponse(ctx, nil, p, nil, nil, remotePeerNetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings.RoutingPeerDNSResolutionEnabled, extraSetting)
+			log.WithContext(ctx).Infof("updateAccountPeers: toSyncResponse took %s", time.Since(start))
+			start = time.Now()
 			am.peersUpdateManager.SendUpdate(ctx, p.ID, &UpdateMessage{Update: update, NetworkMap: remotePeerNetworkMap})
+			log.WithContext(ctx).Infof("updateAccountPeers: sending update toSyncResponse took %s", time.Since(start))
 		}(peer)
 	}
 
 	//
 
 	wg.Wait()
+	log.WithContext(ctx).Infof("updateAccountPeers: waiting for updates to complete took %s", time.Since(globalStart))
+
 	if am.metrics != nil {
-		am.metrics.AccountManagerMetrics().CountUpdateAccountPeersDuration(time.Since(start))
+		am.metrics.AccountManagerMetrics().CountUpdateAccountPeersDuration(time.Since(globalStart))
 	}
 }
 
@@ -1292,7 +1347,7 @@ func (am *DefaultAccountManager) UpdateAccountPeer(ctx context.Context, accountI
 		return
 	}
 
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, accountId)
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, accountId, peerId, account.Peers)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return

management/server/peer/peer.go

@@ -24,7 +24,7 @@ type Peer struct {
 	// Meta is a Peer system meta data
 	Meta PeerSystemMeta `gorm:"embedded;embeddedPrefix:meta_"`
 	// Name is peer's name (machine name)
-	Name string
+	Name string `gorm:"index"`
 	// DNSLabel is the parsed peer name for domain resolution. It is used to form an FQDN by appending the account's
 	// domain to the peer label. e.g. peer-dns-label.netbird.cloud
 	DNSLabel string

management/server/status/error.go

@@ -37,6 +37,8 @@ const (
 
 	// Unauthenticated indicates that user is not authenticated due to absence of valid credentials
 	Unauthenticated Type = 10
+
+	StatusTooManyRequests = 11
 )
 
 // Type is a type of the Error

management/server/store/sql_store.go

@@ -1311,7 +1311,7 @@ func (s *SqlStore) GetAccountPeers(ctx context.Context, lockStrength LockingStre
 	query := s.db.Clauses(clause.Locking{Strength: string(lockStrength)}).Where(accountIDCondition, accountID)
 
 	if nameFilter != "" {
-		query = query.Where("name LIKE ?", "%"+nameFilter+"%")
+		query = query.Where("name = ?", nameFilter)
 	}
 	if ipFilter != "" {
 		query = query.Where("ip LIKE ?", "%"+ipFilter+"%")

management/server/telemetry/app_metrics.go

@@ -29,7 +29,6 @@ type MockAppMetrics struct {
 	StoreMetricsFunc             func() *StoreMetrics
 	UpdateChannelMetricsFunc     func() *UpdateChannelMetrics
 	AddAccountManagerMetricsFunc func() *AccountManagerMetrics
-	EventStreamingMetricsFunc    func() *EventStreamingMetrics
 }
 
 // GetMeter mocks the GetMeter function of the AppMetrics interface
@@ -104,14 +103,6 @@ func (mock *MockAppMetrics) AccountManagerMetrics() *AccountManagerMetrics {
 	return nil
 }
 
-// EventStreamingMetrics mocks the EventStreamingMetrics function of the AppMetrics interface
-func (mock *MockAppMetrics) EventStreamingMetrics() *EventStreamingMetrics {
-	if mock.EventStreamingMetricsFunc != nil {
-		return mock.EventStreamingMetricsFunc()
-	}
-	return nil
-}
-
 // AppMetrics is metrics interface
 type AppMetrics interface {
 	GetMeter() metric2.Meter
@@ -123,7 +114,6 @@ type AppMetrics interface {
 	StoreMetrics() *StoreMetrics
 	UpdateChannelMetrics() *UpdateChannelMetrics
 	AccountManagerMetrics() *AccountManagerMetrics
-	EventStreamingMetrics() *EventStreamingMetrics
 }
 
 // defaultAppMetrics are core application metrics based on OpenTelemetry https://opentelemetry.io/
@@ -138,7 +128,6 @@ type defaultAppMetrics struct {
 	storeMetrics          *StoreMetrics
 	updateChannelMetrics  *UpdateChannelMetrics
 	accountManagerMetrics *AccountManagerMetrics
-	eventStreamingMetrics *EventStreamingMetrics
 }
 
 // IDPMetrics returns metrics for the idp package
@@ -171,16 +160,6 @@ func (appMetrics *defaultAppMetrics) AccountManagerMetrics() *AccountManagerMetr
 	return appMetrics.accountManagerMetrics
 }
 
-// EventStreamingMetrics returns metrics for the event streaming module
-func (appMetrics *defaultAppMetrics) EventStreamingMetrics() *EventStreamingMetrics {
-	return appMetrics.eventStreamingMetrics
-}
-
-// RegisterEventStreamingMetrics sets the event streaming metrics
-func (appMetrics *defaultAppMetrics) RegisterEventStreamingMetrics(metrics *EventStreamingMetrics) {
-	appMetrics.eventStreamingMetrics = metrics
-}
-
 // Close stop application metrics HTTP handler and closes listener.
 func (appMetrics *defaultAppMetrics) Close() error {
 	if appMetrics.listener == nil {
@@ -205,10 +184,10 @@ func (appMetrics *defaultAppMetrics) Expose(ctx context.Context, port int, endpo
 	}
 	appMetrics.listener = listener
 	go func() {
-		if err := http.Serve(listener, rootRouter); err != nil && err != http.ErrServerClosed {
-			log.WithContext(ctx).Errorf("metrics server error: %v", err)
+		err := http.Serve(listener, rootRouter)
+		if err != nil {
+			return
 		}
-		log.WithContext(ctx).Info("metrics server stopped")
 	}()
 
 	log.WithContext(ctx).Infof("enabled application metrics and exposing on http://%s", listener.Addr().String())
@@ -225,7 +204,7 @@ func (appMetrics *defaultAppMetrics) GetMeter() metric2.Meter {
 func NewDefaultAppMetrics(ctx context.Context) (AppMetrics, error) {
 	exporter, err := prometheus.New()
 	if err != nil {
-		return nil, fmt.Errorf("failed to create prometheus exporter: %w", err)
+		return nil, err
 	}
 
 	provider := metric.NewMeterProvider(metric.WithReader(exporter))
@@ -234,32 +213,32 @@ func NewDefaultAppMetrics(ctx context.Context) (AppMetrics, error) {
 
 	idpMetrics, err := NewIDPMetrics(ctx, meter)
 	if err != nil {
-		return nil, fmt.Errorf("failed to initialize IDP metrics: %w", err)
+		return nil, err
 	}
 
 	middleware, err := NewMetricsMiddleware(ctx, meter)
 	if err != nil {
-		return nil, fmt.Errorf("failed to initialize HTTP middleware metrics: %w", err)
+		return nil, err
 	}
 
 	grpcMetrics, err := NewGRPCMetrics(ctx, meter)
 	if err != nil {
-		return nil, fmt.Errorf("failed to initialize gRPC metrics: %w", err)
+		return nil, err
 	}
 
 	storeMetrics, err := NewStoreMetrics(ctx, meter)
 	if err != nil {
-		return nil, fmt.Errorf("failed to initialize store metrics: %w", err)
+		return nil, err
 	}
 
 	updateChannelMetrics, err := NewUpdateChannelMetrics(ctx, meter)
 	if err != nil {
-		return nil, fmt.Errorf("failed to initialize update channel metrics: %w", err)
+		return nil, err
 	}
 
 	accountManagerMetrics, err := NewAccountManagerMetrics(ctx, meter)
 	if err != nil {
-		return nil, fmt.Errorf("failed to initialize account manager metrics: %w", err)
+		return nil, err
 	}
 
 	return &defaultAppMetrics{

sharedsock/sock_linux.go

@@ -234,7 +234,7 @@ func (s *SharedSocket) read(receiver receiver) {
 }
 
 // ReadFrom reads packets received in the packetDemux channel
-func (s *SharedSocket) ReadFrom(b []byte) (int, net.Addr, error) {
+func (s *SharedSocket) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
 	var pkt rcvdPacket
 	select {
 	case <-s.ctx.Done():
@@ -263,7 +263,8 @@ func (s *SharedSocket) ReadFrom(b []byte) (int, net.Addr, error) {
 
 	decodedLayers := make([]gopacket.LayerType, 0, 3)
 
-	if err := parser.DecodeLayers(pkt.buf, &decodedLayers); err != nil {
+	err = parser.DecodeLayers(pkt.buf, &decodedLayers)
+	if err != nil {
 		return 0, nil, err
 	}
 
@@ -272,8 +273,8 @@ func (s *SharedSocket) ReadFrom(b []byte) (int, net.Addr, error) {
 		Port: int(udp.SrcPort),
 	}
 
-	n := copy(b, payload)
-	return n, remoteAddr, nil
+	copy(b, payload)
+	return int(udp.Length), remoteAddr, nil
 }
 
 // WriteTo builds a UDP packet and writes it using the specific IP version writer