From 3f1fb3b52d010d07247c46bf2a9fbf483563b0a9 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Mon, 29 Jun 2026 19:51:25 +0200 Subject: [PATCH 1/6] [ingest] raise duration validation limit to 24 hours (#6598) Peer connection timing fields (signaling_to_connection_seconds) can legitimately exceed 5 minutes during long reconnections; the previous 300 s cap caused valid data points to be rejected. --- client/internal/metrics/infra/ingest/main.go | 2 +- client/internal/metrics/infra/ingest/main_test.go | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/client/internal/metrics/infra/ingest/main.go b/client/internal/metrics/infra/ingest/main.go index 623a17e4d..91405b85f 100644 --- a/client/internal/metrics/infra/ingest/main.go +++ b/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 ) diff --git a/client/internal/metrics/infra/ingest/main_test.go b/client/internal/metrics/infra/ingest/main_test.go index bacaa4588..96287813e 100644 --- a/client/internal/metrics/infra/ingest/main_test.go +++ b/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") From 04c3d19032dd1f3361168e0d1fc706b1e6cbfb42 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Mon, 29 Jun 2026 19:51:50 +0200 Subject: [PATCH 2/6] [client] Skip firewall ruleset rebuild when config is unchanged (#6508) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * [client] Skip firewall ruleset rebuild when config is unchanged ApplyFiltering rebuilt every peer and route ACL and flushed the firewall on every sync, with no guard for an unchanged configuration. Management re-sends the same network map far more often than it actually changes (account-wide updates, peer meta churn), so on busy accounts this is the dominant client-side cost of redundant syncs — especially with a large route set and a userspace firewall. Hash the inputs ApplyFiltering consumes (peer rules, route rules, the empty flag and the dns-route feature flag) and skip the rebuild + flush when the hash matches the last successfully applied update. Mirrors the guard the DNS server already uses (previousConfigHash). The hash is only recorded after apply and flush both succeed, so a failed update is not skipped on the next (possibly identical) sync and gets a chance to reconcile the firewall state. * [client] Include config hash in ACL skip debug log * [client] Include RoutesFirewallRulesIsEmpty in firewall config hash * [client] Add benchmarks for firewall config hash computation --- client/internal/acl/manager.go | 74 ++++++++++++-- client/internal/acl/manager_test.go | 147 ++++++++++++++++++++++++++++ 2 files changed, 212 insertions(+), 9 deletions(-) diff --git a/client/internal/acl/manager.go b/client/internal/acl/manager.go index c54a3e897..d9b179457 100644 --- a/client/internal/acl/manager.go +++ b/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) { diff --git a/client/internal/acl/manager_test.go b/client/internal/acl/manager_test.go index 408ed992f..968654ae9 100644 --- a/client/internal/acl/manager_test.go +++ b/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") +} From 3de889d529a366441788886cfc46970096fd28fd Mon Sep 17 00:00:00 2001 From: Riccardo Manfrin <3090891+riccardomanfrin@users.noreply.github.com> Date: Tue, 30 Jun 2026 08:18:51 +0200 Subject: [PATCH 3/6] [client] bound system info / posture-check gathering with a timeout to prevent sync-loop freeze (#6512) * Wraps syestem info / posture checks into a goroutine with timeout e.checks = checks is set before doing the SyncMeta, so if it fails next time isCheckEquals compares true and bypasses the update. This is to avoid another repeating the 15 seconds hang. The checks will be synced on reconnect or posture checks changes push from mgmt. * Propagate context to OS calls that can leverage its cancellation / timeout * Distinguish timeout from cancellation in logs * Dont log twice * Block on timeout failure and reapply the exclude_ips * Refactor for complexity --- client/internal/engine.go | 59 ++++++++++++++++------------------- client/system/info.go | 44 +++++++++++++++++++++++++- client/system/info_android.go | 2 +- client/system/info_darwin.go | 2 +- client/system/info_ios.go | 2 +- client/system/info_js.go | 2 +- client/system/info_test.go | 15 +++++++++ client/system/process.go | 21 +++++++++---- client/system/process_test.go | 31 ++++++++++++++++-- 9 files changed, 133 insertions(+), 45 deletions(-) diff --git a/client/internal/engine.go b/client/internal/engine.go index f7c7e1862..de151592d 100644 --- a/client/internal/engine.go +++ b/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 diff --git a/client/system/info.go b/client/system/info.go index 27588859e..496b478a3 100644 --- a/client/system/info.go +++ b/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 + } +} diff --git a/client/system/info_android.go b/client/system/info_android.go index 794ff15ed..3c71573bb 100644 --- a/client/system/info_android.go +++ b/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 } diff --git a/client/system/info_darwin.go b/client/system/info_darwin.go index 4a31920ec..e7bf367f6 100644 --- a/client/system/info_darwin.go +++ b/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) diff --git a/client/system/info_ios.go b/client/system/info_ios.go index ad42b1edf..1b0c084b3 100644 --- a/client/system/info_ios.go +++ b/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 } diff --git a/client/system/info_js.go b/client/system/info_js.go index 994d439a7..f32532881 100644 --- a/client/system/info_js.go +++ b/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 } diff --git a/client/system/info_test.go b/client/system/info_test.go index dcda18e61..a7fa02197 100644 --- a/client/system/info_test.go +++ b/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 { diff --git a/client/system/process.go b/client/system/process.go index 87e21eb9d..07f69a212 100644 --- a/client/system/process.go +++ b/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) diff --git a/client/system/process_test.go b/client/system/process_test.go index 505808a9e..44a1c8ba0 100644 --- a/client/system/process_test.go +++ b/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 { From 5b5f11740acdc82d8c7b80b314194bbb7420988a Mon Sep 17 00:00:00 2001 From: Bethuel Mmbaga Date: Tue, 30 Jun 2026 11:34:23 +0300 Subject: [PATCH 4/6] [misc] Require on-premise EULA acceptance in enterprise scripts (#6596) --- .../getting-started-enterprise.sh | 47 +++++++++++++++++++ infrastructure_files/migrate-to-enterprise.sh | 46 ++++++++++++++++++ 2 files changed, 93 insertions(+) diff --git a/infrastructure_files/getting-started-enterprise.sh b/infrastructure_files/getting-started-enterprise.sh index 5d2341cbe..135440180 100755 --- a/infrastructure_files/getting-started-enterprise.sh +++ b/infrastructure_files/getting-started-enterprise.sh @@ -9,6 +9,8 @@ set -o pipefail SED_STRIP_PADDING='s/=//g' +NETBIRD_EULA_URL="https://netbird.io/self-hosted-EULA" + check_docker_compose() { if command -v docker-compose &> /dev/null; then echo "docker-compose" @@ -139,6 +141,43 @@ read_yes_no() { esac } +# Gate the install on explicit acceptance of the NetBird On-Premise EULA. +require_eula_acceptance() { + cat > /dev/stderr < /dev/stderr + return 0 + fi + + local ans="" + echo -n 'Type "accept" to agree, or anything else to abort: ' > /dev/stderr + read -r ans < /dev/tty + if [[ "$ans" != "accept" ]]; then + echo "" > /dev/stderr + echo "EULA not accepted. Aborting installation." > /dev/stderr + exit 1 + fi + echo "" > /dev/stderr +} + wait_postgres() { set +e echo -n "Waiting for postgres to become ready" @@ -174,6 +213,9 @@ init_environment() { exit 1 fi + require_eula_acceptance + NETBIRD_EULA_ACCEPTED_AT=$(date -u +%Y-%m-%dT%H:%M:%SZ) + echo "NetBird Enterprise bootstrap" echo "" echo "Traffic flow:" @@ -260,6 +302,11 @@ render_env() { # Generated by getting-started-enterprise.sh # Holds all configuration and secrets for the stack. Mode 600. +# NetBird On-Premise EULA acceptance +NETBIRD_EULA_ACCEPTED=yes +NETBIRD_EULA_ACCEPTED_AT=${NETBIRD_EULA_ACCEPTED_AT} +NETBIRD_EULA_URL=${NETBIRD_EULA_URL} + # Features (set by the script; don't edit without re-running) NETBIRD_TRAFFIC_FLOW_ENABLED=${NETBIRD_TRAFFIC_FLOW} diff --git a/infrastructure_files/migrate-to-enterprise.sh b/infrastructure_files/migrate-to-enterprise.sh index e8a3ad515..8e8a41114 100755 --- a/infrastructure_files/migrate-to-enterprise.sh +++ b/infrastructure_files/migrate-to-enterprise.sh @@ -25,6 +25,8 @@ set -o pipefail OVERRIDE_FILE="docker-compose.override.yml" ENTERPRISE_CONFIG_FILE="config.yaml.enterprise" +NETBIRD_EULA_URL="https://netbird.io/self-hosted-EULA" + check_docker_compose() { if command -v docker-compose &> /dev/null; then echo "docker-compose" @@ -115,6 +117,43 @@ read_yes_no() { esac } +# Gate the migration on explicit acceptance of the NetBird On-Premise EULA. +require_eula_acceptance() { + cat > /dev/stderr < /dev/stderr + return 0 + fi + + local ans="" + echo -n 'Type "accept" to agree, or anything else to abort: ' > /dev/stderr + read -r ans < /dev/tty + if [[ "$ans" != "accept" ]]; then + echo "" > /dev/stderr + echo "EULA not accepted. Aborting migration." > /dev/stderr + exit 1 + fi + echo "" > /dev/stderr +} + # --------------------------------------------------------------------------- # Detection — read the operator's existing compose to find service names and # paths we need to override. Bail loudly if shape isn't recognised. @@ -436,6 +475,9 @@ init_migration() { echo " Network: $COMPOSE_NETWORK" echo "" + require_eula_acceptance + NETBIRD_EULA_ACCEPTED_AT=$(date -u +%Y-%m-%dT%H:%M:%SZ) + local proceed proceed=$(read_yes_no "Proceed with migration?" "y") if [[ "$proceed" != "yes" ]]; then @@ -529,6 +571,10 @@ apply_changes() { { echo "" echo "# Added by migrate-to-enterprise.sh on $(date -u +%Y-%m-%dT%H:%M:%SZ)" + echo "# NetBird On-Premise EULA accepted at install time" + echo "NETBIRD_EULA_ACCEPTED=yes" + echo "NETBIRD_EULA_ACCEPTED_AT=${NETBIRD_EULA_ACCEPTED_AT}" + echo "NETBIRD_EULA_URL=${NETBIRD_EULA_URL}" echo "NB_LICENSE_KEY=${NB_LICENSE_KEY}" if [[ -n "${NETBIRD_LICENSE_SERVER_BASE_URL:-}" ]]; then echo "NETBIRD_LICENSE_SERVER_BASE_URL=${NETBIRD_LICENSE_SERVER_BASE_URL}" From 4ef65294e970b8559cc222c7f184313fae22f272 Mon Sep 17 00:00:00 2001 From: Viktor Liu <17948409+lixmal@users.noreply.github.com> Date: Tue, 30 Jun 2026 18:22:25 +0900 Subject: [PATCH 5/6] [client] Reinject captured first packet on lazy connection activation (#6572) --- client/iface/wgproxy/bind/proxy.go | 5 ++ client/iface/wgproxy/ebpf/wrapper.go | 11 ++++ client/iface/wgproxy/proxy.go | 5 ++ client/iface/wgproxy/udp/proxy.go | 11 ++++ client/internal/engine_test.go | 4 ++ client/internal/iface_common.go | 1 + .../lazyconn/activity/listener_bind.go | 5 ++ .../lazyconn/activity/listener_bind_test.go | 17 ++++--- .../lazyconn/activity/listener_udp.go | 28 +++++++--- client/internal/lazyconn/activity/manager.go | 20 +++++--- .../lazyconn/activity/manager_test.go | 19 ++++--- client/internal/lazyconn/manager/manager.go | 12 ++--- client/internal/lazyconn/wgiface.go | 1 + client/internal/peer/conn.go | 51 +++++++++++++++++++ client/internal/peerstore/store.go | 15 +++++- 15 files changed, 169 insertions(+), 36 deletions(-) diff --git a/client/iface/wgproxy/bind/proxy.go b/client/iface/wgproxy/bind/proxy.go index be6f3806e..be690ed4f 100644 --- a/client/iface/wgproxy/bind/proxy.go +++ b/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 diff --git a/client/iface/wgproxy/ebpf/wrapper.go b/client/iface/wgproxy/ebpf/wrapper.go index 6e80945c4..a6156a661 100644 --- a/client/iface/wgproxy/ebpf/wrapper.go +++ b/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 { diff --git a/client/iface/wgproxy/proxy.go b/client/iface/wgproxy/proxy.go index 3c8dfd30e..40346bc15 100644 --- a/client/iface/wgproxy/proxy.go +++ b/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 } diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index 6069d1960..783843aba 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/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 { diff --git a/client/internal/engine_test.go b/client/internal/engine_test.go index 1ac9ceff7..fbd47ed74 100644 --- a/client/internal/engine_test.go +++ b/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 } diff --git a/client/internal/iface_common.go b/client/internal/iface_common.go index 2eeac1954..8ffa0b102 100644 --- a/client/internal/iface_common.go +++ b/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 } diff --git a/client/internal/lazyconn/activity/listener_bind.go b/client/internal/lazyconn/activity/listener_bind.go index 666c3bc28..72a0cfc76 100644 --- a/client/internal/lazyconn/activity/listener_bind.go +++ b/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)") diff --git a/client/internal/lazyconn/activity/listener_bind_test.go b/client/internal/lazyconn/activity/listener_bind_test.go index 1baaae6be..7026a9c97 100644 --- a/client/internal/lazyconn/activity/listener_bind_test.go +++ b/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") } diff --git a/client/internal/lazyconn/activity/listener_udp.go b/client/internal/lazyconn/activity/listener_udp.go index e0b09be6c..4b7e0ddf7 100644 --- a/client/internal/lazyconn/activity/listener_udp.go +++ b/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()) diff --git a/client/internal/lazyconn/activity/manager.go b/client/internal/lazyconn/activity/manager.go index cccc0669f..9de8c0fa7 100644 --- a/client/internal/lazyconn/activity/manager.go +++ b/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: } } diff --git a/client/internal/lazyconn/activity/manager_test.go b/client/internal/lazyconn/activity/manager_test.go index 0768d9219..07dd8d84c 100644 --- a/client/internal/lazyconn/activity/manager_test.go +++ b/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") } } diff --git a/client/internal/lazyconn/manager/manager.go b/client/internal/lazyconn/manager/manager.go index fc47bda39..3868e37e8 100644 --- a/client/internal/lazyconn/manager/manager.go +++ b/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{}) { diff --git a/client/internal/lazyconn/wgiface.go b/client/internal/lazyconn/wgiface.go index 0626c1815..f003ab3cf 100644 --- a/client/internal/lazyconn/wgiface.go +++ b/client/internal/lazyconn/wgiface.go @@ -17,4 +17,5 @@ type WGIface interface { IsUserspaceBind() bool Address() wgaddr.Address LastActivities() map[string]monotime.Time + MTU() uint16 } diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 79a513956..85e54ba5f 100644 --- a/client/internal/peer/conn.go +++ b/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() diff --git a/client/internal/peerstore/store.go b/client/internal/peerstore/store.go index 099fe4528..112caa101 100644 --- a/client/internal/peerstore/store.go +++ b/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) { From 3be90f06b25d2e7ff044263f39e01a13e854a65a Mon Sep 17 00:00:00 2001 From: Bethuel Mmbaga Date: Wed, 1 Jul 2026 12:31:46 +0300 Subject: [PATCH 6/6] [management] Add peer expiration reason to activity meta (#6619) --- management/server/account.go | 6 +++--- management/server/peer.go | 11 ++++++++++- management/server/user.go | 8 +++++--- 3 files changed, 18 insertions(+), 7 deletions(-) diff --git a/management/server/account.go b/management/server/account.go index 34220ed3f..2c57c4637 100644 --- a/management/server/account.go +++ b/management/server/account.go @@ -689,7 +689,7 @@ func (am *DefaultAccountManager) peerLoginExpirationJob(ctx context.Context, acc log.WithContext(ctx).Debugf("discovered %d peers to expire for account %s", len(peerIDs), accountID) - if err := am.expireAndUpdatePeers(ctx, accountID, expiredPeers); err != nil { + if err := am.expireAndUpdatePeers(ctx, accountID, expiredPeers, peerExpirationSessionExpired); err != nil { log.WithContext(ctx).Errorf("failed updating account peers while expiring peers for account %s", accountID) return peerSchedulerRetryInterval, true } @@ -724,7 +724,7 @@ func (am *DefaultAccountManager) peerInactivityExpirationJob(ctx context.Context log.Debugf("discovered %d peers to expire for account %s", len(peerIDs), accountID) - if err := am.expireAndUpdatePeers(ctx, accountID, inactivePeers); err != nil { + if err := am.expireAndUpdatePeers(ctx, accountID, inactivePeers, peerExpirationInactivity); err != nil { log.Errorf("failed updating account peers while expiring peers for account %s", accountID) return peerSchedulerRetryInterval, true } @@ -1949,7 +1949,7 @@ func (am *DefaultAccountManager) onPeersInvalidated(ctx context.Context, account } } if len(peers) > 0 { - err := am.expireAndUpdatePeers(ctx, accountID, peers) + err := am.expireAndUpdatePeers(ctx, accountID, peers, peerExpirationValidationFailed) if err != nil { log.WithContext(ctx).Errorf("failed to expire and update invalidated peers for account %s: %v", accountID, err) return diff --git a/management/server/peer.go b/management/server/peer.go index 440e90044..32bf9feea 100644 --- a/management/server/peer.go +++ b/management/server/peer.go @@ -34,7 +34,16 @@ import ( "github.com/netbirdio/netbird/version" ) -const remoteJobsMinVer = "0.64.0" +type peerExpirationReason string + +const ( + remoteJobsMinVer = "0.64.0" + + peerExpirationSessionExpired peerExpirationReason = "session expiration" + peerExpirationInactivity peerExpirationReason = "inactivity timeout" + peerExpirationValidationFailed peerExpirationReason = "failed integration validation" + peerExpirationUserBlocked peerExpirationReason = "blocked owner account" +) // GetPeers returns peers visible to the user within an account. // Users with "peers:read" see all peers. Otherwise, users see only their own peers, or none if restricted by account settings. diff --git a/management/server/user.go b/management/server/user.go index 666d6d178..b4b9ebe01 100644 --- a/management/server/user.go +++ b/management/server/user.go @@ -675,7 +675,7 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID, } if len(peersToExpire) > 0 { - if err := am.expireAndUpdatePeers(ctx, accountID, peersToExpire); err != nil { + if err := am.expireAndUpdatePeers(ctx, accountID, peersToExpire, peerExpirationUserBlocked); err != nil { log.WithContext(ctx).Errorf("failed update expired peers: %s", err) return nil, err } @@ -1118,7 +1118,7 @@ func (am *DefaultAccountManager) BuildUserInfosForAccount(ctx context.Context, a } // expireAndUpdatePeers expires all peers of the given user and updates them in the account -func (am *DefaultAccountManager) expireAndUpdatePeers(ctx context.Context, accountID string, peers []*nbpeer.Peer) error { +func (am *DefaultAccountManager) expireAndUpdatePeers(ctx context.Context, accountID string, peers []*nbpeer.Peer, reason peerExpirationReason) error { log.WithContext(ctx).Debugf("Expiring %d peers for account %s", len(peers), accountID) settings, err := am.Store.GetAccountSettings(ctx, store.LockingStrengthNone, accountID) if err != nil { @@ -1145,10 +1145,12 @@ func (am *DefaultAccountManager) expireAndUpdatePeers(ctx context.Context, accou if err := am.Store.SavePeerStatus(ctx, accountID, peer.ID, *peer.Status); err != nil { return err } + meta := peer.EventMeta(dnsDomain) + meta["reason"] = string(reason) am.StoreEvent( ctx, peer.UserID, peer.ID, accountID, - activity.PeerLoginExpired, peer.EventMeta(dnsDomain), + activity.PeerLoginExpired, meta, ) }