Merge remote-tracking branch 'origin/main' into refactor/permissions-manager

2026-09-13 18:29:07 +02:00 · 2026-09-11 13:55:02 +02:00
parent 9b10d74ab8 0fac1ee638
commit 18c7917c6d
2126 changed files with 284567 additions and 42026 deletions
@@ -0,0 +1,65 @@
+package embed
+
+import (
+	"io"
+
+	"github.com/netbirdio/netbird/client/internal"
+	"github.com/netbirdio/netbird/util/capture"
+)
+
+// CaptureOptions configures a packet capture session.
+type CaptureOptions struct {
+	// Output receives pcap-formatted data. Nil disables pcap output.
+	Output io.Writer
+	// TextOutput receives human-readable packet summaries. Nil disables text output.
+	TextOutput io.Writer
+	// Filter is a BPF-like filter expression (e.g. "host 10.0.0.1 and tcp port 443").
+	// Empty captures all packets.
+	Filter string
+	// Verbose adds seq/ack, TTL, window, and total length to text output.
+	Verbose bool
+	// ASCII dumps transport payload as printable ASCII after each packet line.
+	ASCII bool
+}
+
+// CaptureStats reports capture session counters.
+type CaptureStats struct {
+	Packets int64
+	Bytes   int64
+	Dropped int64
+}
+
+// CaptureSession represents an active packet capture. Call Stop to end the
+// capture and flush buffered packets.
+type CaptureSession struct {
+	sess   *capture.Session
+	engine *internal.Engine
+}
+
+// Stop ends the capture, flushes remaining packets, and detaches from the device.
+// Safe to call multiple times.
+func (cs *CaptureSession) Stop() {
+	if cs.engine != nil {
+		_ = cs.engine.SetCapture(nil)
+		cs.engine = nil
+	}
+	if cs.sess != nil {
+		cs.sess.Stop()
+	}
+}
+
+// Stats returns current capture counters.
+func (cs *CaptureSession) Stats() CaptureStats {
+	s := cs.sess.Stats()
+	return CaptureStats{
+		Packets: s.Packets,
+		Bytes:   s.Bytes,
+		Dropped: s.Dropped,
+	}
+}
+
+// Done returns a channel that is closed when the capture's writer goroutine
+// has fully exited and all buffered packets have been flushed.
+func (cs *CaptureSession) Done() <-chan struct{} {
+	return cs.sess.Done()
+}
@@ -12,6 +12,7 @@ import (
 	"sync"

 	"github.com/sirupsen/logrus"
+	wgdevice "golang.zx2c4.com/wireguard/device"
 	wgnetstack "golang.zx2c4.com/wireguard/tun/netstack"

 	"github.com/netbirdio/netbird/client/iface"
@@ -20,10 +21,12 @@ import (
 	"github.com/netbirdio/netbird/client/internal/auth"
 	"github.com/netbirdio/netbird/client/internal/peer"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
-	sshcommon "github.com/netbirdio/netbird/client/ssh"
+	"github.com/netbirdio/netbird/client/mdm"
+	nbssh "github.com/netbirdio/netbird/client/ssh"
 	"github.com/netbirdio/netbird/client/system"
 	"github.com/netbirdio/netbird/shared/management/domain"
 	mgmProto "github.com/netbirdio/netbird/shared/management/proto"
+	"github.com/netbirdio/netbird/util/capture"
 )

 var (
@@ -65,7 +68,7 @@ type Options struct {
 	PrivateKey string
 	// ManagementURL overrides the default management server URL
 	ManagementURL string
-	// PreSharedKey is the pre-shared key for the WireGuard interface
+	// PreSharedKey is the pre-shared key for the tunnel interface
 	PreSharedKey string
 	// LogOutput is the output destination for logs (defaults to os.Stderr if nil)
 	LogOutput io.Writer
@@ -79,11 +82,31 @@ type Options struct {
 	StatePath string
 	// DisableClientRoutes disables the client routes
 	DisableClientRoutes bool
+	// DisableIPv6 disables IPv6 overlay addressing
+	DisableIPv6 bool
 	// BlockInbound blocks all inbound connections from peers
 	BlockInbound bool
-	// WireguardPort is the port for the WireGuard interface. Use 0 for a random port.
+	// EnableRosenpass enables the Rosenpass post-quantum key exchange.
+	EnableRosenpass bool
+	// RosenpassPermissive lets a Rosenpass-enabled peer still connect to peers
+	// that do not run Rosenpass (falling back to the plain WireGuard PSK).
+	RosenpassPermissive bool
+	// BlockLANAccess blocks the embedded peer from reaching the host's
+	// LAN (RFC 1918, link-local, loopback) when it's used as a routing
+	// peer. Mirrors profilemanager.ConfigInput.BlockLANAccess. Useful
+	// when the embedded client must never act as a stepping stone into
+	// the host's local network (e.g. the proxy's overlay peer).
+	BlockLANAccess bool
+	// LazyConnectionEnabled is a tri-state local override for lazy connections,
+	// mirroring the NB_LAZY_CONN env var. Nil defers to the management feature
+	// flag; a set value overrides it in both directions. A short-lived client
+	// that reaches only a few known peers can set this to false, so its peers
+	// connect eagerly and the first request does not wait for the connection to
+	// be established.
+	LazyConnectionEnabled *bool
+	// WireguardPort is the port for the tunnel interface. Use 0 for a random port.
 	WireguardPort *int
-	// MTU is the MTU for the WireGuard interface.
+	// MTU is the MTU for the tunnel interface.
 	// Valid values are in the range 576..8192 bytes.
 	// If non-nil, this value overrides any value stored in the config file.
 	// If nil, the existing config MTU (if non-zero) is preserved; otherwise it defaults to 1280.
@@ -91,6 +114,26 @@ type Options struct {
 	MTU *uint16
 	// DNSLabels defines additional DNS labels configured in the peer.
 	DNSLabels []string
+	// Performance configures the tunnel's buffer pool cap and batch size.
+	Performance Performance
+}
+
+// Performance configures the embedded client's tunnel memory/throughput knobs.
+//
+// These settings are process-global: any non-nil field also becomes the
+// default for Clients constructed by later embed.New calls in the same
+// process. Nil fields are ignored.
+type Performance struct {
+	// PreallocatedBuffersPerPool caps the per-tunnel buffer pool. Zero
+	// leaves the pool unbounded. Lower values trade throughput for a
+	// tighter memory ceiling. May also be changed on a running Client via
+	// Client.SetPerformance, provided this field was nonzero at construction.
+	PreallocatedBuffersPerPool *uint32
+	// MaxBatchSize overrides the number of packets the tunnel reads or
+	// writes per syscall, which also bounds eager buffer allocation per
+	// worker. Zero uses the platform default. Applied at construction
+	// only; ignored by Client.SetPerformance.
+	MaxBatchSize *uint32
 }

 // validateCredentials checks that exactly one credential type is provided
@@ -170,7 +213,11 @@ func New(opts Options) (*Client, error) {
 		PreSharedKey:        &opts.PreSharedKey,
 		DisableServerRoutes: &t,
 		DisableClientRoutes: &opts.DisableClientRoutes,
+		DisableIPv6:         &opts.DisableIPv6,
 		BlockInbound:        &opts.BlockInbound,
+		BlockLANAccess:      &opts.BlockLANAccess,
+		RosenpassEnabled:    &opts.EnableRosenpass,
+		RosenpassPermissive: &opts.RosenpassPermissive,
 		WireguardPort:       opts.WireguardPort,
 		MTU:                 opts.MTU,
 		DNSLabels:           parsedLabels,
@@ -183,11 +230,31 @@ func New(opts Options) (*Client, error) {
 	if err != nil {
 		return nil, fmt.Errorf("create config: %w", err)
 	}
+	// Embedded path runs without the daemon Server: apply the active
+	// MDM policy explicitly so a forced ManagementURL / PSK / other
+	// managed key takes effect on this embedded engine instance.
+	config.ApplyMDMPolicy(mdm.NewLoader(nil).Load())

 	if opts.PrivateKey != "" {
 		config.PrivateKey = opts.PrivateKey
 	}

+	if opts.LazyConnectionEnabled != nil {
+		// Runtime-only override, read back through lazyconn.ParseState; a set value
+		// wins over the management feature flag in both directions.
+		config.LazyConnection = "off"
+		if *opts.LazyConnectionEnabled {
+			config.LazyConnection = "on"
+		}
+	}
+
+	if opts.Performance.PreallocatedBuffersPerPool != nil {
+		wgdevice.SetPreallocatedBuffersPerPool(*opts.Performance.PreallocatedBuffersPerPool)
+	}
+	if opts.Performance.MaxBatchSize != nil {
+		wgdevice.SetMaxBatchSizeOverride(*opts.Performance.MaxBatchSize)
+	}
+
 	return &Client{
 		deviceName: opts.DeviceName,
 		setupKey:   opts.SetupKey,
@@ -240,6 +307,12 @@ func (c *Client) Start(startCtx context.Context) error {

 	select {
 	case <-startCtx.Done():
+		// ConnectClient.Stop now cancels its own run context and waits for the
+		// run loop to tear the engine down, so this cancel() is no longer
+		// required to break the deadlock and could be removed. It is kept as a
+		// defensive belt-and-suspenders: cancelling the parent context first
+		// guarantees the run loop is unblocked even if Stop's contract regresses.
+		cancel()
 		if stopErr := client.Stop(); stopErr != nil {
 			return fmt.Errorf("stop error after context done. Stop error: %w. Context done: %w", stopErr, startCtx.Err())
 		}
@@ -332,7 +405,7 @@ func (c *Client) ListenTCP(address string) (net.Listener, error) {
 	if err != nil {
 		return nil, fmt.Errorf("split host port: %w", err)
 	}
-	listenAddr := fmt.Sprintf("%s:%s", addr, port)
+	listenAddr := net.JoinHostPort(addr.String(), port)

 	tcpAddr, err := net.ResolveTCPAddr("tcp", listenAddr)
 	if err != nil {
@@ -353,7 +426,7 @@ func (c *Client) ListenUDP(address string) (net.PacketConn, error) {
 	if err != nil {
 		return nil, fmt.Errorf("split host port: %w", err)
 	}
-	listenAddr := fmt.Sprintf("%s:%s", addr, port)
+	listenAddr := net.JoinHostPort(addr.String(), port)

 	udpAddr, err := net.ResolveUDPAddr("udp", listenAddr)
 	if err != nil {
@@ -401,6 +474,21 @@ func (c *Client) Expose(ctx context.Context, req ExposeRequest) (*ExposeSession,
 	}, nil
 }

+// IdentityForIP looks up a remote peer by its tunnel IP using the
+// embedded client's status recorder. Returns the peer's WireGuard public
+// key and FQDN. ok=false means the IP doesn't belong to an active peer
+// — offline roster peers are treated as unknown, same as foreign IPs.
+func (c *Client) IdentityForIP(ip netip.Addr) (pubKey, fqdn string, ok bool) {
+	if !ip.IsValid() || c.recorder == nil {
+		return "", "", false
+	}
+	state, found := c.recorder.PeerStateByIP(ip.String())
+	if !found {
+		return "", "", false
+	}
+	return state.PubKey, state.FQDN, true
+}
+
 // Status returns the current status of the client.
 func (c *Client) Status() (peer.FullStatus, error) {
 	c.mu.Lock()
@@ -410,7 +498,7 @@ func (c *Client) Status() (peer.FullStatus, error) {
 	if connect != nil {
 		engine := connect.Engine()
 		if engine != nil {
-			_ = engine.RunHealthProbes(false)
+			_ = engine.RunHealthProbes(context.Background(), false)
 		}
 	}

@@ -461,12 +549,72 @@ func (c *Client) VerifySSHHostKey(peerAddress string, key []byte) error {
 		return err
 	}

-	storedKey, found := engine.GetPeerSSHKey(peerAddress)
-	if !found {
-		return sshcommon.ErrPeerNotFound
+	return nbssh.PeerKeyLookup(engine.GetPeerSSHKey).VerifySSHHostKey(peerAddress, key)
+}
+
+// SetPerformance retunes a running Client. Only PreallocatedBuffersPerPool
+// takes effect, and only when it was nonzero at construction;
+// MaxBatchSize is construction-only and returns an error if set here.
+//
+// Returns ErrClientNotStarted / ErrEngineNotStarted if the Client is not
+// running yet.
+func (c *Client) SetPerformance(t Performance) error {
+	if t.MaxBatchSize != nil {
+		return errors.New("MaxBatchSize is construction-only and cannot be changed at runtime")
+	}
+	engine, err := c.getEngine()
+	if err != nil {
+		return err
+	}
+	return engine.SetPerformance(internal.Performance{
+		PreallocatedBuffersPerPool: t.PreallocatedBuffersPerPool,
+	})
+}
+
+// StartCapture begins capturing packets on this client's tunnel device.
+// Only one capture can be active at a time; starting a new one stops the previous.
+// Call StopCapture (or CaptureSession.Stop) to end it.
+func (c *Client) StartCapture(opts CaptureOptions) (*CaptureSession, error) {
+	engine, err := c.getEngine()
+	if err != nil {
+		return nil, err
 	}

-	return sshcommon.VerifyHostKey(storedKey, key, peerAddress)
+	var matcher capture.Matcher
+	if opts.Filter != "" {
+		m, err := capture.ParseFilter(opts.Filter)
+		if err != nil {
+			return nil, fmt.Errorf("parse filter: %w", err)
+		}
+		matcher = m
+	}
+
+	sess, err := capture.NewSession(capture.Options{
+		Output:     opts.Output,
+		TextOutput: opts.TextOutput,
+		Matcher:    matcher,
+		Verbose:    opts.Verbose,
+		ASCII:      opts.ASCII,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("create capture session: %w", err)
+	}
+
+	if err := engine.SetCapture(sess); err != nil {
+		sess.Stop()
+		return nil, fmt.Errorf("set capture: %w", err)
+	}
+
+	return &CaptureSession{sess: sess, engine: engine}, nil
+}
+
+// StopCapture stops the active capture session if one is running.
+func (c *Client) StopCapture() error {
+	engine, err := c.getEngine()
+	if err != nil {
+		return err
+	}
+	return engine.SetCapture(nil)
 }

 // getEngine safely retrieves the engine from the client with proper locking.
@@ -0,0 +1,168 @@
+package embed
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+
+	"go.uber.org/mock/gomock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+
+	"github.com/netbirdio/netbird/management/internals/controllers/network_map/controller"
+	"github.com/netbirdio/netbird/management/internals/controllers/network_map/update_channel"
+	"github.com/netbirdio/netbird/management/internals/modules/peers"
+	"github.com/netbirdio/netbird/management/internals/modules/peers/ephemeral/manager"
+	"github.com/netbirdio/netbird/management/internals/server/config"
+	nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
+	mgmt "github.com/netbirdio/netbird/management/server"
+	"github.com/netbirdio/netbird/management/server/activity"
+	nbcache "github.com/netbirdio/netbird/management/server/cache"
+	"github.com/netbirdio/netbird/management/server/groups"
+	"github.com/netbirdio/netbird/management/server/integrations/integrated_validator/validator"
+	"github.com/netbirdio/netbird/management/server/integrations/port_forwarding"
+	"github.com/netbirdio/netbird/management/server/job"
+	"github.com/netbirdio/netbird/management/internals/modules/permissions"
+	"github.com/netbirdio/netbird/management/server/settings"
+	"github.com/netbirdio/netbird/management/server/store"
+	"github.com/netbirdio/netbird/management/server/telemetry"
+	"github.com/netbirdio/netbird/management/server/types"
+	mgmtProto "github.com/netbirdio/netbird/shared/management/proto"
+	"github.com/netbirdio/netbird/util"
+)
+
+const testSetupKey = "A2C8E62B-38F5-4553-B31E-DD66C696CEBB"
+
+// TestClientStartTimeoutRollback reproduces a deadlock between Engine.Start and
+// Engine.Stop. The signal endpoint accepts gRPC connections but never serves the
+// SignalExchange service, so Engine.Start parks in WaitStreamConnected while
+// holding the engine mutex. When the Start context expires, the rollback path
+// calls ConnectClient.Stop, which must not block forever acquiring that mutex.
+func TestClientStartTimeoutRollback(t *testing.T) {
+	signalAddr := startBlackholeSignal(t)
+	mgmAddr := startManagement(t, signalAddr)
+
+	wgPort := 0
+	client, err := New(Options{
+		DeviceName:    "embed-rollback-test",
+		SetupKey:      testSetupKey,
+		ManagementURL: "http://" + mgmAddr,
+		WireguardPort: &wgPort,
+	})
+	require.NoError(t, err, "embed client creation must succeed")
+
+	startCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	startErr := make(chan error, 1)
+	go func() {
+		startErr <- client.Start(startCtx)
+	}()
+
+	select {
+	case err := <-startErr:
+		require.ErrorIs(t, err, context.DeadlineExceeded)
+	case <-time.After(60 * time.Second):
+		t.Fatal("client.Start did not return after its context expired: Engine.Stop deadlocked against Engine.Start waiting for the signal stream")
+	}
+}
+
+// startBlackholeSignal starts a gRPC server without the SignalExchange service
+// registered. Connections succeed, but the signal stream can never be
+// established, which keeps Engine.Start parked in WaitStreamConnected.
+func startBlackholeSignal(t *testing.T) string {
+	t.Helper()
+
+	lis, err := net.Listen("tcp", "localhost:0")
+	require.NoError(t, err)
+
+	s := grpc.NewServer()
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			t.Error(err)
+		}
+	}()
+	t.Cleanup(s.Stop)
+
+	return lis.Addr().String()
+}
+
+func startManagement(t *testing.T, signalAddr string) string {
+	t.Helper()
+
+	cfg := &config.Config{
+		Stuns:      []*config.Host{},
+		TURNConfig: &config.TURNConfig{},
+		Relay: &config.Relay{
+			Addresses:      []string{"127.0.0.1:1234"},
+			CredentialsTTL: util.Duration{Duration: time.Hour},
+			Secret:         "222222222222222222",
+		},
+		Signal: &config.Host{
+			Proto: "http",
+			URI:   signalAddr,
+		},
+		Datadir:    t.TempDir(),
+		HttpConfig: nil,
+	}
+
+	lis, err := net.Listen("tcp", "localhost:0")
+	require.NoError(t, err)
+
+	s := grpc.NewServer()
+
+	testStore, cleanUp, err := store.NewTestStoreFromSQL(context.Background(), "../testdata/store.sql", cfg.Datadir)
+	require.NoError(t, err)
+	t.Cleanup(cleanUp)
+
+	eventStore := &activity.InMemoryEventStore{}
+
+	permissionsManager := permissions.NewManager(testStore)
+	peersManager := peers.NewManager(testStore)
+	jobManager := job.NewJobManager(nil, testStore, peersManager)
+
+	cacheStore, err := nbcache.NewStore(context.Background(), 100*time.Millisecond, 300*time.Millisecond, 100)
+	require.NoError(t, err)
+
+	iv, err := validator.NewIntegratedValidator(context.Background(), peersManager, nil, eventStore, cacheStore)
+	require.NoError(t, err)
+	metrics, err := telemetry.NewDefaultAppMetrics(context.Background())
+	require.NoError(t, err)
+
+	ctrl := gomock.NewController(t)
+	t.Cleanup(ctrl.Finish)
+	settingsMockManager := settings.NewMockManager(ctrl)
+	settingsMockManager.EXPECT().
+		GetSettings(gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(&types.Settings{}, nil).
+		AnyTimes()
+	settingsMockManager.EXPECT().
+		GetExtraSettings(gomock.Any(), gomock.Any()).
+		Return(&types.ExtraSettings{}, nil).
+		AnyTimes()
+
+	groupsManager := groups.NewManagerMock()
+
+	updateManager := update_channel.NewPeersUpdateManager(metrics)
+	requestBuffer := mgmt.NewAccountRequestBuffer(context.Background(), testStore)
+	networkMapController := controller.NewController(context.Background(), testStore, metrics, updateManager, requestBuffer, mgmt.MockIntegratedValidator{}, settingsMockManager, "netbird.selfhosted", port_forwarding.NewControllerMock(), manager.NewEphemeralManager(testStore, peersManager), cfg, nil)
+	accountManager, err := mgmt.BuildManager(context.Background(), cfg, testStore, networkMapController, jobManager, nil, "", eventStore, nil, false, iv, metrics, port_forwarding.NewControllerMock(), settingsMockManager, permissionsManager, false, cacheStore)
+	require.NoError(t, err)
+
+	secretsManager, err := nbgrpc.NewTimeBasedAuthSecretsManager(updateManager, cfg.TURNConfig, cfg.Relay, settingsMockManager, groupsManager)
+	require.NoError(t, err)
+
+	mgmtServer, err := nbgrpc.NewServer(cfg, accountManager, settingsMockManager, jobManager, secretsManager, nil, nil, &mgmt.MockIntegratedValidator{}, networkMapController, nil, nil)
+	require.NoError(t, err)
+	mgmtProto.RegisterManagementServiceServer(s, mgmtServer)
+
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			t.Error(err)
+		}
+	}()
+	t.Cleanup(s.Stop)
+
+	return lis.Addr().String()
+}