Merge remote-tracking branch 'origin/main' into 0.74.7-branch-sync

# Conflicts:
#	client/internal/peer/conn.go
#	client/internal/peer/conn_test.go
#	client/internal/peer/wg_watcher.go
#	client/internal/peer/wg_watcher_test.go
#	go.mod
#	go.sum
This commit is contained in:
Viktor Liu
2026-07-17 16:34:58 +02:00
453 changed files with 52390 additions and 7003 deletions

View File

@@ -3,6 +3,7 @@ package auth
import (
"context"
"net/url"
"strings"
"sync"
"time"
@@ -21,6 +22,25 @@ import (
mgmProto "github.com/netbirdio/netbird/shared/management/proto"
)
// peerLoginExpiredMsg is the exact phrase the management server returns
// when a previously SSO-enrolled peer's login has expired. Sourced from
// shared/management/status/error.go (NewPeerLoginExpiredError). Matched
// by substring so a future server-side rewording that keeps the phrase
// still triggers the friendly fallback in Login().
const peerLoginExpiredMsg = "peer login has expired"
// errSetupKeyOnSSOExpiredPeer replaces the raw management error when the
// user runs `netbird login -k <setup-key>` against a peer that was
// originally enrolled via SSO. Wrapped in a PermissionDenied gRPC status
// so callers' existing isPermissionDenied / isAuthError checks still
// classify it correctly (early-exit from retry backoff, StatusNeedsLogin
// in the server state machine).
var errSetupKeyOnSSOExpiredPeer = status.Error(
codes.PermissionDenied,
"this peer was originally enrolled via SSO and its session has expired. "+
"Setup keys can only enrol new peers — run `netbird up` (interactive SSO) to re-login.",
)
// Auth manages authentication operations with the management server
// It maintains a long-lived connection and automatically handles reconnection with backoff
type Auth struct {
@@ -184,6 +204,15 @@ func (a *Auth) Login(ctx context.Context, setupKey string, jwtToken string) (err
log.Debugf("peer registration required")
_, err = a.registerPeer(client, ctx, setupKey, jwtToken, pubSSHKey)
if err != nil {
// The peer pub-key is already on file with the management
// server (originally enrolled via SSO) and the session has
// expired. The setup-key path can only enrol new peers, so
// retrying with -k will keep failing. Replace the raw mgm
// message with an actionable hint that tells the user to
// re-authenticate via SSO instead.
if setupKey != "" && jwtToken == "" && isPeerLoginExpired(err) {
err = errSetupKeyOnSSOExpiredPeer
}
isAuthError = isPermissionDenied(err)
return err
}
@@ -473,3 +502,16 @@ func isLoginNeeded(err error) bool {
func isRegistrationNeeded(err error) bool {
return isPermissionDenied(err)
}
// isPeerLoginExpired reports whether err is the management server's
// "peer login has expired" PermissionDenied response. Used by Login to
// detect the case where the caller passed a setup-key but the peer is
// actually an SSO-enrolled record whose session needs refreshing — the
// setup-key path cannot help there.
func isPeerLoginExpired(err error) bool {
if !isPermissionDenied(err) {
return false
}
s, _ := status.FromError(err)
return strings.Contains(s.Message(), peerLoginExpiredMsg)
}

View File

@@ -0,0 +1,80 @@
package auth
import (
"errors"
"strings"
"testing"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)
func TestIsPeerLoginExpired(t *testing.T) {
cases := []struct {
name string
err error
want bool
}{
{
name: "nil",
err: nil,
want: false,
},
{
name: "plain error (not a gRPC status)",
err: errors.New("network read: connection reset"),
want: false,
},
{
name: "PermissionDenied with different message",
err: status.Error(codes.PermissionDenied, "user is blocked"),
want: false,
},
{
name: "Unauthenticated with the expected phrase",
// Wrong status code — must still return false.
err: status.Error(codes.Unauthenticated, "peer login has expired, please log in once more"),
want: false,
},
{
name: "exact server message",
err: status.Error(codes.PermissionDenied, "peer login has expired, please log in once more"),
want: true,
},
{
name: "phrase as substring",
// Future-proofing: if mgm reworords but keeps the phrase,
// the friendly fallback must still kick in.
err: status.Error(codes.PermissionDenied, "session refused: peer login has expired (account=foo)"),
want: true,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
if got := isPeerLoginExpired(tc.err); got != tc.want {
t.Fatalf("isPeerLoginExpired(%v) = %v, want %v", tc.err, got, tc.want)
}
})
}
}
func TestErrSetupKeyOnSSOExpiredPeer(t *testing.T) {
// Sentinel must surface as PermissionDenied so the upstream
// isPermissionDenied / isAuthError checks classify it correctly
// (short-circuit retry backoff, set StatusNeedsLogin).
if !isPermissionDenied(errSetupKeyOnSSOExpiredPeer) {
t.Fatalf("errSetupKeyOnSSOExpiredPeer must be a PermissionDenied gRPC error")
}
// Message must actually mention SSO and `netbird up` so it is
// actionable for the end user. Loose substring checks keep the
// test resilient to copy edits.
s, _ := status.FromError(errSetupKeyOnSSOExpiredPeer)
msg := strings.ToLower(s.Message())
for _, want := range []string{"sso", "netbird up"} {
if !strings.Contains(msg, want) {
t.Errorf("sentinel message should contain %q, got %q", want, s.Message())
}
}
}

View File

@@ -259,12 +259,18 @@ func (d *DeviceAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowIn
ticker := time.NewTicker(interval)
defer ticker.Stop()
log.Infof("device flow: waiting for user authorization, polling token endpoint every %s, code expires in %s", interval, timeout)
start := time.Now()
polls := 0
for {
select {
case <-waitCtx.Done():
return TokenInfo{}, waitCtx.Err()
case <-ticker.C:
polls++
tokenResponse, err := d.requestToken(info)
if err != nil {
return TokenInfo{}, fmt.Errorf("parsing token response failed with error: %v", err)
@@ -272,10 +278,12 @@ func (d *DeviceAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowIn
if tokenResponse.Error != "" {
if tokenResponse.Error == "authorization_pending" {
log.Tracef("device flow: authorization still pending after poll %d", polls)
continue
} else if tokenResponse.Error == "slow_down" {
interval += (3 * time.Second)
ticker.Reset(interval)
log.Infof("device flow: IdP requested slow_down, polling interval increased to %s", interval)
continue
}
@@ -296,6 +304,7 @@ func (d *DeviceAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowIn
return TokenInfo{}, fmt.Errorf("validate access token failed with error: %v", err)
}
log.Infof("device flow: user authorization confirmed after %d polls in %s", polls, time.Since(start).Round(time.Second))
return tokenInfo, err
}
}

View File

@@ -0,0 +1,89 @@
package auth
import (
"context"
"sync"
"time"
)
// PendingFlow stores an in-progress OAuth flow between the RPC that
// initiates it (returns the verification URI to the UI) and the RPC
// that waits for the user to complete it. The flow handle, the
// device-code info, and the absolute expiry are kept together so the
// waiting RPC can validate the device code and reuse the same flow.
//
// PendingFlow is safe for concurrent use; callers must not access the
// stored fields directly.
type PendingFlow struct {
mu sync.Mutex
flow OAuthFlow
info AuthFlowInfo
expiresAt time.Time
waitCancel context.CancelFunc
}
// NewPendingFlow returns an empty PendingFlow ready to be populated by Set.
func NewPendingFlow() *PendingFlow {
return &PendingFlow{}
}
// Set stores the flow and its authorization info, computing the absolute
// expiry from info.ExpiresIn (seconds, as returned by the IdP).
func (p *PendingFlow) Set(flow OAuthFlow, info AuthFlowInfo) {
p.mu.Lock()
defer p.mu.Unlock()
p.flow = flow
p.info = info
p.expiresAt = time.Now().Add(time.Duration(info.ExpiresIn) * time.Second)
}
// Get returns the stored flow, info, and whether a flow is currently
// pending. Returns (nil, zero, false) after Clear or before Set.
func (p *PendingFlow) Get() (OAuthFlow, AuthFlowInfo, bool) {
p.mu.Lock()
defer p.mu.Unlock()
if p.flow == nil {
return nil, AuthFlowInfo{}, false
}
return p.flow, p.info, true
}
// ExpiresAt returns the absolute expiry of the pending flow. Returns
// the zero time when no flow is pending.
func (p *PendingFlow) ExpiresAt() time.Time {
p.mu.Lock()
defer p.mu.Unlock()
return p.expiresAt
}
// SetWaitCancel records the cancel function for the goroutine currently
// blocked in WaitToken so a new RequestAuth can preempt it.
func (p *PendingFlow) SetWaitCancel(cancel context.CancelFunc) {
p.mu.Lock()
defer p.mu.Unlock()
p.waitCancel = cancel
}
// CancelWait invokes and clears the stored wait-cancel, if any. Safe to
// call when no wait is in progress.
func (p *PendingFlow) CancelWait() {
p.mu.Lock()
cancel := p.waitCancel
p.waitCancel = nil
p.mu.Unlock()
if cancel != nil {
cancel()
}
}
// Clear resets the pending flow to empty. Any stored wait-cancel is
// dropped without being invoked — call CancelWait first if the waiting
// goroutine must be stopped.
func (p *PendingFlow) Clear() {
p.mu.Lock()
defer p.mu.Unlock()
p.flow = nil
p.info = AuthFlowInfo{}
p.expiresAt = time.Time{}
p.waitCancel = nil
}

View File

@@ -188,6 +188,8 @@ func (p *PKCEAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowInfo
waitCtx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
log.Infof("pkce flow: waiting for authorization callback on %s, timeout %s", p.oAuthConfig.RedirectURL, timeout)
tokenChan := make(chan *oauth2.Token, 1)
errChan := make(chan error, 1)
@@ -221,6 +223,7 @@ func (p *PKCEAuthorizationFlow) WaitToken(ctx context.Context, info AuthFlowInfo
func (p *PKCEAuthorizationFlow) startServer(server *http.Server, tokenChan chan<- *oauth2.Token, errChan chan<- error) {
mux := http.NewServeMux()
mux.HandleFunc("/", func(w http.ResponseWriter, req *http.Request) {
log.Infof("pkce flow: received authorization callback from IdP")
cert := p.providerConfig.ClientCertPair
if cert != nil {
tr := &http.Transport{
@@ -271,11 +274,18 @@ func (p *PKCEAuthorizationFlow) handleRequest(req *http.Request) (*oauth2.Token,
return nil, fmt.Errorf("authentication failed: missing code")
}
return p.oAuthConfig.Exchange(
exchangeStart := time.Now()
token, err := p.oAuthConfig.Exchange(
req.Context(),
code,
oauth2.SetAuthURLParam("code_verifier", p.codeVerifier),
)
if err != nil {
return nil, err
}
log.Infof("pkce flow: authorization code exchanged for token in %s", time.Since(exchangeStart).Round(time.Millisecond))
return token, nil
}
func (p *PKCEAuthorizationFlow) parseOAuthToken(token *oauth2.Token) (TokenInfo, error) {

View File

@@ -0,0 +1,82 @@
package sessionwatch
import (
"strconv"
"time"
)
// internal event kinds are no longer exposed: the watcher drives the Sink
// directly (NotifyStateChange on deadline change/clear, PublishEvent at
// each warning lead). Tests use a mock Sink to observe what the watcher
// emits.
// Metadata keys attached by the daemon to session-warning SystemEvents.
// The UI tray reads these to build a locale-aware notification without
// relying on the daemon's locale-less UserMessage string, and to
// disambiguate the T-WarningLead notification from the T-FinalWarningLead
// fallback that auto-opens the SessionAboutToExpire dialog.
const (
// MetaSessionWarning is set to "true" on both warning events (T-10 and
// T-2) so the UI can detect a session-warning SystemEvent without
// matching on the message text. Use MetaSessionFinal to distinguish
// the two.
MetaSessionWarning = "session_warning"
// MetaSessionFinal is set to "true" on the T-FinalWarningLead event
// only. Consumers that need to auto-open the SessionAboutToExpire
// dialog gate on this; T-WarningLead events leave the field unset.
MetaSessionFinal = "session_final_warning"
// MetaSessionExpiresAt carries the absolute UTC deadline encoded with
// FormatExpiresAt; consumers must decode with ParseExpiresAt so a
// future format change stays a single edit.
MetaSessionExpiresAt = "session_expires_at"
// MetaSessionLeadMinutes carries the lead in whole minutes (WarningLead
// for the T-10 event, FinalWarningLead for the T-2 event) so the UI
// can show "expires in ~N minutes" without hardcoding either constant.
MetaSessionLeadMinutes = "lead_minutes"
// MetaSessionDeadlineRejected is attached to the ERROR/AUTHENTICATION
// SystemEvent the daemon emits when it discards a deadline from the
// management server (pre-epoch, too far in the future, or past the
// clock-skew tolerance). The value is the rejection reason string.
// userMessage is left empty; the UI detects the event via this key
// and builds a localized notification — same pattern as the session
// warnings above.
MetaSessionDeadlineRejected = "session_deadline_rejected"
)
// expiresAtLayout is the wire format used for MetaSessionExpiresAt.
// Producer and consumers both go through FormatExpiresAt/ParseExpiresAt
// so this layout stays a single source of truth.
const expiresAtLayout = time.RFC3339
// FormatExpiresAt encodes a deadline for MetaSessionExpiresAt. Always
// emits UTC so a consumer in another timezone reads the same wall-clock
// deadline.
func FormatExpiresAt(t time.Time) string {
return t.UTC().Format(expiresAtLayout)
}
// ParseExpiresAt decodes the MetaSessionExpiresAt value back to a UTC
// time. Returns an error when the field is empty or malformed; the
// caller decides whether to fall back (zero value) or propagate.
func ParseExpiresAt(s string) (time.Time, error) {
t, err := time.Parse(expiresAtLayout, s)
if err != nil {
return time.Time{}, err
}
return t.UTC(), nil
}
// FormatLeadMinutes encodes a lead duration for MetaSessionLeadMinutes
// as the integer count of whole minutes. Sub-minute residuals are
// truncated — the field is informational ("expires in ~N minutes") and
// fractional minutes don't change what the UI displays.
func FormatLeadMinutes(d time.Duration) string {
return strconv.Itoa(int(d / time.Minute))
}
// ParseLeadMinutes decodes a MetaSessionLeadMinutes value. Returns 0
// and the parse error for malformed input; consumers that prefer a
// silent fallback can simply ignore the error.
func ParseLeadMinutes(s string) (int, error) {
return strconv.Atoi(s)
}

View File

@@ -0,0 +1,387 @@
// Package sessionwatch tracks the SSO session expiry deadline that the
// management server publishes via LoginResponse / SyncResponse and fires
// two warning events at fixed lead times before expiry: an interactive
// T-WarningLead notification and a dismiss-gated T-FinalWarningLead
// fallback dialog.
//
// The watcher is idempotent: Update may be called as often as the network
// map snapshots arrive. Repeating the same deadline is a no-op; a new
// deadline reschedules the timers and arms a fresh warning cycle.
//
// Warning firing is edge-detected. Each unique deadline value fires each
// warning callback at most once.
package sessionwatch
import (
"errors"
"fmt"
"sync"
"time"
log "github.com/sirupsen/logrus"
cProto "github.com/netbirdio/netbird/client/proto"
)
const (
// Skew tolerates a small clock difference between the management
// server and this peer before treating a deadline as "in the past".
// Slightly above typical NTP drift; tight enough that the UI doesn't
// paint a stale expiry as if it were valid.
Skew = 30 * time.Second
// maxDeadlineHorizon caps how far in the future an accepted deadline
// can sit. A timestamp beyond this is almost certainly a protocol
// glitch, and silently arming a 100-year timer would hide the bug.
maxDeadlineHorizon = 10 * 365 * 24 * time.Hour
// WarningLead is how far before expiry the first (interactive)
// warning fires. Drives the T-10 OS notification with
// Extend/Dismiss actions.
WarningLead = 10 * time.Minute
// FinalWarningLead is how far before expiry the fallback final
// warning fires. Drives the auto-opened SessionAboutToExpire dialog,
// but only when the user has not dismissed the T-WarningLead warning
// for the same deadline. Must be strictly less than WarningLead.
FinalWarningLead = 2 * time.Minute
)
var (
// ErrDeadlineBeforeEpoch is returned by Update when the supplied
// deadline pre-dates 1970-01-01.
ErrDeadlineBeforeEpoch = errors.New("session deadline before unix epoch")
// ErrDeadlineTooFarFuture is returned by Update when the supplied
// deadline is more than maxDeadlineHorizon in the future.
ErrDeadlineTooFarFuture = errors.New("session deadline too far in the future")
// ErrDeadlineInPast is returned by Update when the supplied deadline
// is more than Skew in the past.
ErrDeadlineInPast = errors.New("session deadline in the past")
)
// StatusRecorder is the side-effect surface the watcher drives on every
// state transition. Production wires this to peer.Status (SetSessionExpiresAt
// for deadline change/clear, PublishEvent for the two warnings); tests pass
// a fake recorder so the same surface is observable without an engine.
//
// The watcher is the single owner of the deadline propagated to the
// recorder: every set, clear, sanity-check rejection and Close routes the
// value through SetSessionExpiresAt, so the SubscribeStatus snapshot the UI
// reads can never drift from the watcher's timer state. (SetSessionExpiresAt
// fans out its own state-change notification, so no separate notify is
// needed.) The recorder is server-scoped and outlives this engine-scoped
// watcher — without the Close-time clear a teardown (Down, or the Down+Up of
// a profile switch) would leave the next session showing the previous one's
// stale "expires in" value.
//
// PublishEvent's signature mirrors peer.Status.PublishEvent: the watcher
// composes the metadata internally so the wire format (MetaSession*) is
// owned by sessionwatch, not the caller.
type StatusRecorder interface {
SetSessionExpiresAt(deadline time.Time)
PublishEvent(
severity cProto.SystemEvent_Severity,
category cProto.SystemEvent_Category,
message string,
userMessage string,
metadata map[string]string,
)
}
// Watcher observes the latest session deadline and fires two warnings
// before it expires: the interactive T-WarningLead notification, and the
// fallback T-FinalWarningLead dialog (suppressed when the user dismissed
// the first one for the same deadline). Safe for concurrent use.
type Watcher struct {
lead time.Duration
finalLead time.Duration
mu sync.Mutex
current time.Time
timer *time.Timer
finalTimer *time.Timer
firedAt time.Time // deadline value the T-WarningLead callback last fired against
finalFiredAt time.Time // deadline value the T-FinalWarningLead callback last fired against
dismissedAt time.Time // deadline value the user dismissed via Dismiss(); gates fireFinal
closed bool
recorder StatusRecorder
}
// New returns a watcher with the package defaults WarningLead and
// FinalWarningLead. Pass nil for recorder to silence side effects (handy
// in unit tests that exercise sanity checks without observing the publish
// path).
func New(recorder StatusRecorder) *Watcher {
return NewWithLeads(WarningLead, FinalWarningLead, recorder)
}
// NewWithLeads returns a watcher with custom lead times. Useful for tests.
// final must be strictly less than lead; otherwise both timers fire in the
// wrong order or simultaneously and the UI flow breaks. A zero final lead
// disables the final-warning timer entirely (see armTimerLocked) so a
// millisecond-scale deadline doesn't flush both timers in one tick.
func NewWithLeads(lead, final time.Duration, recorder StatusRecorder) *Watcher {
return &Watcher{
lead: lead,
finalLead: final,
recorder: recorder,
}
}
// Update sets the latest deadline. Pass the zero time to clear (e.g. when
// a Sync push from the server omits the field because login expiration
// was disabled).
//
// Same-value updates are no-ops. A different non-zero value cancels any
// pending timer, resets the "already fired" guard, and arms a new one.
//
// Returns one of the sentinel Err* values when the deadline fails the
// sanity checks (pre-epoch, far future, or in the past beyond Skew).
// In every error case the watcher first clears its state so it stays
// consistent with what the caller will push into its other sinks (e.g.
// applySessionDeadline forces a zero deadline into the status recorder
// after a non-nil error).
func (w *Watcher) Update(deadline time.Time) error {
w.mu.Lock()
if w.closed {
w.mu.Unlock()
return nil
}
if deadline.IsZero() {
w.clearLocked()
return nil
}
now := time.Now()
switch {
case deadline.Before(time.Unix(0, 0)):
w.clearLocked()
return fmt.Errorf("%w: %v", ErrDeadlineBeforeEpoch, deadline)
case deadline.After(now.Add(maxDeadlineHorizon)):
w.clearLocked()
return fmt.Errorf("%w: %v", ErrDeadlineTooFarFuture, deadline)
case deadline.Before(now.Add(-Skew)):
w.clearLocked()
return fmt.Errorf("%w: %v (now=%v)", ErrDeadlineInPast, deadline, now)
}
if deadline.Equal(w.current) {
w.mu.Unlock()
return nil
}
w.stopTimerLocked()
w.current = deadline
// Reset every per-deadline guard so a refreshed deadline arms a fresh
// warning cycle: both edge triggers and the user Dismiss decision
// (the user agreed to the old deadline expiring; a new deadline
// restarts the contract).
w.firedAt = time.Time{}
w.finalFiredAt = time.Time{}
w.dismissedAt = time.Time{}
w.armTimerLocked(deadline)
recorder := w.recorder
w.mu.Unlock()
if recorder != nil {
recorder.SetSessionExpiresAt(deadline)
}
log.Infof("auth session deadline set to: %s (in %s)", deadline.Format(time.RFC3339), time.Until(deadline).Round(time.Second))
return nil
}
// Deadline returns the most recently observed deadline. Zero when no
// deadline is currently tracked.
func (w *Watcher) Deadline() time.Time {
w.mu.Lock()
defer w.mu.Unlock()
return w.current
}
// Dismiss records the user's "Dismiss" action against the current deadline
// and suppresses the upcoming final-warning callback for that deadline.
// Idempotent: repeated calls are no-ops. A subsequent Update with a fresh
// deadline resets the dismissal so the final-warning cycle re-arms.
//
// No-op when the watcher holds no deadline or has been closed.
func (w *Watcher) Dismiss() {
w.mu.Lock()
defer w.mu.Unlock()
if w.closed || w.current.IsZero() {
return
}
if w.dismissedAt.Equal(w.current) {
return
}
w.dismissedAt = w.current
// Cancel the armed final-warning timer eagerly. fireFinal would also
// gate on dismissedAt, but stopping the timer avoids a wakeup with
// nothing to do and makes the intent visible.
if w.finalTimer != nil {
w.finalTimer.Stop()
w.finalTimer = nil
}
log.Infof("auth session final-warning dismissed for deadline %s", w.current.Format(time.RFC3339))
}
// Close stops any pending timer and drops the deadline on the status
// recorder. Update calls after Close are ignored. Clearing the recorder
// here is what keeps a teardown (Down, or the Down+Up of a profile switch)
// from leaving the next session showing this one's stale "expires in"
// value — the recorder is server-scoped and outlives this engine-scoped
// watcher, so nothing else drops the anchor on teardown.
func (w *Watcher) Close() {
w.mu.Lock()
if w.closed {
w.mu.Unlock()
return
}
w.closed = true
w.stopTimerLocked()
hadDeadline := !w.current.IsZero()
w.current = time.Time{}
w.firedAt = time.Time{}
w.finalFiredAt = time.Time{}
w.dismissedAt = time.Time{}
recorder := w.recorder
w.mu.Unlock()
if recorder != nil && hadDeadline {
recorder.SetSessionExpiresAt(time.Time{})
}
}
// clearLocked drops the tracked deadline and notifies the recorder so
// downstream consumers (SubscribeStatus stream, UI) drop their anchor.
// The caller must hold w.mu; this helper releases it before invoking
// the recorder.
func (w *Watcher) clearLocked() {
if w.current.IsZero() {
w.mu.Unlock()
return
}
w.stopTimerLocked()
w.current = time.Time{}
w.firedAt = time.Time{}
w.finalFiredAt = time.Time{}
w.dismissedAt = time.Time{}
recorder := w.recorder
w.mu.Unlock()
if recorder != nil {
recorder.SetSessionExpiresAt(time.Time{})
}
log.Infof("auth session deadline cleared")
}
func (w *Watcher) stopTimerLocked() {
if w.timer != nil {
w.timer.Stop()
w.timer = nil
}
if w.finalTimer != nil {
w.finalTimer.Stop()
w.finalTimer = nil
}
}
func (w *Watcher) armTimerLocked(deadline time.Time) {
w.timer = armOneShotLocked(deadline.Add(-w.lead), func() { w.fire(deadline) })
// finalLead <= 0 disables the final-warning timer entirely. Used by
// tests that predate the final-warning fallback so a millisecond-scale
// deadline does not flush both timers at once.
if w.finalLead > 0 {
w.finalTimer = armOneShotLocked(deadline.Add(-w.finalLead), func() { w.fireFinal(deadline) })
}
}
func (w *Watcher) fire(armedFor time.Time) {
w.mu.Lock()
if w.closed || !w.current.Equal(armedFor) {
// Deadline moved while we were waiting (e.g. a successful extend).
// The reschedule path armed a fresh timer; this one is stale.
w.mu.Unlock()
return
}
if !w.firedAt.IsZero() && w.firedAt.Equal(armedFor) {
w.mu.Unlock()
return
}
w.firedAt = armedFor
recorder := w.recorder
w.mu.Unlock()
if recorder == nil {
return
}
log.Infof("auth session expiry soon warning fired")
publishWarning(recorder, armedFor, false)
}
// fireFinal mirrors fire for the T-FinalWarningLead timer with an extra
// dismiss-gate: if the user dismissed the T-WarningLead notification for
// this deadline, the final warning is suppressed entirely.
func (w *Watcher) fireFinal(armedFor time.Time) {
w.mu.Lock()
if w.closed || !w.current.Equal(armedFor) {
w.mu.Unlock()
return
}
if !w.finalFiredAt.IsZero() && w.finalFiredAt.Equal(armedFor) {
w.mu.Unlock()
return
}
if w.dismissedAt.Equal(armedFor) {
w.mu.Unlock()
log.Infof("auth session final-warning skipped (dismissed by user)")
return
}
w.finalFiredAt = armedFor
recorder := w.recorder
w.mu.Unlock()
if recorder == nil {
return
}
log.Infof("auth session final-warning fired")
publishWarning(recorder, armedFor, true)
}
// armOneShotLocked schedules cb at fireAt. When fireAt is already in the
// past it dispatches on the next scheduler tick so a state-change recorder
// notification (invoked after w.mu is released) lands first. Caller must
// hold w.mu.
func armOneShotLocked(fireAt time.Time, cb func()) *time.Timer {
delay := time.Until(fireAt)
if delay <= 0 {
return time.AfterFunc(0, cb)
}
return time.AfterFunc(delay, cb)
}
// publishWarning composes the SystemEvent for a watcher-fired warning and
// pushes it through the recorder. Severity is CRITICAL on both — bypassing
// the user's Notifications toggle is deliberate: missing the warning
// window forces the post-mortem SessionExpired flow (tunnel torn down,
// lock icon, manual re-login), which is the UX we are trying to avoid.
func publishWarning(recorder StatusRecorder, deadline time.Time, final bool) {
lead := WarningLead
message := "session expiry warning"
meta := map[string]string{
MetaSessionWarning: "true",
MetaSessionExpiresAt: FormatExpiresAt(deadline),
}
if final {
lead = FinalWarningLead
message = "session expiry final warning"
meta[MetaSessionFinal] = "true"
}
meta[MetaSessionLeadMinutes] = FormatLeadMinutes(lead)
recorder.PublishEvent(
cProto.SystemEvent_CRITICAL,
cProto.SystemEvent_AUTHENTICATION,
message,
"",
meta,
)
}

View File

@@ -0,0 +1,519 @@
package sessionwatch
import (
"errors"
"sync"
"testing"
"time"
cProto "github.com/netbirdio/netbird/client/proto"
)
// fakeRecorder satisfies StatusRecorder and records every call so tests
// can observe what the watcher emits. SetSessionExpiresAt and PublishEvent
// land in the same ordered events slice (with the Kind distinguishing
// them) so tests that care about ordering still work. lastDeadline holds
// the most recent value passed to SetSessionExpiresAt so tests can assert
// the recorder ended up cleared/set as expected.
type fakeRecorder struct {
mu sync.Mutex
events []event
lastDeadline time.Time
}
type eventKind int
const (
stateChange eventKind = iota
publish
)
type event struct {
kind eventKind
// Set only for publish events.
severity cProto.SystemEvent_Severity
category cProto.SystemEvent_Category
message string
meta map[string]string
}
// SetSessionExpiresAt mirrors peer.Status: a same-value write is a no-op,
// a real change records the new value and fans out a state-change (the
// production recorder calls notifyStateChange internally). The baseline
// is the zero time, so an initial clear before any deadline is set emits
// nothing — matching the real recorder.
func (r *fakeRecorder) SetSessionExpiresAt(deadline time.Time) {
r.mu.Lock()
defer r.mu.Unlock()
if r.lastDeadline.Equal(deadline) {
return
}
r.lastDeadline = deadline
r.events = append(r.events, event{kind: stateChange})
}
func (r *fakeRecorder) deadline() time.Time {
r.mu.Lock()
defer r.mu.Unlock()
return r.lastDeadline
}
func (r *fakeRecorder) PublishEvent(
severity cProto.SystemEvent_Severity,
category cProto.SystemEvent_Category,
message string,
_ string,
metadata map[string]string,
) {
r.mu.Lock()
defer r.mu.Unlock()
r.events = append(r.events, event{
kind: publish,
severity: severity,
category: category,
message: message,
meta: metadata,
})
}
func (r *fakeRecorder) snapshot() []event {
r.mu.Lock()
defer r.mu.Unlock()
out := make([]event, len(r.events))
copy(out, r.events)
return out
}
func (e event) isFinalWarning() bool {
return e.kind == publish && e.meta[MetaSessionFinal] == "true"
}
func (e event) isWarning() bool {
return e.kind == publish && e.meta[MetaSessionWarning] == "true" && e.meta[MetaSessionFinal] != "true"
}
func countWhere(events []event, pred func(event) bool) int {
n := 0
for _, e := range events {
if pred(e) {
n++
}
}
return n
}
func waitForEvents(t *testing.T, r *fakeRecorder, want int) []event {
t.Helper()
deadline := time.Now().Add(500 * time.Millisecond)
for time.Now().Before(deadline) {
if got := r.snapshot(); len(got) >= want {
return got
}
time.Sleep(5 * time.Millisecond)
}
got := r.snapshot()
t.Fatalf("timed out waiting for %d events, got %d: %+v", want, len(got), got)
return nil
}
// newWatcher builds a watcher with the final timer disabled (finalLead=0),
// matching the lead-only behaviour the pre-final-warning tests assume.
func newWatcher(lead time.Duration, r *fakeRecorder) *Watcher {
return NewWithLeads(lead, 0, r)
}
func TestUpdateZeroBeforeAnythingIsNoop(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
defer w.Close()
_ = w.Update(time.Time{})
if got := r.snapshot(); len(got) != 0 {
t.Fatalf("expected no events on initial zero, got %+v", got)
}
}
func TestUpdateNonZeroFiresStateChange(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
defer w.Close()
d := time.Now().Add(time.Hour)
_ = w.Update(d)
events := waitForEvents(t, r, 1)
if events[0].kind != stateChange {
t.Fatalf("expected stateChange, got %+v", events[0])
}
if !w.Deadline().Equal(d) {
t.Fatalf("deadline mismatch: %v vs %v", w.Deadline(), d)
}
}
func TestSameDeadlineIsNoop(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
defer w.Close()
d := time.Now().Add(time.Hour)
_ = w.Update(d)
_ = w.Update(d)
_ = w.Update(d)
events := waitForEvents(t, r, 1)
if len(events) != 1 {
t.Fatalf("expected exactly 1 event for repeated same deadline, got %d: %+v", len(events), events)
}
}
func TestWarningFiresOnceWithinLeadWindow(t *testing.T) {
r := &fakeRecorder{}
lead := 50 * time.Millisecond
w := newWatcher(lead, r)
defer w.Close()
// Deadline 80ms out — warning should fire after ~30ms.
d := time.Now().Add(80 * time.Millisecond)
_ = w.Update(d)
events := waitForEvents(t, r, 2)
if events[0].kind != stateChange {
t.Fatalf("event[0] should be stateChange, got %+v", events[0])
}
if !events[1].isWarning() {
t.Fatalf("event[1] should be a warning publish, got %+v", events[1])
}
}
func TestWarningFiresImmediatelyWhenAlreadyInsideWindow(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(time.Hour, r) // lead > delta => fire immediately
defer w.Close()
d := time.Now().Add(10 * time.Millisecond)
_ = w.Update(d)
events := waitForEvents(t, r, 2)
if !events[1].isWarning() {
t.Fatalf("expected immediate warning publish, got %+v", events[1])
}
}
func TestNewDeadlineCancelsPriorTimer(t *testing.T) {
r := &fakeRecorder{}
lead := 50 * time.Millisecond
w := newWatcher(lead, r)
defer w.Close()
first := time.Now().Add(80 * time.Millisecond) // would fire warning ~30ms in
_ = w.Update(first)
// Replace with a far-future deadline before the warning fires.
time.Sleep(5 * time.Millisecond)
second := time.Now().Add(time.Hour)
_ = w.Update(second)
// Wait past when first's warning would have fired.
time.Sleep(80 * time.Millisecond)
if n := countWhere(r.snapshot(), event.isWarning); n != 0 {
t.Fatalf("warning fired for cancelled deadline: %+v", r.snapshot())
}
}
func TestRefreshAfterFireArmsNewWarning(t *testing.T) {
r := &fakeRecorder{}
lead := 30 * time.Millisecond
w := newWatcher(lead, r)
defer w.Close()
first := time.Now().Add(50 * time.Millisecond)
_ = w.Update(first)
// Wait for stateChange + warning of the first cycle.
waitForEvents(t, r, 2)
// Simulate a successful extend: brand new deadline.
second := time.Now().Add(60 * time.Millisecond)
_ = w.Update(second)
// 4 events total: stateChange, warning (first), stateChange, warning (second).
events := waitForEvents(t, r, 4)
if events[2].kind != stateChange {
t.Fatalf("event[2] should be stateChange for the new deadline, got %+v", events[2])
}
if !events[3].isWarning() {
t.Fatalf("event[3] should be a warning publish for the new deadline, got %+v", events[3])
}
}
func TestUpdateZeroAfterNonZeroClearsState(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(time.Hour, r)
defer w.Close()
d := time.Now().Add(2 * time.Hour)
_ = w.Update(d)
waitForEvents(t, r, 1)
_ = w.Update(time.Time{})
events := waitForEvents(t, r, 2)
if events[1].kind != stateChange {
t.Fatalf("expected stateChange on clear, got %+v", events[1])
}
if !w.Deadline().IsZero() {
t.Fatalf("Deadline should be zero after clear")
}
}
func TestUpdateRejectsBeforeEpoch(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
defer w.Close()
good := time.Now().Add(time.Hour)
if err := w.Update(good); err != nil {
t.Fatalf("seed Update: %v", err)
}
err := w.Update(time.Unix(-100, 0))
if !errors.Is(err, ErrDeadlineBeforeEpoch) {
t.Fatalf("want ErrDeadlineBeforeEpoch, got %v", err)
}
if !w.Deadline().IsZero() {
t.Fatalf("rejected pre-epoch update must clear deadline; got %v", w.Deadline())
}
}
func TestUpdateRejectsTooFarFuture(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
defer w.Close()
good := time.Now().Add(time.Hour)
if err := w.Update(good); err != nil {
t.Fatalf("seed Update: %v", err)
}
err := w.Update(time.Now().Add(50 * 365 * 24 * time.Hour))
if !errors.Is(err, ErrDeadlineTooFarFuture) {
t.Fatalf("want ErrDeadlineTooFarFuture, got %v", err)
}
if !w.Deadline().IsZero() {
t.Fatalf("rejected far-future update must clear deadline; got %v", w.Deadline())
}
}
func TestUpdateInPastClearsDeadline(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
defer w.Close()
good := time.Now().Add(time.Hour)
if err := w.Update(good); err != nil {
t.Fatalf("seed Update: %v", err)
}
// Drain the stateChange from the seed.
waitForEvents(t, r, 1)
err := w.Update(time.Now().Add(-1 * time.Hour))
if !errors.Is(err, ErrDeadlineInPast) {
t.Fatalf("want ErrDeadlineInPast, got %v", err)
}
if !w.Deadline().IsZero() {
t.Fatalf("in-past update must clear the deadline, got %v", w.Deadline())
}
events := waitForEvents(t, r, 2)
if events[1].kind != stateChange {
t.Fatalf("expected stateChange on clear, got %+v", events[1])
}
}
func TestUpdateWithinSkewAccepted(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
defer w.Close()
// 5 seconds in the past is within the 30s Skew tolerance — accept it.
d := time.Now().Add(-5 * time.Second)
if err := w.Update(d); err != nil {
t.Fatalf("within-skew Update should succeed, got %v", err)
}
if !w.Deadline().Equal(d) {
t.Fatalf("expected deadline to be applied, got %v want %v", w.Deadline(), d)
}
}
func TestCloseSilencesUpdates(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(50*time.Millisecond, r)
w.Close()
_ = w.Update(time.Now().Add(time.Hour))
time.Sleep(20 * time.Millisecond)
if got := r.snapshot(); len(got) != 0 {
t.Fatalf("expected no events after Close, got %+v", got)
}
}
// TestCloseClearsRecorderDeadline pins the profile-switch fix: a watcher
// holding a live deadline must zero the recorder on Close so the next
// engine's watcher (and the UI reading the shared server-scoped recorder)
// doesn't start out showing the previous session's stale "expires in".
func TestCloseClearsRecorderDeadline(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(time.Hour, r)
d := time.Now().Add(2 * time.Hour)
if err := w.Update(d); err != nil {
t.Fatalf("seed Update: %v", err)
}
if got := r.deadline(); !got.Equal(d) {
t.Fatalf("recorder deadline after Update = %v, want %v", got, d)
}
w.Close()
if got := r.deadline(); !got.IsZero() {
t.Fatalf("recorder deadline after Close = %v, want zero", got)
}
}
// TestCloseWithoutDeadlineLeavesRecorderUntouched guards the symmetric
// case: closing a watcher that never held a deadline must not emit a
// redundant clear (the recorder may legitimately hold a value written by
// some other path; the watcher only owns what it set).
func TestCloseWithoutDeadlineLeavesRecorderUntouched(t *testing.T) {
r := &fakeRecorder{}
w := newWatcher(time.Hour, r)
w.Close()
if got := r.snapshot(); len(got) != 0 {
t.Fatalf("expected no events from Close on an empty watcher, got %+v", got)
}
}
func TestFinalWarningFiresAfterRegularWarning(t *testing.T) {
r := &fakeRecorder{}
// Warning fires at deadline-80ms, final at deadline-30ms.
w := NewWithLeads(80*time.Millisecond, 30*time.Millisecond, r)
defer w.Close()
d := time.Now().Add(100 * time.Millisecond)
_ = w.Update(d)
// Expect stateChange + warning + final-warning.
events := waitForEvents(t, r, 3)
if countWhere(events, func(e event) bool { return e.kind == stateChange }) != 1 {
t.Fatalf("expected exactly 1 stateChange, got %+v", events)
}
if countWhere(events, event.isWarning) != 1 {
t.Fatalf("expected exactly 1 warning publish, got %+v", events)
}
if countWhere(events, event.isFinalWarning) != 1 {
t.Fatalf("expected exactly 1 final-warning publish, got %+v", events)
}
// Warning must precede final (same deadline, longer lead fires first).
var wIdx, fIdx int
for i, e := range events {
switch {
case e.isWarning():
wIdx = i
case e.isFinalWarning():
fIdx = i
}
}
if wIdx > fIdx {
t.Fatalf("warning must publish before final-warning, got order %+v", events)
}
}
func TestDismissSuppressesFinalWarning(t *testing.T) {
r := &fakeRecorder{}
w := NewWithLeads(80*time.Millisecond, 30*time.Millisecond, r)
defer w.Close()
d := time.Now().Add(100 * time.Millisecond)
_ = w.Update(d)
// Wait for the warning publish so we know we're inside the warning
// window, then dismiss before the final timer would fire.
deadline := time.Now().Add(500 * time.Millisecond)
for time.Now().Before(deadline) {
if countWhere(r.snapshot(), event.isWarning) >= 1 {
break
}
time.Sleep(2 * time.Millisecond)
}
if countWhere(r.snapshot(), event.isWarning) < 1 {
t.Fatalf("warning did not publish in time, events=%+v", r.snapshot())
}
w.Dismiss()
// Now wait past when the final would have fired.
time.Sleep(120 * time.Millisecond)
if n := countWhere(r.snapshot(), event.isFinalWarning); n != 0 {
t.Fatalf("final-warning published after Dismiss(), events=%+v", r.snapshot())
}
}
func TestDismissResetByNewDeadline(t *testing.T) {
r := &fakeRecorder{}
w := NewWithLeads(80*time.Millisecond, 30*time.Millisecond, r)
defer w.Close()
first := time.Now().Add(100 * time.Millisecond)
_ = w.Update(first)
// Dismiss against the first deadline.
w.Dismiss()
// Replace with a fresh deadline before the first's timers complete.
time.Sleep(10 * time.Millisecond)
second := time.Now().Add(100 * time.Millisecond)
_ = w.Update(second)
// The second cycle must publish a final-warning (the dismiss state
// did not carry over).
deadline := time.Now().Add(500 * time.Millisecond)
for time.Now().Before(deadline) {
if countWhere(r.snapshot(), event.isFinalWarning) >= 1 {
break
}
time.Sleep(5 * time.Millisecond)
}
if countWhere(r.snapshot(), event.isFinalWarning) < 1 {
t.Fatalf("final-warning did not publish on fresh deadline after Dismiss reset, events=%+v", r.snapshot())
}
}
func TestDismissBeforeUpdateIsNoop(t *testing.T) {
r := &fakeRecorder{}
w := NewWithLeads(80*time.Millisecond, 30*time.Millisecond, r)
defer w.Close()
// No deadline tracked yet; Dismiss must be a no-op (no panic, no state).
w.Dismiss()
d := time.Now().Add(100 * time.Millisecond)
_ = w.Update(d)
// Final warning should still publish — Dismiss only acts on the current
// deadline, and there was none at the time of the call.
deadline := time.Now().Add(500 * time.Millisecond)
for time.Now().Before(deadline) {
if countWhere(r.snapshot(), event.isFinalWarning) >= 1 {
return
}
time.Sleep(5 * time.Millisecond)
}
t.Fatalf("final-warning did not publish after no-op pre-Update Dismiss, events=%+v", r.snapshot())
}

View File

@@ -109,7 +109,7 @@ func (e *ConnMgr) UpdatedRemoteFeatureFlag(ctx context.Context, enabled bool) er
return nil
}
log.Warnf("lazy connection manager is enabled by management feature flag")
log.Infof("lazy connection manager is enabled by the management feature flag")
e.initLazyManager(ctx)
e.statusRecorder.UpdateLazyConnection(true)
return e.addPeersToLazyConnManager()

View File

@@ -277,6 +277,15 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
log.Debugf("connecting to the Management service %s", c.config.ManagementURL.Host)
mgmClient, err := mgm.NewClient(engineCtx, c.config.ManagementURL.Host, myPrivateKey, mgmTlsEnabled)
if err != nil {
// On daemon shutdown / Down() the parent context is cancelled
// and the dial fails with "context canceled". Wrapping that
// into state would leave the snapshot stuck at Connecting+err
// until the backoff loop wakes up — instead let the operation
// return cleanly so the deferred state.Set(StatusIdle) takes
// effect on the next iteration.
if c.ctx.Err() != nil {
return nil
}
return wrapErr(gstatus.Errorf(codes.FailedPrecondition, "failed connecting to Management Service : %s", err))
}
mgmNotifier := statusRecorderToMgmConnStateNotifier(c.statusRecorder)
@@ -415,6 +424,10 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
return wrapErr(err)
}
// Seed the session-expiry deadline from the LoginResponse. Subsequent
// changes flow in through SyncResponse and are applied in handleSync.
engine.ApplySessionDeadline(loginResp.GetSessionExpiresAt())
log.Infof("Netbird engine started, the IP is: %s", peerConfig.GetAddress())
state.Set(StatusConnected)
@@ -451,6 +464,10 @@ func (c *ConnectClient) run(mobileDependency MobileDependency, runningChan chan
}
c.statusRecorder.ClientStart()
// Wrap the backoff with c.ctx so Down()/actCancel propagates into the
// inter-attempt sleep — otherwise a 15s MaxInterval can keep the retry
// loop alive long after the caller asked to give up, leaving the
// status stream stuck at Connecting.
err = backoff.Retry(operation, backoff.WithContext(backOff, c.ctx))
if err != nil {
log.Debugf("exiting client retry loop due to unrecoverable error: %s", err)

View File

@@ -229,9 +229,16 @@ scutil_dns.txt (macOS only):
const (
clientLogFile = "client.log"
uiLogFile = "gui-client.log"
errorLogFile = "netbird.err"
stdoutLogFile = "netbird.out"
// Rotated-log glob prefixes (base log name without extension) passed to
// addRotatedLogFiles. The daemon's own log and the GUI log live in the same
// dir, so the prefixes must be disjoint to keep their rotated siblings apart.
clientLogPrefix = "client"
uiLogPrefix = "gui-client"
darwinErrorLogPath = "/var/log/netbird.out.log"
darwinStdoutLogPath = "/var/log/netbird.err.log"
)
@@ -249,6 +256,7 @@ type BundleGenerator struct {
statusRecorder *peer.Status
syncResponse *mgmProto.SyncResponse
logPath string
uiLogPath string
tempDir string
statePath string
cpuProfile []byte
@@ -276,6 +284,7 @@ type GeneratorDependencies struct {
StatusRecorder *peer.Status
SyncResponse *mgmProto.SyncResponse
LogPath string
UILogPath string // Absolute path to the desktop UI's gui-client.log, reported via RegisterUILog. Empty if no UI registered one.
TempDir string // Directory for temporary bundle zip files. If empty, os.TempDir() is used.
StatePath string // Path to the state file. If empty, the ServiceManager default path is used.
CPUProfile []byte
@@ -300,6 +309,7 @@ func NewBundleGenerator(deps GeneratorDependencies, cfg BundleConfig) *BundleGen
statusRecorder: deps.StatusRecorder,
syncResponse: deps.SyncResponse,
logPath: deps.LogPath,
uiLogPath: deps.UILogPath,
tempDir: deps.TempDir,
statePath: deps.StatePath,
cpuProfile: deps.CPUProfile,
@@ -411,6 +421,10 @@ func (g *BundleGenerator) createArchive() error {
log.Errorf("failed to add logs to debug bundle: %v", err)
}
if err := g.addUILog(); err != nil {
log.Errorf("failed to add UI log to debug bundle: %v", err)
}
if err := g.addUpdateLogs(); err != nil {
log.Errorf("failed to add updater logs: %v", err)
}
@@ -466,7 +480,6 @@ func (g *BundleGenerator) addStatus() error {
fullStatus := g.statusRecorder.GetFullStatus()
protoFullStatus := nbstatus.ToProtoFullStatus(fullStatus)
protoFullStatus.Events = g.statusRecorder.GetEventHistory()
overview := nbstatus.ConvertToStatusOutputOverview(protoFullStatus, nbstatus.ConvertOptions{
Anonymize: g.anonymize,
ProfileName: profName,
@@ -986,7 +999,7 @@ func (g *BundleGenerator) addLogfile() error {
return fmt.Errorf("add client log file to zip: %w", err)
}
g.addRotatedLogFiles(logDir)
g.addRotatedLogFiles(logDir, clientLogPrefix)
stdErrLogPath := filepath.Join(logDir, errorLogFile)
stdoutLogPath := filepath.Join(logDir, stdoutLogFile)
@@ -1006,6 +1019,25 @@ func (g *BundleGenerator) addLogfile() error {
return nil
}
// addUILog adds the desktop UI's gui-client.log (and its rotated siblings) to
// the bundle. The path is reported by the UI via RegisterUILog; empty when no
// UI registered one (e.g. headless / server). Missing file is non-fatal — the
// UI only writes it while the daemon is in debug, so it's often absent.
func (g *BundleGenerator) addUILog() error {
if g.uiLogPath == "" {
log.Debugf("no UI log path registered, skipping in debug bundle")
return nil
}
if err := g.addSingleLogfile(g.uiLogPath, uiLogFile); err != nil {
return fmt.Errorf("add UI log file to zip: %w", err)
}
g.addRotatedLogFiles(filepath.Dir(g.uiLogPath), uiLogPrefix)
return nil
}
// addSingleLogfile adds a single log file to the archive
func (g *BundleGenerator) addSingleLogfile(logPath, targetName string) error {
logFile, err := os.Open(logPath)
@@ -1078,14 +1110,16 @@ func (g *BundleGenerator) addSingleLogFileGz(logPath, targetName string) error {
return nil
}
// addRotatedLogFiles adds rotated log files to the bundle based on logFileCount
func (g *BundleGenerator) addRotatedLogFiles(logDir string) {
// addRotatedLogFiles adds rotated log files to the bundle based on logFileCount.
// prefix is the base log name without extension (e.g. "client", "gui-client");
// the glob matches both files rotated by us and by logrotate on linux.
func (g *BundleGenerator) addRotatedLogFiles(logDir, prefix string) {
if g.logFileCount == 0 {
return
}
// This regex will match both logs rotated by us and logrotate on linux
pattern := filepath.Join(logDir, "client*.log.*")
// This pattern matches both logs rotated by us and logrotate on linux
pattern := filepath.Join(logDir, prefix+"*.log.*")
files, err := filepath.Glob(pattern)
if err != nil {
log.Warnf("failed to glob rotated logs: %v", err)

View File

@@ -40,6 +40,25 @@ func TestAddRotatedLogFiles_PicksUpAllVariants(t *testing.T) {
require.NotContains(t, names, "other.log", "unrelated files should not be in bundle")
}
// TestAddRotatedLogFiles_GUIPrefix asserts the prefix parameter scopes the glob
// to the GUI log: gui-client.log.* rotated siblings are picked up and the
// daemon's own client.log.* are not (and vice versa, covered above). This is
// the load-bearing check for the gui-client.log bundle collection — the old
// "client*.log.*" glob would have missed gui-client rotations.
func TestAddRotatedLogFiles_GUIPrefix(t *testing.T) {
dir := t.TempDir()
writeFile(t, filepath.Join(dir, "gui-client.log.1"), "gui rotated\n")
writeGzFile(t, filepath.Join(dir, "gui-client.log.2.gz"), "gui rotated gz\n")
writeFile(t, filepath.Join(dir, "client.log.1"), "daemon rotated\n")
names := runAddRotatedLogFilesPrefix(t, dir, "gui-client", 10)
require.Contains(t, names, "gui-client.log.1", "gui-client rotated file should be in bundle")
require.Contains(t, names, "gui-client.log.2.gz", "gui-client gz rotated file should be in bundle")
require.NotContains(t, names, "client.log.1", "daemon rotated file must not match the gui-client prefix")
}
// TestAddRotatedLogFiles_RespectsLogFileCount asserts that only the newest
// logFileCount rotated files are bundled, ordered by mtime.
func TestAddRotatedLogFiles_RespectsLogFileCount(t *testing.T) {
@@ -67,6 +86,10 @@ func TestAddRotatedLogFiles_RespectsLogFileCount(t *testing.T) {
// runAddRotatedLogFiles calls addRotatedLogFiles against a fresh in-memory
// zip writer and returns the set of entry names that ended up in the archive.
func runAddRotatedLogFiles(t *testing.T, dir string, logFileCount uint32) map[string]struct{} {
return runAddRotatedLogFilesPrefix(t, dir, "client", logFileCount)
}
func runAddRotatedLogFilesPrefix(t *testing.T, dir, prefix string, logFileCount uint32) map[string]struct{} {
t.Helper()
var buf bytes.Buffer
@@ -74,7 +97,7 @@ func runAddRotatedLogFiles(t *testing.T, dir string, logFileCount uint32) map[st
archive: zip.NewWriter(&buf),
logFileCount: logFileCount,
}
g.addRotatedLogFiles(dir)
g.addRotatedLogFiles(dir, prefix)
require.NoError(t, g.archive.Close())
zr, err := zip.NewReader(bytes.NewReader(buf.Bytes()), int64(buf.Len()))

View File

@@ -292,18 +292,16 @@ func (s *serviceViaListener) generateFreePort() (uint16, error) {
return customPort, nil
}
udpAddr := net.UDPAddrFromAddrPort(netip.MustParseAddrPort("0.0.0.0:0"))
probeListener, err := net.ListenUDP("udp", udpAddr)
probeListener, err := net.ListenUDP("udp4", &net.UDPAddr{})
if err != nil {
log.Debugf("failed to bind random port for DNS: %s", err)
return 0, err
}
addrPort := netip.MustParseAddrPort(probeListener.LocalAddr().String()) // might panic if address is incorrect
err = probeListener.Close()
if err != nil {
port := uint16(probeListener.LocalAddr().(*net.UDPAddr).Port)
if err = probeListener.Close(); err != nil {
log.Debugf("failed to free up DNS port: %s", err)
return 0, err
}
return addrPort.Port(), nil
return port, nil
}

View File

@@ -274,6 +274,20 @@ type Engine struct {
jobExecutorWG sync.WaitGroup
exposeManager *expose.Manager
sessionWatcher sessionDeadlineWatcher
}
// sessionDeadlineWatcher is the engine-facing surface of the SSO session
// expiry watcher. The concrete implementation (sessionwatch.Watcher) is wired
// in via newSessionWatcher, which is build-tagged so the js/wasm build links a
// no-op stub instead of pulling the full sessionwatch package (and its timer
// machinery) into the binary — the wasm client never runs the engine's
// session-warning flow.
type sessionDeadlineWatcher interface {
Update(deadline time.Time) error
Dismiss()
Close()
}
// Peer is an instance of the Connection Peer
@@ -325,6 +339,17 @@ func NewEngine(
updateManager: services.UpdateManager,
syncStoreDir: config.StateDir,
}
// sessionWatcher keeps the SubscribeStatus consumers in sync with the
// session expiry deadline. Deadline-change ticks come for free via
// Status.SetSessionExpiresAt; the watcher exists to push a wake-up at
// T-WarningLead and T-FinalWarningLead so the UI repaints the remaining
// time / warning state even when nothing else changed, and to publish
// two SystemEvents (the warning composition lives in sessionwatch so
// the wire format stays owned by one package):
// - T-WarningLead → interactive "Extend now / Dismiss" notification
// - T-FinalWarningLead → auto-opened SessionAboutToExpire dialog,
// suppressed when the user dismissed the earlier warning
engine.sessionWatcher = newSessionWatcher(engine.statusRecorder)
log.Infof("I am: %s", config.WgPrivateKey.PublicKey().String())
return engine
@@ -391,6 +416,10 @@ func (e *Engine) stopLocked() {
e.srWatcher.Close()
}
if e.sessionWatcher != nil {
e.sessionWatcher.Close()
}
if e.updateManager != nil {
e.updateManager.SetDownloadOnly()
}
@@ -932,6 +961,8 @@ func (e *Engine) handleSync(update *mgmProto.SyncResponse) error {
return e.ctx.Err()
}
e.ApplySessionDeadline(update.GetSessionExpiresAt())
if update.NetworkMap != nil && update.NetworkMap.PeerConfig != nil {
e.handleAutoUpdateVersion(update.NetworkMap.PeerConfig.AutoUpdate)
}
@@ -1267,7 +1298,7 @@ func (e *Engine) handleBundle(params *mgmProto.BundleParameters) (*mgmProto.JobR
ClientMetrics: e.clientMetrics,
DaemonVersion: version.NetbirdVersion(),
RefreshStatus: func() {
e.RunHealthProbes(true)
e.RunHealthProbes(e.ctx, true)
},
}
@@ -2193,7 +2224,20 @@ func (e *Engine) getRosenpassAddr() string {
// RunHealthProbes executes health checks for Signal, Management, Relay, and WireGuard services
// and updates the status recorder with the latest states.
func (e *Engine) RunHealthProbes(waitForResult bool) bool {
//
// ctx scopes the (potentially slow) STUN/TURN probing: a caller that gives up —
// e.g. a Status RPC whose client disconnected — cancels its ctx and the probe
// returns instead of running to its per-component timeout. The engine's own
// lifetime ctx still applies independently, so an engine shutdown aborts the
// probe even if the caller's ctx is context.Background().
func (e *Engine) RunHealthProbes(ctx context.Context, waitForResult bool) bool {
// Tie the caller's ctx to the engine lifetime: either cancelling aborts
// the probe below.
ctx, cancel := context.WithCancel(ctx)
defer cancel()
stop := context.AfterFunc(e.ctx, cancel)
defer stop()
e.syncMsgMux.Lock()
signalHealthy := e.signal.IsHealthy()
@@ -2216,9 +2260,9 @@ func (e *Engine) RunHealthProbes(waitForResult bool) bool {
if runtime.GOOS != "js" {
var results []relay.ProbeResult
if waitForResult {
results = e.probeStunTurn.ProbeAllWaitResult(e.ctx, stuns, turns)
results = e.probeStunTurn.ProbeAllWaitResult(ctx, stuns, turns)
} else {
results = e.probeStunTurn.ProbeAll(e.ctx, stuns, turns)
results = e.probeStunTurn.ProbeAll(ctx, stuns, turns)
}
e.statusRecorder.UpdateRelayStates(results)
@@ -2561,13 +2605,14 @@ func (e *Engine) updateForwardRules(rules []*mgmProto.ForwardingRule) ([]firewal
func (e *Engine) toExcludedLazyPeers(rules []firewallManager.ForwardRule, peers []*mgmProto.RemotePeerConfig) map[string]bool {
excludedPeers := make(map[string]bool)
// Ingress forward targets: inbound forwarded traffic is initiated remotely and
// cannot wake a lazy connection, so the peer routing the target must stay
// permanently connected. AllowedIPs are already parsed on the peer conn, so
// reuse those typed prefixes instead of re-parsing the network map strings.
for _, r := range rules {
ip := r.TranslatedAddress
for _, p := range peers {
for _, allowedIP := range p.GetAllowedIps() {
if allowedIP != ip.String() {
continue
}
if e.peerRoutesAddr(p, r.TranslatedAddress) {
log.Infof("exclude forwarder peer from lazy connection: %s", p.GetWgPubKey())
excludedPeers[p.GetWgPubKey()] = true
}
@@ -2577,6 +2622,27 @@ func (e *Engine) toExcludedLazyPeers(rules []firewallManager.ForwardRule, peers
return excludedPeers
}
// peerRoutesAddr reports whether the peer is a router for addr, matched against
// the peer's already-parsed AllowedIPs from the store (the same typed value the
// lazy manager consumes) rather than re-parsing the network map strings.
func (e *Engine) peerRoutesAddr(p *mgmProto.RemotePeerConfig, addr netip.Addr) bool {
prefixes, ok := e.peerStore.AllowedIPs(p.GetWgPubKey())
if !ok {
return false
}
return prefixesContain(prefixes, addr)
}
// prefixesContain reports whether addr falls within any of the prefixes.
func prefixesContain(prefixes []netip.Prefix, addr netip.Addr) bool {
for _, prefix := range prefixes {
if prefix.Contains(addr) {
return true
}
}
return false
}
// isChecksEqual checks if two slices of checks are equal.
func isChecksEqual(checks1, checks2 []*mgmProto.Checks) bool {
normalize := func(checks []*mgmProto.Checks) []string {

View File

@@ -0,0 +1,108 @@
package internal
import (
"context"
"errors"
"fmt"
"time"
log "github.com/sirupsen/logrus"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/netbirdio/netbird/client/internal/auth/sessionwatch"
cProto "github.com/netbirdio/netbird/client/proto"
"github.com/netbirdio/netbird/client/system"
)
// ApplySessionDeadline propagates the absolute SSO session deadline carried on
// LoginResponse / SyncResponse to both the watcher (for the edge-triggered
// warning) and the status recorder (for the SubscribeStatus / Status RPC
// snapshot the UI consumes).
//
// The wire field is 3-state:
// - nil → snapshot carries no info; keep the
// previously-anchored deadline (no-op)
// - explicit zero (s=0, n=0) → peer is not SSO-registered or expiry is
// disabled; clear both sinks
// - valid timestamp → new deadline; arm watcher, expose on
// status recorder
//
// Deadline sanity-checks live in sessionwatch.Watcher.Update. Any rejected
// value is treated as a clear on both sinks: the alternative — leaving the
// previously-known deadline in place — risks the UI confidently displaying
// a stale "expires in X" while the server has actually invalidated it.
func (e *Engine) ApplySessionDeadline(ts *timestamppb.Timestamp) {
if ts == nil {
return
}
var deadline time.Time
// Explicit zero (seconds=0 AND nanos=0) is the sentinel for "disabled".
// Everything else flows through Watcher.Update, whose sanity-checks
// reject out-of-range / pre-epoch / far-future / too-stale values and
// clear on rejection.
if ts.GetSeconds() != 0 || ts.GetNanos() != 0 {
deadline = ts.AsTime().UTC()
}
if e.sessionWatcher == nil {
return
}
// Watcher.Update owns the propagation to the status recorder (the
// SubscribeStatus / Status snapshot the UI reads): a set writes the
// deadline, a clear or a sanity-check rejection writes the zero value.
// Keeping a single writer is what stops the recorder from drifting out
// of sync with the warning timers.
if err := e.sessionWatcher.Update(deadline); err != nil {
log.Errorf("auth session deadline rejected: %v, clearing", err)
e.statusRecorder.PublishEvent(
cProto.SystemEvent_ERROR,
cProto.SystemEvent_AUTHENTICATION,
"session deadline rejected",
"",
map[string]string{sessionwatch.MetaSessionDeadlineRejected: err.Error()},
)
}
}
// DismissSessionWarning records the user's "Dismiss" click on the
// T-WarningLead interactive notification and suppresses the upcoming
// T-FinalWarningLead fallback for the current deadline. No-op when the
// watcher is not running or holds no deadline.
func (e *Engine) DismissSessionWarning() {
if e.sessionWatcher == nil {
return
}
e.sessionWatcher.Dismiss()
}
// ExtendAuthSession asks the management server to refresh the SSO session
// expiry deadline using the supplied JWT, then mirrors the new deadline into
// the daemon's state. The tunnel is untouched; no resync, no reconnect.
//
// Returns the new absolute UTC deadline (or zero time when the server
// reports the peer is not eligible for extension).
func (e *Engine) ExtendAuthSession(ctx context.Context, jwtToken string) (time.Time, error) {
if jwtToken == "" {
return time.Time{}, errors.New("jwt token is required")
}
if e.mgmClient == nil {
return time.Time{}, errors.New("management client is not initialised")
}
info, err := system.GetInfoWithChecks(ctx, e.checks)
if err != nil {
log.Warnf("failed to collect system info for session extend: %v", err)
info = system.GetInfo(ctx)
}
resp, err := e.mgmClient.ExtendAuthSession(info, jwtToken)
if err != nil {
return time.Time{}, fmt.Errorf("extend auth session on management: %w", err)
}
e.ApplySessionDeadline(resp.GetSessionExpiresAt())
if resp.GetSessionExpiresAt().IsValid() {
return resp.GetSessionExpiresAt().AsTime().UTC(), nil
}
return time.Time{}, nil
}

View File

@@ -0,0 +1,87 @@
package internal
import (
"net/netip"
"testing"
"github.com/stretchr/testify/require"
firewallManager "github.com/netbirdio/netbird/client/firewall/manager"
"github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/client/internal/peerstore"
mgmProto "github.com/netbirdio/netbird/shared/management/proto"
)
func TestPrefixesContain(t *testing.T) {
tests := []struct {
name string
prefixes []string
addr string
want bool
}{
{name: "own overlay /32 matches", prefixes: []string{"100.110.8.145/32"}, addr: "100.110.8.145", want: true},
{name: "addr inside routed subnet", prefixes: []string{"10.121.0.0/16"}, addr: "10.121.208.4", want: true},
{name: "addr outside subnet", prefixes: []string{"10.121.0.0/16"}, addr: "10.122.0.1", want: false},
{name: "different /32", prefixes: []string{"100.110.8.145/32"}, addr: "100.110.8.146", want: false},
{name: "ipv6 /128 matches", prefixes: []string{"fd00::1/128"}, addr: "fd00::1", want: true},
{name: "no prefixes", prefixes: nil, addr: "10.121.208.4", want: false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
prefixes := make([]netip.Prefix, 0, len(tt.prefixes))
for _, p := range tt.prefixes {
prefixes = append(prefixes, netip.MustParsePrefix(p))
}
require.Equal(t, tt.want, prefixesContain(prefixes, netip.MustParseAddr(tt.addr)))
})
}
}
// TestToExcludedLazyPeers_ForwardTarget guards a regression: the forward-target
// peer (the peer routing a ForwardRule.TranslatedAddress) must be excluded from
// lazy connections, matched via the peer's already-parsed AllowedIPs.
func TestToExcludedLazyPeers_ForwardTarget(t *testing.T) {
const targetPeerKey = "cccccccccccccccccccccccccccccccccccccccccc0="
const otherPeerKey = "dddddddddddddddddddddddddddddddddddddddddd0="
store := peerstore.NewConnStore()
store.AddPeerConn(targetPeerKey, newTestConn(t, targetPeerKey, "100.110.8.145/32"))
store.AddPeerConn(otherPeerKey, newTestConn(t, otherPeerKey, "100.110.9.10/32"))
e := &Engine{peerStore: store}
peers := []*mgmProto.RemotePeerConfig{
{WgPubKey: targetPeerKey, AllowedIps: []string{"100.110.8.145/32"}},
{WgPubKey: otherPeerKey, AllowedIps: []string{"100.110.9.10/32"}},
}
rules := []firewallManager.ForwardRule{
{TranslatedAddress: netip.MustParseAddr("100.110.8.145")},
}
excluded := e.toExcludedLazyPeers(rules, peers)
require.True(t, excluded[targetPeerKey], "forward-target peer must be excluded from lazy connections")
require.False(t, excluded[otherPeerKey], "non-target peer must not be excluded")
require.Len(t, excluded, 1)
}
func TestToExcludedLazyPeers_NoRules(t *testing.T) {
e := &Engine{peerStore: peerstore.NewConnStore()}
peers := []*mgmProto.RemotePeerConfig{
{WgPubKey: "peer-a", AllowedIps: []string{"100.110.8.145/32"}},
}
require.Empty(t, e.toExcludedLazyPeers(nil, peers))
}
func newTestConn(t *testing.T, key, allowedIP string) *peer.Conn {
t.Helper()
conn, err := peer.NewConn(peer.ConnConfig{
Key: key,
WgConfig: peer.WgConfig{AllowedIps: []netip.Prefix{netip.MustParsePrefix(allowedIP)}},
}, peer.ServiceDependencies{})
require.NoError(t, err)
return conn
}

View File

@@ -0,0 +1,78 @@
package internal
import (
"testing"
"time"
"github.com/stretchr/testify/require"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/netbirdio/netbird/client/internal/auth/sessionwatch"
"github.com/netbirdio/netbird/client/internal/peer"
)
// TestApplySessionDeadline_ThreeState pins down the 3-state semantics of the
// wire field carried on LoginResponse / SyncResponse:
//
// - nil pointer → no info; previously-anchored deadline survives
// - explicit zero value → "expiry disabled" sentinel; both sinks cleared
// - valid future timestamp → new deadline propagated to both sinks
func TestApplySessionDeadline_ThreeState(t *testing.T) {
newEngine := func() *Engine {
recorder := peer.NewRecorder("")
return &Engine{
statusRecorder: recorder,
sessionWatcher: sessionwatch.New(recorder),
}
}
t.Run("valid timestamp sets deadline on both sinks", func(t *testing.T) {
e := newEngine()
deadline := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
e.ApplySessionDeadline(timestamppb.New(deadline))
require.True(t, e.statusRecorder.GetSessionExpiresAt().Equal(deadline),
"status recorder should hold the new deadline")
})
t.Run("nil is a no-op and preserves previous deadline", func(t *testing.T) {
e := newEngine()
seeded := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
e.ApplySessionDeadline(timestamppb.New(seeded))
require.True(t, e.statusRecorder.GetSessionExpiresAt().Equal(seeded))
e.ApplySessionDeadline(nil)
require.True(t, e.statusRecorder.GetSessionExpiresAt().Equal(seeded),
"nil snapshot must not disturb the existing deadline")
})
t.Run("explicit zero clears a previously-anchored deadline", func(t *testing.T) {
e := newEngine()
seeded := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
e.ApplySessionDeadline(timestamppb.New(seeded))
require.True(t, e.statusRecorder.GetSessionExpiresAt().Equal(seeded))
// Explicit zero Timestamp{} (seconds=0, nanos=0) is the
// "expiry disabled / not SSO" sentinel.
e.ApplySessionDeadline(&timestamppb.Timestamp{})
require.True(t, e.statusRecorder.GetSessionExpiresAt().IsZero(),
"explicit zero sentinel must clear the deadline")
})
t.Run("invalid timestamp clears the deadline", func(t *testing.T) {
e := newEngine()
seeded := time.Now().Add(time.Hour).UTC().Truncate(time.Second)
e.ApplySessionDeadline(timestamppb.New(seeded))
require.True(t, e.statusRecorder.GetSessionExpiresAt().Equal(seeded))
// Out-of-range nanos → IsValid()==false; same-meaning as the
// disabled sentinel for downstream sinks.
e.ApplySessionDeadline(&timestamppb.Timestamp{Seconds: 1, Nanos: -1})
require.True(t, e.statusRecorder.GetSessionExpiresAt().IsZero(),
"invalid timestamp must clear the deadline")
})
}

View File

@@ -0,0 +1,16 @@
//go:build !js
package internal
import (
"github.com/netbirdio/netbird/client/internal/auth/sessionwatch"
"github.com/netbirdio/netbird/client/internal/peer"
)
// newSessionWatcher returns the real SSO session expiry watcher for every
// non-wasm build. The js/wasm build gets a no-op stub from
// engine_sessionwatch_js.go so the sessionwatch package (and its timer
// machinery) never links into the wasm binary.
func newSessionWatcher(recorder *peer.Status) sessionDeadlineWatcher {
return sessionwatch.New(recorder)
}

View File

@@ -0,0 +1,44 @@
//go:build js
package internal
import (
"time"
"github.com/netbirdio/netbird/client/internal/peer"
)
// noopSessionWatcher is the js/wasm stand-in for sessionwatch.Watcher. The
// wasm client never runs the engine's session-warning flow (the interactive
// T-WarningLead notification and the T-FinalWarningLead fallback dialog live
// in the desktop UI), so linking the full sessionwatch package (timers, event
// composition) would only bloat the binary.
//
// It still mirrors the deadline into the status recorder so the SubscribeStatus
// / Status snapshot the UI consumes stays correct — only the timer-driven
// warnings are dropped.
type noopSessionWatcher struct {
recorder *peer.Status
}
func newSessionWatcher(recorder *peer.Status) sessionDeadlineWatcher {
return noopSessionWatcher{recorder: recorder}
}
// Update mirrors the real watcher's recorder propagation without the timers or
// sanity-check sentinels: a valid deadline is exposed on the status snapshot,
// the zero time clears it.
func (w noopSessionWatcher) Update(deadline time.Time) error {
if w.recorder != nil {
w.recorder.SetSessionExpiresAt(deadline)
}
return nil
}
func (noopSessionWatcher) Dismiss() {
// No-op: only suppresses the timer-driven final-warning, which this stub never arms.
}
func (noopSessionWatcher) Close() {
// No-op: no timers to stop and no state to unwind; the recorder is cleared via Update(zero).
}

View File

@@ -27,7 +27,7 @@ type Logger struct {
wgIfaceNetV6 netip.Prefix
dnsCollection atomic.Bool
exitNodeCollection atomic.Bool
Store types.Store
Store types.AggregatingStore
}
func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix) *Logger {
@@ -35,7 +35,7 @@ func New(statusRecorder *peer.Status, wgIfaceIPNet, wgIfaceIPNetV6 netip.Prefix)
statusRecorder: statusRecorder,
wgIfaceNet: wgIfaceIPNet,
wgIfaceNetV6: wgIfaceIPNetV6,
Store: store.NewMemoryStore(),
Store: store.NewAggregatingMemoryStore(),
}
}
@@ -125,6 +125,10 @@ func (l *Logger) stop() {
l.mux.Unlock()
}
func (l *Logger) ResetAggregationWindow() types.FlowEventAggregator {
return l.Store.ResetAggregationWindow()
}
func (l *Logger) GetEvents() []*types.Event {
return l.Store.GetEvents()
}

View File

@@ -9,12 +9,14 @@ import (
"sync"
"time"
"github.com/cenkalti/backoff/v4"
"github.com/google/uuid"
log "github.com/sirupsen/logrus"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/netbirdio/netbird/client/internal/netflow/conntrack"
"github.com/netbirdio/netbird/client/internal/netflow/logger"
"github.com/netbirdio/netbird/client/internal/netflow/store"
nftypes "github.com/netbirdio/netbird/client/internal/netflow/types"
"github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/flow/client"
@@ -23,14 +25,16 @@ import (
// Manager handles netflow tracking and logging
type Manager struct {
mux sync.Mutex
shutdownWg sync.WaitGroup
logger nftypes.FlowLogger
flowConfig *nftypes.FlowConfig
conntrack nftypes.ConnTracker
receiverClient *client.GRPCClient
publicKey []byte
cancel context.CancelFunc
mux sync.Mutex
shutdownWg sync.WaitGroup
logger nftypes.FlowLogger
flowConfig *nftypes.FlowConfig
conntrack nftypes.ConnTracker
receiverClient *client.GRPCClient
eventsWithoutAcks nftypes.Store
publicKey []byte
cancel context.CancelFunc
retryInterval time.Duration
}
// NewManager creates a new netflow manager
@@ -48,9 +52,11 @@ func NewManager(iface nftypes.IFaceMapper, publicKey []byte, statusRecorder *pee
}
return &Manager{
logger: flowLogger,
conntrack: ct,
publicKey: publicKey,
logger: flowLogger,
conntrack: ct,
publicKey: publicKey,
retryInterval: time.Second,
eventsWithoutAcks: store.NewMemoryStore(),
}
}
@@ -66,6 +72,7 @@ func (m *Manager) needsNewClient(previous *nftypes.FlowConfig) bool {
}
// enableFlow starts components for flow tracking
// must be called under m.mux lock
func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
// first make sender ready so events don't pile up
if m.needsNewClient(previous) {
@@ -85,6 +92,7 @@ func (m *Manager) enableFlow(previous *nftypes.FlowConfig) error {
return nil
}
// must be called under m.mux lock
func (m *Manager) resetClient() error {
if m.receiverClient != nil {
if err := m.receiverClient.Close(); err != nil {
@@ -107,14 +115,19 @@ func (m *Manager) resetClient() error {
ctx, cancel := context.WithCancel(context.Background())
m.cancel = cancel
m.shutdownWg.Add(2)
m.shutdownWg.Add(3)
flowConfigInterval := m.flowConfig.Interval
go func() {
defer m.shutdownWg.Done()
m.receiveACKs(ctx, flowClient)
m.receiveACKs(ctx, flowClient, flowConfigInterval)
}()
go func() {
defer m.shutdownWg.Done()
m.startSender(ctx)
m.startSender(ctx, flowConfigInterval)
}()
go func() {
defer m.shutdownWg.Done()
m.startRetries(ctx, flowConfigInterval)
}()
return nil
@@ -198,8 +211,8 @@ func (m *Manager) GetLogger() nftypes.FlowLogger {
return m.logger
}
func (m *Manager) startSender(ctx context.Context) {
ticker := time.NewTicker(m.flowConfig.Interval)
func (m *Manager) startSender(ctx context.Context, flowConfigInterval time.Duration) {
ticker := time.NewTicker(flowConfigInterval)
defer ticker.Stop()
for {
@@ -207,27 +220,29 @@ func (m *Manager) startSender(ctx context.Context) {
case <-ctx.Done():
return
case <-ticker.C:
events := m.logger.GetEvents()
collectedEvents := m.logger.ResetAggregationWindow()
events := collectedEvents.GetAggregatedEvents()
for _, event := range events {
m.eventsWithoutAcks.StoreEvent(event)
if err := m.send(event); err != nil {
log.Errorf("failed to send flow event to server: %v", err)
continue
} else {
log.Tracef("sent flow event: %s", event.ID)
}
log.Tracef("sent flow event: %s", event.ID)
}
}
}
}
func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
err := client.Receive(ctx, m.flowConfig.Interval, func(ack *proto.FlowEventAck) error {
func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient, flowConfigInterval time.Duration) {
err := client.Receive(ctx, flowConfigInterval, func(ack *proto.FlowEventAck) error {
id, err := uuid.FromBytes(ack.EventId)
if err != nil {
log.Warnf("failed to convert ack event id to uuid: %v", err)
return nil
}
log.Tracef("received flow event ack: %s", id)
m.logger.DeleteEvents([]uuid.UUID{id})
m.eventsWithoutAcks.DeleteEvents([]uuid.UUID{id})
return nil
})
@@ -236,6 +251,51 @@ func (m *Manager) receiveACKs(ctx context.Context, client *client.GRPCClient) {
}
}
// We effectively never drop events (see MaxInterval), which makes eventsWithoutAcks unbounded.
// We may want to limit the max size of the store, and start dropping oldest events when the threshold is reached.
func (m *Manager) startRetries(ctx context.Context, flowConfigInterval time.Duration) {
timer := time.NewTimer(m.retryInterval)
retryBackoff := backoff.WithContext(&backoff.ExponentialBackOff{
InitialInterval: 1 * time.Second,
RandomizationFactor: 0.5,
Multiplier: 1.7,
MaxInterval: flowConfigInterval / 2,
MaxElapsedTime: 3 * 30 * 24 * time.Hour, // 3 months
Stop: backoff.Stop,
Clock: backoff.SystemClock,
}, ctx)
defer timer.Stop()
for {
select {
case <-ctx.Done():
return
case <-timer.C:
resetBackoff := true
for _, e := range m.eventsWithoutAcks.GetEvents() {
if e.Timestamp.Add(time.Second).After(time.Now()) {
// grace period on retries to avoid early retries
// do not retry if the event is less than 1 sec old
continue
}
if err := m.send(e); err != nil {
if nextBackoff := retryBackoff.NextBackOff(); nextBackoff != backoff.Stop {
timer = time.NewTimer(nextBackoff)
resetBackoff = false
} else {
resetBackoff = true // we exhausted retries, reset retry loop
}
break
}
}
if resetBackoff { // use regular retry interval in absence of network errors
retryBackoff.Reset()
timer = time.NewTimer(m.retryInterval)
}
}
}
}
func (m *Manager) send(event *nftypes.Event) error {
m.mux.Lock()
client := m.receiverClient
@@ -250,9 +310,11 @@ func (m *Manager) send(event *nftypes.Event) error {
func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
protoEvent := &proto.FlowEvent{
EventId: event.ID[:],
Timestamp: timestamppb.New(event.Timestamp),
PublicKey: publicKey,
EventId: event.ID[:],
Timestamp: timestamppb.New(event.Timestamp),
PublicKey: publicKey,
WindowStart: timestamppb.New(event.WindowStart),
WindowEnd: timestamppb.New(event.WindowEnd),
FlowFields: &proto.FlowFields{
FlowId: event.FlowID[:],
RuleId: event.RuleID,
@@ -267,6 +329,9 @@ func toProtoEvent(publicKey []byte, event *nftypes.Event) *proto.FlowEvent {
TxBytes: event.TxBytes,
SourceResourceId: event.SourceResourceID,
DestResourceId: event.DestResourceID,
NumOfStarts: event.NumOfStarts,
NumOfEnds: event.NumOfEnds,
NumOfDrops: event.NumOfDrops,
},
}

View File

@@ -0,0 +1,291 @@
package netflow
import (
"context"
"errors"
"fmt"
"net"
"net/netip"
"slices"
"testing"
"time"
"github.com/google/uuid"
"github.com/netbirdio/netbird/client/iface/wgaddr"
"github.com/netbirdio/netbird/client/internal/netflow/types"
"github.com/netbirdio/netbird/flow/proto"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
)
type testServer struct {
proto.UnimplementedFlowServiceServer
events chan *proto.FlowEvent
acks chan *proto.FlowEventAck
grpcSrv *grpc.Server
addr string
handlerDone chan struct{} // signaled each time Events() exits
handlerStarted chan struct{} // signaled each time Events() begins
}
func newTestServer(t *testing.T) *testServer {
listener, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
s := &testServer{
events: make(chan *proto.FlowEvent, 100),
acks: make(chan *proto.FlowEventAck, 100),
grpcSrv: grpc.NewServer(),
addr: listener.Addr().String(),
handlerDone: make(chan struct{}, 10),
handlerStarted: make(chan struct{}, 10),
}
proto.RegisterFlowServiceServer(s.grpcSrv, s)
go func() {
if err := s.grpcSrv.Serve(listener); err != nil && !errors.Is(err, grpc.ErrServerStopped) {
t.Logf("server error: %v", err)
}
}()
t.Cleanup(func() {
s.grpcSrv.Stop()
})
return s
}
func (s *testServer) Events(stream proto.FlowService_EventsServer) error {
defer func() {
select {
case s.handlerDone <- struct{}{}:
default:
}
}()
err := stream.Send(&proto.FlowEventAck{IsInitiator: true})
if err != nil {
return err
}
select {
case s.handlerStarted <- struct{}{}:
default:
}
ctx, cancel := context.WithCancel(stream.Context())
defer cancel()
go func() {
defer cancel()
for {
event, err := stream.Recv()
if err != nil {
return
}
if !event.IsInitiator {
select {
case s.events <- event:
case <-ctx.Done():
return
}
}
}
}()
for {
select {
case ack := <-s.acks:
if err := stream.Send(ack); err != nil {
return err
}
case <-ctx.Done():
return ctx.Err()
}
}
}
func TestSendEventReceiveAck(t *testing.T) {
_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
t.Cleanup(cancel)
server := newTestServer(t)
manager := createManager(t, server.addr, 60*time.Second) // set high to prevent retries in this test
defer manager.Close()
assert.Eventually(t, func() bool {
select {
case <-server.handlerStarted:
return true
default:
return false
}
}, 3*time.Second, 100*time.Millisecond)
event1 := types.EventFields{
FlowID: uuid.New(),
Type: types.TypeStart,
Direction: types.Ingress,
DestIP: ipAddr("172.16.1.2"),
DestPort: 2345,
Protocol: 6,
}
manager.logger.StoreEvent(event1)
event2 := types.EventFields{
FlowID: uuid.New(),
Type: types.TypeStart,
Direction: types.Ingress,
DestIP: ipAddr("172.16.1.1"),
DestPort: 1234,
Protocol: 6,
}
manager.logger.StoreEvent(event2)
// verify the server received logged events
serverSideEvents := make([]*proto.FlowEvent, 0)
assert.Eventually(t, func() bool {
select {
case event := <-server.events:
serverSideEvents = append(serverSideEvents, event)
if len(serverSideEvents) == 2 {
return true
}
default:
if len(serverSideEvents) == 2 {
return true
}
}
return false
}, 5*time.Second, 100*time.Millisecond)
serverSideFlowIds := make([]uuid.UUID, 0, 2)
slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
id, err := uuid.FromBytes(e.FlowFields.FlowId)
assert.NoError(t, err)
serverSideFlowIds = append(serverSideFlowIds, id)
return true
})
assert.ElementsMatch(t, []uuid.UUID{event1.FlowID, event2.FlowID}, serverSideFlowIds)
// verify the manager tracks un-acked events
unackedEvents := manager.eventsWithoutAcks.GetEvents()
assert.Len(t, unackedEvents, 2)
flowIds := make([]uuid.UUID, 0)
slices.Values(unackedEvents)(func(e *types.Event) bool {
flowIds = append(flowIds, e.FlowID)
return true
})
assert.ElementsMatch(t, flowIds, []uuid.UUID{event1.FlowID, event2.FlowID})
}
// verify handling of retries:
// - unacked events are retried
// - when acks arrive, events are removed from the un-acked event tracker
func TestRetryEvents(t *testing.T) {
_, cancel := context.WithTimeout(context.Background(), 10*time.Second)
t.Cleanup(cancel)
server := newTestServer(t)
manager := createManager(t, server.addr, time.Second) // set low to start retries sooner
defer manager.Close()
assert.Eventually(t, func() bool {
select {
case <-server.handlerStarted:
return true
default:
return false
}
}, 3*time.Second, 100*time.Millisecond)
event1 := types.EventFields{
FlowID: uuid.New(),
Type: types.TypeStart,
Direction: types.Ingress,
DestIP: ipAddr("172.16.1.2"),
DestPort: 2345,
Protocol: 6,
}
manager.logger.StoreEvent(event1)
event2 := types.EventFields{
FlowID: uuid.New(),
Type: types.TypeStart,
Direction: types.Ingress,
DestIP: ipAddr("172.16.1.1"),
DestPort: 1234,
Protocol: 6,
}
manager.logger.StoreEvent(event2)
// verify the server received retries of logged events
serverSideEvents := make([]*proto.FlowEvent, 0)
func() {
c := time.After(2500 * time.Millisecond)
for {
select {
case event := <-server.events:
serverSideEvents = append(serverSideEvents, event)
case <-c:
return
}
}
}()
assert.True(t, len(serverSideEvents) > 2) // must see retries
uniqueServerSideEvents := make(map[uuid.UUID]*proto.FlowEvent)
slices.Values(serverSideEvents)(func(e *proto.FlowEvent) bool {
id, err := uuid.FromBytes(e.FlowFields.FlowId)
assert.NoError(t, err)
uniqueServerSideEvents[id] = e
return true
})
assert.Contains(t, uniqueServerSideEvents, event1.FlowID)
assert.Contains(t, uniqueServerSideEvents, event2.FlowID)
// ack events
server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event1.FlowID].EventId}
server.acks <- &proto.FlowEventAck{EventId: uniqueServerSideEvents[event2.FlowID].EventId}
assert.EventuallyWithT(t, func(c *assert.CollectT) {
unackedEvents := manager.eventsWithoutAcks.GetEvents()
assert.Empty(c, unackedEvents)
}, 3*time.Second, 100*time.Millisecond)
}
func createManager(t *testing.T, serverAddr string, retryInterval time.Duration) *Manager {
t.Helper()
mockIFace := &mockIFaceMapper{
address: wgaddr.Address{
Network: netip.MustParsePrefix("192.168.1.1/32"),
},
isUserspaceBind: true,
}
publicKey := []byte("test-public-key")
manager := NewManager(mockIFace, publicKey, nil)
manager.retryInterval = retryInterval
initialConfig := &types.FlowConfig{
Enabled: true,
URL: fmt.Sprintf("http://%s", serverAddr),
TokenPayload: "initial-payload",
TokenSignature: "initial-signature",
Interval: 500 * time.Millisecond,
}
err := manager.Update(initialConfig)
require.NoError(t, err)
return manager
}
func ipAddr(a string) netip.Addr {
addr, _ := netip.ParseAddr(a)
return addr
}

View File

@@ -0,0 +1,365 @@
package store
import (
"math/rand"
"net/netip"
"testing"
"time"
"github.com/google/uuid"
"github.com/netbirdio/netbird/client/internal/netflow/types"
"github.com/stretchr/testify/assert"
)
var random = rand.New(rand.NewSource(time.Now().UnixNano()))
func TestFlowAggregation(t *testing.T) {
var protocols = []types.Protocol{types.ICMP, types.ICMPv6, types.TCP, types.UDP}
var tests = []struct {
description string
addresses [][]netip.Addr
dstPort uint16
eventTypes []types.Type
}{
{
description: "start and stop",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeStart, types.TypeEnd},
},
{
description: "start and drop",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeStart, types.TypeDrop},
},
{
description: "start only",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeStart},
},
{
description: "drop only",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeDrop},
}}
for _, protocol := range protocols {
for _, tt := range tests {
t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
store := NewAggregatingMemoryStore()
store.WindowEnd = time.Now().Add(5 * time.Second)
allExpected := make([]*types.Event, 0)
for _, srcAndDst := range tt.addresses {
inEvents, expected := generateEvents(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, protocol, types.Ingress, 0, store.WindowStart, store.WindowEnd)
for _, e := range inEvents {
store.StoreEvent(e)
}
allExpected = append(allExpected, expected)
}
events := store.GetAggregatedEvents()
assert.ElementsMatch(t, events, allExpected)
})
}
}
}
func TestIcmpEventAggregation(t *testing.T) {
var protocols = []types.Protocol{types.ICMP, types.ICMPv6}
var icmpTypes = []uint8{1, 2, 3}
var tests = []struct {
description string
addresses [][]netip.Addr
eventTypes []types.Type
}{
{
description: "start and stop",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
eventTypes: []types.Type{types.TypeStart, types.TypeEnd},
},
{
description: "start and drop",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
eventTypes: []types.Type{types.TypeStart, types.TypeDrop},
},
{
description: "start only",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
eventTypes: []types.Type{types.TypeStart},
},
{
description: "drop only",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}},
eventTypes: []types.Type{types.TypeDrop},
}}
for _, protocol := range protocols {
for _, tt := range tests {
t.Run(tt.description+" "+protocol.String(), func(t *testing.T) {
store := NewAggregatingMemoryStore()
store.WindowEnd = time.Now().Add(5 * time.Second)
allExpected := make([]*types.Event, 0)
for _, icmpType := range icmpTypes {
events, expected := generateEvents(tt.addresses[0][0], tt.addresses[0][1], 0, tt.eventTypes, protocol, types.Ingress, icmpType, store.WindowStart, store.WindowEnd)
for _, e := range events {
store.StoreEvent(e)
}
allExpected = append(allExpected, expected)
}
aggregatedEvents := store.GetAggregatedEvents()
assert.Len(t, aggregatedEvents, len(allExpected))
assert.ElementsMatch(t, aggregatedEvents, allExpected)
})
}
}
}
func TestFlowAggregationOfUnknownProtocols(t *testing.T) {
var tests = []struct {
description string
addresses [][]netip.Addr
dstPort uint16
eventTypes []types.Type
}{
{
description: "start and stop",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeStart, types.TypeEnd},
},
{
description: "start and drop",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeStart, types.TypeDrop},
},
{
description: "start only",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeStart},
},
{
description: "drop only",
addresses: [][]netip.Addr{{netip.MustParseAddr("1.1.1.1"), netip.MustParseAddr("2.2.2.2")}, {netip.MustParseAddr("3.3.3.3"), netip.MustParseAddr("2.2.2.2")}},
dstPort: uint16(random.Uint32() >> 16),
eventTypes: []types.Type{types.TypeDrop},
}}
for _, tt := range tests {
t.Run(tt.description+" "+types.ProtocolUnknown.String(), func(t *testing.T) {
store := NewAggregatingMemoryStore()
store.WindowEnd = time.Now().Add(5 * time.Second)
allExpected := make([]*types.Event, 0)
for _, srcAndDst := range tt.addresses {
inEvents, expected := generateEventsForUnknownProtocol(srcAndDst[0], srcAndDst[1], tt.dstPort, tt.eventTypes, types.ProtocolUnknown, types.Ingress, store.WindowStart, store.WindowEnd)
for _, e := range inEvents {
store.StoreEvent(e)
}
allExpected = append(allExpected, expected...)
}
events := store.GetAggregatedEvents()
assert.ElementsMatch(t, events, allExpected)
})
}
}
func TestResetAggregationWindow(t *testing.T) {
now := time.Now()
nowFunc := func() time.Time { return now }
store := NewAggregatingMemoryStoreWithTimeFunc(nowFunc)
store.StoreEvent(&types.Event{
ID: uuid.New(),
Timestamp: time.Now(),
EventFields: types.EventFields{
FlowID: uuid.New(),
Type: types.TypeStart,
Protocol: types.TCP,
RuleID: []byte("rule-id-1"),
Direction: types.Ingress,
SourceIP: netip.MustParseAddr("1.1.1.1"),
SourcePort: 1234,
DestIP: netip.MustParseAddr("2.2.2.2"),
DestPort: 5678,
SourceResourceID: []byte("source-resource-id"),
DestResourceID: []byte("dest-resource-id"),
RxPackets: random.Uint64(),
TxPackets: random.Uint64(),
RxBytes: random.Uint64(),
TxBytes: random.Uint64(),
},
})
now = now.Add(1 * time.Second)
reset := store.ResetAggregationWindow()
previousEvents, ok := reset.(*AggregatingMemory)
assert.True(t, ok)
assert.NotEqual(t, previousEvents.WindowStart, store.WindowStart)
assert.Equal(t, previousEvents.WindowEnd, store.WindowStart)
assert.NotEmpty(t, previousEvents.events)
assert.Empty(t, store.events)
}
func generateEvents(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
direction types.Direction, icmpType uint8, windowStart, windowEnd time.Time) ([]*types.Event, *types.Event) {
var rxPackets, txPackets, rxBytes, txBytes uint64
inEvents := make([]*types.Event, 0)
ts := time.Now()
flowId := uuid.New()
srcPort := uint16(random.Uint32() >> 16)
for idx, eventType := range eventTypes {
e := &types.Event{
ID: uuid.New(),
Timestamp: ts.Add(time.Duration(idx) * time.Second),
EventFields: types.EventFields{
FlowID: flowId,
Type: eventType,
Protocol: protocol,
RuleID: []byte("rule-id-1"),
Direction: direction,
SourceIP: srcIp,
SourcePort: srcPort,
DestIP: dstIp,
DestPort: dstPort,
SourceResourceID: []byte("source-resource-id"),
DestResourceID: []byte("dest-resource-id"),
RxPackets: random.Uint64(),
TxPackets: random.Uint64(),
RxBytes: random.Uint64(),
TxBytes: random.Uint64(),
}}
rxBytes += e.RxBytes
txBytes += e.TxBytes
rxPackets += e.RxPackets
txPackets += e.TxPackets
inEvents = append(inEvents, e)
if protocol == types.ICMP || protocol == types.ICMPv6 {
e.ICMPType = icmpType
}
}
var start, end, drop uint64
for _, eventType := range eventTypes {
switch eventType {
case types.TypeStart:
start += 1
case types.TypeDrop:
drop += 1
case types.TypeEnd:
end += 1
}
}
aggregatedEvent := &types.Event{
ID: inEvents[0].ID,
Timestamp: inEvents[0].Timestamp,
WindowStart: windowStart,
WindowEnd: windowEnd,
EventFields: types.EventFields{
FlowID: flowId,
Type: types.TypeUnknown,
Protocol: inEvents[0].Protocol,
RuleID: []byte("rule-id-1"),
Direction: inEvents[0].Direction,
SourceIP: srcIp,
SourcePort: srcPort,
DestIP: dstIp,
DestPort: dstPort,
SourceResourceID: []byte("source-resource-id"),
DestResourceID: []byte("dest-resource-id"),
RxPackets: rxPackets,
TxPackets: txPackets,
RxBytes: rxBytes,
TxBytes: txBytes,
NumOfStarts: start,
NumOfEnds: end,
NumOfDrops: drop,
}}
if protocol == types.ICMP || protocol == types.ICMPv6 {
aggregatedEvent.ICMPType = icmpType
}
return inEvents, aggregatedEvent
}
func generateEventsForUnknownProtocol(srcIp, dstIp netip.Addr, dstPort uint16, eventTypes []types.Type, protocol types.Protocol,
direction types.Direction, windowStart, windowEnd time.Time) ([]*types.Event, []*types.Event) {
inEvents := make([]*types.Event, 0)
expectedEvents := make([]*types.Event, 0)
ts := time.Now()
flowId := uuid.New()
srcPort := uint16(random.Uint32() >> 16)
for idx, eventType := range eventTypes {
e := &types.Event{
ID: uuid.New(),
Timestamp: ts.Add(time.Duration(idx) * time.Second),
EventFields: types.EventFields{
FlowID: flowId,
Type: eventType,
Protocol: protocol,
RuleID: []byte("rule-id-1"),
Direction: direction,
SourceIP: srcIp,
SourcePort: srcPort,
DestIP: dstIp,
DestPort: dstPort,
SourceResourceID: []byte("source-resource-id"),
DestResourceID: []byte("dest-resource-id"),
RxPackets: random.Uint64(),
TxPackets: random.Uint64(),
RxBytes: random.Uint64(),
TxBytes: random.Uint64(),
}}
inEvents = append(inEvents, e)
var start, end, drop uint64
switch eventType {
case types.TypeStart:
start = 1
case types.TypeDrop:
drop = 1
case types.TypeEnd:
end = 1
}
expectedEvents = append(expectedEvents, &types.Event{
ID: e.ID,
Timestamp: e.Timestamp,
WindowStart: windowStart,
WindowEnd: windowEnd,
EventFields: types.EventFields{
FlowID: flowId,
Type: types.TypeUnknown,
Protocol: e.Protocol,
RuleID: []byte("rule-id-1"),
Direction: e.Direction,
SourceIP: srcIp,
SourcePort: srcPort,
DestIP: dstIp,
DestPort: dstPort,
SourceResourceID: []byte("source-resource-id"),
DestResourceID: []byte("dest-resource-id"),
RxPackets: e.RxPackets,
TxPackets: e.TxPackets,
RxBytes: e.RxBytes,
TxBytes: e.TxBytes,
NumOfStarts: start,
NumOfEnds: end,
NumOfDrops: drop,
}})
}
return inEvents, expectedEvents
}

View File

@@ -1,10 +1,15 @@
package store
import (
"maps"
"math/rand"
v2 "math/rand/v2"
"net/netip"
"slices"
"sync"
"time"
"github.com/google/uuid"
"github.com/netbirdio/netbird/client/internal/netflow/types"
)
@@ -19,6 +24,14 @@ type Memory struct {
events map[uuid.UUID]*types.Event
}
type AggregatingMemory struct {
Memory
WindowStart time.Time
WindowEnd time.Time
rnd *v2.PCG
nowFunc func() time.Time
}
func (m *Memory) StoreEvent(event *types.Event) {
m.mux.Lock()
defer m.mux.Unlock()
@@ -48,3 +61,104 @@ func (m *Memory) DeleteEvents(ids []uuid.UUID) {
delete(m.events, id)
}
}
func NewAggregatingMemoryStore() *AggregatingMemory {
return NewAggregatingMemoryStoreWithTimeFunc(defaultNowFunc)
}
// used in tests when deterministic (less random) time intervals are required
func NewAggregatingMemoryStoreWithTimeFunc(nowFunc func() time.Time) *AggregatingMemory {
return &AggregatingMemory{WindowStart: nowFunc(), Memory: Memory{events: make(map[uuid.UUID]*types.Event)}, nowFunc: nowFunc, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())}
}
func (am *AggregatingMemory) ResetAggregationWindow() types.FlowEventAggregator {
am.mux.Lock()
defer am.mux.Unlock()
now := am.nowFunc()
toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: now, Memory: Memory{events: am.events}, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())}
am.events = make(map[uuid.UUID]*types.Event)
am.WindowStart = now
return &toret
}
type aggregationKey struct {
srcAddr netip.Addr
destAddr netip.Addr
destPort uint16
direction int
protocol uint8
icmpType uint8
unique uint64 // used to prevent aggregation on non icmp/udp/tcp events
}
func (am *AggregatingMemory) GetAggregatedEvents() []*types.Event {
am.mux.Lock()
defer am.mux.Unlock()
aggregated := make(map[aggregationKey]*types.Event)
for _, v := range am.events {
lookupKey := aggregationKey{srcAddr: v.SourceIP, destAddr: v.DestIP, destPort: v.DestPort, direction: int(v.Direction), protocol: uint8(v.Protocol), icmpType: v.ICMPType}
if _, ok := aggregated[lookupKey]; !ok {
event := v.Clone()
switch event.Type {
case types.TypeStart:
event.NumOfStarts += 1
case types.TypeDrop:
event.NumOfDrops += 1
case types.TypeEnd:
event.NumOfEnds += 1
}
event.Type = types.TypeUnknown
// Please note that ICMPCode field isn't propagated by the manager (see flow/proto/flow.pb.go, FlowFields struct)
// so the field value in an icmp event in the "aggregated" doesn't matter
event.WindowStart = am.WindowStart
event.WindowEnd = am.WindowEnd
if event.Protocol != types.ICMP && event.Protocol != types.ICMPv6 && event.Protocol != types.UDP && event.Protocol != types.TCP {
lookupKey.unique = am.rnd.Uint64() // to make the lookup key unique so we don't aggregate on it
}
aggregated[lookupKey] = event
continue
}
aggregatedEvent := aggregated[lookupKey]
if aggregatedEvent.Protocol != types.ICMP && aggregatedEvent.Protocol != types.ICMPv6 && aggregatedEvent.Protocol != types.UDP && aggregatedEvent.Protocol != types.TCP {
continue // we don't aggregate this type of events; shouldn't ever get here
}
// track the number of connections, duration?, open and close events?
aggregatedEvent.RxBytes += v.RxBytes
aggregatedEvent.RxPackets += v.RxPackets
aggregatedEvent.TxBytes += v.TxBytes
aggregatedEvent.TxPackets += v.TxPackets
switch v.Type {
case types.TypeStart:
aggregatedEvent.NumOfStarts += 1
case types.TypeDrop:
aggregatedEvent.NumOfDrops += 1
case types.TypeEnd:
aggregatedEvent.NumOfEnds += 1
}
if aggregatedEvent.Timestamp.Compare(v.Timestamp) > 0 {
aggregatedEvent.Timestamp = v.Timestamp
aggregatedEvent.ID = v.ID
aggregatedEvent.SourcePort = v.SourcePort
}
if len(aggregatedEvent.RuleID) == 0 && len(v.RuleID) != 0 {
aggregatedEvent.RuleID = slices.Clone(v.RuleID)
}
}
return slices.Collect(maps.Values(aggregated)) // could return an iterator instead here
}
func defaultNowFunc() time.Time {
return time.Now()
}

View File

@@ -2,6 +2,7 @@ package types
import (
"net/netip"
"slices"
"strconv"
"time"
@@ -69,8 +70,10 @@ const (
)
type Event struct {
ID uuid.UUID
Timestamp time.Time
ID uuid.UUID
Timestamp time.Time
WindowStart time.Time
WindowEnd time.Time
EventFields
}
@@ -92,6 +95,17 @@ type EventFields struct {
TxPackets uint64
RxBytes uint64
TxBytes uint64
NumOfStarts uint64
NumOfEnds uint64
NumOfDrops uint64
}
func (e *Event) Clone() *Event {
toret := *e
toret.RuleID = slices.Clone(e.RuleID)
toret.SourceResourceID = slices.Clone(e.SourceResourceID)
toret.DestResourceID = slices.Clone(e.DestResourceID)
return &toret
}
type FlowConfig struct {
@@ -114,13 +128,15 @@ type FlowManager interface {
GetLogger() FlowLogger
}
type FlowEventAggregator interface {
ResetAggregationWindow() FlowEventAggregator
GetAggregatedEvents() []*Event
}
type FlowLogger interface {
ResetAggregationWindow() FlowEventAggregator
// StoreEvent stores a flow event
StoreEvent(flowEvent EventFields)
// GetEvents returns all stored events
GetEvents() []*Event
// DeleteEvents deletes events from the store
DeleteEvents([]uuid.UUID)
// Close closes the logger
Close()
// Enable enables the flow logger receiver
@@ -140,6 +156,11 @@ type Store interface {
Close()
}
type AggregatingStore interface {
FlowEventAggregator
Store
}
// ConnTracker defines the interface for connection tracking functionality
type ConnTracker interface {
// Start begins tracking connections by listening for conntrack events.

View File

@@ -203,7 +203,6 @@ func NewConn(config ConnConfig, services ServiceDependencies) (*Conn, error) {
statusICE: worker.NewAtomicStatus(),
dumpState: dumpState,
endpointUpdater: NewEndpointUpdater(connLog, config.WgConfig, isController(config)),
wgWatcher: NewWGWatcher(connLog, config.WgConfig.WgInterface, config.Key, dumpState),
metricsRecorder: services.MetricsRecorder,
}
@@ -671,11 +670,12 @@ func (conn *Conn) onGuardEvent() {
}
}
func (conn *Conn) onWGDisconnected() {
func (conn *Conn) onWGDisconnected(watcherCtx context.Context) {
conn.mu.Lock()
defer conn.mu.Unlock()
if conn.ctx.Err() != nil {
// watcherCtx guards against a stale watcher tearing down a connection that already superseded it.
if conn.ctx.Err() != nil || watcherCtx.Err() != nil {
return
}
@@ -833,25 +833,39 @@ func (conn *Conn) isConnectedOnAllWay() (status guard.ConnStatus) {
})
}
// enableWgWatcherIfNeeded starts a fresh watcher instance per connection attempt, so its
// lifecycle stays bound to conn.mu and enable/disable can't race an old goroutine's shutdown.
// Caller must hold conn.mu.
func (conn *Conn) enableWgWatcherIfNeeded(enabledTime time.Time) {
if !conn.wgWatcher.PrepareInitialHandshake() {
if conn.wgWatcher != nil {
return
}
watcher := NewWGWatcher(conn.Log, conn.config.WgConfig.WgInterface, conn.config.Key, conn.dumpState)
watcher.PrepareInitialHandshake()
wgWatcherCtx, wgWatcherCancel := context.WithCancel(conn.ctx)
conn.wgWatcher = watcher
conn.wgWatcherCancel = wgWatcherCancel
conn.wgWatcherWg.Add(1)
go func() {
defer conn.wgWatcherWg.Done()
conn.wgWatcher.EnableWgWatcher(wgWatcherCtx, enabledTime, conn.onWGDisconnected, conn.onWGHandshakeSuccess, conn.onWGCheckSuccess)
onDisconnected := func() { conn.onWGDisconnected(wgWatcherCtx) }
watcher.EnableWgWatcher(wgWatcherCtx, enabledTime, onDisconnected, conn.onWGHandshakeSuccess, conn.onWGCheckSuccess)
}()
}
// disableWgWatcherIfNeeded cancels and drops the watcher once no transport is active. It never
// waits for the goroutine: the timeout path reentrantly calls back here under conn.mu, so
// blocking would deadlock. Caller must hold conn.mu.
func (conn *Conn) disableWgWatcherIfNeeded() {
if conn.currentConnPriority == conntype.None && conn.wgWatcherCancel != nil {
conn.wgWatcherCancel()
conn.wgWatcherCancel = nil
if conn.currentConnPriority != conntype.None || conn.wgWatcher == nil {
return
}
conn.wgWatcherCancel()
conn.wgWatcher = nil
conn.wgWatcherCancel = nil
}
func (conn *Conn) newProxy(remoteConn net.Conn) (wgproxy.Proxy, error) {
@@ -874,7 +888,9 @@ func (conn *Conn) resetEndpoint() {
return
}
conn.Log.Infof("reset wg endpoint")
conn.wgWatcher.Reset()
if conn.wgWatcher != nil {
conn.wgWatcher.Reset()
}
if err := conn.endpointUpdater.RemoveEndpointAddress(); err != nil {
conn.Log.Warnf("failed to remove endpoint address before update: %v", err)
}

View File

@@ -339,20 +339,20 @@ func TestConn_onWGDisconnected_EscalatesToRosenpassReset(t *testing.T) {
conn := newWGTimeoutTestConn(true, &disconnected)
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
conn.onWGDisconnected()
conn.onWGDisconnected(conn.ctx)
}
assert.Empty(t, disconnected, "escalation must not fire below the threshold")
conn.onWGDisconnected()
conn.onWGDisconnected(conn.ctx)
assert.Equal(t, []string{conn.config.WgConfig.RemoteKey}, disconnected,
"reaching the threshold must report the peer disconnected once")
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
conn.onWGDisconnected()
conn.onWGDisconnected(conn.ctx)
}
assert.Len(t, disconnected, 1, "escalation must restart counting after firing")
conn.onWGDisconnected()
conn.onWGDisconnected(conn.ctx)
assert.Len(t, disconnected, 2, "continued timeouts must escalate again")
}
@@ -364,12 +364,12 @@ func TestConn_onWGDisconnected_CheckSuccessResetsEscalation(t *testing.T) {
conn := newWGTimeoutTestConn(true, &disconnected)
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
conn.onWGDisconnected()
conn.onWGDisconnected(conn.ctx)
}
conn.onWGCheckSuccess()
for i := 0; i < wgTimeoutEscalationThreshold-1; i++ {
conn.onWGDisconnected()
conn.onWGDisconnected(conn.ctx)
}
assert.Empty(t, disconnected, "handshake success must reset the timeout count")
}
@@ -382,7 +382,7 @@ func TestConn_onWGDisconnected_NoEscalationWithoutRosenpass(t *testing.T) {
conn := newWGTimeoutTestConn(false, &disconnected)
for i := 0; i < wgTimeoutEscalationThreshold*3; i++ {
conn.onWGDisconnected()
conn.onWGDisconnected(conn.ctx)
}
assert.Empty(t, disconnected, "escalation must be limited to rosenpass connections")
}

View File

@@ -7,6 +7,7 @@ import (
"net/netip"
"slices"
"sync"
"sync/atomic"
"time"
"github.com/google/uuid"
@@ -191,23 +192,30 @@ func (s *StatusChangeSubscription) Events() chan map[string]RouterState {
// every private-service request) don't contend against each other.
// Pure read methods take RLock; anything that mutates state takes Lock.
type Status struct {
mux sync.RWMutex
muxRelays sync.RWMutex
peers map[string]State
ipToKey map[string]string
changeNotify map[string]map[string]*StatusChangeSubscription // map[peerID]map[subscriptionID]*StatusChangeSubscription
signalState bool
signalError error
managementState bool
managementError error
relayStates []relay.ProbeResult
localPeer LocalPeerState
offlinePeers []State
mgmAddress string
signalAddress string
notifier *notifier
rosenpassEnabled bool
rosenpassPermissive bool
mux sync.RWMutex
muxRelays sync.RWMutex
peers map[string]State
ipToKey map[string]string
changeNotify map[string]map[string]*StatusChangeSubscription // map[peerID]map[subscriptionID]*StatusChangeSubscription
signalState bool
signalError error
managementState bool
managementError error
relayStates []relay.ProbeResult
localPeer LocalPeerState
offlinePeers []State
mgmAddress string
signalAddress string
notifier *notifier
rosenpassEnabled bool
rosenpassPermissive bool
// sessionExpiresAt is the absolute UTC instant at which the peer's SSO
// session expires. Zero when the peer is not SSO-tracked or login
// expiration is disabled. Populated from management LoginResponse /
// SyncResponse and exposed via the daemon's Status / SubscribeStatus RPC
// so the UI can show remaining time without itself talking to mgm.
sessionExpiresAt time.Time
nsGroupStates []NSGroupState
resolvedDomainsStates map[domain.Domain]ResolvedDomainInfo
lazyConnectionEnabled bool
@@ -223,6 +231,21 @@ type Status struct {
eventStreams map[string]chan *proto.SystemEvent
eventQueue *EventQueue
// stateChangeStreams fan-out connection-state changes (connected /
// disconnected / connecting / address change / peers list change) to
// every active SubscribeStatus gRPC stream. Each subscriber gets a
// buffered chan; the notifier non-blockingly pings them so a slow
// consumer can never stall the daemon.
stateChangeMux sync.Mutex
stateChangeStreams map[string]chan struct{}
// networksRevision bumps whenever the routed-networks set or their
// selected state changes (driven by the route manager). Surfaced in the
// status snapshot so the UI can fingerprint on it and re-fetch
// ListNetworks only on a real change. Atomic so the snapshot builder can
// read it without taking mux.
networksRevision atomic.Uint64
ingressGwMgr *ingressgw.Manager
routeIDLookup routeIDLookup
@@ -237,6 +260,7 @@ func NewRecorder(mgmAddress string) *Status {
changeNotify: make(map[string]map[string]*StatusChangeSubscription),
eventStreams: make(map[string]chan *proto.SystemEvent),
eventQueue: NewEventQueue(eventQueueSize),
stateChangeStreams: make(map[string]chan struct{}),
offlinePeers: make([]State, 0),
notifier: newNotifier(),
mgmAddress: mgmAddress,
@@ -401,6 +425,7 @@ func (d *Status) UpdatePeerState(receivedState State) error {
if notifyRouter {
d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
}
d.notifyStateChange()
return nil
}
@@ -426,6 +451,7 @@ func (d *Status) AddPeerStateRoute(peer string, route string, resourceId route.R
// todo: consider to make sense of this notification or not
d.notifier.peerListChanged(numPeers)
d.notifyStateChange()
return nil
}
@@ -451,6 +477,7 @@ func (d *Status) RemovePeerStateRoute(peer string, route string) error {
// todo: consider to make sense of this notification or not
d.notifier.peerListChanged(numPeers)
d.notifyStateChange()
return nil
}
@@ -500,6 +527,7 @@ func (d *Status) UpdatePeerICEState(receivedState State) error {
if notifyRouter {
d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
}
d.notifyStateChange()
return nil
}
@@ -536,6 +564,7 @@ func (d *Status) UpdatePeerRelayedState(receivedState State) error {
if notifyRouter {
d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
}
d.notifyStateChange()
return nil
}
@@ -571,6 +600,7 @@ func (d *Status) UpdatePeerRelayedStateToDisconnected(receivedState State) error
if notifyRouter {
d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
}
d.notifyStateChange()
return nil
}
@@ -609,6 +639,7 @@ func (d *Status) UpdatePeerICEStateToDisconnected(receivedState State) error {
if notifyRouter {
d.dispatchRouterPeers(receivedState.PubKey, routerSnapshot)
}
d.notifyStateChange()
return nil
}
@@ -702,6 +733,7 @@ func (d *Status) FinishPeerListModifications() {
for _, rd := range dispatches {
d.dispatchRouterPeers(rd.peerID, rd.snapshot)
}
d.notifyStateChange()
}
func (d *Status) SubscribeToPeerStateChanges(ctx context.Context, peerID string) *StatusChangeSubscription {
@@ -760,6 +792,41 @@ func (d *Status) UpdateLocalPeerState(localPeerState LocalPeerState) {
d.mux.Unlock()
d.notifier.localAddressChanged(fqdn, ip)
d.notifyStateChange()
}
// SetSessionExpiresAt records the absolute UTC instant at which the peer's
// SSO session is set to expire. Pass the zero value to clear (e.g. when the
// management server stops publishing a deadline because login expiration was
// disabled or the peer is not SSO-tracked). Same-value updates are no-ops;
// real changes fan out via notifyStateChange so SubscribeStatus consumers
// pick up the new deadline on their next read.
func (d *Status) SetSessionExpiresAt(deadline time.Time) {
d.mux.Lock()
if d.sessionExpiresAt.Equal(deadline) {
d.mux.Unlock()
return
}
d.sessionExpiresAt = deadline
d.mux.Unlock()
d.notifyStateChange()
}
// GetSessionExpiresAt returns the most recently recorded SSO session deadline,
// or the zero value when no deadline is tracked. A deadline that has already
// slipped into the past reports as "none": once the session has expired it is
// no longer a meaningful countdown, and the sessionwatch.Watcher does not
// arm a timer at the deadline itself to clear it (only the two pre-expiry
// warnings). Without this guard the UI would keep painting a stale
// "expires in …" against a moment that has passed until the next login,
// extend, or teardown rewrote the value.
func (d *Status) GetSessionExpiresAt() time.Time {
d.mux.Lock()
defer d.mux.Unlock()
if !d.sessionExpiresAt.IsZero() && d.sessionExpiresAt.Before(time.Now()) {
return time.Time{}
}
return d.sessionExpiresAt
}
// AddLocalPeerStateRoute adds a route to the local peer state
@@ -828,11 +895,19 @@ func (d *Status) CleanLocalPeerState() {
d.mux.Unlock()
d.notifier.localAddressChanged(fqdn, ip)
d.notifyStateChange()
}
// MarkManagementDisconnected sets ManagementState to disconnected
func (d *Status) MarkManagementDisconnected(err error) {
d.mux.Lock()
// Health checks re-mark the same state on every probe; skip the fan-out
// when nothing actually changed so we don't flood SubscribeStatus
// consumers with identical snapshots.
if !d.managementState && errors.Is(d.managementError, err) {
d.mux.Unlock()
return
}
d.managementState = false
d.managementError = err
mgm := d.managementState
@@ -840,11 +915,16 @@ func (d *Status) MarkManagementDisconnected(err error) {
d.mux.Unlock()
d.notifier.updateServerStates(mgm, sig)
d.notifyStateChange()
}
// MarkManagementConnected sets ManagementState to connected
func (d *Status) MarkManagementConnected() {
d.mux.Lock()
if d.managementState && d.managementError == nil {
d.mux.Unlock()
return
}
d.managementState = true
d.managementError = nil
mgm := d.managementState
@@ -852,6 +932,7 @@ func (d *Status) MarkManagementConnected() {
d.mux.Unlock()
d.notifier.updateServerStates(mgm, sig)
d.notifyStateChange()
}
// UpdateSignalAddress update the address of the signal server
@@ -885,6 +966,10 @@ func (d *Status) UpdateLazyConnection(enabled bool) {
// MarkSignalDisconnected sets SignalState to disconnected
func (d *Status) MarkSignalDisconnected(err error) {
d.mux.Lock()
if !d.signalState && errors.Is(d.signalError, err) {
d.mux.Unlock()
return
}
d.signalState = false
d.signalError = err
mgm := d.managementState
@@ -892,11 +977,16 @@ func (d *Status) MarkSignalDisconnected(err error) {
d.mux.Unlock()
d.notifier.updateServerStates(mgm, sig)
d.notifyStateChange()
}
// MarkSignalConnected sets SignalState to connected
func (d *Status) MarkSignalConnected() {
d.mux.Lock()
if d.signalState && d.signalError == nil {
d.mux.Unlock()
return
}
d.signalState = true
d.signalError = nil
mgm := d.managementState
@@ -904,6 +994,7 @@ func (d *Status) MarkSignalConnected() {
d.mux.Unlock()
d.notifier.updateServerStates(mgm, sig)
d.notifyStateChange()
}
func (d *Status) UpdateRelayStates(relayResults []relay.ProbeResult) {
@@ -1110,16 +1201,19 @@ func (d *Status) GetFullStatus() FullStatus {
// ClientStart will notify all listeners about the new service state
func (d *Status) ClientStart() {
d.notifier.clientStart()
d.notifyStateChange()
}
// ClientStop will notify all listeners about the new service state
func (d *Status) ClientStop() {
d.notifier.clientStop()
d.notifyStateChange()
}
// ClientTeardown will notify all listeners about the service is under teardown
func (d *Status) ClientTeardown() {
d.notifier.clientTearDown()
d.notifyStateChange()
}
// SetConnectionListener set a listener to the notifier
@@ -1261,6 +1355,79 @@ func (d *Status) GetEventHistory() []*proto.SystemEvent {
return d.eventQueue.GetAll()
}
// SubscribeToStateChanges hands back a channel that receives a tick on
// every connection-state change (connected / disconnected / connecting /
// address change / peers-list change). The channel is buffered to one
// pending tick so a coalesced burst still wakes the consumer exactly
// once. Pass the returned id to UnsubscribeFromStateChanges to detach.
func (d *Status) SubscribeToStateChanges() (string, <-chan struct{}) {
d.stateChangeMux.Lock()
defer d.stateChangeMux.Unlock()
id := uuid.New().String()
ch := make(chan struct{}, 1)
d.stateChangeStreams[id] = ch
return id, ch
}
// UnsubscribeFromStateChanges releases a SubscribeToStateChanges channel
// and closes it so any consumer goroutine selecting on the channel
// unblocks cleanly.
func (d *Status) UnsubscribeFromStateChanges(id string) {
d.stateChangeMux.Lock()
defer d.stateChangeMux.Unlock()
if ch, ok := d.stateChangeStreams[id]; ok {
close(ch)
delete(d.stateChangeStreams, id)
}
}
// notifyStateChange wakes every SubscribeToStateChanges subscriber. Drops
// the tick if a subscriber's buffer is full — by definition the consumer
// is already going to fetch the latest snapshot, so multiple pending ticks
// would be redundant.
func (d *Status) notifyStateChange() {
d.stateChangeMux.Lock()
defer d.stateChangeMux.Unlock()
for _, ch := range d.stateChangeStreams {
select {
case ch <- struct{}{}:
default:
}
}
}
// NotifyStateChange is the public wake-the-subscribers entry point used by
// callers that mutate state outside the peer recorder — most importantly
// the connect-state machine, which writes StatusNeedsLogin into the
// shared contextState (client/internal/state.go) without touching any
// recorder field. Without this push the SubscribeStatus stream stays on
// the previous snapshot until an unrelated peer/management/signal
// change happens to fire notifyStateChange, leaving the UI's status
// out of sync with the daemon.
func (d *Status) NotifyStateChange() {
d.notifyStateChange()
}
// BumpNetworksRevision increments the routed-networks revision and wakes every
// SubscribeStatus subscriber. The route manager calls it when a network map
// changes the available routes or when a selection is applied — the peer
// status itself only records actively-routed (chosen) networks, so without
// this bump a candidate route appearing/disappearing would never reach the UI.
func (d *Status) BumpNetworksRevision() {
d.networksRevision.Add(1)
d.notifyStateChange()
}
// GetNetworksRevision returns the current routed-networks revision, surfaced in
// the status snapshot so the UI can detect route/selection changes (see
// BumpNetworksRevision).
func (d *Status) GetNetworksRevision() uint64 {
return d.networksRevision.Load()
}
func (d *Status) SetWgIface(wgInterface WGIfaceStatus) {
d.mux.Lock()
defer d.mux.Unlock()

View File

@@ -314,3 +314,39 @@ func TestGetFullStatus(t *testing.T) {
assert.Equal(t, signalState, fullStatus.SignalState, "signal status should be equal")
assert.ElementsMatch(t, []State{peerState1, peerState2}, fullStatus.Peers, "peers states should match")
}
// notified reports whether a state-change tick is pending on ch, draining it.
func notified(ch <-chan struct{}) bool {
select {
case <-ch:
return true
default:
return false
}
}
func TestMarkServerStateDoesNotNotifyWhenUnchanged(t *testing.T) {
status := NewRecorder("https://mgm")
_, ch := status.SubscribeToStateChanges()
// First transition is a real change and must notify.
status.MarkManagementConnected()
require.True(t, notified(ch), "first connect should notify")
// Re-marking the same state must not notify again.
status.MarkManagementConnected()
assert.False(t, notified(ch), "redundant connect should not notify")
// Same for signal.
status.MarkSignalConnected()
require.True(t, notified(ch), "first signal connect should notify")
status.MarkSignalConnected()
assert.False(t, notified(ch), "redundant signal connect should not notify")
// A genuine change (disconnect with an error) notifies again.
err := errors.New("boom")
status.MarkManagementDisconnected(err)
require.True(t, notified(ch), "disconnect should notify")
status.MarkManagementDisconnected(err)
assert.False(t, notified(ch), "redundant disconnect should not notify")
}

View File

@@ -3,7 +3,6 @@ package peer
import (
"context"
"fmt"
"sync"
"time"
log "github.com/sirupsen/logrus"
@@ -24,14 +23,14 @@ type WGInterfaceStater interface {
GetStats() (map[string]configurer.WGStats, error)
}
// WGWatcher is single-shot: one instance per connection attempt, run once, then discarded.
// Lifecycle is owned by Conn under conn.mu, so it keeps no "enabled" state to go stale.
type WGWatcher struct {
log *log.Entry
wgIfaceStater WGInterfaceStater
peerKey string
stateDump *stateDump
enabled bool
muEnabled sync.Mutex
// initialHandshake is not thread-safe; never call PrepareInitialHandshake and EnableWgWatcher concurrently.
initialHandshake time.Time
@@ -48,25 +47,14 @@ func NewWGWatcher(log *log.Entry, wgIfaceStater WGInterfaceStater, peerKey strin
}
}
// PrepareInitialHandshake reserves the watcher and reads the peer's current WireGuard
// handshake time. It must be called before the peer is (re)configured on the WireGuard
// interface, so the captured baseline reflects the state prior to this connection attempt
// instead of racing with that configuration. Returns ok=false if the watcher is already
// running, in which case EnableWgWatcher must not be called.
func (w *WGWatcher) PrepareInitialHandshake() (ok bool) {
w.muEnabled.Lock()
if w.enabled {
w.muEnabled.Unlock()
return false
}
// PrepareInitialHandshake reads the peer's current WireGuard handshake time. It must be
// called before the peer is (re)configured on the WireGuard interface, so the captured
// baseline reflects the state prior to this connection attempt instead of racing with
// that configuration.
func (w *WGWatcher) PrepareInitialHandshake() {
w.log.Debugf("enable WireGuard watcher")
w.enabled = true
w.muEnabled.Unlock()
handshake, _ := w.wgState()
w.initialHandshake = handshake
return true
}
// EnableWgWatcher runs the WireGuard watcher loop using the handshake baseline captured by
@@ -76,10 +64,6 @@ func (w *WGWatcher) PrepareInitialHandshake() (ok bool) {
// handshake, including the first.
func (w *WGWatcher) EnableWgWatcher(ctx context.Context, enabledTime time.Time, onDisconnectedFn func(), onHandshakeSuccessFn func(when time.Time), onCheckSuccessFn func()) {
w.periodicHandshakeCheck(ctx, onDisconnectedFn, onHandshakeSuccessFn, onCheckSuccessFn, enabledTime, w.initialHandshake)
w.muEnabled.Lock()
w.enabled = false
w.muEnabled.Unlock()
}
// Reset signals the watcher that the WireGuard peer has been reset and a new
@@ -105,6 +89,7 @@ func (w *WGWatcher) periodicHandshakeCheck(ctx context.Context, onDisconnectedFn
case <-timer.C:
handshake, ok := w.handshakeCheck(lastHandshake)
if !ok {
// early ctx cancel check return
if ctx.Err() != nil {
return
}
@@ -153,9 +138,9 @@ func (w *WGWatcher) handshakeCheck(lastHandshake time.Time) (*time.Time, bool) {
w.log.Tracef("previous handshake, handshake: %v, %v", lastHandshake, handshake)
// the current know handshake did not change
// the current known handshake did not change
if handshake.Equal(lastHandshake) {
w.log.Warnf("WireGuard handshake timed out: %v", handshake)
w.log.Warnf("WireGuard handshake not updated: %v", handshake)
return nil, false
}

View File

@@ -7,7 +7,6 @@ import (
"time"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/client/iface/configurer"
)
@@ -62,7 +61,7 @@ func TestWGWatcher_CheckSuccessCallback(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
require.True(t, watcher.PrepareInitialHandshake())
watcher.PrepareInitialHandshake()
firstHandshake := make(chan struct{}, 1)
checkSuccess := make(chan struct{}, 1)
@@ -101,8 +100,7 @@ func TestWGWatcher_EnableWgWatcher(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
ok := watcher.PrepareInitialHandshake()
require.True(t, ok, "watcher should not be enabled yet")
watcher.PrepareInitialHandshake()
onDisconnected := make(chan struct{}, 1)
go watcher.EnableWgWatcher(ctx, time.Now(), func() {
@@ -132,8 +130,7 @@ func TestWGWatcher_ReEnable(t *testing.T) {
watcher := NewWGWatcher(mlog, mocWgIface, "", newStateDump("peer", mlog, &Status{}))
ctx, cancel := context.WithCancel(context.Background())
ok := watcher.PrepareInitialHandshake()
require.True(t, ok, "watcher should not be enabled yet")
watcher.PrepareInitialHandshake()
wg := &sync.WaitGroup{}
wg.Add(1)
@@ -149,8 +146,7 @@ func TestWGWatcher_ReEnable(t *testing.T) {
ctx, cancel = context.WithCancel(context.Background())
defer cancel()
ok = watcher.PrepareInitialHandshake()
require.True(t, ok, "watcher should be re-enabled after the previous run stopped")
watcher.PrepareInitialHandshake()
onDisconnected := make(chan struct{}, 1)
go watcher.EnableWgWatcher(ctx, time.Now(), func() {

View File

@@ -11,6 +11,7 @@ import (
"runtime"
"sort"
"strings"
"syscall"
log "github.com/sirupsen/logrus"
@@ -439,7 +440,11 @@ func (s *ServiceManager) GetStatePath() string {
activeProf, err := s.GetActiveProfileState()
if err != nil {
log.Warnf("failed to get active profile state: %v", err)
if errors.Is(err, syscall.ENOSYS) {
log.Debugf("active profile state unavailable on this platform: %v", err)
} else {
log.Warnf("failed to get active profile state: %v", err)
}
return defaultStatePath
}

View File

@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
"github.com/netbirdio/netbird/util"
@@ -71,3 +72,22 @@ func (pm *ProfileManager) SetActiveProfileState(state *ProfileState) error {
return nil
}
// RemoveProfileState deletes the per-profile state file (which holds the
// account email used for the SSO login hint and the UI display). Called after
// a successful logout so a logged-out profile no longer shows a stale account
// email. The state file only stores the email, so deleting it is equivalent to
// clearing it; the next SSO login recreates it. A missing file is not an error.
func (pm *ProfileManager) RemoveProfileState(profileName string) error {
configDir, err := getConfigDir()
if err != nil {
return fmt.Errorf("get config directory: %w", err)
}
stateFile := filepath.Join(configDir, profileName+".state.json")
if err := os.Remove(stateFile); err != nil && !os.IsNotExist(err) {
return fmt.Errorf("remove profile state: %w", err)
}
return nil
}

View File

@@ -12,6 +12,7 @@ import (
"strings"
"sync"
"sync/atomic"
"syscall"
"time"
"github.com/google/uuid"
@@ -264,7 +265,11 @@ func (m *DefaultManager) initSelector() *routeselector.RouteSelector {
// restore selector state if it exists
if err := m.stateManager.LoadState(state); err != nil {
log.Warnf("failed to load state: %v", err)
if errors.Is(err, syscall.ENOSYS) {
log.Debugf("route selector state unavailable on this platform: %v", err)
} else {
log.Warnf("failed to load state: %v", err)
}
return routeselector.NewRouteSelector()
}
@@ -442,6 +447,11 @@ func (m *DefaultManager) UpdateRoutes(
m.updateClientNetworks(updateSerial, filteredClientRoutes)
m.notifier.OnNewRoutes(filteredClientRoutes)
// A new network map can add or drop route/exit-node candidates without
// touching any peer's chosen-route state, so the peer status alone
// wouldn't notify SubscribeStatus subscribers. Bump the revision so the
// UI re-fetches ListNetworks.
m.statusRecorder.BumpNetworksRevision()
}
m.clientRoutes = clientRoutes
@@ -582,6 +592,10 @@ func (m *DefaultManager) TriggerSelection(networks route.HAMap) {
if err := m.stateManager.UpdateState((*SelectorState)(m.routeSelector)); err != nil {
log.Errorf("failed to update state: %v", err)
}
// A selection change flips Network.selected without altering the candidate
// set, so bump the revision to push the new state to the UI.
m.statusRecorder.BumpNetworksRevision()
}
// stopObsoleteClients stops the client network watcher for the networks that are not in the new list

View File

@@ -859,3 +859,31 @@ func TestRouteSelector_ComplexScenarios(t *testing.T) {
})
}
}
// TestRouteSelector_EnableExitNodeKeepsOtherRoutes is a regression test for the
// tray exit-node toggle disabling every non-exit routed network. The tray used
// to Select an exit node with append=false, which the RouteSelector treats as
// "drop the whole current selection" (default-on semantics) — so enabling an
// exit node also turned off every LAN/route the user had on. The fix sends
// append=true and lets the daemon's SelectNetworks handler deselect only the
// sibling exit nodes. This test models that handler sequence against the
// selector: SelectRoutes(exit, append=true) followed by DeselectRoutes(other
// exit nodes) must leave non-exit routes untouched.
func TestRouteSelector_EnableExitNodeKeepsOtherRoutes(t *testing.T) {
rs := routeselector.NewRouteSelector()
all := []route.NetID{"exitA", "exitB", "lan1", "lan2"}
// User has two LAN routes on (default-on: nothing deselected => all selected).
require.True(t, rs.IsSelected("lan1"))
require.True(t, rs.IsSelected("lan2"))
// Tray enables exitA: SelectNetworks handler does SelectRoutes(append=true)
// then deselects sibling exit nodes (exitB), never the LAN routes.
require.NoError(t, rs.SelectRoutes([]route.NetID{"exitA"}, true, all))
require.NoError(t, rs.DeselectRoutes([]route.NetID{"exitB"}, all))
assert.True(t, rs.IsSelected("exitA"), "selected exit node stays on")
assert.False(t, rs.IsSelected("exitB"), "sibling exit node is deselected")
assert.True(t, rs.IsSelected("lan1"), "non-exit route must stay selected")
assert.True(t, rs.IsSelected("lan2"), "non-exit route must stay selected")
}

View File

@@ -33,17 +33,34 @@ func CtxGetState(ctx context.Context) *contextState {
}
type contextState struct {
err error
status StatusType
mutex sync.Mutex
err error
status StatusType
mutex sync.Mutex
onChange func()
}
// SetOnChange installs a callback fired after every successful Set. Used by
// the daemon to wire the status recorder's notifyStateChange so any
// state.Set in the connect/login paths pushes a fresh snapshot to
// SubscribeStatus subscribers without each callsite having to opt in.
// The callback runs outside the contextState mutex to avoid a lock-order
// dependency with the recorder's stateChangeMux.
func (c *contextState) SetOnChange(fn func()) {
c.mutex.Lock()
c.onChange = fn
c.mutex.Unlock()
}
func (c *contextState) Set(update StatusType) {
c.mutex.Lock()
defer c.mutex.Unlock()
c.status = update
c.err = nil
cb := c.onChange
c.mutex.Unlock()
if cb != nil {
cb()
}
}
func (c *contextState) Status() (StatusType, error) {
@@ -57,6 +74,17 @@ func (c *contextState) Status() (StatusType, error) {
return c.status, nil
}
// CurrentStatus returns the last status set via Set, ignoring any wrapped
// error. Use when the status is needed for reporting purposes (e.g. the
// status snapshot stream) and a transient wrapped error from a retry loop
// shouldn't blank out the underlying status.
func (c *contextState) CurrentStatus() StatusType {
c.mutex.Lock()
defer c.mutex.Unlock()
return c.status
}
func (c *contextState) Wrap(err error) error {
c.mutex.Lock()
defer c.mutex.Unlock()

View File

@@ -1,6 +1,7 @@
package statemanager
import (
"bytes"
"context"
"encoding/json"
"errors"
@@ -305,6 +306,11 @@ func (m *Manager) loadStateFile(deleteCorrupt bool) (map[string]json.RawMessage,
var rawStates map[string]json.RawMessage
if err := json.Unmarshal(data, &rawStates); err != nil {
if len(bytes.TrimSpace(data)) == 0 {
log.Warnf("state file %s is empty (%d bytes)", m.filePath, len(data))
} else {
log.Warnf("state file %s has malformed content (%d bytes)", m.filePath, len(data))
}
m.handleCorruptedState(deleteCorrupt)
return nil, fmt.Errorf("unmarshal states: %w", err)
}