[client] Gather fresh system info on every management sync stream connect (#7409)

* Gather fresh system info on every management sync stream connect The engine collected the peer meta once at start and reused the same Info for every Sync stream reconnect, so a mobile network switch that redials management kept reporting the old local network addresses. The peer network range posture check was then evaluated against stale data until the client restarted. Sync now takes a gatherer that runs at each stream connect. The gatherer is cheap: GetInfo plus the cached posture check file results, kept in the new system.InfoSource, which the engine refreshes whenever the checks list changes. No process enumeration runs on the reconnect path. Also fix the management mock server calling itself instead of SyncFunc. * Evaluate the login response posture checks before the first sync connect The engine starts with the checks the login response carried, and the first sync stream request used to send their evaluated file results. After moving the gather into InfoSource, the stream opened with an empty cache and the first sync response did not refill it, because its checks equal the ones the engine already holds. Desktop peers therefore never reported process or file posture results. Seed the cache once before the first connect, where the old gather ran, so a timed out evaluation still falls through to the address-only info. * Harden the sync info source against nil callbacks and shared slices A nil getInfo opens the stream without metadata, as a nil sysInfo did before. The cached posture results are a copy, so the Info returned by Refresh cannot alias the snapshot later Current calls report. The exclusion test asserts the remaining address count so it cannot pass vacuously on a single-address host. * Retry a posture check refresh that timed out or failed to sync The checks list was recorded before the gather ran, so once the gather timed out or SyncMeta failed, the next sync response carrying the same list matched the recorded one and nothing retried. The peer kept reporting the previous posture results until the list changed again. Record the checks only after the meta reached management, so a failed cycle is repeated on the next sync response. * Log the skipped posture refresh, let the mock Sync return errors and deflake the reconnect test * Drop the nil guard around the sync info callback * Send the refreshed info on the first sync connect instead of gathering it twice
2026-09-17 12:19:07 +02:00 · 2026-09-04 15:07:01 +02:00
parent 0bdfa4277e
commit 825389818c
10 changed files with 323 additions and 22 deletions
@@ -265,6 +265,8 @@ type Engine struct {
 	// checks are the client-applied posture checks that need to be evaluated on the client
 	checks []*mgmProto.Checks

+	infoSource system.InfoSource
+
 	relayManager       *relayClient.Manager
 	stateManager       *statemanager.Manager
 	portForwardManager *portforward.Manager
@@ -1241,9 +1243,7 @@ func (e *Engine) updateChecksIfNew(checks []*mgmProto.Checks) error {
 	if isChecksEqual(e.checks, checks) {
 		return nil
 	}
-	e.checks = checks
-
-	info, ok := system.GetInfoWithChecksTimeout(e.ctx, systemInfoTimeout, checks, e.overlayAddresses()...)
+	info, ok := e.infoSource.Refresh(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.
@@ -1254,6 +1254,7 @@ func (e *Engine) updateChecksIfNew(checks []*mgmProto.Checks) error {
 	if err := e.mgmClient.SyncMeta(info); err != nil {
 		return fmt.Errorf("could not sync meta: error %s", err)
 	}
+	e.checks = checks
 	return nil
 }

@@ -1280,6 +1281,28 @@ func (e *Engine) applyInfoFlags(info *system.Info) {
 	)
 }

+func (e *Engine) currentSystemInfo(ctx context.Context) *system.Info {
+	info := e.infoSource.Current(ctx, e.overlayAddresses()...)
+	e.applyInfoFlags(info)
+	return info
+}
+
+// syncInfoFunc returns the info callback for the management sync stream. The
+// first connect sends the info refreshed right before it instead of gathering
+// again; every reconnect gathers a fresh one. The stream retry loop calls the
+// callback sequentially, so the handoff needs no synchronization.
+func (e *Engine) syncInfoFunc(refreshed *system.Info) func(ctx context.Context) *system.Info {
+	return func(ctx context.Context) *system.Info {
+		if refreshed == nil {
+			return e.currentSystemInfo(ctx)
+		}
+		info := refreshed
+		refreshed = nil
+		e.applyInfoFlags(info)
+		return info
+	}
+}
+
 // overlayAddresses returns our own WireGuard overlay address (v4 and v6) so it
 // can be excluded from the reported network addresses; the interface coming and
 // going otherwise churns the peer meta on the management server.
@@ -1473,15 +1496,11 @@ func (e *Engine) receiveManagementEvents() {
 	e.shutdownWg.Add(1)
 	go func() {
 		defer e.shutdownWg.Done()
-		info, ok := system.GetInfoWithChecksTimeout(e.ctx, systemInfoTimeout, e.checks, e.overlayAddresses()...)
+		info, ok := e.infoSource.Refresh(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)
+			log.Warnf("posture checks not refreshed before the sync connect, sending the previous results")
 		}
-		e.applyInfoFlags(info)
-
-		err := e.mgmClient.Sync(e.ctx, info, e.handleSync)
+		err := e.mgmClient.Sync(e.ctx, e.syncInfoFunc(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
@@ -193,7 +193,7 @@ func TestEngine_Sync(t *testing.T) {
 	// feed updates to Engine via mocked Management client
 	updates := make(chan *mgmtProto.SyncResponse)
 	defer close(updates)
-	syncFunc := func(ctx context.Context, info *system.Info, msgHandler func(msg *mgmtProto.SyncResponse) error) error {
+	syncFunc := func(ctx context.Context, _ func(context.Context) *system.Info, msgHandler func(msg *mgmtProto.SyncResponse) error) error {
 		for msg := range updates {
 			err := msgHandler(msg)
 			if err != nil {
@@ -2,6 +2,7 @@ package internal

 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"net/netip"
@@ -31,6 +32,7 @@ import (
 	icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
 	"github.com/netbirdio/netbird/client/internal/routemanager"
+	"github.com/netbirdio/netbird/client/system"
 	nbdns "github.com/netbirdio/netbird/dns"
 	"github.com/netbirdio/netbird/monotime"
 	"github.com/netbirdio/netbird/route"
@@ -253,6 +255,118 @@ func TestEngine_SSHServerConsistency(t *testing.T) {
 	})
 }

+func TestEngine_FirstSyncInfoCarriesLoginChecks(t *testing.T) {
+	key, err := wgtypes.GeneratePrivateKey()
+	require.NoError(t, err)
+
+	exe, err := os.Executable()
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(CtxInitState(context.Background()))
+	defer cancel()
+
+	infos := make(chan *system.Info, 1)
+	mgmClient := &mgmt.MockClient{
+		SyncFunc: func(ctx context.Context, getInfo func(context.Context) *system.Info, _ func(*mgmtProto.SyncResponse) error) error {
+			infos <- getInfo(ctx)
+			return nil
+		},
+	}
+
+	relayMgr := relayClient.NewManager(ctx, nil, key.PublicKey().String(), iface.DefaultMTU)
+	engine := NewEngine(ctx, cancel, &EngineConfig{
+		WgIfaceName:  "utun104",
+		WgAddr:       wgaddr.MustParseWGAddress("100.64.0.1/24"),
+		WgPrivateKey: key,
+		WgPort:       33100,
+		MTU:          iface.DefaultMTU,
+	}, EngineServices{
+		SignalClient:   &signal.MockClient{},
+		MgmClient:      mgmClient,
+		RelayManager:   relayMgr,
+		StatusRecorder: peer.NewRecorder("https://mgm"),
+		Checks:         []*mgmtProto.Checks{{Files: []string{exe}}},
+	}, MobileDependency{})
+
+	engine.receiveManagementEvents()
+
+	select {
+	case info := <-infos:
+		require.Len(t, info.Files, 1)
+		assert.Equal(t, exe, info.Files[0].Path)
+		assert.True(t, info.Files[0].Exist)
+	case <-time.After(20 * time.Second):
+		t.Fatal("timeout waiting for the first sync info")
+	}
+	engine.shutdownWg.Wait()
+}
+
+func TestEngine_SyncInfoFuncReusesRefreshedInfoOnce(t *testing.T) {
+	engine := &Engine{config: &EngineConfig{}}
+
+	refreshed := &system.Info{Hostname: "from-refresh"}
+	getInfo := engine.syncInfoFunc(refreshed)
+
+	first := getInfo(context.Background())
+	assert.Same(t, refreshed, first, "the first connect should send the refreshed info instead of gathering again")
+
+	second := getInfo(context.Background())
+	assert.NotSame(t, refreshed, second, "the reconnect should gather a fresh info")
+	assert.NotEqual(t, "from-refresh", second.Hostname, "the fresh info should not carry the refreshed hostname")
+}
+
+func TestEngine_SyncInfoFuncGathersWhenRefreshFailed(t *testing.T) {
+	engine := &Engine{config: &EngineConfig{}}
+
+	info := engine.syncInfoFunc(nil)(context.Background())
+	require.NotNil(t, info, "a failed refresh should fall back to gathering the info")
+	assert.NotEmpty(t, info.Hostname, "the gathered info should carry the hostname")
+}
+
+func TestEngine_UpdateChecksIfNewRetriesAfterFailedSyncMeta(t *testing.T) {
+	key, err := wgtypes.GeneratePrivateKey()
+	require.NoError(t, err)
+
+	exe, err := os.Executable()
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(CtxInitState(context.Background()))
+	defer cancel()
+
+	syncMetaCalls := 0
+	mgmClient := &mgmt.MockClient{
+		SyncMetaFunc: func(*system.Info) error {
+			syncMetaCalls++
+			if syncMetaCalls == 1 {
+				return errors.New("management unavailable")
+			}
+			return nil
+		},
+	}
+
+	relayMgr := relayClient.NewManager(ctx, nil, key.PublicKey().String(), iface.DefaultMTU)
+	engine := NewEngine(ctx, cancel, &EngineConfig{
+		WgIfaceName:  "utun105",
+		WgAddr:       wgaddr.MustParseWGAddress("100.64.0.1/24"),
+		WgPrivateKey: key,
+		WgPort:       33100,
+		MTU:          iface.DefaultMTU,
+	}, EngineServices{
+		SignalClient:   &signal.MockClient{},
+		MgmClient:      mgmClient,
+		RelayManager:   relayMgr,
+		StatusRecorder: peer.NewRecorder("https://mgm"),
+	}, MobileDependency{})
+
+	checks := []*mgmtProto.Checks{{Files: []string{exe}}}
+
+	require.Error(t, engine.updateChecksIfNew(checks))
+	require.NoError(t, engine.updateChecksIfNew(checks))
+	require.NoError(t, engine.updateChecksIfNew(checks))
+
+	assert.Equal(t, 2, syncMetaCalls)
+}
+
 func TestEngine_UpdateNetworkMap(t *testing.T) {
 	// test setup
 	key, err := wgtypes.GeneratePrivateKey()
@@ -0,0 +1,38 @@
+package system
+
+import (
+	"context"
+	"net/netip"
+	"slices"
+	"sync/atomic"
+	"time"
+
+	"github.com/netbirdio/netbird/shared/management/proto"
+)
+
+// InfoSource gathers the system info sent to management, keeping the posture
+// check results from the last Refresh for the cheap Current snapshots.
+type InfoSource struct {
+	files atomic.Pointer[[]File]
+}
+
+// Refresh gathers the info with the posture checks evaluated, bounded by timeout.
+func (s *InfoSource) Refresh(ctx context.Context, timeout time.Duration, checks []*proto.Checks, excludeIPs ...netip.Addr) (*Info, bool) {
+	info, ok := GetInfoWithChecksTimeout(ctx, timeout, checks, excludeIPs...)
+	if !ok {
+		return nil, false
+	}
+	files := slices.Clone(info.Files)
+	s.files.Store(&files)
+	return info, true
+}
+
+// Current gathers the info without evaluating the checks, reusing the last Refresh results.
+func (s *InfoSource) Current(ctx context.Context, excludeIPs ...netip.Addr) *Info {
+	info := GetInfo(ctx)
+	info.removeAddresses(excludeIPs...)
+	if files := s.files.Load(); files != nil {
+		info.Files = *files
+	}
+	return info
+}
@@ -0,0 +1,59 @@
+package system
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/netbirdio/netbird/shared/management/proto"
+)
+
+func TestInfoSource_CurrentBeforeRefresh(t *testing.T) {
+	var src InfoSource
+
+	info := src.Current(context.Background())
+
+	assert.Empty(t, info.Files)
+}
+
+func TestInfoSource_CurrentReusesRefreshedFiles(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "agent")
+	require.NoError(t, os.WriteFile(path, nil, 0o600))
+	checks := []*proto.Checks{{Files: []string{path}}}
+
+	var src InfoSource
+	refreshed, ok := src.Refresh(context.Background(), 15*time.Second, checks)
+	require.True(t, ok)
+	require.Equal(t, []File{{Path: path, Exist: true}}, refreshed.Files)
+
+	info := src.Current(context.Background())
+
+	assert.Equal(t, refreshed.Files, info.Files)
+}
+
+func TestInfoSource_CurrentExcludesAddresses(t *testing.T) {
+	addrs := GetInfo(context.Background()).NetworkAddresses
+	if len(addrs) == 0 {
+		t.Skip("no network addresses on this host")
+	}
+	excluded := addrs[0].NetIP.Addr()
+	matching := 0
+	for _, addr := range addrs {
+		if addr.NetIP.Addr() == excluded {
+			matching++
+		}
+	}
+
+	var src InfoSource
+	info := src.Current(context.Background(), excluded)
+
+	assert.Len(t, info.NetworkAddresses, len(addrs)-matching)
+	for _, addr := range info.NetworkAddresses {
+		assert.NotEqual(t, excluded, addr.NetIP.Addr())
+	}
+}