mirror of
https://github.com/netbirdio/netbird.git
synced 2026-07-30 04:11:28 +02:00
Compare commits
2 Commits
feature/an
...
fix/grpc-g
| Author | SHA1 | Date | |
|---|---|---|---|
|
ed58cfc63a | ||
|
|
63c320b6a9 |
@@ -75,24 +75,6 @@ type Client struct {
|
||||
connectClient *internal.ConnectClient
|
||||
config *profilemanager.Config
|
||||
cacheDir string
|
||||
|
||||
stateChangeMu sync.Mutex
|
||||
stateChangeSubID string
|
||||
eventSub *peer.EventSubscription
|
||||
// Closed to stop the watch goroutines from delivering buffered items to a
|
||||
// listener that has been removed or replaced. See stopStateChangeWatchLocked.
|
||||
stateChangeDone chan struct{}
|
||||
|
||||
// Latched "the server wants an interactive login": survives the engine
|
||||
// restarts that replace the run loop's context state. See Client.Status.
|
||||
// Guarded by loginRequiredMu together with loginCleared, which counts
|
||||
// clears so a stale observation cannot re-latch over one.
|
||||
loginRequiredMu sync.Mutex
|
||||
loginRequired bool
|
||||
loginCleared uint64
|
||||
|
||||
extendMu sync.Mutex
|
||||
extendCancel context.CancelFunc
|
||||
}
|
||||
|
||||
func (c *Client) setState(cfg *profilemanager.Config, cacheDir string, cc *internal.ConnectClient) {
|
||||
@@ -166,16 +148,11 @@ func (c *Client) Run(platformFiles PlatformFiles, urlOpener URLOpener, isAndroid
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// todo do not throw error in case of cancelled context
|
||||
ctx = internal.CtxInitState(ctx)
|
||||
connectClient := internal.NewConnectClient(ctx, cfg, c.recorder)
|
||||
c.setState(cfg, cacheDir, connectClient)
|
||||
// This path runs the interactive SSO flow, so reaching here means the peer
|
||||
// is authenticated again — release the latch Status() reports from. Clear
|
||||
// only once the fresh connect client is installed: until then Status()
|
||||
// still reads the previous run's context state, which holds the NeedsLogin
|
||||
// that prompted this login, and would re-latch what was just cleared.
|
||||
c.clearLoginRequired()
|
||||
return connectClient.RunOnAndroid(c.tunAdapter, c.iFaceDiscover, c.networkChangeListener, slices.Clone(dns.items), dnsReadyListener, stateFile, cacheDir)
|
||||
}
|
||||
|
||||
|
||||
@@ -1,309 +0,0 @@
|
||||
//go:build android
|
||||
|
||||
package android
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal"
|
||||
"github.com/netbirdio/netbird/client/internal/auth"
|
||||
"github.com/netbirdio/netbird/client/internal/auth/sessionwatch"
|
||||
"github.com/netbirdio/netbird/client/internal/peer"
|
||||
cProto "github.com/netbirdio/netbird/client/proto"
|
||||
)
|
||||
|
||||
// StateChangeListener receives client state notifications.
|
||||
//
|
||||
// OnStateChanged is a payload-free wake-up whenever the state snapshot
|
||||
// changed: connection state, the run-loop status label (e.g. NeedsLogin) or
|
||||
// the session deadline. It mirrors the daemon's SubscribeStatus stream
|
||||
// trigger — on each signal the consumer pulls the fresh values via
|
||||
// Status() / SessionExpiresAtUnix().
|
||||
//
|
||||
// OnSessionExpiring forwards the engine's session-expiry warnings, fired at
|
||||
// sessionwatch.WarningLead before the deadline and again at FinalWarningLead
|
||||
// (finalWarning true). The second one is suppressed when the user dismissed
|
||||
// the first via DismissSessionWarning. The daemon turns the same events into
|
||||
// its tray notification.
|
||||
type StateChangeListener interface {
|
||||
OnStateChanged()
|
||||
OnSessionExpiring(expiresAtUnix int64, leadMinutes int64, finalWarning bool)
|
||||
}
|
||||
|
||||
// Status returns the connect run-loop's status label — the same value the
|
||||
// desktop daemon serves in StatusResponse.Status. "NeedsLogin" means the
|
||||
// management server rejected the peer and an interactive login is required.
|
||||
//
|
||||
// The label is latched: the run loop keeps its status in a per-run context
|
||||
// state, which a restart replaces with a fresh Idle one, so an engine restart
|
||||
// (network change, always-on) would otherwise erase the fact that the peer
|
||||
// still needs to log in. Only a successful interactive login or extend clears
|
||||
// it — see clearLoginRequired.
|
||||
func (c *Client) Status() string {
|
||||
latched, generation := c.loginRequiredState()
|
||||
if latched {
|
||||
return string(internal.StatusNeedsLogin)
|
||||
}
|
||||
cc := c.getConnectClient()
|
||||
if cc == nil {
|
||||
return string(internal.StatusIdle)
|
||||
}
|
||||
status := cc.Status()
|
||||
if status == internal.StatusNeedsLogin {
|
||||
c.latchLoginRequired(generation)
|
||||
}
|
||||
return string(status)
|
||||
}
|
||||
|
||||
func (c *Client) loginRequiredState() (bool, uint64) {
|
||||
c.loginRequiredMu.Lock()
|
||||
defer c.loginRequiredMu.Unlock()
|
||||
return c.loginRequired, c.loginCleared
|
||||
}
|
||||
|
||||
// latchLoginRequired records a NeedsLogin observation, unless a clear landed
|
||||
// while the caller was reading the run loop's status: cc.Status() is read
|
||||
// outside the lock, so a login or extend completing in that window would
|
||||
// otherwise be undone by this stale observation, stranding the UI on
|
||||
// "login required" over a healthy session.
|
||||
func (c *Client) latchLoginRequired(observedGeneration uint64) {
|
||||
c.loginRequiredMu.Lock()
|
||||
defer c.loginRequiredMu.Unlock()
|
||||
if c.loginCleared != observedGeneration {
|
||||
return
|
||||
}
|
||||
c.loginRequired = true
|
||||
}
|
||||
|
||||
// clearLoginRequired releases the latch after a successful interactive login
|
||||
// or session extend, and invalidates any observation already in flight.
|
||||
func (c *Client) clearLoginRequired() {
|
||||
c.loginRequiredMu.Lock()
|
||||
defer c.loginRequiredMu.Unlock()
|
||||
c.loginRequired = false
|
||||
c.loginCleared++
|
||||
}
|
||||
|
||||
// SessionExpiresAtUnix returns the SSO session deadline as unix seconds, or 0
|
||||
// when no deadline is known (not SSO-registered, expiry disabled, or the
|
||||
// engine has not received one yet). A past value means the session expired.
|
||||
// Mirror of StatusResponse.sessionExpiresAt on the desktop daemon.
|
||||
func (c *Client) SessionExpiresAtUnix() int64 {
|
||||
deadline := c.recorder.GetSessionExpiresAt()
|
||||
if deadline.IsZero() {
|
||||
return 0
|
||||
}
|
||||
return deadline.Unix()
|
||||
}
|
||||
|
||||
// SetStateChangeListener registers the state notification listener.
|
||||
// Replaces any previously registered listener; remove it with
|
||||
// RemoveStateChangeListener.
|
||||
func (c *Client) SetStateChangeListener(listener StateChangeListener) {
|
||||
c.stateChangeMu.Lock()
|
||||
defer c.stateChangeMu.Unlock()
|
||||
c.stopStateChangeWatchLocked()
|
||||
if listener == nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Both subscriptions are buffered (one pending tick, ten pending events),
|
||||
// so unsubscribing is not enough to stop callbacks: the loops would drain
|
||||
// what is already queued and deliver it to a listener the caller has
|
||||
// already removed or replaced. Gate every callback on this registration's
|
||||
// own signal, which is closed before unsubscribing.
|
||||
done := make(chan struct{})
|
||||
c.stateChangeDone = done
|
||||
|
||||
id, ch := c.recorder.SubscribeToStateChanges()
|
||||
c.stateChangeSubID = id
|
||||
// The channel is closed by UnsubscribeFromStateChanges, which ends the
|
||||
// goroutine. Ticks are coalesced (buffer of one), so a burst of changes
|
||||
// wakes the listener once.
|
||||
go func() {
|
||||
for range ch {
|
||||
select {
|
||||
case <-done:
|
||||
return
|
||||
default:
|
||||
}
|
||||
listener.OnStateChanged()
|
||||
}
|
||||
}()
|
||||
|
||||
c.eventSub = c.recorder.SubscribeToEvents()
|
||||
go watchSessionWarnings(c.eventSub, listener, done)
|
||||
}
|
||||
|
||||
// RemoveStateChangeListener unregisters the state notification listener.
|
||||
func (c *Client) RemoveStateChangeListener() {
|
||||
c.stateChangeMu.Lock()
|
||||
defer c.stateChangeMu.Unlock()
|
||||
c.stopStateChangeWatchLocked()
|
||||
}
|
||||
|
||||
// DismissSessionWarning records the user's "Dismiss" on the first expiry
|
||||
// warning and suppresses the final one for the current deadline. A refreshed
|
||||
// deadline re-arms both. No-op while the engine is not running.
|
||||
func (c *Client) DismissSessionWarning() {
|
||||
cc := c.getConnectClient()
|
||||
if cc == nil {
|
||||
return
|
||||
}
|
||||
engine := cc.Engine()
|
||||
if engine == nil {
|
||||
return
|
||||
}
|
||||
engine.DismissSessionWarning()
|
||||
}
|
||||
|
||||
// ExtendAuthSession runs the interactive SSO flow to obtain a fresh JWT and
|
||||
// asks the management server to extend the session deadline. The tunnel is
|
||||
// untouched: no resync, no reconnect. Async; the result arrives on the
|
||||
// listener. Mirror of the daemon's RequestExtendAuthSession /
|
||||
// WaitExtendAuthSession RPC pair, with URLOpener playing the "UI opens the
|
||||
// browser" role.
|
||||
//
|
||||
// Only one flow may be in flight: the PKCE step binds a fixed loopback port,
|
||||
// so a second concurrent flow would fail on that bind. Call
|
||||
// CancelExtendAuthSession when the user abandons the browser.
|
||||
func (c *Client) ExtendAuthSession(urlOpener URLOpener, isAndroidTV bool, resultListener ErrListener) {
|
||||
ctx, err := c.beginExtend()
|
||||
if err != nil {
|
||||
resultListener.OnError(err)
|
||||
return
|
||||
}
|
||||
|
||||
go func() {
|
||||
defer c.endExtend()
|
||||
if err := c.extendAuthSession(ctx, urlOpener, isAndroidTV); err != nil {
|
||||
resultListener.OnError(err)
|
||||
return
|
||||
}
|
||||
resultListener.OnSuccess()
|
||||
}()
|
||||
}
|
||||
|
||||
// CancelExtendAuthSession aborts an in-flight ExtendAuthSession. The tunnel is
|
||||
// left alone — unlike the login flow, which cancels the whole client context
|
||||
// by stopping the engine. Without this the abandoned PKCE wait keeps its
|
||||
// loopback port for the full flow timeout and blocks every later attempt.
|
||||
// No-op when no flow is running.
|
||||
func (c *Client) CancelExtendAuthSession() {
|
||||
c.extendMu.Lock()
|
||||
defer c.extendMu.Unlock()
|
||||
if c.extendCancel != nil {
|
||||
c.extendCancel()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) stopStateChangeWatchLocked() {
|
||||
// Signal first, unsubscribe second: closing the channels only stops new
|
||||
// items, and the loops would still hand whatever is buffered to a listener
|
||||
// that is no longer registered.
|
||||
if c.stateChangeDone != nil {
|
||||
close(c.stateChangeDone)
|
||||
c.stateChangeDone = nil
|
||||
}
|
||||
if c.stateChangeSubID != "" {
|
||||
c.recorder.UnsubscribeFromStateChanges(c.stateChangeSubID)
|
||||
c.stateChangeSubID = ""
|
||||
}
|
||||
if c.eventSub != nil {
|
||||
// Closes the channel, which ends watchSessionWarnings.
|
||||
c.recorder.UnsubscribeFromEvents(c.eventSub)
|
||||
c.eventSub = nil
|
||||
}
|
||||
}
|
||||
|
||||
// watchSessionWarnings forwards the engine's session-expiry warnings to the
|
||||
// listener. The event stream also carries unrelated traffic — network-map
|
||||
// updates on every sync, DNS and route errors — so everything but an
|
||||
// AUTHENTICATION event carrying the session-warning marker is dropped. Exits
|
||||
// when the subscription is closed by UnsubscribeFromEvents, or earlier when
|
||||
// done is closed — the stream buffers up to ten events, and a deregistered
|
||||
// listener must not receive the ones already queued.
|
||||
func watchSessionWarnings(sub *peer.EventSubscription, listener StateChangeListener, done <-chan struct{}) {
|
||||
for ev := range sub.Events() {
|
||||
select {
|
||||
case <-done:
|
||||
return
|
||||
default:
|
||||
}
|
||||
if ev.GetCategory() != cProto.SystemEvent_AUTHENTICATION {
|
||||
continue
|
||||
}
|
||||
meta := ev.GetMetadata()
|
||||
if meta[sessionwatch.MetaSessionWarning] != "true" {
|
||||
// Other AUTHENTICATION events exist (e.g. a deadline rejected as
|
||||
// out of range); they carry no warning marker.
|
||||
continue
|
||||
}
|
||||
deadline, err := sessionwatch.ParseExpiresAt(meta[sessionwatch.MetaSessionExpiresAt])
|
||||
if err != nil {
|
||||
log.Warnf("session warning event with unparsable deadline: %v", err)
|
||||
continue
|
||||
}
|
||||
lead, err := sessionwatch.ParseLeadMinutes(meta[sessionwatch.MetaSessionLeadMinutes])
|
||||
if err != nil {
|
||||
// Informational only — the deadline above is what drives the UI.
|
||||
lead = 0
|
||||
}
|
||||
listener.OnSessionExpiring(deadline.Unix(), int64(lead),
|
||||
meta[sessionwatch.MetaSessionFinal] == "true")
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) beginExtend() (context.Context, error) {
|
||||
c.extendMu.Lock()
|
||||
defer c.extendMu.Unlock()
|
||||
if c.extendCancel != nil {
|
||||
return nil, fmt.Errorf("session extend already in progress")
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
c.extendCancel = cancel
|
||||
return ctx, nil
|
||||
}
|
||||
|
||||
func (c *Client) endExtend() {
|
||||
c.extendMu.Lock()
|
||||
defer c.extendMu.Unlock()
|
||||
if c.extendCancel != nil {
|
||||
c.extendCancel()
|
||||
c.extendCancel = nil
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) extendAuthSession(ctx context.Context, urlOpener URLOpener, isAndroidTV bool) error {
|
||||
cfg, _, cc := c.stateSnapshot()
|
||||
if cfg == nil || cc == nil {
|
||||
return fmt.Errorf("engine is not running")
|
||||
}
|
||||
engine := cc.Engine()
|
||||
if engine == nil {
|
||||
return fmt.Errorf("engine is not initialized")
|
||||
}
|
||||
|
||||
authClient, err := auth.NewAuth(ctx, cfg.PrivateKey, cfg.ManagementURL, cfg)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create auth client: %v", err)
|
||||
}
|
||||
defer authClient.Close()
|
||||
|
||||
a := &Auth{ctx: ctx, config: cfg}
|
||||
tokenInfo, err := a.foregroundGetTokenInfo(authClient, urlOpener, isAndroidTV)
|
||||
if err != nil {
|
||||
return fmt.Errorf("interactive sso login failed: %v", err)
|
||||
}
|
||||
|
||||
if _, err := engine.ExtendAuthSession(ctx, tokenInfo.GetTokenToUse()); err != nil {
|
||||
return err
|
||||
}
|
||||
c.clearLoginRequired()
|
||||
|
||||
go urlOpener.OnLoginSuccess()
|
||||
return nil
|
||||
}
|
||||
@@ -34,6 +34,7 @@ import (
|
||||
"github.com/netbirdio/netbird/client/internal/profilemanager"
|
||||
"github.com/netbirdio/netbird/client/internal/statemanager"
|
||||
"github.com/netbirdio/netbird/client/internal/stdnet"
|
||||
"github.com/netbirdio/netbird/client/internal/tunnelnotifier"
|
||||
"github.com/netbirdio/netbird/client/internal/updater"
|
||||
"github.com/netbirdio/netbird/client/internal/updater/installer"
|
||||
nbnet "github.com/netbirdio/netbird/client/net"
|
||||
@@ -136,10 +137,13 @@ func (c *ConnectClient) RunOniOS(
|
||||
// Set GC percent to 5% to reduce memory usage as iOS only allows 50MB of memory for the extension.
|
||||
debug.SetGCPercent(5)
|
||||
|
||||
notifier := tunnelnotifier.New(networkChangeListener, dnsManager)
|
||||
defer notifier.Close()
|
||||
|
||||
mobileDependency := MobileDependency{
|
||||
FileDescriptor: fileDescriptor,
|
||||
NetworkChangeListener: networkChangeListener,
|
||||
DnsManager: dnsManager,
|
||||
NetworkChangeListener: notifier,
|
||||
DnsManager: notifier,
|
||||
StateFilePath: stateFilePath,
|
||||
TempDir: cacheDir,
|
||||
}
|
||||
|
||||
@@ -11,12 +11,14 @@ import (
|
||||
|
||||
// MobileDependency collect all dependencies for mobile platform
|
||||
type MobileDependency struct {
|
||||
// Android only
|
||||
TunAdapter device.TunAdapter
|
||||
IFaceDiscover stdnet.ExternalIFaceDiscover
|
||||
// Android and iOS
|
||||
NetworkChangeListener listener.NetworkChangeListener
|
||||
HostDNSAddresses []netip.AddrPort
|
||||
DnsReadyListener dns.ReadyListener
|
||||
|
||||
// Android only
|
||||
TunAdapter device.TunAdapter
|
||||
IFaceDiscover stdnet.ExternalIFaceDiscover
|
||||
HostDNSAddresses []netip.AddrPort
|
||||
DnsReadyListener dns.ReadyListener
|
||||
|
||||
// iOS only
|
||||
DnsManager dns.IosDnsManager
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
package notifier
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
"net/netip"
|
||||
"slices"
|
||||
"sort"
|
||||
@@ -16,20 +15,12 @@ import (
|
||||
|
||||
type Notifier struct {
|
||||
mu sync.Mutex
|
||||
cond *sync.Cond
|
||||
currentPrefixes []string
|
||||
listener listener.NetworkChangeListener
|
||||
queue *list.List
|
||||
closed bool
|
||||
}
|
||||
|
||||
func NewNotifier() *Notifier {
|
||||
n := &Notifier{
|
||||
queue: list.New(),
|
||||
}
|
||||
n.cond = sync.NewCond(&n.mu)
|
||||
go n.deliverLoop()
|
||||
return n
|
||||
return &Notifier{}
|
||||
}
|
||||
|
||||
func (n *Notifier) SetListener(listener listener.NetworkChangeListener) {
|
||||
@@ -59,44 +50,19 @@ func (n *Notifier) OnNewPrefixes(prefixes []netip.Prefix) {
|
||||
sort.Strings(newNets)
|
||||
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
if slices.Equal(n.currentPrefixes, newNets) {
|
||||
n.mu.Unlock()
|
||||
return
|
||||
}
|
||||
n.currentPrefixes = newNets
|
||||
routes := strings.Join(n.currentPrefixes, ",")
|
||||
n.queue.PushBack(routes)
|
||||
n.cond.Signal()
|
||||
n.mu.Unlock()
|
||||
if n.listener != nil {
|
||||
n.listener.OnNetworkChanged(strings.Join(n.currentPrefixes, ","))
|
||||
}
|
||||
}
|
||||
|
||||
func (n *Notifier) Close() {
|
||||
n.mu.Lock()
|
||||
n.closed = true
|
||||
n.cond.Signal()
|
||||
n.mu.Unlock()
|
||||
}
|
||||
|
||||
func (n *Notifier) GetInitialRouteRanges() []string {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (n *Notifier) deliverLoop() {
|
||||
for {
|
||||
n.mu.Lock()
|
||||
for n.queue.Len() == 0 && !n.closed {
|
||||
n.cond.Wait()
|
||||
}
|
||||
if n.closed && n.queue.Len() == 0 {
|
||||
n.mu.Unlock()
|
||||
return
|
||||
}
|
||||
routes := n.queue.Remove(n.queue.Front()).(string)
|
||||
l := n.listener
|
||||
n.mu.Unlock()
|
||||
|
||||
if l != nil {
|
||||
l.OnNetworkChanged(routes)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
124
client/internal/tunnelnotifier/notifier.go
Normal file
124
client/internal/tunnelnotifier/notifier.go
Normal file
@@ -0,0 +1,124 @@
|
||||
package tunnelnotifier
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
"sync"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/dns"
|
||||
"github.com/netbirdio/netbird/client/internal/listener"
|
||||
)
|
||||
|
||||
type eventKind int
|
||||
|
||||
const (
|
||||
eventRoutes eventKind = iota
|
||||
eventIfaceIP
|
||||
eventIfaceIPv6
|
||||
eventDNS
|
||||
)
|
||||
|
||||
var (
|
||||
_ listener.NetworkChangeListener = (*Notifier)(nil)
|
||||
_ dns.IosDnsManager = (*Notifier)(nil)
|
||||
)
|
||||
|
||||
type event struct {
|
||||
kind eventKind
|
||||
payload string
|
||||
}
|
||||
|
||||
type Notifier struct {
|
||||
mu sync.Mutex
|
||||
cond *sync.Cond
|
||||
queue *list.List
|
||||
closed bool
|
||||
done chan struct{}
|
||||
|
||||
listener listener.NetworkChangeListener
|
||||
dnsManager dns.IosDnsManager
|
||||
}
|
||||
|
||||
func New(l listener.NetworkChangeListener, dm dns.IosDnsManager) *Notifier {
|
||||
n := &Notifier{
|
||||
queue: list.New(),
|
||||
done: make(chan struct{}),
|
||||
listener: l,
|
||||
dnsManager: dm,
|
||||
}
|
||||
n.cond = sync.NewCond(&n.mu)
|
||||
go n.deliverLoop()
|
||||
return n
|
||||
}
|
||||
|
||||
func (n *Notifier) OnNetworkChanged(routes string) {
|
||||
n.enqueue(event{kind: eventRoutes, payload: routes})
|
||||
}
|
||||
|
||||
func (n *Notifier) SetInterfaceIP(ip string) {
|
||||
n.enqueue(event{kind: eventIfaceIP, payload: ip})
|
||||
}
|
||||
|
||||
func (n *Notifier) SetInterfaceIPv6(ip string) {
|
||||
n.enqueue(event{kind: eventIfaceIPv6, payload: ip})
|
||||
}
|
||||
|
||||
func (n *Notifier) ApplyDns(config string) {
|
||||
n.enqueue(event{kind: eventDNS, payload: config})
|
||||
}
|
||||
|
||||
// Close stops accepting new events and blocks until the delivery loop has
|
||||
// drained all queued events and exited.
|
||||
func (n *Notifier) Close() {
|
||||
n.mu.Lock()
|
||||
n.closed = true
|
||||
n.cond.Signal()
|
||||
n.mu.Unlock()
|
||||
<-n.done
|
||||
}
|
||||
|
||||
func (n *Notifier) enqueue(ev event) {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
if n.closed {
|
||||
return
|
||||
}
|
||||
n.queue.PushBack(ev)
|
||||
n.cond.Signal()
|
||||
}
|
||||
|
||||
func (n *Notifier) deliverLoop() {
|
||||
defer close(n.done)
|
||||
for {
|
||||
n.mu.Lock()
|
||||
for n.queue.Len() == 0 && !n.closed {
|
||||
n.cond.Wait()
|
||||
}
|
||||
if n.closed && n.queue.Len() == 0 {
|
||||
n.mu.Unlock()
|
||||
return
|
||||
}
|
||||
ev := n.queue.Remove(n.queue.Front()).(event)
|
||||
l := n.listener
|
||||
dm := n.dnsManager
|
||||
n.mu.Unlock()
|
||||
|
||||
switch ev.kind {
|
||||
case eventRoutes:
|
||||
if l != nil {
|
||||
l.OnNetworkChanged(ev.payload)
|
||||
}
|
||||
case eventIfaceIP:
|
||||
if l != nil {
|
||||
l.SetInterfaceIP(ev.payload)
|
||||
}
|
||||
case eventIfaceIPv6:
|
||||
if l != nil {
|
||||
l.SetInterfaceIPv6(ev.payload)
|
||||
}
|
||||
case eventDNS:
|
||||
if dm != nil {
|
||||
dm.ApplyDns(ev.payload)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
192
client/internal/tunnelnotifier/notifier_test.go
Normal file
192
client/internal/tunnelnotifier/notifier_test.go
Normal file
@@ -0,0 +1,192 @@
|
||||
package tunnelnotifier
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type call struct {
|
||||
kind string
|
||||
payload string
|
||||
}
|
||||
|
||||
type recorder struct {
|
||||
mu sync.Mutex
|
||||
calls []call
|
||||
inFlight atomic.Int32
|
||||
overlap atomic.Bool
|
||||
delay time.Duration
|
||||
}
|
||||
|
||||
func (r *recorder) record(kind, payload string) {
|
||||
if r.inFlight.Add(1) != 1 {
|
||||
r.overlap.Store(true)
|
||||
}
|
||||
if r.delay > 0 {
|
||||
time.Sleep(r.delay)
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.calls = append(r.calls, call{kind: kind, payload: payload})
|
||||
r.mu.Unlock()
|
||||
r.inFlight.Add(-1)
|
||||
}
|
||||
|
||||
func (r *recorder) count() int {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return len(r.calls)
|
||||
}
|
||||
|
||||
func (r *recorder) snapshot() []call {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
out := make([]call, len(r.calls))
|
||||
copy(out, r.calls)
|
||||
return out
|
||||
}
|
||||
|
||||
type fakeListener struct {
|
||||
rec *recorder
|
||||
}
|
||||
|
||||
func (f *fakeListener) OnNetworkChanged(routes string) {
|
||||
f.rec.record("routes", routes)
|
||||
}
|
||||
|
||||
func (f *fakeListener) SetInterfaceIP(ip string) {
|
||||
f.rec.record("ip", ip)
|
||||
}
|
||||
|
||||
func (f *fakeListener) SetInterfaceIPv6(ip string) {
|
||||
f.rec.record("ipv6", ip)
|
||||
}
|
||||
|
||||
type fakeDNSManager struct {
|
||||
rec *recorder
|
||||
}
|
||||
|
||||
func (f *fakeDNSManager) ApplyDns(config string) {
|
||||
f.rec.record("dns", config)
|
||||
}
|
||||
|
||||
func TestFIFOOrder(t *testing.T) {
|
||||
rec := &recorder{}
|
||||
n := New(&fakeListener{rec: rec}, &fakeDNSManager{rec: rec})
|
||||
defer n.Close()
|
||||
|
||||
n.SetInterfaceIP("10.0.0.1")
|
||||
n.SetInterfaceIPv6("fd00::1")
|
||||
n.ApplyDns(`{"domains":[]}`)
|
||||
n.OnNetworkChanged("10.0.0.0/8,192.168.0.0/16")
|
||||
n.ApplyDns(`{"domains":["example.com"]}`)
|
||||
|
||||
require.Eventually(t, func() bool { return rec.count() == 5 }, time.Second, time.Millisecond)
|
||||
|
||||
expected := []call{
|
||||
{kind: "ip", payload: "10.0.0.1"},
|
||||
{kind: "ipv6", payload: "fd00::1"},
|
||||
{kind: "dns", payload: `{"domains":[]}`},
|
||||
{kind: "routes", payload: "10.0.0.0/8,192.168.0.0/16"},
|
||||
{kind: "dns", payload: `{"domains":["example.com"]}`},
|
||||
}
|
||||
assert.Equal(t, expected, rec.snapshot())
|
||||
}
|
||||
|
||||
func TestNoOverlappingCalls(t *testing.T) {
|
||||
rec := &recorder{delay: 100 * time.Microsecond}
|
||||
n := New(&fakeListener{rec: rec}, &fakeDNSManager{rec: rec})
|
||||
defer n.Close()
|
||||
|
||||
const producers = 8
|
||||
const perProducer = 25
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < producers; i++ {
|
||||
wg.Add(1)
|
||||
go func(id int) {
|
||||
defer wg.Done()
|
||||
for j := 0; j < perProducer; j++ {
|
||||
payload := fmt.Sprintf("%d-%d", id, j)
|
||||
switch j % 4 {
|
||||
case 0:
|
||||
n.OnNetworkChanged(payload)
|
||||
case 1:
|
||||
n.SetInterfaceIP(payload)
|
||||
case 2:
|
||||
n.SetInterfaceIPv6(payload)
|
||||
case 3:
|
||||
n.ApplyDns(payload)
|
||||
}
|
||||
}
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
require.Eventually(t, func() bool { return rec.count() == producers*perProducer }, 5*time.Second, time.Millisecond)
|
||||
assert.False(t, rec.overlap.Load())
|
||||
}
|
||||
|
||||
func TestDNSAndRoutesInterleaved(t *testing.T) {
|
||||
rec := &recorder{delay: 100 * time.Microsecond}
|
||||
n := New(&fakeListener{rec: rec}, &fakeDNSManager{rec: rec})
|
||||
defer n.Close()
|
||||
|
||||
const events = 50
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < events; i++ {
|
||||
n.ApplyDns(fmt.Sprintf("dns-%d", i))
|
||||
}
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for i := 0; i < events; i++ {
|
||||
n.OnNetworkChanged(fmt.Sprintf("routes-%d", i))
|
||||
}
|
||||
}()
|
||||
wg.Wait()
|
||||
|
||||
require.Eventually(t, func() bool { return rec.count() == 2*events }, 5*time.Second, time.Millisecond)
|
||||
assert.False(t, rec.overlap.Load())
|
||||
|
||||
var dnsSeen, routesSeen int
|
||||
for _, c := range rec.snapshot() {
|
||||
switch c.kind {
|
||||
case "dns":
|
||||
assert.Equal(t, fmt.Sprintf("dns-%d", dnsSeen), c.payload)
|
||||
dnsSeen++
|
||||
case "routes":
|
||||
assert.Equal(t, fmt.Sprintf("routes-%d", routesSeen), c.payload)
|
||||
routesSeen++
|
||||
}
|
||||
}
|
||||
assert.Equal(t, events, dnsSeen)
|
||||
assert.Equal(t, events, routesSeen)
|
||||
}
|
||||
|
||||
func TestCloseDrainsQueue(t *testing.T) {
|
||||
rec := &recorder{delay: time.Millisecond}
|
||||
n := New(&fakeListener{rec: rec}, &fakeDNSManager{rec: rec})
|
||||
|
||||
const events = 20
|
||||
for i := 0; i < events; i++ {
|
||||
n.OnNetworkChanged(fmt.Sprintf("routes-%d", i))
|
||||
}
|
||||
n.Close()
|
||||
|
||||
require.Equal(t, events, rec.count(), "Close must not return before all queued events are delivered")
|
||||
|
||||
n.OnNetworkChanged("after-close")
|
||||
n.ApplyDns("after-close")
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
assert.Equal(t, events, rec.count())
|
||||
}
|
||||
@@ -25,6 +25,7 @@ import (
|
||||
"github.com/netbirdio/netbird/encryption"
|
||||
"github.com/netbirdio/netbird/shared/management/domain"
|
||||
nbmgmtgrpc "github.com/netbirdio/netbird/shared/management/grpc"
|
||||
nbnetworkmap "github.com/netbirdio/netbird/shared/management/networkmap"
|
||||
"github.com/netbirdio/netbird/shared/management/proto"
|
||||
"github.com/netbirdio/netbird/util/wsproxy"
|
||||
)
|
||||
@@ -472,11 +473,48 @@ func (c *GrpcClient) GetNetworkMap(sysInfo *system.Info) (*proto.NetworkMap, err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if decryptedResp.GetNetworkMap() == nil {
|
||||
var nm *proto.NetworkMap
|
||||
if version := decryptedResp.GetVersion(); version == int32(nbmgmtgrpc.ComponentNetworkMap) {
|
||||
// Components-format peer: decode the envelope back to typed
|
||||
// components, run Calculate() locally, and convert to the wire
|
||||
// NetworkMap shape the rest of the engine consumes. Components are
|
||||
// retained so future incremental updates can apply deltas instead
|
||||
// of doing a full reconstruction.
|
||||
envelope := decryptedResp.GetNetworkMapEnvelope()
|
||||
if envelope == nil {
|
||||
return nil, fmt.Errorf("received a SyncReponse indicating use of components network map, but components are missing")
|
||||
}
|
||||
|
||||
localKey := c.key.String()
|
||||
dnsName := ""
|
||||
if pc := decryptedResp.GetPeerConfig(); pc != nil {
|
||||
// PeerConfig.Fqdn = "<dns_label>.<dns_domain>" — extract the
|
||||
// shared domain by stripping the peer's own label prefix. Falls
|
||||
// back to empty if the FQDN doesn't have the expected shape.
|
||||
dnsName = extractDNSDomainFromFQDN(pc.GetFqdn())
|
||||
}
|
||||
result, err := nbnetworkmap.EnvelopeToNetworkMap(c.ctx, envelope, localKey, dnsName)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("decode network map envelope: %w", err)
|
||||
}
|
||||
nm = result.NetworkMap
|
||||
} else {
|
||||
nm = decryptedResp.GetNetworkMap()
|
||||
}
|
||||
if nm == nil {
|
||||
return nil, fmt.Errorf("invalid msg, required network map")
|
||||
}
|
||||
|
||||
return decryptedResp.GetNetworkMap(), nil
|
||||
return nm, nil
|
||||
}
|
||||
|
||||
func extractDNSDomainFromFQDN(fqdn string) string {
|
||||
for i := 0; i < len(fqdn); i++ {
|
||||
if fqdn[i] == '.' && i+1 < len(fqdn) {
|
||||
return fqdn[i+1:]
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (c *GrpcClient) connectToSyncStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info) (proto.ManagementService_SyncClient, error) {
|
||||
|
||||
Reference in New Issue
Block a user