Compare commits

...

3 Commits

Author SHA1 Message Date
Zoltán Papp
71df67a2b8 [client] Clear the login-required latch after installing the new client
Run() released the latch before setState() swapped in the fresh connect
client, so Status() calls landing in that window still read the previous
run's context state — which holds the NeedsLogin that prompted the login
— and re-latched what had just been cleared. The generation guard does
not catch this: the clear precedes the observation, so the generation
matches and the store counts as current. The state-change goroutine
calls Status() on every recorder tick, and the login path emits several,
so the window is ordinary traffic rather than a rare interleaving.

Clear once the replacement client is installed instead.
2026-07-28 17:42:04 +02:00
Zoltán Papp
408301714e [client] Fix two concurrency bugs in the Android session binding
Status() read the run loop's label outside any synchronization and then
stored it, so a login or extend completing in that window was undone by
the stale observation, stranding the UI on "login required" over a
healthy session. Guard the latch with a mutex and a generation counter,
and route both clears through the same helper, so a clear that lands
mid-observation wins.

The watch goroutines kept delivering to removed and replaced listeners:
both subscriptions are buffered (one pending tick, ten pending events),
and unsubscribing only stops new items while the loops drain what is
already queued. On Android those callbacks cross into Java, so a
delivery after teardown can reach a listener whose collaborators are
gone. Give each registration a done signal, close it before
unsubscribing, and check it immediately before every callback.
2026-07-28 17:17:43 +02:00
Zoltán Papp
d00068bd89 [client] Expose the SSO auth session to the Android binding
The Android client had no way to see or refresh the peer's SSO session:
the core tracked the deadline and published expiry warnings, but none of
it was exported, so an expired session surfaced only as a raw error
string from the engine run loop.

Mirror the surface the daemon serves its tray:

- Status() and SessionExpiresAtUnix() report the run-loop status label
  and the tracked deadline, the two values StatusResponse carries.
- StateChangeListener signals state changes (payload-free, consumers
  re-read the getters) and forwards the session-expiry warnings from
  the engine's watcher, filtered out of the shared event stream.
- ExtendAuthSession() runs the interactive SSO flow and refreshes the
  deadline without touching the tunnel, with CancelExtendAuthSession()
  for an abandoned browser round-trip — its PKCE wait would otherwise
  hold the loopback port until it timed out and block every retry.
- DismissSessionWarning() suppresses the final warning.

Status() latches NeedsLogin: the run loop keeps its status in a
per-run context state that a restart replaces with a fresh Idle one, so
an engine restart would otherwise erase the fact that the peer still
needs to log in. Only a successful login or extend clears it.
2026-07-28 16:47:53 +02:00
2 changed files with 333 additions and 1 deletions

View File

@@ -75,6 +75,24 @@ type Client struct {
connectClient *internal.ConnectClient
config *profilemanager.Config
cacheDir string
stateChangeMu sync.Mutex
stateChangeSubID string
eventSub *peer.EventSubscription
// Closed to stop the watch goroutines from delivering buffered items to a
// listener that has been removed or replaced. See stopStateChangeWatchLocked.
stateChangeDone chan struct{}
// Latched "the server wants an interactive login": survives the engine
// restarts that replace the run loop's context state. See Client.Status.
// Guarded by loginRequiredMu together with loginCleared, which counts
// clears so a stale observation cannot re-latch over one.
loginRequiredMu sync.Mutex
loginRequired bool
loginCleared uint64
extendMu sync.Mutex
extendCancel context.CancelFunc
}
func (c *Client) setState(cfg *profilemanager.Config, cacheDir string, cc *internal.ConnectClient) {
@@ -148,11 +166,16 @@ func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroid
if err != nil {
return err
}
// todo do not throw error in case of cancelled context
ctx = internal.CtxInitState(ctx)
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder)
c.setState(cfg, cacheDir, connectClient)
// This path runs the interactive SSO flow, so reaching here means the peer
// is authenticated again — release the latch Status() reports from. Clear
// only once the fresh connect client is installed: until then Status()
// still reads the previous run's context state, which holds the NeedsLogin
// that prompted this login, and would re-latch what was just cleared.
c.clearLoginRequired()
return connectClient.RunOnAndroid(c.tunAdapter, c.iFaceDiscover, c.networkChangeListener, slices.Clone(dns.items), dnsReadyListener, stateFile, cacheDir)
}

309
client/android/session.go Normal file
View File

@@ -0,0 +1,309 @@
//go:build android
package android
import (
"context"
"fmt"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/internal"
"github.com/netbirdio/netbird/client/internal/auth"
"github.com/netbirdio/netbird/client/internal/auth/sessionwatch"
"github.com/netbirdio/netbird/client/internal/peer"
cProto "github.com/netbirdio/netbird/client/proto"
)
// StateChangeListener receives client state notifications.
//
// OnStateChanged is a payload-free wake-up whenever the state snapshot
// changed: connection state, the run-loop status label (e.g. NeedsLogin) or
// the session deadline. It mirrors the daemon's SubscribeStatus stream
// trigger — on each signal the consumer pulls the fresh values via
// Status() / SessionExpiresAtUnix().
//
// OnSessionExpiring forwards the engine's session-expiry warnings, fired at
// sessionwatch.WarningLead before the deadline and again at FinalWarningLead
// (finalWarning true). The second one is suppressed when the user dismissed
// the first via DismissSessionWarning. The daemon turns the same events into
// its tray notification.
type StateChangeListener interface {
OnStateChanged()
OnSessionExpiring(expiresAtUnix int64, leadMinutes int64, finalWarning bool)
}
// Status returns the connect run-loop's status label — the same value the
// desktop daemon serves in StatusResponse.Status. "NeedsLogin" means the
// management server rejected the peer and an interactive login is required.
//
// The label is latched: the run loop keeps its status in a per-run context
// state, which a restart replaces with a fresh Idle one, so an engine restart
// (network change, always-on) would otherwise erase the fact that the peer
// still needs to log in. Only a successful interactive login or extend clears
// it — see clearLoginRequired.
func (c *Client) Status() string {
latched, generation := c.loginRequiredState()
if latched {
return string(internal.StatusNeedsLogin)
}
cc := c.getConnectClient()
if cc == nil {
return string(internal.StatusIdle)
}
status := cc.Status()
if status == internal.StatusNeedsLogin {
c.latchLoginRequired(generation)
}
return string(status)
}
func (c *Client) loginRequiredState() (bool, uint64) {
c.loginRequiredMu.Lock()
defer c.loginRequiredMu.Unlock()
return c.loginRequired, c.loginCleared
}
// latchLoginRequired records a NeedsLogin observation, unless a clear landed
// while the caller was reading the run loop's status: cc.Status() is read
// outside the lock, so a login or extend completing in that window would
// otherwise be undone by this stale observation, stranding the UI on
// "login required" over a healthy session.
func (c *Client) latchLoginRequired(observedGeneration uint64) {
c.loginRequiredMu.Lock()
defer c.loginRequiredMu.Unlock()
if c.loginCleared != observedGeneration {
return
}
c.loginRequired = true
}
// clearLoginRequired releases the latch after a successful interactive login
// or session extend, and invalidates any observation already in flight.
func (c *Client) clearLoginRequired() {
c.loginRequiredMu.Lock()
defer c.loginRequiredMu.Unlock()
c.loginRequired = false
c.loginCleared++
}
// SessionExpiresAtUnix returns the SSO session deadline as unix seconds, or 0
// when no deadline is known (not SSO-registered, expiry disabled, or the
// engine has not received one yet). A past value means the session expired.
// Mirror of StatusResponse.sessionExpiresAt on the desktop daemon.
func (c *Client) SessionExpiresAtUnix() int64 {
deadline := c.recorder.GetSessionExpiresAt()
if deadline.IsZero() {
return 0
}
return deadline.Unix()
}
// SetStateChangeListener registers the state notification listener.
// Replaces any previously registered listener; remove it with
// RemoveStateChangeListener.
func (c *Client) SetStateChangeListener(listener StateChangeListener) {
c.stateChangeMu.Lock()
defer c.stateChangeMu.Unlock()
c.stopStateChangeWatchLocked()
if listener == nil {
return
}
// Both subscriptions are buffered (one pending tick, ten pending events),
// so unsubscribing is not enough to stop callbacks: the loops would drain
// what is already queued and deliver it to a listener the caller has
// already removed or replaced. Gate every callback on this registration's
// own signal, which is closed before unsubscribing.
done := make(chan struct{})
c.stateChangeDone = done
id, ch := c.recorder.SubscribeToStateChanges()
c.stateChangeSubID = id
// The channel is closed by UnsubscribeFromStateChanges, which ends the
// goroutine. Ticks are coalesced (buffer of one), so a burst of changes
// wakes the listener once.
go func() {
for range ch {
select {
case <-done:
return
default:
}
listener.OnStateChanged()
}
}()
c.eventSub = c.recorder.SubscribeToEvents()
go watchSessionWarnings(c.eventSub, listener, done)
}
// RemoveStateChangeListener unregisters the state notification listener.
func (c *Client) RemoveStateChangeListener() {
c.stateChangeMu.Lock()
defer c.stateChangeMu.Unlock()
c.stopStateChangeWatchLocked()
}
// DismissSessionWarning records the user's "Dismiss" on the first expiry
// warning and suppresses the final one for the current deadline. A refreshed
// deadline re-arms both. No-op while the engine is not running.
func (c *Client) DismissSessionWarning() {
cc := c.getConnectClient()
if cc == nil {
return
}
engine := cc.Engine()
if engine == nil {
return
}
engine.DismissSessionWarning()
}
// ExtendAuthSession runs the interactive SSO flow to obtain a fresh JWT and
// asks the management server to extend the session deadline. The tunnel is
// untouched: no resync, no reconnect. Async; the result arrives on the
// listener. Mirror of the daemon's RequestExtendAuthSession /
// WaitExtendAuthSession RPC pair, with URLOpener playing the "UI opens the
// browser" role.
//
// Only one flow may be in flight: the PKCE step binds a fixed loopback port,
// so a second concurrent flow would fail on that bind. Call
// CancelExtendAuthSession when the user abandons the browser.
func (c *Client) ExtendAuthSession(urlOpener URLOpener, isAndroidTV bool, resultListener ErrListener) {
ctx, err := c.beginExtend()
if err != nil {
resultListener.OnError(err)
return
}
go func() {
defer c.endExtend()
if err := c.extendAuthSession(ctx, urlOpener, isAndroidTV); err != nil {
resultListener.OnError(err)
return
}
resultListener.OnSuccess()
}()
}
// CancelExtendAuthSession aborts an in-flight ExtendAuthSession. The tunnel is
// left alone — unlike the login flow, which cancels the whole client context
// by stopping the engine. Without this the abandoned PKCE wait keeps its
// loopback port for the full flow timeout and blocks every later attempt.
// No-op when no flow is running.
func (c *Client) CancelExtendAuthSession() {
c.extendMu.Lock()
defer c.extendMu.Unlock()
if c.extendCancel != nil {
c.extendCancel()
}
}
func (c *Client) stopStateChangeWatchLocked() {
// Signal first, unsubscribe second: closing the channels only stops new
// items, and the loops would still hand whatever is buffered to a listener
// that is no longer registered.
if c.stateChangeDone != nil {
close(c.stateChangeDone)
c.stateChangeDone = nil
}
if c.stateChangeSubID != "" {
c.recorder.UnsubscribeFromStateChanges(c.stateChangeSubID)
c.stateChangeSubID = ""
}
if c.eventSub != nil {
// Closes the channel, which ends watchSessionWarnings.
c.recorder.UnsubscribeFromEvents(c.eventSub)
c.eventSub = nil
}
}
// watchSessionWarnings forwards the engine's session-expiry warnings to the
// listener. The event stream also carries unrelated traffic — network-map
// updates on every sync, DNS and route errors — so everything but an
// AUTHENTICATION event carrying the session-warning marker is dropped. Exits
// when the subscription is closed by UnsubscribeFromEvents, or earlier when
// done is closed — the stream buffers up to ten events, and a deregistered
// listener must not receive the ones already queued.
func watchSessionWarnings(sub *peer.EventSubscription, listener StateChangeListener, done <-chan struct{}) {
for ev := range sub.Events() {
select {
case <-done:
return
default:
}
if ev.GetCategory() != cProto.SystemEvent_AUTHENTICATION {
continue
}
meta := ev.GetMetadata()
if meta[sessionwatch.MetaSessionWarning] != "true" {
// Other AUTHENTICATION events exist (e.g. a deadline rejected as
// out of range); they carry no warning marker.
continue
}
deadline, err := sessionwatch.ParseExpiresAt(meta[sessionwatch.MetaSessionExpiresAt])
if err != nil {
log.Warnf("session warning event with unparsable deadline: %v", err)
continue
}
lead, err := sessionwatch.ParseLeadMinutes(meta[sessionwatch.MetaSessionLeadMinutes])
if err != nil {
// Informational only — the deadline above is what drives the UI.
lead = 0
}
listener.OnSessionExpiring(deadline.Unix(), int64(lead),
meta[sessionwatch.MetaSessionFinal] == "true")
}
}
func (c *Client) beginExtend() (context.Context, error) {
c.extendMu.Lock()
defer c.extendMu.Unlock()
if c.extendCancel != nil {
return nil, fmt.Errorf("session extend already in progress")
}
ctx, cancel := context.WithCancel(context.Background())
c.extendCancel = cancel
return ctx, nil
}
func (c *Client) endExtend() {
c.extendMu.Lock()
defer c.extendMu.Unlock()
if c.extendCancel != nil {
c.extendCancel()
c.extendCancel = nil
}
}
func (c *Client) extendAuthSession(ctx context.Context, urlOpener URLOpener, isAndroidTV bool) error {
cfg, _, cc := c.stateSnapshot()
if cfg == nil || cc == nil {
return fmt.Errorf("engine is not running")
}
engine := cc.Engine()
if engine == nil {
return fmt.Errorf("engine is not initialized")
}
authClient, err := auth.NewAuth(ctx, cfg.PrivateKey, cfg.ManagementURL, cfg)
if err != nil {
return fmt.Errorf("failed to create auth client: %v", err)
}
defer authClient.Close()
a := &Auth{ctx: ctx, config: cfg}
tokenInfo, err := a.foregroundGetTokenInfo(authClient, urlOpener, isAndroidTV)
if err != nil {
return fmt.Errorf("interactive sso login failed: %v", err)
}
if _, err := engine.ExtendAuthSession(ctx, tokenInfo.GetTokenToUse()); err != nil {
return err
}
c.clearLoginRequired()
go urlOpener.OnLoginSuccess()
return nil
}