Merge remote-tracking branch 'origin/main' into oauth-flow-fallback

# Conflicts: # client/server/server.go
2026-10-30 08:19:09 +01:00 · 2026-09-02 18:32:51 +02:00
parent 384102d017 8a5e940c84
commit 66b36e6c81
616 changed files with 43349 additions and 7079 deletions
@@ -23,6 +23,10 @@ import (

 	"github.com/netbirdio/netbird/client/internal/auth"
 	"github.com/netbirdio/netbird/client/internal/expose"
+	"github.com/netbirdio/netbird/client/internal/ipcauth"
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/netbirdio/netbird/client/internal/localmetrics"
 	"github.com/netbirdio/netbird/client/internal/profilemanager"
 	sleephandler "github.com/netbirdio/netbird/client/internal/sleep/handler"
 	"github.com/netbirdio/netbird/client/mdm"
@@ -35,6 +39,7 @@ import (
 	"github.com/netbirdio/netbird/client/internal/statemanager"
 	"github.com/netbirdio/netbird/client/internal/updater"
 	"github.com/netbirdio/netbird/client/proto"
+	"github.com/netbirdio/netbird/util"
 	"github.com/netbirdio/netbird/util/capture"
 	"github.com/netbirdio/netbird/version"
 )
@@ -108,6 +113,7 @@ type Server struct {

 	statusRecorder *peer.Status
 	sessionWatcher *internal.SessionWatcher
+	localMetrics   *localmetrics.Manager

 	probeThrottle       *probeThrottle
 	persistSyncResponse bool
@@ -145,9 +151,17 @@ type Server struct {
 }

 type oauthAuthFlow struct {
-	expiresAt  time.Time
-	flow       auth.OAuthFlow
-	info       auth.AuthFlowInfo
+	expiresAt time.Time
+	flow      auth.OAuthFlow
+	info      auth.AuthFlowInfo
+
+	// cacheGeneration is the SSH JWT cache's generation as of the start of the
+	// request that created this flow. The flow outlives a profile switch, so
+	// reading the generation any later — when the IdP has answered, or when the
+	// token finally arrives — would read the new session's one and let the old
+	// session's token into the new session's cache.
+	cacheGeneration uint64
+
 	waitCancel context.CancelFunc
 }

@@ -171,9 +185,28 @@ func New(ctx context.Context, logFile string, configFile string, profilesDisable
 	s.sleepHandler = sleephandler.New(agent)
 	s.startSleepDetector()

+	s.localMetrics = localmetrics.NewManager(ctx, s.statusRecorder, s.clientMetricsGatherer)
+
 	return s
 }

+// clientMetricsGatherer returns the Prometheus gatherer of the running
+// engine's client metrics, or nil when no engine is running.
+func (s *Server) clientMetricsGatherer() prometheus.Gatherer {
+	s.mutex.Lock()
+	connectClient := s.connectClient
+	s.mutex.Unlock()
+
+	if connectClient == nil {
+		return nil
+	}
+	engine := connectClient.Engine()
+	if engine == nil {
+		return nil
+	}
+	return engine.GetClientMetrics().PrometheusGatherer()
+}
+
 func (s *Server) Start() error {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
@@ -254,6 +287,7 @@ func (s *Server) Start() error {

 	s.statusRecorder.UpdateManagementAddress(config.ManagementURL.String())
 	s.statusRecorder.UpdateRosenpass(config.RosenpassEnabled, config.RosenpassPermissive)
+	s.localMetrics.Reconcile(config.LocalMetricsEnabled, config.LocalMetricsAddress)

 	if s.sessionWatcher == nil {
 		s.sessionWatcher = internal.NewSessionWatcher(s.rootCtx, s.statusRecorder)
@@ -477,11 +511,18 @@ func (s *Server) SetConfig(callerCtx context.Context, msg *proto.SetConfigReques
 		return nil, err
 	}

-	if _, err := profilemanager.UpdateConfig(config); err != nil {
+	updatedConf, err := profilemanager.UpdateConfig(config)
+	if err != nil {
 		log.Errorf("failed to update profile config: %v", err)
 		return nil, fmt.Errorf("failed to update profile config: %w", err)
 	}

+	if activeProf, err := s.profileManager.GetActiveProfileState(); err == nil {
+		if activePath, err := activeProf.FilePath(); err == nil && activePath == config.ConfigPath {
+			s.localMetrics.Reconcile(updatedConf.LocalMetricsEnabled, updatedConf.LocalMetricsAddress)
+		}
+	}
+
 	return &proto.SetConfigResponse{}, nil
 }

@@ -551,8 +592,11 @@ func (s *Server) setConfigInputFromRequest(msg *proto.SetConfigRequest) (profile

 	config.RosenpassEnabled = msg.RosenpassEnabled
 	config.RosenpassPermissive = msg.RosenpassPermissive
+	config.LocalMetricsEnabled = msg.EnableLocalMetrics
+	config.LocalMetricsAddress = msg.LocalMetricsAddress
 	config.DisableAutoConnect = msg.DisableAutoConnect
 	config.ServerSSHAllowed = msg.ServerSSHAllowed
+	config.RemoteJobsAllowed = msg.RemoteJobsAllowed
 	config.NetworkMonitor = msg.NetworkMonitor
 	config.DisableClientRoutes = msg.DisableClientRoutes
 	config.DisableServerRoutes = msg.DisableServerRoutes
@@ -618,6 +662,11 @@ func (s *Server) Login(callerCtx context.Context, msg *proto.LoginRequest) (*pro
 	}

 	state := internal.CtxGetState(s.rootCtx)
+	status := state.CurrentStatus()
+	if status == internal.StatusConnected {
+		return &proto.LoginResponse{}, nil
+	}
+
 	defer func() {
 		status, err := state.Status()
 		if err != nil || (status != internal.StatusNeedsLogin && status != internal.StatusLoginFailed) {
@@ -657,6 +706,8 @@ func (s *Server) Login(callerCtx context.Context, msg *proto.LoginRequest) (*pro
 	s.config = config
 	s.mutex.Unlock()

+	s.localMetrics.Reconcile(config.LocalMetricsEnabled, config.LocalMetricsAddress)
+
 	// A probe that errors leaves the login undecided: Management unreachable, a
 	// restart mid-request, an internal error. Those are returned for the caller
 	// to retry, because turning them into an SSO prompt asks the user to solve
@@ -675,7 +726,7 @@ func (s *Server) Login(callerCtx context.Context, msg *proto.LoginRequest) (*pro
 	}

 	if msg.SetupKey == "" {
-		return s.ssoLogin(ctx, config, msg)
+		return s.beginSSOLogin(ctx, config, msg)
 	}

 	// Setup-key path: we are about to dial Management with the key, so the
@@ -691,17 +742,15 @@ func (s *Server) Login(callerCtx context.Context, msg *proto.LoginRequest) (*pro
 	return &proto.LoginResponse{}, nil
 }

-// ssoLogin starts the browser leg of a login that carries no setup key: it builds the OAuth
-// flow, reuses the one already in flight when it targets the same client, and returns the
-// verification URI for the caller to open.
-func (s *Server) ssoLogin(ctx context.Context, config *profilemanager.Config, msg *proto.LoginRequest) (*proto.LoginResponse, error) {
+// beginSSOLogin starts the browser leg of a login that carries no setup key and
+// returns the response that parks the caller on it.
+func (s *Server) beginSSOLogin(ctx context.Context, config *profilemanager.Config, msg *proto.LoginRequest) (*proto.LoginResponse, error) {
 	state := internal.CtxGetState(s.rootCtx)

-	var hint string
+	hint := ""
 	if msg.Hint != nil {
 		hint = *msg.Hint
 	}
-
 	oAuthFlow, err := auth.NewOAuthFlow(ctx, config, msg.IsUnixDesktopClient, false, hint)
 	if err != nil {
 		state.Set(internal.StatusLoginFailed)
@@ -713,7 +762,7 @@ func (s *Server) ssoLogin(ctx context.Context, config *profilemanager.Config, ms
 		return nil, err
 	}

-	if resp := s.reusableOAuthFlow(ctx, oAuthFlow); resp != nil {
+	if resp := s.pendingOAuthFlowResponse(ctx, oAuthFlow); resp != nil {
 		state.Set(internal.StatusNeedsLogin)
 		return resp, nil
 	}
@@ -740,30 +789,32 @@ func (s *Server) ssoLogin(ctx context.Context, config *profilemanager.Config, ms
 	}, nil
 }

-// reusableOAuthFlow returns the response for a flow that is still in flight for the same client,
-// or nil when there is none to reuse. A flow too close to expiry is dropped along with the
-// execution waiting on it, so the caller starts a fresh one.
-func (s *Server) reusableOAuthFlow(ctx context.Context, oAuthFlow auth.OAuthFlow) *proto.LoginResponse {
+// pendingOAuthFlowResponse returns the in-flight flow's response when it
+// targets the same IdP client and has enough time left for the user to finish
+// the browser leg, so a second login joins the pending flow instead of opening
+// a competing one. A flow too close to expiry has its waiter cancelled and nil
+// returned, leaving the caller to start a fresh flow.
+func (s *Server) pendingOAuthFlowResponse(ctx context.Context, oAuthFlow auth.OAuthFlow) *proto.LoginResponse {
 	if s.oauthAuthFlow.flow == nil || s.oauthAuthFlow.flow.GetClientID(ctx) != oAuthFlow.GetClientID(ctx) {
 		return nil
 	}

-	if !s.oauthAuthFlow.expiresAt.After(time.Now().Add(90 * time.Second)) {
-		log.Warnf("canceling previous waiting execution")
-		if s.oauthAuthFlow.waitCancel != nil {
-			s.oauthAuthFlow.waitCancel()
+	if s.oauthAuthFlow.expiresAt.After(time.Now().Add(90 * time.Second)) {
+		log.Debugf("using previous oauth flow info")
+		return &proto.LoginResponse{
+			NeedsSSOLogin:           true,
+			VerificationURI:         s.oauthAuthFlow.info.VerificationURI,
+			VerificationURIComplete: s.oauthAuthFlow.info.VerificationURIComplete,
+			UserCode:                s.oauthAuthFlow.info.UserCode,
 		}
-		return nil
 	}

-	log.Debugf("using previous oauth flow info")
-
-	return &proto.LoginResponse{
-		NeedsSSOLogin:           true,
-		VerificationURI:         s.oauthAuthFlow.info.VerificationURI,
-		VerificationURIComplete: s.oauthAuthFlow.info.VerificationURIComplete,
-		UserCode:                s.oauthAuthFlow.info.UserCode,
+	log.Warnf("canceling previous waiting execution")
+	if s.oauthAuthFlow.waitCancel != nil {
+		s.oauthAuthFlow.waitCancel()
 	}
+
+	return nil
 }

 // WaitSSOLogin validates the supplied userCode against the in-flight OAuth
@@ -1035,6 +1086,7 @@ func (s *Server) Up(callerCtx context.Context, msg *proto.UpRequest) (*proto.UpR

 	s.statusRecorder.UpdateManagementAddress(s.config.ManagementURL.String())
 	s.statusRecorder.UpdateRosenpass(s.config.RosenpassEnabled, s.config.RosenpassPermissive)
+	s.localMetrics.Reconcile(s.config.LocalMetricsEnabled, s.config.LocalMetricsAddress)

 	s.clientRunning = true
 	s.clientRunningChan = make(chan struct{})
@@ -1212,6 +1264,9 @@ func (s *Server) SwitchProfile(callerCtx context.Context, msg *proto.SwitchProfi
 	}

 	s.config = config
+	s.localMetrics.Reconcile(config.LocalMetricsEnabled, config.LocalMetricsAddress)
+
+	s.jwtCache.clear()

 	if msg != nil && msg.ProfileName != nil {
 		s.publishProfileListChanged(*msg.ProfileName)
@@ -1338,11 +1393,16 @@ func (s *Server) handleProfileLogout(ctx context.Context, msg *proto.LogoutReque
 		return nil, err
 	}

-	if err := s.validateProfileOperation(resolved.ID, true); err != nil {
+	activeProf, err := s.profileManager.GetActiveProfileState()
+	if err != nil {
+		return nil, gstatus.Errorf(codes.FailedPrecondition, "failed to get active profile state: %v", err)
+	}
+
+	if err := s.validateProfileLogout(resolved.ID, isActiveProfile(activeProf, resolved.ID, username)); err != nil {
 		return nil, err
 	}

-	if err := s.logoutFromProfile(ctx, resolved); err != nil {
+	if err := s.logoutFromProfile(ctx, resolved, username); err != nil {
 		log.Errorf("failed to logout from profile %s: %v", resolved.ID, err)
 		// A refused deregistration is already a status error carrying the reason
 		// and the command to run; rewrapping it as Internal would flatten both
@@ -1353,18 +1413,36 @@ func (s *Server) handleProfileLogout(ctx context.Context, msg *proto.LogoutReque
 		return nil, gstatus.Errorf(codes.Internal, "logout: %v", err)
 	}

-	activeProf, _ := s.profileManager.GetActiveProfileState()
-	if activeProf != nil && activeProf.ID == resolved.ID {
-		if err := s.cleanupConnection(); err != nil && !errors.Is(err, ErrServiceNotUp) {
-			log.Errorf("failed to cleanup connection: %v", err)
-		}
-		state := internal.CtxGetState(s.rootCtx)
-		state.Set(internal.StatusNeedsLogin)
-	}
+	s.cleanupAfterProfileLogout(resolved.ID, username)

 	return &proto.LogoutResponse{}, nil
 }

+// cleanupAfterProfileLogout tears the connection down and asks for a new login
+// when the profile that was just deregistered is the one the daemon is running.
+// The active profile is read again here rather than reused from the pre-flight
+// check: Login switches profiles under guardedConfigMu, which this path does not
+// hold, so a login that landed meanwhile must not have its fresh connection
+// dropped by a logout that targeted the profile it replaced.
+func (s *Server) cleanupAfterProfileLogout(id profilemanager.ID, username string) {
+	activeProf, err := s.profileManager.GetActiveProfileState()
+	if err != nil {
+		log.Errorf("failed to get active profile state after logout from profile %s: %v", id, err)
+		return
+	}
+
+	if !isActiveProfile(activeProf, id, username) {
+		return
+	}
+
+	if err := s.cleanupConnection(); err != nil && !errors.Is(err, ErrServiceNotUp) {
+		log.Errorf("failed to cleanup connection: %v", err)
+	}
+	s.jwtCache.clear()
+	state := internal.CtxGetState(s.rootCtx)
+	state.Set(internal.StatusNeedsLogin)
+}
+
 func (s *Server) handleActiveProfileLogout(ctx context.Context) (*proto.LogoutResponse, error) {
 	if s.config == nil {
 		activeProf, err := s.profileManager.GetActiveProfileState()
@@ -1389,6 +1467,7 @@ func (s *Server) handleActiveProfileLogout(ctx context.Context) (*proto.LogoutRe
 		log.Errorf("failed to cleanup connection: %v", err)
 		return nil, err
 	}
+	s.jwtCache.clear()

 	state := internal.CtxGetState(s.rootCtx)
 	state.Set(internal.StatusNeedsLogin)
@@ -1416,40 +1495,47 @@ func (s *Server) getConfig(activeProf *profilemanager.ActiveProfileState) (*prof
 	return config, configExisted, nil
 }

-func (s *Server) canRemoveProfile(id profilemanager.ID) error {
-	if id == profilemanager.DefaultProfileName {
-		return fmt.Errorf("remove profile with reserved name: %s", profilemanager.DefaultProfileName)
-	}
-
-	activeProf, err := s.profileManager.GetActiveProfileState()
-	if err == nil && activeProf.ID == id {
-		return fmt.Errorf("remove active profile: %s", id)
-	}
-
-	return nil
-}
-
-func (s *Server) validateProfileOperation(id profilemanager.ID, allowActiveProfile bool) error {
-	if s.checkProfilesDisabled() {
-		return gstatus.Errorf(codes.Unavailable, errProfilesDisabled)
-	}
-
+// validateProfileLogout gates a profile-addressed logout. Deregistering the
+// profile the daemon already runs is what a plain `netbird logout` does, so the
+// profiles-disabled kill switch must not block it. Logging out of any other
+// profile is profile management and stays gated.
+func (s *Server) validateProfileLogout(id profilemanager.ID, isActive bool) error {
 	if id == "" {
 		return gstatus.Errorf(codes.InvalidArgument, "profile name must be provided")
 	}

-	if !allowActiveProfile {
-		if err := s.canRemoveProfile(id); err != nil {
-			return gstatus.Errorf(codes.InvalidArgument, "%v", err)
-		}
+	if isActive {
+		return nil
+	}
+
+	if s.checkProfilesDisabled() {
+		return gstatus.Errorf(codes.Unavailable, errProfilesDisabled)
 	}

 	return nil
 }

-func (s *Server) logoutFromProfile(ctx context.Context, profile *profilemanager.Profile) error {
+// isActiveProfile reports whether id is the profile the daemon runs for
+// username. The username is part of the comparison because legacy profile IDs
+// are display names, which two users can both hold; the default profile is
+// shared by every user and carries no username.
+func isActiveProfile(activeProf *profilemanager.ActiveProfileState, id profilemanager.ID, username string) bool {
+	if activeProf == nil || activeProf.ID != id {
+		return false
+	}
+
+	return id == profilemanager.DefaultProfileName || activeProf.Username == username
+}
+
+// logoutFromProfile deregisters profile, reusing the running config when
+// profile is the one the daemon is connected with. The username takes part in
+// that decision for the same reason it does in the logout gate: a legacy
+// profile ID is a display name two users can share, and sending the running
+// config for a namesake would deregister the active peer instead of the
+// requested one.
+func (s *Server) logoutFromProfile(ctx context.Context, profile *profilemanager.Profile, username string) error {
 	activeProf, err := s.profileManager.GetActiveProfileState()
-	if err == nil && activeProf.ID == profile.ID && s.connectClient != nil {
+	if err == nil && isActiveProfile(activeProf, profile.ID, username) && s.connectClient != nil {
 		return s.sendLogoutRequest(ctx)
 	}

@@ -1713,6 +1799,20 @@ func (s *Server) getJWTCacheTTL() time.Duration {
 	return ttl
 }

+// cachedJWT returns the cached SSH JWT to the identity that obtained it, and a
+// miss on a control channel that carries no caller identity.
+func (s *Server) cachedJWT(ctx context.Context) (string, bool) {
+	caller, ok := ipcauth.CallerIdentity(ctx)
+	if !ok {
+		// Expected and handled on a control channel with no peer identity: the
+		// caller re-authenticates. daemonServerOptions warns about it once at
+		// startup, so this stays out of the per-request log.
+		log.Debug("not serving the cached SSH JWT: the caller's identity cannot be verified on this control channel")
+		return "", false
+	}
+	return s.jwtCache.get(caller)
+}
+
 // RequestJWTAuth initiates JWT authentication flow for SSH
 func (s *Server) RequestJWTAuth(
 	ctx context.Context,
@@ -1722,8 +1822,14 @@ func (s *Server) RequestJWTAuth(
 		return nil, ctx.Err()
 	}

+	// The generation is read here, with the config and under the same lock, not
+	// where the flow is stored below: RequestAuthInfo talks to the IdP in
+	// between, and a switch or a logout during that call would otherwise be
+	// read as the generation this flow belongs to. SwitchProfile holds
+	// s.mutex across its own clear(), so the pair cannot be torn.
 	s.mutex.Lock()
 	config := s.config
+	cacheGeneration := s.jwtCache.currentGeneration()
 	s.mutex.Unlock()

 	if config == nil {
@@ -1732,7 +1838,7 @@ func (s *Server) RequestJWTAuth(

 	jwtCacheTTL := s.getJWTCacheTTL()
 	if jwtCacheTTL > 0 {
-		if cachedToken, found := s.jwtCache.get(); found {
+		if cachedToken, found := s.cachedJWT(ctx); found {
 			log.Debugf("JWT token found in cache, returning cached token for SSH authentication")

 			return &proto.RequestJWTAuthResponse{
@@ -1766,6 +1872,7 @@ func (s *Server) RequestJWTAuth(
 	s.oauthAuthFlow.flow = oAuthFlow
 	s.oauthAuthFlow.info = authInfo
 	s.oauthAuthFlow.expiresAt = time.Now().Add(time.Duration(authInfo.ExpiresIn) * time.Second)
+	s.oauthAuthFlow.cacheGeneration = cacheGeneration
 	s.mutex.Unlock()

 	return &proto.RequestJWTAuthResponse{
@@ -1790,6 +1897,10 @@ func (s *Server) WaitJWTToken(
 	s.mutex.Lock()
 	oAuthFlow := s.oauthAuthFlow.flow
 	authInfo := s.oauthAuthFlow.info
+	// Recorded when the flow was created, not read here: the flow survives a
+	// profile switch, and everything from RequestJWTAuth to the IdP answering
+	// has to count as the same session for the cache.
+	generation := s.oauthAuthFlow.cacheGeneration
 	s.mutex.Unlock()

 	if oAuthFlow == nil || authInfo.DeviceCode != req.DeviceCode {
@@ -1804,11 +1915,17 @@ func (s *Server) WaitJWTToken(
 	token := tokenInfo.GetTokenToUse()

 	jwtCacheTTL := s.getJWTCacheTTL()
-	if jwtCacheTTL > 0 {
-		s.jwtCache.store(token, jwtCacheTTL)
-		log.Debugf("JWT token cached for SSH authentication, TTL: %v", jwtCacheTTL)
-	} else {
+	switch caller, ok := ipcauth.CallerIdentity(ctx); {
+	case jwtCacheTTL <= 0:
 		log.Debug("JWT caching disabled, not storing token")
+	case !ok:
+		log.Debug("not caching the SSH JWT: the caller's identity cannot be verified on this control channel")
+	default:
+		if s.jwtCache.store(token, caller, jwtCacheTTL, generation) {
+			log.Debugf("JWT token cached for SSH authentication, TTL: %v", jwtCacheTTL)
+		} else {
+			log.Debug("not caching the SSH JWT: the session it was obtained under ended while the IdP was polled")
+		}
 	}

 	s.mutex.Lock()
@@ -2128,6 +2245,7 @@ func (s *Server) GetConfig(ctx context.Context, req *proto.GetConfigRequest) (*p
 		Mtu:                           int64(cfg.MTU),
 		DisableAutoConnect:            cfg.DisableAutoConnect,
 		ServerSSHAllowed:              *cfg.ServerSSHAllowed,
+		RemoteJobsAllowed:             util.ReturnBoolWithDefaultFalse(cfg.RemoteJobsAllowed),
 		RosenpassEnabled:              cfg.RosenpassEnabled,
 		RosenpassPermissive:           cfg.RosenpassPermissive,
 		BlockInbound:                  cfg.BlockInbound,
@@ -2218,7 +2336,7 @@ func (s *Server) RemoveProfile(ctx context.Context, msg *proto.RemoveProfileRequ
 		return nil, err
 	}

-	if err := s.logoutFromProfile(ctx, resolved); err != nil {
+	if err := s.logoutFromProfile(ctx, resolved, msg.Username); err != nil {
 		// Deregistration is best-effort here: the local profile is removed
 		// either way, so an unprivileged caller leaves the peer registered on
 		// the management server rather than being blocked from removing it.