Merge remote-tracking branch 'origin/main' into dmitri-event-aggregation

Signed-off-by: Dmitri Dolguikh <dmitri.external@netbird.io>

client/iface/wgproxy/bind/proxy.go

@@ -136,6 +136,11 @@ func (p *ProxyBind) CloseConn() error {
 	return p.close()
 }
 
+// InjectPacket is a no-op for the userspace proxy: first-packet reinjection is kernel-only.
+func (p *ProxyBind) InjectPacket(_ []byte) error {
+	return nil
+}
+
 func (p *ProxyBind) close() error {
 	if p.remoteConn == nil {
 		return nil

client/iface/wgproxy/ebpf/wrapper.go

@@ -219,6 +219,17 @@ func (p *ProxyWrapper) RedirectAs(endpoint *net.UDPAddr) {
 	p.pausedCond.L.Unlock()
 }
 
+// InjectPacket writes b to the remote peer over the underlying transport.
+func (p *ProxyWrapper) InjectPacket(b []byte) error {
+	if p.remoteConn == nil {
+		return errors.New("proxy not started")
+	}
+	if _, err := p.remoteConn.Write(b); err != nil {
+		return err
+	}
+	return nil
+}
+
 // CloseConn close the remoteConn and automatically remove the conn instance from the map
 func (p *ProxyWrapper) CloseConn() error {
 	if p.cancel == nil {

client/iface/wgproxy/proxy.go

@@ -18,4 +18,9 @@ type Proxy interface {
 	RedirectAs(endpoint *net.UDPAddr)
 	CloseConn() error
 	SetDisconnectListener(disconnected func())
+
+	// InjectPacket writes a raw packet directly to the remote peer over the underlying transport,
+	// bypassing WireGuard. Used to replay the captured lazyconn handshake initiation. Only the
+	// kernel-mode proxies act on it; the userspace proxy is a no-op since reinjection is kernel-only.
+	InjectPacket(b []byte) error
 }

client/iface/wgproxy/udp/proxy.go

@@ -147,6 +147,17 @@ func (p *WGUDPProxy) RedirectAs(endpoint *net.UDPAddr) {
 	p.sendPkg = p.srcFakerConn.SendPkg
 }
 
+// InjectPacket writes b to the remote peer over the underlying transport.
+func (p *WGUDPProxy) InjectPacket(b []byte) error {
+	if p.remoteConn == nil {
+		return errors.New("proxy not started")
+	}
+	if _, err := p.remoteConn.Write(b); err != nil {
+		return err
+	}
+	return nil
+}
+
 // CloseConn close the localConn
 func (p *WGUDPProxy) CloseConn() error {
 	if p.cancel == nil {

client/internal/acl/manager.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/hashicorp/go-multierror"
+	"github.com/mitchellh/hashstructure/v2"
 	log "github.com/sirupsen/logrus"
 
 	nberrors "github.com/netbirdio/netbird/client/errors"
@@ -30,11 +31,13 @@ type Manager interface {
 
 // DefaultManager uses firewall manager to handle
 type DefaultManager struct {
-	firewall       firewall.Manager
-	ipsetCounter   int
-	peerRulesPairs map[id.RuleID][]firewall.Rule
-	routeRules     map[id.RuleID]struct{}
-	mutex          sync.Mutex
+	firewall           firewall.Manager
+	ipsetCounter       int
+	peerRulesPairs     map[id.RuleID][]firewall.Rule
+	routeRules         map[id.RuleID]struct{}
+	previousConfigHash uint64
+	hasAppliedConfig   bool
+	mutex              sync.Mutex
 }
 
 func NewDefaultManager(fm firewall.Manager) *DefaultManager {
@@ -57,6 +60,23 @@ func (d *DefaultManager) ApplyFiltering(networkMap *mgmProto.NetworkMap, dnsRout
 		return
 	}
 
+	// Skip the full rebuild + flush when the inputs that drive the firewall
+	// state are byte-for-byte identical to the last successfully applied
+	// update. Management re-sends the same network map far more often than it
+	// actually changes (account-wide updates, peer meta churn), and rebuilding
+	// every peer/route ACL and flushing the firewall on every such sync is the
+	// dominant client-side cost when nothing changed. Mirrors the same guard the
+	// DNS server already uses (previousConfigHash). Only the fields ApplyFiltering
+	// consumes participate in the hash, so an unrelated map change cannot mask a
+	// real ACL change.
+	hash, err := d.firewallConfigHash(networkMap, dnsRouteFeatureFlag)
+	if err != nil {
+		log.Errorf("unable to hash firewall configuration, applying unconditionally: %v", err)
+	} else if d.hasAppliedConfig && d.previousConfigHash == hash {
+		log.Debugf("not applying the firewall configuration update as there is nothing new (hash: %d)", hash)
+		return
+	}
+
 	start := time.Now()
 	defer func() {
 		total := 0
@@ -70,13 +90,49 @@ func (d *DefaultManager) ApplyFiltering(networkMap *mgmProto.NetworkMap, dnsRout
 
 	d.applyPeerACLs(networkMap)
 
-	if err := d.applyRouteACLs(networkMap.RoutesFirewallRules, dnsRouteFeatureFlag); err != nil {
-		log.Errorf("Failed to apply route ACLs: %v", err)
+	routeErr := d.applyRouteACLs(networkMap.RoutesFirewallRules, dnsRouteFeatureFlag)
+	if routeErr != nil {
+		log.Errorf("Failed to apply route ACLs: %v", routeErr)
 	}
 
-	if err := d.firewall.Flush(); err != nil {
-		log.Error("failed to flush firewall rules: ", err)
+	flushErr := d.firewall.Flush()
+	if flushErr != nil {
+		log.Error("failed to flush firewall rules: ", flushErr)
 	}
+
+	// Only remember the hash once the firewall actually reflects this config.
+	// If applying or flushing failed, leave the previous hash untouched so the
+	// next (possibly identical) update is not skipped and gets a chance to
+	// reconcile the firewall state.
+	if err == nil && routeErr == nil && flushErr == nil {
+		d.previousConfigHash = hash
+		d.hasAppliedConfig = true
+	} else {
+		d.hasAppliedConfig = false
+	}
+}
+
+// firewallConfigHash hashes exactly the inputs ApplyFiltering uses to build the
+// firewall state, so an identical hash means an identical resulting ruleset.
+func (d *DefaultManager) firewallConfigHash(networkMap *mgmProto.NetworkMap, dnsRouteFeatureFlag bool) (uint64, error) {
+	return hashstructure.Hash(struct {
+		PeerRules            []*mgmProto.FirewallRule
+		PeerRulesIsEmpty     bool
+		RouteRules           []*mgmProto.RouteFirewallRule
+		RouteRulesIsEmpty    bool
+		DNSRouteFeatureFlag  bool
+	}{
+		PeerRules:           networkMap.GetFirewallRules(),
+		PeerRulesIsEmpty:    networkMap.GetFirewallRulesIsEmpty(),
+		RouteRules:          networkMap.GetRoutesFirewallRules(),
+		RouteRulesIsEmpty:   networkMap.GetRoutesFirewallRulesIsEmpty(),
+		DNSRouteFeatureFlag: dnsRouteFeatureFlag,
+	}, hashstructure.FormatV2, &hashstructure.HashOptions{
+		ZeroNil:         true,
+		IgnoreZeroValue: true,
+		SlicesAsSets:    true,
+		UseStringer:     true,
+	})
 }
 
 func (d *DefaultManager) applyPeerACLs(networkMap *mgmProto.NetworkMap) {

client/internal/acl/manager_test.go

@@ -1,6 +1,7 @@
 package acl
 
 import (
+	"fmt"
 	"net/netip"
 	"testing"
 
@@ -485,3 +486,149 @@ func TestPortInfoEmpty(t *testing.T) {
 		})
 	}
 }
+
+// TestApplyFilteringSkipsUnchangedConfig verifies that an identical network map
+// re-applied is recognized as a no-op (hash unchanged), while a real change to
+// any firewall-relevant input forces a re-apply (hash changes). This is the
+// guard that prevents a full ruleset rebuild + flush on every redundant sync.
+func TestApplyFilteringSkipsUnchangedConfig(t *testing.T) {
+	t.Setenv("NB_WG_KERNEL_DISABLED", "true")
+	t.Setenv(firewall.EnvForceUserspaceFirewall, "true")
+
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+
+	ifaceMock := mocks.NewMockIFaceMapper(ctrl)
+	ifaceMock.EXPECT().IsUserspaceBind().Return(true).AnyTimes()
+	ifaceMock.EXPECT().SetFilter(gomock.Any())
+	network := netip.MustParsePrefix("172.0.0.1/32")
+	ifaceMock.EXPECT().Name().Return("lo").AnyTimes()
+	ifaceMock.EXPECT().Address().Return(wgaddr.Address{
+		IP:      network.Addr(),
+		Network: network,
+	}).AnyTimes()
+	ifaceMock.EXPECT().GetWGDevice().Return(nil).AnyTimes()
+
+	fw, err := firewall.NewFirewall(ifaceMock, nil, flowLogger, false, iface.DefaultMTU)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, fw.Close(nil))
+	}()
+
+	acl := NewDefaultManager(fw)
+
+	networkMap := &mgmProto.NetworkMap{
+		FirewallRules: []*mgmProto.FirewallRule{
+			{
+				PeerIP:    "10.93.0.1",
+				Direction: mgmProto.RuleDirection_IN,
+				Action:    mgmProto.RuleAction_ACCEPT,
+				Protocol:  mgmProto.RuleProtocol_TCP,
+				Port:      "22",
+			},
+		},
+		FirewallRulesIsEmpty: false,
+	}
+
+	acl.ApplyFiltering(networkMap, false)
+	require.True(t, acl.hasAppliedConfig, "config should be marked applied after first apply")
+	firstHash := acl.previousConfigHash
+	require.NotZero(t, firstHash)
+
+	// Re-applying the identical map must not change the recorded hash: the
+	// expensive rebuild path was skipped.
+	acl.ApplyFiltering(networkMap, false)
+	assert.Equal(t, firstHash, acl.previousConfigHash,
+		"identical re-apply must be a no-op (hash unchanged)")
+
+	// A real change must produce a different hash and re-apply.
+	networkMap.FirewallRules[0].Action = mgmProto.RuleAction_DROP
+	acl.ApplyFiltering(networkMap, false)
+	assert.NotEqual(t, firstHash, acl.previousConfigHash,
+		"changing a rule's action must force a re-apply (hash changed)")
+
+	// The dnsRouteFeatureFlag also participates in the hash.
+	changedHash := acl.previousConfigHash
+	acl.ApplyFiltering(networkMap, true)
+	assert.NotEqual(t, changedHash, acl.previousConfigHash,
+		"flipping dnsRouteFeatureFlag must force a re-apply (hash changed)")
+}
+
+func buildNetworkMap(peerRules, routeRules int) *mgmProto.NetworkMap {
+	nm := &mgmProto.NetworkMap{
+		FirewallRulesIsEmpty:      peerRules == 0,
+		RoutesFirewallRulesIsEmpty: routeRules == 0,
+	}
+	for i := range peerRules {
+		nm.FirewallRules = append(nm.FirewallRules, &mgmProto.FirewallRule{
+			PeerIP:    fmt.Sprintf("10.%d.%d.%d", i>>16&0xff, i>>8&0xff, i&0xff),
+			Direction: mgmProto.RuleDirection_IN,
+			Action:    mgmProto.RuleAction_ACCEPT,
+			Protocol:  mgmProto.RuleProtocol_TCP,
+			Port:      fmt.Sprintf("%d", 1024+i%64511),
+		})
+	}
+	for i := range routeRules {
+		nm.RoutesFirewallRules = append(nm.RoutesFirewallRules, &mgmProto.RouteFirewallRule{
+			Destination:  fmt.Sprintf("192.168.%d.0/24", i%256),
+			SourceRanges: []string{fmt.Sprintf("10.0.%d.0/24", i%256)},
+			Action:       mgmProto.RuleAction_ACCEPT,
+			Protocol:     mgmProto.RuleProtocol_ALL,
+		})
+	}
+	return nm
+}
+
+func BenchmarkFirewallConfigHash_Small(b *testing.B) {
+	d := &DefaultManager{}
+	nm := buildNetworkMap(10, 5)
+	b.ResetTimer()
+	for b.Loop() {
+		_, _ = d.firewallConfigHash(nm, false)
+	}
+}
+
+func BenchmarkFirewallConfigHash_Medium(b *testing.B) {
+	d := &DefaultManager{}
+	nm := buildNetworkMap(100, 50)
+	b.ResetTimer()
+	for b.Loop() {
+		_, _ = d.firewallConfigHash(nm, false)
+	}
+}
+
+func BenchmarkFirewallConfigHash_Large(b *testing.B) {
+	d := &DefaultManager{}
+	nm := buildNetworkMap(1000, 200)
+	b.ResetTimer()
+	for b.Loop() {
+		_, _ = d.firewallConfigHash(nm, false)
+	}
+}
+
+// TestFirewallConfigHashDeterministic verifies the hash is stable for equal
+// inputs and order-independent for the rule slices (management does not
+// guarantee rule order).
+func TestFirewallConfigHashDeterministic(t *testing.T) {
+	d := &DefaultManager{}
+
+	nm1 := &mgmProto.NetworkMap{
+		FirewallRules: []*mgmProto.FirewallRule{
+			{PeerIP: "10.0.0.1", Direction: mgmProto.RuleDirection_IN, Action: mgmProto.RuleAction_ACCEPT, Protocol: mgmProto.RuleProtocol_TCP, Port: "22"},
+			{PeerIP: "10.0.0.2", Direction: mgmProto.RuleDirection_IN, Action: mgmProto.RuleAction_DROP, Protocol: mgmProto.RuleProtocol_TCP, Port: "80"},
+		},
+	}
+	// Same rules, reversed order.
+	nm2 := &mgmProto.NetworkMap{
+		FirewallRules: []*mgmProto.FirewallRule{
+			nm1.FirewallRules[1],
+			nm1.FirewallRules[0],
+		},
+	}
+
+	h1, err := d.firewallConfigHash(nm1, false)
+	require.NoError(t, err)
+	h2, err := d.firewallConfigHash(nm2, false)
+	require.NoError(t, err)
+	assert.Equal(t, h1, h2, "hash must be order-independent for rule slices")
+}

client/internal/engine.go

@@ -82,6 +82,12 @@ const (
 	PeerConnectionTimeoutMax = 45000 // ms
 	PeerConnectionTimeoutMin = 30000 // ms
 	disableAutoUpdate        = "disabled"
+
+	// systemInfoTimeout bounds how long the sync loop waits for system info / posture
+	// check gathering. The gathering runs uncancellable system calls (process scan,
+	// exec, os.Stat); without this bound a single stuck call freezes handleSync, and
+	// thus syncMsgMux, for as long as the call hangs (observed multi-minute freezes).
+	systemInfoTimeout = 15 * time.Second
 )
 
 var ErrResetConnection = fmt.Errorf("reset connection")
@@ -1084,11 +1090,22 @@ func (e *Engine) updateChecksIfNew(checks []*mgmProto.Checks) error {
 	}
 	e.checks = checks
 
-	info, err := system.GetInfoWithChecks(e.ctx, checks, e.overlayAddresses()...)
-	if err != nil {
-		log.Warnf("failed to get system info with checks: %v", err)
-		info = system.GetInfo(e.ctx)
+	info, ok := system.GetInfoWithChecksTimeout(e.ctx, systemInfoTimeout, checks, e.overlayAddresses()...)
+	if !ok {
+		// Gathering timed out; skip the meta sync this cycle rather than blocking the
+		// sync loop (and syncMsgMux) on a stuck system call. A later sync will retry.
+		return nil
 	}
+	e.applyInfoFlags(info)
+
+	if err := e.mgmClient.SyncMeta(info); err != nil {
+		return fmt.Errorf("could not sync meta: error %s", err)
+	}
+	return nil
+}
+
+// applyInfoFlags sets the engine's config-derived feature flags on the gathered system info.
+func (e *Engine) applyInfoFlags(info *system.Info) {
 	info.SetFlags(
 		e.config.RosenpassEnabled,
 		e.config.RosenpassPermissive,
@@ -1107,12 +1124,6 @@ func (e *Engine) updateChecksIfNew(checks []*mgmProto.Checks) error {
 		e.config.EnableSSHRemotePortForwarding,
 		e.config.DisableSSHAuth,
 	)
-
-	if err := e.mgmClient.SyncMeta(info); err != nil {
-		log.Errorf("could not sync meta: error %s", err)
-		return err
-	}
-	return nil
 }
 
 // overlayAddresses returns our own WireGuard overlay address (v4 and v6) so it
@@ -1272,31 +1283,15 @@ func (e *Engine) receiveManagementEvents() {
 	e.shutdownWg.Add(1)
 	go func() {
 		defer e.shutdownWg.Done()
-		info, err := system.GetInfoWithChecks(e.ctx, e.checks, e.overlayAddresses()...)
-		if err != nil {
-			log.Warnf("failed to get system info with checks: %v", err)
+		info, ok := system.GetInfoWithChecksTimeout(e.ctx, systemInfoTimeout, e.checks, e.overlayAddresses()...)
+		if !ok {
+			// Gathering timed out; connect the stream with base info so management
+			// connectivity still comes up rather than blocking here.
 			info = system.GetInfo(e.ctx)
 		}
-		info.SetFlags(
-			e.config.RosenpassEnabled,
-			e.config.RosenpassPermissive,
-			&e.config.ServerSSHAllowed,
-			e.config.DisableClientRoutes,
-			e.config.DisableServerRoutes,
-			e.config.DisableDNS,
-			e.config.DisableFirewall,
-			e.config.BlockLANAccess,
-			e.config.BlockInbound,
-			e.config.DisableIPv6,
-			e.config.LazyConnectionEnabled,
-			e.config.EnableSSHRoot,
-			e.config.EnableSSHSFTP,
-			e.config.EnableSSHLocalPortForwarding,
-			e.config.EnableSSHRemotePortForwarding,
-			e.config.DisableSSHAuth,
-		)
+		e.applyInfoFlags(info)
 
-		err = e.mgmClient.Sync(e.ctx, info, e.handleSync)
+		err := e.mgmClient.Sync(e.ctx, info, e.handleSync)
 		if err != nil {
 			// happens if management is unavailable for a long time.
 			// We want to cancel the operation of the whole client

client/internal/engine_test.go

@@ -178,6 +178,10 @@ func (m *MockWGIface) LastActivities() map[string]monotime.Time {
 	return nil
 }
 
+func (m *MockWGIface) MTU() uint16 {
+	return 1280
+}
+
 func (m *MockWGIface) SetPresharedKey(peerKey string, psk wgtypes.Key, updateOnly bool) error {
 	return nil
 }

client/internal/iface_common.go

@@ -44,4 +44,5 @@ type wgIfaceBase interface {
 	FullStats() (*configurer.Stats, error)
 	LastActivities() map[string]monotime.Time
 	SetPresharedKey(peerKey string, psk wgtypes.Key, updateOnly bool) error
+	MTU() uint16
 }

client/internal/lazyconn/activity/listener_bind.go

@@ -124,6 +124,11 @@ func (d *BindListener) ReadPackets() {
 	d.done.Done()
 }
 
+// CapturedPacket is unused in userspace bind mode: first-packet reinjection is kernel-only.
+func (d *BindListener) CapturedPacket() []byte {
+	return nil
+}
+
 // Close stops the listener and cleans up resources.
 func (d *BindListener) Close() {
 	d.peerCfg.Log.Infof("closing activity listener (LazyConn)")

client/internal/lazyconn/activity/listener_bind_test.go

@@ -45,10 +45,6 @@ type MockWGIfaceBind struct {
 	endpointMgr *mockEndpointManager
 }
 
-func (m *MockWGIfaceBind) RemovePeer(string) error {
-	return nil
-}
-
 func (m *MockWGIfaceBind) UpdatePeer(string, []netip.Prefix, time.Duration, *net.UDPAddr, *wgtypes.Key) error {
 	return nil
 }
@@ -68,6 +64,10 @@ func (m *MockWGIfaceBind) GetBind() device.EndpointManager {
 	return m.endpointMgr
 }
 
+func (m *MockWGIfaceBind) MTU() uint16 {
+	return 1280
+}
+
 func TestBindListener_Creation(t *testing.T) {
 	mockEndpointMgr := newMockEndpointManager()
 	mockIface := &MockWGIfaceBind{endpointMgr: mockEndpointMgr}
@@ -207,8 +207,9 @@ func TestManager_BindMode(t *testing.T) {
 	require.NoError(t, err)
 
 	select {
-	case peerConnID := <-mgr.OnActivityChan:
-		assert.Equal(t, cfg.PeerConnID, peerConnID, "Received peer connection ID should match")
+	case ev := <-mgr.OnActivityChan:
+		assert.Equal(t, cfg.PeerConnID, ev.PeerConnID, "Received peer connection ID should match")
+		assert.Nil(t, ev.FirstPacket, "Bind mode does not capture packets: reinjection is kernel-only")
 	case <-time.After(2 * time.Second):
 		t.Fatal("timeout waiting for activity notification")
 	}
@@ -266,8 +267,8 @@ func TestManager_BindMode_MultiplePeers(t *testing.T) {
 	receivedPeers := make(map[peerid.ConnID]bool)
 	for i := 0; i < 2; i++ {
 		select {
-		case peerConnID := <-mgr.OnActivityChan:
-			receivedPeers[peerConnID] = true
+		case ev := <-mgr.OnActivityChan:
+			receivedPeers[ev.PeerConnID] = true
 		case <-time.After(2 * time.Second):
 			t.Fatal("timeout waiting for activity notifications")
 		}

client/internal/lazyconn/activity/listener_udp.go

@@ -3,11 +3,13 @@ package activity
 import (
 	"fmt"
 	"net"
+	"slices"
 	"sync"
 	"sync/atomic"
 
 	log "github.com/sirupsen/logrus"
 
+	"github.com/netbirdio/netbird/client/iface/bufsize"
 	"github.com/netbirdio/netbird/client/internal/lazyconn"
 )
 
@@ -20,6 +22,8 @@ type UDPListener struct {
 	done     sync.Mutex
 
 	isClosed atomic.Bool
+
+	capturedPacket []byte
 }
 
 // NewUDPListener creates a listener that detects activity via UDP socket reads.
@@ -46,9 +50,13 @@ func NewUDPListener(wgIface WgInterface, cfg lazyconn.PeerConfig) (*UDPListener,
 }
 
 // ReadPackets blocks reading from the UDP socket until activity is detected or the listener is closed.
+// The first packet that triggers activity is captured so it can be reinjected through the real
+// transport once it is established. Without this, kernel WireGuard's handshake initiation would be
+// dropped and WG would only retry after REKEY_TIMEOUT.
 func (d *UDPListener) ReadPackets() {
 	for {
-		n, remoteAddr, err := d.conn.ReadFromUDP(make([]byte, 1))
+		buf := make([]byte, int(d.wgIface.MTU())+bufsize.WGBufferOverhead)
+		n, remoteAddr, err := d.conn.ReadFromUDP(buf)
 		if err != nil {
 			if d.isClosed.Load() {
 				d.peerCfg.Log.Infof("exit from activity listener")
@@ -62,20 +70,24 @@ func (d *UDPListener) ReadPackets() {
 			d.peerCfg.Log.Warnf("received %d bytes from %s, too short", n, remoteAddr)
 			continue
 		}
-		d.peerCfg.Log.Infof("activity detected")
+		d.capturedPacket = slices.Clone(buf[:n])
+		d.peerCfg.Log.Infof("activity detected, captured %d bytes for reinjection", n)
 		break
 	}
 
-	d.peerCfg.Log.Debugf("removing lazy endpoint: %s", d.endpoint.String())
-	if err := d.wgIface.RemovePeer(d.peerCfg.PublicKey); err != nil {
-		d.peerCfg.Log.Errorf("failed to remove endpoint: %s", err)
-	}
-
-	// Ignore close error as it may return "use of closed network connection" if already closed.
+	// Leave the peer in place. ConfigureWGEndpoint will UpdatePeer with the real endpoint;
+	// removing the peer here wipes kernel WG's staged queue and drops the user packet that
+	// triggered activation.
 	_ = d.conn.Close()
 	d.done.Unlock()
 }
 
+// CapturedPacket returns the first packet that triggered activity, or nil if none was captured.
+// Safe to call after ReadPackets returns.
+func (d *UDPListener) CapturedPacket() []byte {
+	return d.capturedPacket
+}
+
 // Close stops the listener and cleans up resources.
 func (d *UDPListener) Close() {
 	d.peerCfg.Log.Infof("closing activity listener: %s", d.conn.LocalAddr().String())

client/internal/lazyconn/activity/manager.go

@@ -19,17 +19,25 @@ import (
 type listener interface {
 	ReadPackets()
 	Close()
+	CapturedPacket() []byte
+}
+
+// Event reports activity on a managed peer. FirstPacket is the bytes that triggered activation,
+// captured for reinjection through the real transport.
+type Event struct {
+	PeerConnID  peerid.ConnID
+	FirstPacket []byte
 }
 
 type WgInterface interface {
-	RemovePeer(peerKey string) error
 	UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
 	IsUserspaceBind() bool
 	Address() wgaddr.Address
+	MTU() uint16
 }
 
 type Manager struct {
-	OnActivityChan chan peerid.ConnID
+	OnActivityChan chan Event
 
 	wgIface WgInterface
 
@@ -41,7 +49,7 @@ type Manager struct {
 
 func NewManager(wgIface WgInterface) *Manager {
 	m := &Manager{
-		OnActivityChan: make(chan peerid.ConnID, 1),
+		OnActivityChan: make(chan Event, 1),
 		wgIface:        wgIface,
 		peers:          make(map[peerid.ConnID]listener),
 		done:           make(chan struct{}),
@@ -116,12 +124,12 @@ func (m *Manager) waitForTraffic(l listener, peerConnID peerid.ConnID) {
 	delete(m.peers, peerConnID)
 	m.mu.Unlock()
 
-	m.notify(peerConnID)
+	m.notify(Event{PeerConnID: peerConnID, FirstPacket: l.CapturedPacket()})
 }
 
-func (m *Manager) notify(peerConnID peerid.ConnID) {
+func (m *Manager) notify(ev Event) {
 	select {
 	case <-m.done:
-	case m.OnActivityChan <- peerConnID:
+	case m.OnActivityChan <- ev:
 	}
 }

client/internal/lazyconn/activity/manager_test.go

@@ -1,6 +1,7 @@
 package activity
 
 import (
+	"bytes"
 	"net"
 	"net/netip"
 	"testing"
@@ -25,10 +26,6 @@ func (m *MocPeer) ConnID() peerid.ConnID {
 type MocWGIface struct {
 }
 
-func (m MocWGIface) RemovePeer(string) error {
-	return nil
-}
-
 func (m MocWGIface) UpdatePeer(string, []netip.Prefix, time.Duration, *net.UDPAddr, *wgtypes.Key) error {
 	return nil
 }
@@ -44,6 +41,10 @@ func (m MocWGIface) Address() wgaddr.Address {
 	}
 }
 
+func (m MocWGIface) MTU() uint16 {
+	return 1280
+}
+
 // GetPeerListener is a test helper to access listeners
 func (m *Manager) GetPeerListener(peerConnID peerid.ConnID) (listener, bool) {
 	m.mu.Lock()
@@ -86,11 +87,15 @@ func TestManager_MonitorPeerActivity(t *testing.T) {
 	}
 
 	select {
-	case peerConnID := <-mgr.OnActivityChan:
-		if peerConnID != peerCfg1.PeerConnID {
-			t.Fatalf("unexpected peerConnID: %v", peerConnID)
+	case ev := <-mgr.OnActivityChan:
+		if ev.PeerConnID != peerCfg1.PeerConnID {
+			t.Fatalf("unexpected peerConnID: %v", ev.PeerConnID)
+		}
+		if !bytes.Equal(ev.FirstPacket, []byte{0x01, 0x02, 0x03, 0x04, 0x05}) {
+			t.Fatalf("unexpected first packet: %v", ev.FirstPacket)
 		}
 	case <-time.After(1 * time.Second):
+		t.Fatal("timed out waiting for activity")
 	}
 }
 

client/internal/lazyconn/manager/manager.go

@@ -130,8 +130,8 @@ func (m *Manager) Start(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case peerConnID := <-m.activityManager.OnActivityChan:
-			m.onPeerActivity(peerConnID)
+		case ev := <-m.activityManager.OnActivityChan:
+			m.onPeerActivity(ev)
 		case peerIDs := <-m.inactivityManager.InactivePeersChan():
 			m.onPeerInactivityTimedOut(peerIDs)
 		}
@@ -513,13 +513,13 @@ func (m *Manager) checkHaGroupActivity(haGroup route.HAUniqueID, peerID string,
 	return false
 }
 
-func (m *Manager) onPeerActivity(peerConnID peerid.ConnID) {
+func (m *Manager) onPeerActivity(ev activity.Event) {
 	m.managedPeersMu.Lock()
 	defer m.managedPeersMu.Unlock()
 
-	mp, ok := m.managedPeersByConnID[peerConnID]
+	mp, ok := m.managedPeersByConnID[ev.PeerConnID]
 	if !ok {
-		log.Errorf("peer not found by conn id: %v", peerConnID)
+		log.Errorf("peer not found by conn id: %v", ev.PeerConnID)
 		return
 	}
 
@@ -536,7 +536,7 @@ func (m *Manager) onPeerActivity(peerConnID peerid.ConnID) {
 
 	m.activateHAGroupPeers(mp.peerCfg)
 
-	m.peerStore.PeerConnOpen(m.engineCtx, mp.peerCfg.PublicKey)
+	m.peerStore.PeerConnOpenWithFirstPacket(m.engineCtx, mp.peerCfg.PublicKey, ev.FirstPacket)
 }
 
 func (m *Manager) onPeerInactivityTimedOut(peerIDs map[string]struct{}) {

client/internal/lazyconn/wgiface.go

@@ -17,4 +17,5 @@ type WGIface interface {
 	IsUserspaceBind() bool
 	Address() wgaddr.Address
 	LastActivities() map[string]monotime.Time
+	MTU() uint16
 }

client/internal/metrics/infra/ingest/main.go

@@ -19,7 +19,7 @@ const (
 	defaultListenAddr  = ":8087"
 	defaultInfluxDBURL = "http://influxdb:8086/api/v2/write?org=netbird&bucket=metrics&precision=ns"
 	maxBodySize        = 50 * 1024 * 1024 // 50 MB max request body
-	maxDurationSeconds = 300.0            // reject any duration field > 5 minutes
+	maxDurationSeconds = 86400.0          // reject any duration field > 24 hours
 	peerIDLength       = 16               // truncated SHA-256: 8 bytes = 16 hex chars
 	maxTagValueLength  = 64               // reject tag values longer than this
 )

client/internal/metrics/infra/ingest/main_test.go

@@ -53,14 +53,14 @@ func TestValidateLine_NegativeValue(t *testing.T) {
 }
 
 func TestValidateLine_DurationTooLarge(t *testing.T) {
-	line := `netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=999 1234567890`
+	line := `netbird_sync,deployment_type=cloud,version=1.0.0,os=linux,arch=amd64,peer_id=abc duration_seconds=100000 1234567890`
 	err := validateLine(line)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "too large")
 }
 
 func TestValidateLine_TotalSecondsTooLarge(t *testing.T) {
-	line := `netbird_peer_connection,deployment_type=cloud,connection_type=ice,attempt_type=initial,version=1.0.0,os=linux,arch=amd64,peer_id=abc,connection_pair_id=pair total_seconds=500 1234567890`
+	line := `netbird_peer_connection,deployment_type=cloud,connection_type=ice,attempt_type=initial,version=1.0.0,os=linux,arch=amd64,peer_id=abc,connection_pair_id=pair total_seconds=100000 1234567890`
 	err := validateLine(line)
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "too large")

client/internal/peer/conn.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"net/netip"
 	"runtime"
+	"slices"
 	"sync"
 	"time"
 
@@ -136,6 +137,39 @@ type Conn struct {
 	// Connection stage timestamps for metrics
 	metricsRecorder MetricsRecorder
 	metricsStages   *MetricsStages
+
+	// pendingFirstPacket is the lazyconn-captured handshake init, replayed once the real
+	// transport is up.
+	pendingFirstPacket []byte
+}
+
+// injectPendingFirstPacket replays the captured handshake through the proxy if present, else
+// directly through the ICE conn. The packet is cleared only after a successful write, so a failed
+// or transport-less attempt leaves it available for a later reinjection. Caller must hold conn.mu.
+func (conn *Conn) injectPendingFirstPacket(proxy wgproxy.Proxy, directConn net.Conn) {
+	pkt := conn.pendingFirstPacket
+	if len(pkt) == 0 {
+		return
+	}
+
+	switch {
+	case proxy != nil:
+		if err := proxy.InjectPacket(pkt); err != nil {
+			conn.Log.Debugf("failed to reinject captured first packet via proxy: %v", err)
+			return
+		}
+	case directConn != nil:
+		if _, err := directConn.Write(pkt); err != nil {
+			conn.Log.Debugf("failed to reinject captured first packet via direct conn: %v", err)
+			return
+		}
+	default:
+		conn.Log.Debugf("no transport available to reinject captured first packet")
+		return
+	}
+
+	conn.pendingFirstPacket = nil
+	conn.Log.Debugf("reinjected captured first packet (%d bytes)", len(pkt))
 }
 
 // NewConn creates a new not opened Conn to the remote peer.
@@ -172,6 +206,16 @@ func NewConn(config ConnConfig, services ServiceDependencies) (*Conn, error) {
 // It will try to establish a connection using ICE and in parallel with relay. The higher priority connection type will
 // be used.
 func (conn *Conn) Open(engineCtx context.Context) error {
+	return conn.open(engineCtx, nil)
+}
+
+// OpenWithFirstPacket opens the connection like Open and stashes firstPacket to be replayed once
+// the real transport is established. The packet is retained only on a successful open.
+func (conn *Conn) OpenWithFirstPacket(engineCtx context.Context, firstPacket []byte) error {
+	return conn.open(engineCtx, firstPacket)
+}
+
+func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
 
@@ -227,6 +271,9 @@ func (conn *Conn) Open(engineCtx context.Context) error {
 		defer conn.wg.Done()
 		conn.guard.Start(conn.ctx, conn.onGuardEvent)
 	}()
+	if len(firstPacket) > 0 {
+		conn.pendingFirstPacket = slices.Clone(firstPacket)
+	}
 	conn.opened = true
 	return nil
 }
@@ -423,6 +470,8 @@ func (conn *Conn) onICEConnectionIsReady(priority conntype.ConnPriority, iceConn
 		conn.wgProxyRelay.RedirectAs(ep)
 	}
 
+	conn.injectPendingFirstPacket(wgProxy, iceConnInfo.RemoteConn)
+
 	conn.currentConnPriority = priority
 	conn.statusICE.SetConnected()
 	conn.updateIceState(iceConnInfo, updateTime)
@@ -546,6 +595,8 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) {
 
 	wgConfigWorkaround()
 
+	conn.injectPendingFirstPacket(wgProxy, nil)
+
 	conn.rosenpassRemoteKey = rci.rosenpassPubKey
 	conn.currentConnPriority = conntype.Relay
 	conn.statusRelay.SetConnected()

client/internal/peerstore/store.go

@@ -88,11 +88,24 @@ func (s *Store) PeerConnOpen(ctx context.Context, pubKey string) {
 	if !ok {
 		return
 	}
-	// this can be blocked because of the connect open limiter semaphore
 	if err := p.Open(ctx); err != nil {
 		p.Log.Errorf("failed to open peer connection: %v", err)
 	}
+}
 
+// PeerConnOpenWithFirstPacket opens the peer connection and stashes a first packet to be
+// reinjected once the real transport is established.
+func (s *Store) PeerConnOpenWithFirstPacket(ctx context.Context, pubKey string, firstPacket []byte) {
+	s.peerConnsMu.RLock()
+	defer s.peerConnsMu.RUnlock()
+
+	p, ok := s.peerConns[pubKey]
+	if !ok {
+		return
+	}
+	if err := p.OpenWithFirstPacket(ctx, firstPacket); err != nil {
+		p.Log.Errorf("failed to open peer connection: %v", err)
+	}
 }
 
 func (s *Store) PeerConnIdle(pubKey string) {

client/system/info.go

@@ -2,9 +2,11 @@ package system
 
 import (
 	"context"
+	"errors"
 	"net/netip"
 	"slices"
 	"strings"
+	"time"
 
 	log "github.com/sirupsen/logrus"
 	"google.golang.org/grpc/metadata"
@@ -174,7 +176,7 @@ func GetInfoWithChecks(ctx context.Context, checks []*proto.Checks, excludeIPs .
 		processCheckPaths = append(processCheckPaths, check.GetFiles()...)
 	}
 
-	files, err := checkFileAndProcess(processCheckPaths)
+	files, err := checkFileAndProcess(ctx, processCheckPaths)
 	if err != nil {
 		return nil, err
 	}
@@ -187,3 +189,43 @@ func GetInfoWithChecks(ctx context.Context, checks []*proto.Checks, excludeIPs .
 	log.Debugf("all system information gathered successfully")
 	return info, nil
 }
+
+// GetInfoWithChecksTimeout is GetInfoWithChecks bounded by timeout. Posture-check gathering
+// runs uncancellable system calls (process enumeration, os.Stat), so calling it inline can
+// block the caller for as long as such a call hangs. It runs in a goroutine instead: if it
+// does not return within timeout the caller gets (nil, false) and should proceed with
+// degraded behavior rather than block. On a gathering error it falls back to base GetInfo.
+//
+// The buffered channel lets the abandoned goroutine finish and exit once its blocking call
+// returns, so it does not leak beyond the duration of that call.
+func GetInfoWithChecksTimeout(ctx context.Context, timeout time.Duration, checks []*proto.Checks, excludeIPs ...netip.Addr) (*Info, bool) {
+	ctx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	infoCh := make(chan *Info, 1)
+	go func() {
+		info, err := GetInfoWithChecks(ctx, checks, excludeIPs...)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			log.Warnf("failed to get system info with checks: %v", err)
+			info = GetInfo(ctx)
+			info.removeAddresses(excludeIPs...)
+		}
+		infoCh <- info
+	}()
+
+	select {
+	case info := <-infoCh:
+		return info, true
+	case <-ctx.Done():
+		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			log.Warnf("gathering system info with checks timed out after %s", timeout)
+		} else {
+			// Parent context canceled (e.g. shutdown), not a timeout.
+			log.Warnf("gathering system info with checks canceled: %v", ctx.Err())
+		}
+		return nil, false
+	}
+}

client/system/info_android.go

@@ -50,7 +50,7 @@ func GetInfo(ctx context.Context) *Info {
 }
 
 // checkFileAndProcess checks if the file path exists and if a process is running at that path.
-func checkFileAndProcess(paths []string) ([]File, error) {
+func checkFileAndProcess(_ context.Context, _ []string) ([]File, error) {
 	return []File{}, nil
 }
 

client/system/info_darwin.go

@@ -32,7 +32,7 @@ func GetInfo(ctx context.Context) *Info {
 	sysName := string(bytes.Split(utsname.Sysname[:], []byte{0})[0])
 	machine := string(bytes.Split(utsname.Machine[:], []byte{0})[0])
 	release := string(bytes.Split(utsname.Release[:], []byte{0})[0])
-	swVersion, err := exec.Command("sw_vers", "-productVersion").Output()
+	swVersion, err := exec.CommandContext(ctx, "sw_vers", "-productVersion").Output()
 	if err != nil {
 		log.Warnf("got an error while retrieving macOS version with sw_vers, error: %s. Using darwin version instead.\n", err)
 		swVersion = []byte(release)

client/system/info_ios.go

@@ -105,7 +105,7 @@ func isDuplicated(addresses []NetworkAddress, addr NetworkAddress) bool {
 }
 
 // checkFileAndProcess checks if the file path exists and if a process is running at that path.
-func checkFileAndProcess(paths []string) ([]File, error) {
+func checkFileAndProcess(_ context.Context, _ []string) ([]File, error) {
 	return []File{}, nil
 }
 

client/system/info_js.go

@@ -103,7 +103,7 @@ func collectLocationInfo(info *Info) {
 	}
 }
 
-func checkFileAndProcess(_ []string) ([]File, error) {
+func checkFileAndProcess(_ context.Context, _ []string) ([]File, error) {
 	return []File{}, nil
 }
 

client/system/info_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"net/netip"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc/metadata"
@@ -35,6 +36,20 @@ func Test_CustomHostname(t *testing.T) {
 	assert.Equal(t, want, got.Hostname)
 }
 
+func TestGetInfoWithChecksTimeout_Success(t *testing.T) {
+	info, ok := GetInfoWithChecksTimeout(context.Background(), 30*time.Second, nil)
+	assert.True(t, ok, "expected gathering to complete within the timeout")
+	assert.NotNil(t, info)
+}
+
+func TestGetInfoWithChecksTimeout_Timeout(t *testing.T) {
+	// A 1ns budget expires before the (real) system-info gathering can finish, so the
+	// caller must get (nil, false) instead of blocking on the in-flight goroutine.
+	info, ok := GetInfoWithChecksTimeout(context.Background(), time.Nanosecond, nil)
+	assert.False(t, ok, "expected timeout to be reported")
+	assert.Nil(t, info)
+}
+
 func Test_NetAddresses(t *testing.T) {
 	addr, err := networkAddresses()
 	if err != nil {

client/system/process.go

@@ -3,24 +3,30 @@
 package system
 
 import (
+	"context"
 	"os"
 	"slices"
 
 	"github.com/shirou/gopsutil/v3/process"
 )
 
-// getRunningProcesses returns a list of running process paths.
-func getRunningProcesses() ([]string, error) {
-	processIDs, err := process.Pids()
+// getRunningProcesses returns a list of running process paths. The context bounds the work:
+// the per-PID loop bails as soon as ctx is done, and the gopsutil calls honor it where they
+// can, so a stuck enumeration cannot run unbounded.
+func getRunningProcesses(ctx context.Context) ([]string, error) {
+	processIDs, err := process.PidsWithContext(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	processMap := make(map[string]bool)
 	for _, pID := range processIDs {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
 		p := &process.Process{Pid: pID}
 
-		path, _ := p.Exe()
+		path, _ := p.ExeWithContext(ctx)
 		if path != "" {
 			processMap[path] = false
 		}
@@ -35,18 +41,21 @@ func getRunningProcesses() ([]string, error) {
 }
 
 // checkFileAndProcess checks if the file path exists and if a process is running at that path.
-func checkFileAndProcess(paths []string) ([]File, error) {
+func checkFileAndProcess(ctx context.Context, paths []string) ([]File, error) {
 	files := make([]File, len(paths))
 	if len(paths) == 0 {
 		return files, nil
 	}
 
-	runningProcesses, err := getRunningProcesses()
+	runningProcesses, err := getRunningProcesses(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	for i, path := range paths {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
 		file := File{Path: path}
 
 		_, err := os.Stat(path)

client/system/process_test.go

@@ -1,6 +1,7 @@
 package system
 
 import (
+	"context"
 	"testing"
 
 	"github.com/shirou/gopsutil/v3/process"
@@ -9,7 +10,7 @@ import (
 func Benchmark_getRunningProcesses(b *testing.B) {
 	b.Run("getRunningProcesses new", func(b *testing.B) {
 		for i := 0; i < b.N; i++ {
-			ps, err := getRunningProcesses()
+			ps, err := getRunningProcesses(context.Background())
 			if err != nil {
 				b.Fatalf("unexpected error: %v", err)
 			}
@@ -29,12 +30,38 @@ func Benchmark_getRunningProcesses(b *testing.B) {
 			}
 		}
 	})
-	s, _ := getRunningProcesses()
+	s, _ := getRunningProcesses(context.Background())
 	b.Logf("getRunningProcesses returned %d processes", len(s))
 	s, _ = getRunningProcessesOld()
 	b.Logf("getRunningProcessesOld returned %d processes", len(s))
 }
 
+func TestCheckFileAndProcess_ContextCanceled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	// With a canceled context and non-empty paths the gathering must bail with an error
+	// instead of running the (potentially blocking) process scan / stat loop.
+	if _, err := checkFileAndProcess(ctx, []string{"/does/not/exist"}); err == nil {
+		t.Fatal("expected error on canceled context, got nil")
+	}
+}
+
+func TestCheckFileAndProcess_EmptyPaths(t *testing.T) {
+	// No check paths means no work to do: it must return immediately with no error,
+	// even on a canceled context (nothing to scan or stat).
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	files, err := checkFileAndProcess(ctx, nil)
+	if err != nil {
+		t.Fatalf("unexpected error for empty paths: %v", err)
+	}
+	if len(files) != 0 {
+		t.Fatalf("expected no files, got %d", len(files))
+	}
+}
+
 func getRunningProcessesOld() ([]string, error) {
 	processes, err := process.Processes()
 	if err != nil {