Compare commits

..

10 Commits

Author SHA1 Message Date
Theodor S. Midtlien
8a46353ae2 Add tests 2026-07-21 15:40:39 +02:00
Zoltán Papp
d663da8f82 Merge branch 'main' into refactor/relay-foreign-cache
# Conflicts:
#	shared/relay/client/manager.go
2026-07-21 11:27:09 +02:00
Zoltan Papp
cf101c44b4 [relay] Count winning attempt before draining losers
The success path in handleResult left settled at 0, so drainLoser
waited on a result that was already consumed and blocked forever,
leaking a goroutine and the results channel on every successful race.
Increment settled before stop() so drainLoser only waits for actual
started losers.
2026-07-02 12:22:01 +02:00
Zoltan Papp
79b51a79e4 [relay] Release home relay lock before the connection race
OpenConn held relayClientMu for the whole call, including the network-bound
FallbackOpener.Run, which blocked onServerDisconnected/storeClient from taking
the write lock and delayed reconnect by up to the race timeout. Snapshot the
home relay client under the read lock, release it, and run the foreign check
and the race on the snapshot. isForeignServer now takes the client so it uses
the snapshot instead of re-reading the field after the lock is dropped.
2026-07-01 19:22:41 +02:00
Zoltan Papp
a737504ec9 Fix timeout 2026-07-01 17:11:23 +02:00
Zoltan Papp
3b1beb3497 [relay] Return raceOutcome struct from handleResult to avoid nilnil
handleResult returned (net.Conn, error, bool) and produced (nil, nil, false)
on the non-terminal path, which the nilnil linter rejects. Return a
raceOutcome{conn, err, done} struct instead so the not-done case carries no
ambiguous nil value / nil error pair.
2026-07-01 17:05:46 +02:00
Zoltan Papp
49c0aeb6ce [relay] Update relay client tests for the new OpenConn signature
Manager.OpenConn now takes a RelayServer struct and a preferForeign
flag instead of a server address string and a serverIP. Adjust the
manager tests accordingly: bundle addr/IP into RelayServer and set
preferForeign=true for foreign-relay dials, false for home-relay dials.
2026-07-01 17:00:20 +02:00
Zoltan Papp
6774a43eae [relay] Rename ConnRacer to FallbackOpener and align file names with types
Rename the ConnRacer type to FallbackOpener and its constructor to
NewFallbackOpener. Rename fallback.go to fallback_opener.go and
foreign_relays.go to foreign_relays_store.go so file names mirror their
primary struct.
2026-07-01 16:53:42 +02:00
Zoltan Papp
50a29c07ce [relay] Race home and foreign relay for peer connection setup
When a peer advertises a relay different from the home relay, open the
peer connection by racing the home relay and the remote peer's relay in
parallel: start the preferred one (controller prefers home, otherwise the
remote relay), fall back to the other after a delay, use whichever connects
first and drain the loser.

Expose the foreign relay cache as ForeignRelaysStore and the race as
ConnRacer. The manager passes the remote relay as a RelayServer{Addr, IP}.
OpenConn no longer returns the winning server address; the worker derives it
from the returned conn's RemoteAddr to register the close listener, matching
the previous main-branch behavior.
2026-06-30 18:26:13 +02:00
Zoltan Papp
7d8e20030b [relay] Extract foreign relay client cache into a dedicated type
Move the foreign-relay client cache out of Manager into a foreignRelays
type. Concurrent first-time connects to the same server are deduplicated
with singleflight, so the cache mutex is never held during a network
connect (removing the previous stall where a slow connect blocked all map
operations). A per-entry in-use refcount prevents the cleanup loop from
closing a client while a connection is being opened on it.

This drops RelayTrack, its per-track lock and the hand-over-hand locking
between the map lock and the track lock. The exported API is unchanged.
2026-06-29 00:41:47 +02:00
56 changed files with 994 additions and 1849 deletions

View File

@@ -34,8 +34,6 @@ const (
// - Handling connection establishment based on peer signaling
//
// The implementation is not thread-safe; it is protected by engine.syncMsgMux.
// The only exception is ActivatePeer, which is safe for concurrent use so the
// DNS warm-up path can call it without contending on the engine mutex.
type ConnMgr struct {
peerStore *peerstore.Store
statusRecorder *peer.Status
@@ -44,10 +42,6 @@ type ConnMgr struct {
rosenpassEnabled bool
lazyConnMgr *manager.Manager
// lazyConnMgrMu guards the lazyConnMgr pointer for readers outside the
// engine loop (ActivatePeer). Writers hold it in addition to
// engine.syncMsgMux; all other reads stay under engine.syncMsgMux only.
lazyConnMgrMu sync.RWMutex
wg sync.WaitGroup
lazyCtx context.Context
@@ -244,20 +238,12 @@ func (e *ConnMgr) RemovePeerConn(peerKey string) {
conn.Log.Infof("removed peer from lazy conn manager")
}
// ActivatePeer wakes an idle lazy connection. Unlike the rest of ConnMgr it is
// safe for concurrent use: the lazy manager pointer is read under lazyConnMgrMu
// and the manager itself is internally synchronized, so callers outside the
// engine loop (DNS warm-up) do not need engine.syncMsgMux.
func (e *ConnMgr) ActivatePeer(ctx context.Context, conn *peer.Conn) {
e.lazyConnMgrMu.RLock()
lazyConnMgr := e.lazyConnMgr
started := lazyConnMgr != nil && e.lazyCtxCancel != nil
e.lazyConnMgrMu.RUnlock()
if !started {
if !e.isStartedWithLazyMgr() {
return
}
if found := lazyConnMgr.ActivatePeer(conn.GetKey()); found {
if found := e.lazyConnMgr.ActivatePeer(conn.GetKey()); found {
if err := conn.Open(ctx); err != nil {
conn.Log.Errorf("failed to open connection: %v", err)
}
@@ -282,21 +268,16 @@ func (e *ConnMgr) Close() {
e.lazyCtxCancel()
e.wg.Wait()
e.lazyConnMgrMu.Lock()
e.lazyConnMgr = nil
e.lazyConnMgrMu.Unlock()
}
func (e *ConnMgr) initLazyManager(engineCtx context.Context) {
cfg := manager.Config{
InactivityThreshold: inactivityThresholdEnv(),
}
e.lazyConnMgrMu.Lock()
e.lazyConnMgr = manager.NewManager(cfg, engineCtx, e.peerStore, e.iface)
e.lazyCtx, e.lazyCtxCancel = context.WithCancel(engineCtx)
e.lazyConnMgrMu.Unlock()
e.wg.Add(1)
go func() {
@@ -335,10 +316,7 @@ func (e *ConnMgr) closeManager(ctx context.Context) {
e.lazyCtxCancel()
e.wg.Wait()
e.lazyConnMgrMu.Lock()
e.lazyConnMgr = nil
e.lazyConnMgrMu.Unlock()
for _, peerID := range e.peerStore.PeersPubKey() {
e.peerStore.PeerConnOpen(ctx, peerID)

View File

@@ -1,21 +1,10 @@
package internal
import (
"context"
"net"
"net/netip"
"os"
"sync"
"testing"
"time"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface/wgaddr"
"github.com/netbirdio/netbird/client/internal/lazyconn"
"github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/client/internal/peerstore"
"github.com/netbirdio/netbird/monotime"
)
func TestResolveLazyForce(t *testing.T) {
@@ -49,58 +38,3 @@ func TestResolveLazyForce(t *testing.T) {
})
}
}
type mockLazyWGIface struct{}
func (mockLazyWGIface) RemovePeer(string) error { return nil }
func (mockLazyWGIface) UpdatePeer(string, []netip.Prefix, time.Duration, *net.UDPAddr, *wgtypes.Key) error {
return nil
}
func (mockLazyWGIface) IsUserspaceBind() bool { return false }
func (mockLazyWGIface) Address() wgaddr.Address { return wgaddr.Address{} }
func (mockLazyWGIface) LastActivities() map[string]monotime.Time { return nil }
func (mockLazyWGIface) MTU() uint16 { return 1280 }
// TestConnMgr_ActivatePeerConcurrentWithLifecycle exercises ActivatePeer from
// non-engine goroutines (the DNS warm-up path) racing the manager lifecycle,
// which stays on the engine loop. Run with -race: it fails if ActivatePeer
// still requires engine.syncMsgMux for safety.
func TestConnMgr_ActivatePeerConcurrentWithLifecycle(t *testing.T) {
t.Setenv(lazyconn.EnvLazyConn, "on")
status := peer.NewRecorder("https://mgm")
store := peerstore.NewConnStore()
connMgr := NewConnMgr(&EngineConfig{}, status, store, mockLazyWGIface{})
conn := newTestPeerConn(t, "peerA")
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
connMgr.Start(ctx)
done := make(chan struct{})
var wg sync.WaitGroup
for range 4 {
wg.Add(1)
go func() {
defer wg.Done()
for {
select {
case <-done:
return
default:
connMgr.ActivatePeer(ctx, conn)
}
}
}()
}
// Let the activators spin against the started manager, then tear it down
// underneath them and let them spin against the stopped manager.
time.Sleep(100 * time.Millisecond)
connMgr.Close()
time.Sleep(50 * time.Millisecond)
close(done)
wg.Wait()
}

View File

@@ -6,7 +6,6 @@ import (
"fmt"
"net"
"net/netip"
"os"
"slices"
"strings"
"sync"
@@ -37,43 +36,7 @@ type resolver interface {
// record is left alone (it points at something outside our mesh, e.g.
// a non-peer upstream).
type PeerConnectivity interface {
IsConnectedByIP(ip netip.Addr) (known, connected bool)
}
// PeerActivator wakes lazy-connection peers on demand. The local resolver calls
// it with the tunnel IPs an answer points at, so a peer that is idle (lazily
// disconnected) starts connecting at DNS-resolution time rather than racing the
// client's first request packet. nil disables warm-up.
type PeerActivator interface {
// ActivatePeersByIP triggers wake-up for the peer(s) owning addrs and blocks
// until one is connected or ctx (a short per-query budget) expires. It is a
// fast no-op for unknown or already-connected addresses.
ActivatePeersByIP(ctx context.Context, addrs []netip.Addr)
}
const (
defaultLazyWarmupTimeout = 2 * time.Second
envLazyWarmupTimeout = "NB_DNS_LAZY_WARMUP_TIMEOUT"
)
// lazyWarmupTimeoutFromEnv returns the per-query budget for waking a
// lazy-connection peer a DNS answer points at. Tunable via
// NB_DNS_LAZY_WARMUP_TIMEOUT (a Go duration). Parsed once at construction time.
func lazyWarmupTimeoutFromEnv() time.Duration {
v := os.Getenv(envLazyWarmupTimeout)
if v == "" {
return defaultLazyWarmupTimeout
}
d, err := time.ParseDuration(v)
if err != nil {
log.Warnf("invalid %s value %q, using default %s: %v", envLazyWarmupTimeout, v, defaultLazyWarmupTimeout, err)
return defaultLazyWarmupTimeout
}
if d <= 0 {
log.Warnf("non-positive %s value %q, using default %s", envLazyWarmupTimeout, v, defaultLazyWarmupTimeout)
return defaultLazyWarmupTimeout
}
return d
IsConnectedByIP(ip string) (known, connected bool)
}
type Resolver struct {
@@ -88,12 +51,6 @@ type Resolver struct {
// filter and preserves the legacy "return whatever is registered"
// behaviour for callers that never wire a status source.
peerConn PeerConnectivity
// peerActivator, when non-nil, is called at resolution time to warm the
// lazy connection to the peer(s) an answer points at. nil disables warm-up.
peerActivator PeerActivator
// warmupTimeout is the per-query budget for the lazy-connection warm-up
// wait, resolved from the environment once at construction time.
warmupTimeout time.Duration
ctx context.Context
cancel context.CancelFunc
@@ -102,12 +59,11 @@ type Resolver struct {
func NewResolver() *Resolver {
ctx, cancel := context.WithCancel(context.Background())
return &Resolver{
records: make(map[dns.Question][]dns.RR),
domains: make(map[domain.Domain]struct{}),
zones: make(map[domain.Domain]bool),
warmupTimeout: lazyWarmupTimeoutFromEnv(),
ctx: ctx,
cancel: cancel,
records: make(map[dns.Question][]dns.RR),
domains: make(map[domain.Domain]struct{}),
zones: make(map[domain.Domain]bool),
ctx: ctx,
cancel: cancel,
}
}
@@ -120,14 +76,6 @@ func (d *Resolver) SetPeerConnectivity(p PeerConnectivity) {
d.peerConn = p
}
// SetPeerActivator wires the DNS-time lazy-connection warm-up. Pass nil to
// disable. Safe to call multiple times; the latest value wins.
func (d *Resolver) SetPeerActivator(a PeerActivator) {
d.mu.Lock()
defer d.mu.Unlock()
d.peerActivator = a
}
func (d *Resolver) MatchSubdomains() bool {
return true
}
@@ -174,9 +122,6 @@ func (d *Resolver) ServeDNS(w dns.ResponseWriter, r *dns.Msg) {
replyMessage.RecursionAvailable = true
result := d.lookupRecords(logger, question)
// Warm before filtering: activation flips a lazily-idle target to connected,
// which then lets it survive the disconnected-peer filter below.
d.warmLazyPeers(question, result.records)
result.records = d.filterDisconnectedPeerAnswers(logger, question, result.records)
replyMessage.Authoritative = !result.hasExternalData
replyMessage.Answer = result.records
@@ -550,8 +495,8 @@ func (d *Resolver) filterDisconnectedPeerAnswers(logger *log.Entry, question dns
kept := make([]dns.RR, 0, len(records))
var dropped int
for _, rr := range records {
ip, ok := extractRecordAddr(rr)
if !ok {
ip := extractRecordIP(rr)
if ip == "" {
kept = append(kept, rr)
continue
}
@@ -573,54 +518,22 @@ func (d *Resolver) filterDisconnectedPeerAnswers(logger *log.Entry, question dns
return kept
}
// warmLazyPeers triggers lazy-connection wake-up for the peers a resolved
// answer points at and waits briefly for one to connect, so the caller's first
// request doesn't race the connection establishment. Warm-up is scoped to
// match-only (non-authoritative) zones — the synthesized private-service zones
// and user-created zones whose records point at specific peers. The account's
// peer zone is authoritative, so plain peer-name lookups never trigger warm-up;
// otherwise resolving any peer's name would wake its idle connection, defeating
// laziness mesh-wide. No-op when no activator is wired (lazy connections
// disabled) or the answer carries no peer IPs.
func (d *Resolver) warmLazyPeers(question dns.Question, records []dns.RR) {
d.mu.RLock()
activator := d.peerActivator
var nonAuth, found bool
if activator != nil {
nonAuth, found = d.findZone(question.Name)
}
d.mu.RUnlock()
if activator == nil || !found || !nonAuth {
return
}
var addrs []netip.Addr
for _, rr := range records {
if addr, ok := extractRecordAddr(rr); ok {
addrs = append(addrs, addr)
}
}
if len(addrs) == 0 {
return
}
ctx, cancel := context.WithTimeout(d.ctx, d.warmupTimeout)
defer cancel()
activator.ActivatePeersByIP(ctx, addrs)
}
// extractRecordAddr returns the IP address carried by an A or AAAA record.
// ok is false for any other record type or a record with no address.
func extractRecordAddr(rr dns.RR) (netip.Addr, bool) {
// extractRecordIP returns the dotted-decimal / colon-hex IP carried by
// an A or AAAA record, or "" for any other record type.
func extractRecordIP(rr dns.RR) string {
switch r := rr.(type) {
case *dns.A:
addr, ok := netip.AddrFromSlice(r.A)
return addr.Unmap(), ok
if r.A == nil {
return ""
}
return r.A.String()
case *dns.AAAA:
addr, ok := netip.AddrFromSlice(r.AAAA)
return addr.Unmap(), ok
if r.AAAA == nil {
return ""
}
return r.AAAA.String()
}
return netip.Addr{}, false
return ""
}
// Update replaces all zones and their records

View File

@@ -37,8 +37,8 @@ type mockPeerConnectivity struct {
byIP map[string]struct{ known, connected bool }
}
func (m mockPeerConnectivity) IsConnectedByIP(ip netip.Addr) (known, connected bool) {
v, ok := m.byIP[ip.String()]
func (m mockPeerConnectivity) IsConnectedByIP(ip string) (known, connected bool) {
v, ok := m.byIP[ip]
if !ok {
return false, false
}

View File

@@ -1,172 +0,0 @@
package local
import (
"context"
"net"
"net/netip"
"sync"
"testing"
"time"
"github.com/miekg/dns"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/client/internal/dns/test"
nbdns "github.com/netbirdio/netbird/dns"
)
// recordingActivator records the addresses it was asked to warm and returns
// immediately, so ServeDNS is not blocked by the test.
type recordingActivator struct {
mu sync.Mutex
called bool
addrs []netip.Addr
}
func (r *recordingActivator) ActivatePeersByIP(_ context.Context, addrs []netip.Addr) {
r.mu.Lock()
defer r.mu.Unlock()
r.called = true
r.addrs = append(r.addrs, addrs...)
}
func serveA(t *testing.T, resolver *Resolver, name string) *dns.Msg {
t.Helper()
var resp *dns.Msg
w := &test.MockResponseWriter{WriteMsgFunc: func(m *dns.Msg) error { resp = m; return nil }}
resolver.ServeDNS(w, new(dns.Msg).SetQuestion(name, dns.TypeA))
return resp
}
// serviceZone registers rec in a match-only (non-authoritative) zone, the shape
// the synthesized private-service zones arrive in.
func serviceZone(t *testing.T, resolver *Resolver, zone string, records ...nbdns.SimpleRecord) {
t.Helper()
resolver.Update([]nbdns.CustomZone{{
Domain: zone,
Records: records,
NonAuthoritative: true,
}})
}
func TestLocalResolver_WarmsLazyPeerOnResolve(t *testing.T) {
rec := nbdns.SimpleRecord{Name: "svc.proxy.netbird.cloud.", Type: 1, Class: nbdns.DefaultClass, TTL: 300, RData: "100.64.0.7"}
resolver := NewResolver()
serviceZone(t, resolver, "proxy.netbird.cloud", rec)
act := &recordingActivator{}
resolver.SetPeerActivator(act)
resp := serveA(t, resolver, rec.Name)
require.NotNil(t, resp, "resolver must answer")
require.NotEmpty(t, resp.Answer, "answer must carry the A record")
act.mu.Lock()
defer act.mu.Unlock()
assert.True(t, act.called, "activator must be invoked for a service-zone A answer")
assert.Contains(t, act.addrs, netip.MustParseAddr("100.64.0.7"), "activator must receive the answer's peer IP")
}
func TestLocalResolver_NoActivatorNoWarmup(t *testing.T) {
// With no activator wired the resolver behaves exactly as before.
rec := nbdns.SimpleRecord{Name: "svc.proxy.netbird.cloud.", Type: 1, Class: nbdns.DefaultClass, TTL: 300, RData: "100.64.0.7"}
resolver := NewResolver()
serviceZone(t, resolver, "proxy.netbird.cloud", rec)
resp := serveA(t, resolver, rec.Name)
require.NotNil(t, resp, "resolver must still answer without an activator")
require.NotEmpty(t, resp.Answer, "answer must carry the A record")
}
func TestLocalResolver_NoWarmupForMissingRecord(t *testing.T) {
// A query that resolves to nothing must not invoke the activator (no IPs).
resolver := NewResolver()
serviceZone(t, resolver, "proxy.netbird.cloud",
nbdns.SimpleRecord{Name: "svc.proxy.netbird.cloud.", Type: 1, Class: nbdns.DefaultClass, TTL: 300, RData: "100.64.0.7"})
act := &recordingActivator{}
resolver.SetPeerActivator(act)
serveA(t, resolver, "absent.proxy.netbird.cloud.")
act.mu.Lock()
defer act.mu.Unlock()
assert.False(t, act.called, "activator must not be invoked when there is no answer")
}
func TestLocalResolver_NoWarmupInAuthoritativeZone(t *testing.T) {
// The account's peer zone is authoritative; resolving a peer's name there
// must not wake its lazy connection — warm-up is scoped to match-only
// (non-authoritative) zones such as the synthesized private-service zones.
rec := nbdns.SimpleRecord{Name: "peer.netbird.cloud.", Type: 1, Class: nbdns.DefaultClass, TTL: 300, RData: "100.64.0.9"}
resolver := NewResolver()
resolver.Update([]nbdns.CustomZone{{
Domain: "netbird.cloud",
Records: []nbdns.SimpleRecord{rec},
}})
act := &recordingActivator{}
resolver.SetPeerActivator(act)
resp := serveA(t, resolver, rec.Name)
require.NotNil(t, resp, "resolver must answer")
require.NotEmpty(t, resp.Answer, "answer must carry the A record")
act.mu.Lock()
defer act.mu.Unlock()
assert.False(t, act.called, "activator must not be invoked for authoritative-zone answers")
}
func TestLazyWarmupTimeoutFromEnv(t *testing.T) {
tests := []struct {
name string
value string
envSet bool
want time.Duration
}{
{name: "unset uses default", want: defaultLazyWarmupTimeout},
{name: "valid overrides", value: "5s", envSet: true, want: 5 * time.Second},
{name: "invalid falls back", value: "not-a-duration", envSet: true, want: defaultLazyWarmupTimeout},
{name: "negative falls back", value: "-1s", envSet: true, want: defaultLazyWarmupTimeout},
{name: "zero falls back", value: "0s", envSet: true, want: defaultLazyWarmupTimeout},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
if tt.envSet {
t.Setenv(envLazyWarmupTimeout, tt.value)
}
assert.Equal(t, tt.want, lazyWarmupTimeoutFromEnv())
assert.Equal(t, tt.want, NewResolver().warmupTimeout, "constructor must resolve the timeout once")
})
}
}
func TestExtractRecordAddr(t *testing.T) {
t.Run("A record yields unmapped v4", func(t *testing.T) {
// net.ParseIP returns the 16-byte v4-in-v6 form, the same shape
// miekg/dns stores after parsing an A record; the extracted address
// must compare equal to a plain v4 netip.Addr.
addr, ok := extractRecordAddr(&dns.A{A: net.ParseIP("100.64.0.7")})
require.True(t, ok)
assert.True(t, addr.Is4())
assert.Equal(t, netip.MustParseAddr("100.64.0.7"), addr)
})
t.Run("AAAA record yields v6", func(t *testing.T) {
addr, ok := extractRecordAddr(&dns.AAAA{AAAA: net.ParseIP("fd00::1")})
require.True(t, ok)
assert.Equal(t, netip.MustParseAddr("fd00::1"), addr)
})
t.Run("A record without address", func(t *testing.T) {
_, ok := extractRecordAddr(&dns.A{})
assert.False(t, ok)
})
t.Run("non-address record", func(t *testing.T) {
_, ok := extractRecordAddr(&dns.CNAME{Target: "target.netbird.cloud."})
assert.False(t, ok)
})
}

View File

@@ -8,7 +8,6 @@ import (
"github.com/miekg/dns"
dnsconfig "github.com/netbirdio/netbird/client/internal/dns/config"
"github.com/netbirdio/netbird/client/internal/dns/local"
nbdns "github.com/netbirdio/netbird/dns"
"github.com/netbirdio/netbird/route"
"github.com/netbirdio/netbird/shared/management/domain"
@@ -93,11 +92,6 @@ func (m *MockServer) SetFirewall(Firewall) {
// Mock implementation - no-op
}
// SetPeerActivator mock implementation of SetPeerActivator from Server interface
func (m *MockServer) SetPeerActivator(local.PeerActivator) {
// Mock implementation - no-op
}
// BeginBatch mock implementation of BeginBatch from Server interface
func (m *MockServer) BeginBatch() {
// Mock implementation - no-op

View File

@@ -82,7 +82,6 @@ type Server interface {
PopulateManagementDomain(mgmtURL *url.URL) error
SetRouteSources(selected, active func() route.HAMap)
SetFirewall(Firewall)
SetPeerActivator(local.PeerActivator)
}
type nsGroupsByDomain struct {
@@ -492,13 +491,6 @@ func (s *DefaultServer) SetFirewall(fw Firewall) {
}
}
// SetPeerActivator wires the DNS-time lazy-connection warm-up on the local
// resolver. Injected after the connection manager exists (it does not at
// DNS-server construction time). Pass nil to disable.
func (s *DefaultServer) SetPeerActivator(a local.PeerActivator) {
s.localResolver.SetPeerActivator(a)
}
// Stop stops the server
func (s *DefaultServer) Stop() {
s.ctxCancel()
@@ -1443,11 +1435,11 @@ type localPeerConnectivity struct {
// IsConnectedByIP looks the IP up in the peerstore and surfaces both
// the known and connected bits. Used by Resolver.filterDisconnectedPeerAnswers.
func (l localPeerConnectivity) IsConnectedByIP(ip netip.Addr) (known, connected bool) {
func (l localPeerConnectivity) IsConnectedByIP(ip string) (known, connected bool) {
if l.status == nil {
return false, false
}
state, ok := l.status.PeerStateByIP(ip.String())
state, ok := l.status.PeerStateByIP(ip)
if !ok {
return false, false
}

View File

@@ -1,76 +0,0 @@
package internal
import (
"context"
"net/netip"
"time"
"github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/client/internal/peerstore"
)
const dnsActivationPollInterval = 50 * time.Millisecond
// dnsPeerActivator wakes lazy-connection peers from the DNS resolution path. It
// implements dns/local.PeerActivator. DNS queries run on their own goroutines,
// so it only touches state that is safe for concurrent use — ConnMgr.ActivatePeer,
// peerstore.Store and peer.Status — and never takes the engine's syncMsgMux,
// keeping DNS resolution from contending with network-map processing.
type dnsPeerActivator struct {
connMgr *ConnMgr
peerStore *peerstore.Store
status *peer.Status
// ctx is the engine's long-lived context. The connection dial is tied to it
// (not the per-query DNS wait budget) so a handshake that outlasts the wait
// still completes in the background rather than being cancelled at the deadline.
ctx context.Context
}
// ActivatePeersByIP triggers wake-up for the peer(s) owning addrs and waits
// until one is connected or ctx (the per-query DNS wait budget) expires.
// Activation itself is tied to the engine's long-lived context so the dial
// survives a wait that times out. Unknown or already-connected addresses are
// skipped, so the steady-state (warm) path adds no latency.
func (a *dnsPeerActivator) ActivatePeersByIP(ctx context.Context, addrs []netip.Addr) {
if a == nil || a.connMgr == nil {
return
}
var pending []string
for _, addr := range addrs {
ip := addr.String()
st, ok := a.status.PeerStateByIP(ip)
if !ok || st.ConnStatus == peer.StatusConnected {
continue
}
conn, ok := a.peerStore.PeerConn(st.PubKey)
if !ok {
continue
}
a.connMgr.ActivatePeer(a.ctx, conn)
pending = append(pending, ip)
}
if len(pending) == 0 {
return
}
a.waitConnected(ctx, pending)
}
// waitConnected blocks until any of ips reports a connected peer or ctx expires.
func (a *dnsPeerActivator) waitConnected(ctx context.Context, ips []string) {
ticker := time.NewTicker(dnsActivationPollInterval)
defer ticker.Stop()
for {
for _, ip := range ips {
if st, ok := a.status.PeerStateByIP(ip); ok && st.ConnStatus == peer.StatusConnected {
return
}
}
select {
case <-ctx.Done():
return
case <-ticker.C:
}
}
}

View File

@@ -1,129 +0,0 @@
package internal
import (
"context"
"net/netip"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/client/internal/peerstore"
)
func newTestPeerConn(t *testing.T, key string) *peer.Conn {
t.Helper()
conn, err := peer.NewConn(peer.ConnConfig{
Key: key,
LocalKey: "local",
WgConfig: peer.WgConfig{
AllowedIps: []netip.Prefix{netip.MustParsePrefix("100.64.0.1/32")},
},
}, peer.ServiceDependencies{})
require.NoError(t, err)
return conn
}
func newTestDNSPeerActivator(t *testing.T) (*dnsPeerActivator, *peer.Status, *peerstore.Store) {
t.Helper()
status := peer.NewRecorder("https://mgm")
store := peerstore.NewConnStore()
// ConnMgr without Start: the lazy manager is nil, so ActivatePeer is a
// no-op — these tests exercise the activator's skip/wait logic.
connMgr := NewConnMgr(&EngineConfig{}, status, store, nil)
return &dnsPeerActivator{
connMgr: connMgr,
peerStore: store,
status: status,
ctx: context.Background(),
}, status, store
}
func TestDNSPeerActivator_NilSafe(t *testing.T) {
var a *dnsPeerActivator
a.ActivatePeersByIP(context.Background(), []netip.Addr{netip.MustParseAddr("100.64.0.1")})
}
// TestDNSPeerActivator_SkipsUnknownAndConnectedPeers verifies the steady-state
// (warm) path adds no latency: already-connected and unknown addresses never
// enter the wait loop.
func TestDNSPeerActivator_SkipsUnknownAndConnectedPeers(t *testing.T) {
a, status, store := newTestDNSPeerActivator(t)
require.NoError(t, status.AddPeer("peerA", "a.netbird.cloud", "100.64.0.1", "fd00::1"))
require.NoError(t, status.UpdatePeerState(peer.State{PubKey: "peerA", ConnStatus: peer.StatusConnected}))
store.AddPeerConn("peerA", newTestPeerConn(t, "peerA"))
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
start := time.Now()
a.ActivatePeersByIP(ctx, []netip.Addr{
netip.MustParseAddr("100.64.0.1"), // known, connected -> skipped
netip.MustParseAddr("fd00::1"), // known via IPv6, connected -> skipped
netip.MustParseAddr("100.64.0.99"), // unknown -> skipped
})
require.Less(t, time.Since(start), time.Second, "no pending peer must mean no wait")
}
// TestDNSPeerActivator_WaitsForPendingPeerToConnect verifies the wait loop
// returns as soon as a pending peer reports connected, well before the
// per-query budget expires.
func TestDNSPeerActivator_WaitsForPendingPeerToConnect(t *testing.T) {
a, status, store := newTestDNSPeerActivator(t)
require.NoError(t, status.AddPeer("peerA", "a.netbird.cloud", "100.64.0.1", ""))
store.AddPeerConn("peerA", newTestPeerConn(t, "peerA"))
go func() {
time.Sleep(150 * time.Millisecond)
_ = status.UpdatePeerState(peer.State{PubKey: "peerA", ConnStatus: peer.StatusConnected})
}()
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
start := time.Now()
a.ActivatePeersByIP(ctx, []netip.Addr{netip.MustParseAddr("100.64.0.1")})
elapsed := time.Since(start)
require.GreaterOrEqual(t, elapsed, 100*time.Millisecond, "must wait for the pending peer")
require.Less(t, elapsed, 5*time.Second, "must return on connect, not at the deadline")
}
// TestDNSPeerActivator_ReturnsAtBudgetWhenPeerStaysIdle verifies a peer that
// never connects releases the DNS response at the per-query budget instead of
// blocking it indefinitely.
func TestDNSPeerActivator_ReturnsAtBudgetWhenPeerStaysIdle(t *testing.T) {
a, status, store := newTestDNSPeerActivator(t)
require.NoError(t, status.AddPeer("peerA", "a.netbird.cloud", "100.64.0.1", ""))
store.AddPeerConn("peerA", newTestPeerConn(t, "peerA"))
ctx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
defer cancel()
start := time.Now()
a.ActivatePeersByIP(ctx, []netip.Addr{netip.MustParseAddr("100.64.0.1")})
elapsed := time.Since(start)
require.GreaterOrEqual(t, elapsed, 250*time.Millisecond, "must wait out the budget for a pending peer")
require.Less(t, elapsed, 5*time.Second, "must not block past the budget")
}
// TestDNSPeerActivator_NoWaitWithoutPeerConn verifies a known-but-idle peer
// with no connection object in the store is not waited on: there is nothing to
// activate, so waiting could only ever time out.
func TestDNSPeerActivator_NoWaitWithoutPeerConn(t *testing.T) {
a, status, _ := newTestDNSPeerActivator(t)
require.NoError(t, status.AddPeer("peerA", "a.netbird.cloud", "100.64.0.1", ""))
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
start := time.Now()
a.ActivatePeersByIP(ctx, []netip.Addr{netip.MustParseAddr("100.64.0.1")})
require.Less(t, time.Since(start), time.Second, "peer without a conn must not be waited on")
}

View File

@@ -654,16 +654,6 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL)
e.connMgr = NewConnMgr(e.config, e.statusRecorder, e.peerStore, wgIface)
e.connMgr.Start(e.ctx)
// Wire DNS-time lazy-connection warm-up now that the connection manager
// exists (it does not at DNS-server construction time). A DNS answer that
// points at an idle peer then wakes it before the client's first request.
e.dnsServer.SetPeerActivator(&dnsPeerActivator{
connMgr: e.connMgr,
peerStore: e.peerStore,
status: e.statusRecorder,
ctx: e.ctx,
})
e.srWatcher = guard.NewSRWatcher(e.signal, e.relayManager, e.mobileDep.IFaceDiscover, iceCfg)
e.srWatcher.Start(peer.IsForceRelayed())

View File

@@ -54,19 +54,15 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
w.relaySupportedOnRemotePeer.Store(true)
// the relayManager will return with error in case if the connection has lost with relay server
currentRelayAddress, _, err := w.relayManager.RelayInstanceAddress()
_, _, err := w.relayManager.RelayInstanceAddress()
if err != nil {
w.log.Errorf("failed to handle new offer: %s", err)
return
}
srv := w.preferredRelayServer(currentRelayAddress, remoteOfferAnswer.RelaySrvAddress)
var serverIP netip.Addr
if srv == remoteOfferAnswer.RelaySrvAddress {
serverIP = remoteOfferAnswer.RelaySrvIP
}
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key, serverIP)
preferForeign := !w.isController
remoteRelayServer := relayClient.RelayServer{Addr: remoteOfferAnswer.RelaySrvAddress, IP: remoteOfferAnswer.RelaySrvIP}
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, remoteRelayServer, w.config.Key, preferForeign)
if err != nil {
if errors.Is(err, relayClient.ErrConnAlreadyExists) {
w.log.Debugf("handled offer by reusing existing relay connection")
@@ -80,14 +76,13 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
w.relayedConn = relayedConn
w.relayLock.Unlock()
err = w.relayManager.AddCloseListener(srv, w.onRelayClientDisconnected)
if err != nil {
log.Errorf("failed to add close listener: %s", err)
if err := w.relayManager.AddCloseListener(relayedConn.RemoteAddr().String(), w.onRelayClientDisconnected); err != nil {
w.log.Errorf("failed to add close listener: %s", err)
_ = relayedConn.Close()
return
}
w.log.Debugf("peer conn opened via Relay: %s", srv)
w.log.Debugf("peer conn opened via Relay: %s", relayedConn.RemoteAddr())
go w.conn.onRelayConnectionIsReady(RelayConnInfo{
relayedConn: relayedConn,
rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
@@ -126,13 +121,6 @@ func (w *WorkerRelay) isRelaySupported(answer *OfferAnswer) bool {
return answer.RelaySrvAddress != ""
}
func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress string) string {
if w.isController {
return myRelayAddress
}
return remoteRelayAddress
}
func (w *WorkerRelay) onRelayClientDisconnected() {
go w.conn.onRelayDisconnected()
}

View File

@@ -185,7 +185,7 @@ func (r *Route) startResolver(ctx context.Context) {
}
func (r *Route) update(ctx context.Context) error {
resolved, err := r.resolveDomains(ctx)
resolved, err := r.resolveDomains()
if err != nil {
if len(resolved) == 0 {
return fmt.Errorf("resolve domains: %w", err)
@@ -199,9 +199,9 @@ func (r *Route) update(ctx context.Context) error {
return nil
}
func (r *Route) resolveDomains(ctx context.Context) (domainMap, error) {
func (r *Route) resolveDomains() (domainMap, error) {
results := make(chan resolveResult)
go r.resolve(ctx, results)
go r.resolve(results)
resolved := domainMap{}
var merr *multierror.Error
@@ -217,7 +217,7 @@ func (r *Route) resolveDomains(ctx context.Context) (domainMap, error) {
return resolved, nberrors.FormatErrorOrNil(merr)
}
func (r *Route) resolve(ctx context.Context, results chan resolveResult) {
func (r *Route) resolve(results chan resolveResult) {
var wg sync.WaitGroup
for _, d := range r.route.Domains {
@@ -225,10 +225,10 @@ func (r *Route) resolve(ctx context.Context, results chan resolveResult) {
go func(domain domain.Domain) {
defer wg.Done()
ips, err := r.getIPsFromResolver(ctx, domain)
ips, err := r.getIPsFromResolver(domain)
if err != nil {
log.Tracef("Failed to resolve domain %s with private resolver: %v", domain.SafeString(), err)
ips, err = lookupHostIPs(ctx, domain)
ips, err = net.LookupIP(domain.PunycodeString())
if err != nil {
results <- resolveResult{domain: domain, err: fmt.Errorf("resolve d %s: %w", domain.SafeString(), err)}
return
@@ -364,20 +364,6 @@ func determinePrefixChanges(oldPrefixes, newPrefixes []netip.Prefix) (toAdd, toR
return
}
// lookupHostIPs resolves d via the system resolver, honoring ctx cancellation.
func lookupHostIPs(ctx context.Context, d domain.Domain) ([]net.IP, error) {
addrs, err := net.DefaultResolver.LookupIPAddr(ctx, d.PunycodeString())
if err != nil {
return nil, err
}
ips := make([]net.IP, 0, len(addrs))
for _, addr := range addrs {
ips = append(ips, addr.IP)
}
return ips, nil
}
func combinePrefixes(oldPrefixes, removedPrefixes, addedPrefixes []netip.Prefix) []netip.Prefix {
prefixSet := make(map[netip.Prefix]struct{})
for _, prefix := range oldPrefixes {

View File

@@ -3,12 +3,11 @@
package dynamic
import (
"context"
"net"
"github.com/netbirdio/netbird/shared/management/domain"
)
func (r *Route) getIPsFromResolver(ctx context.Context, domain domain.Domain) ([]net.IP, error) {
return lookupHostIPs(ctx, domain)
func (r *Route) getIPsFromResolver(domain domain.Domain) ([]net.IP, error) {
return net.LookupIP(domain.PunycodeString())
}

View File

@@ -3,7 +3,6 @@
package dynamic
import (
"context"
"fmt"
"net"
"time"
@@ -17,7 +16,7 @@ import (
const dialTimeout = 10 * time.Second
func (r *Route) getIPsFromResolver(ctx context.Context, domain domain.Domain) ([]net.IP, error) {
func (r *Route) getIPsFromResolver(domain domain.Domain) ([]net.IP, error) {
privateClient, err := nbdns.GetClientPrivate(r.wgInterface, r.resolverAddr.Addr(), dialTimeout)
if err != nil {
return nil, fmt.Errorf("error while creating private client: %s", err)
@@ -33,7 +32,7 @@ func (r *Route) getIPsFromResolver(ctx context.Context, domain domain.Domain) ([
msg := new(dns.Msg)
msg.SetQuestion(fqdn, qtype)
response, _, err := nbdns.ExchangeWithFallback(ctx, privateClient, msg, r.resolverAddr.String())
response, _, err := nbdns.ExchangeWithFallback(nil, privateClient, msg, r.resolverAddr.String())
if err != nil {
if queryErr == nil {
queryErr = fmt.Errorf("DNS query for %s (type %d) after %s: %w", domain.SafeString(), qtype, time.Since(startTime), err)

View File

@@ -1081,10 +1081,7 @@ func (s *Server) Down(ctx context.Context, _ *proto.DownRequest) (*proto.DownRes
if err := s.cleanupConnection(); err != nil {
s.mutex.Unlock()
if errors.Is(err, ErrServiceNotUp) {
log.Debugf("Down called while service not up: %v", err)
return nil, err
}
// todo review to update the status in case any type of error
log.Errorf("failed to shut down properly: %v", err)
return nil, err
}
@@ -1157,7 +1154,7 @@ func (s *Server) cleanupConnection() error {
// making the run loop the sole owner of engine shutdown.
if engine != nil {
if err := engine.Stop(); err != nil {
log.Errorf("failed to stop engine during cleanup: %v", err)
return err
}
}

View File

@@ -1,13 +1,10 @@
import { useEffect, useState } from "react";
import { useTranslation } from "react-i18next";
import { AlertTriangleIcon, DownloadIcon } from "lucide-react";
import { Browser } from "@wailsio/runtime";
import { Version } from "@bindings/services";
import { Button } from "@/components/buttons/Button";
import { useStatus } from "@/contexts/StatusContext.tsx";
const RELEASES_URL = "https://github.com/netbirdio/netbird/releases/latest";
const RC_RELEASES_URL = "https://pkgs.netbird.io/releases/rc";
function openUrl(url: string) {
Browser.OpenURL(url).catch(() => globalThis.open(url, "_blank"));
@@ -15,26 +12,7 @@ function openUrl(url: string) {
export const DaemonOutdatedOverlay = () => {
const { t } = useTranslation();
const { status, isDaemonOutdated } = useStatus();
const [guiVersion, setGuiVersion] = useState<string>("-");
const clientVersion = status?.daemonVersion ?? "—";
const isRc = /-rc/i.test(guiVersion) || /-rc/i.test(clientVersion);
const downloadUrl = isRc ? RC_RELEASES_URL : RELEASES_URL;
useEffect(() => {
if (!isDaemonOutdated) return;
let cancelled = false;
Version.GUI()
.then((v) => {
if (!cancelled) setGuiVersion(v);
})
.catch((err) => console.error("[DaemonOutdatedOverlay] GUI version error", err));
return () => {
cancelled = true;
};
}, [isDaemonOutdated]);
const { isDaemonOutdated } = useStatus();
if (!isDaemonOutdated) return null;
@@ -60,37 +38,10 @@ export const DaemonOutdatedOverlay = () => {
<p className={"text-sm text-nb-gray-300"}>{t("daemon.outdated.description")}</p>
</div>
<div className={"flex flex-col items-center gap-0.5 text-center"}>
<p className={"text-sm font-semibold text-nb-gray-100"}>
{clientVersion === "development" ? (
<span>
{t("settings.about.clientName")}{" "}
<span className={"font-mono text-yellow-400"}>
{t("settings.about.development")}
</span>
</span>
) : (
t("settings.about.client", { version: clientVersion })
)}
</p>
<p className={"text-sm font-medium text-nb-gray-250"}>
{guiVersion === "development" ? (
<span>
{t("settings.about.guiName")}{" "}
<span className={"font-mono text-yellow-400"}>
{t("settings.about.development")}
</span>
</span>
) : (
t("settings.about.gui", { version: guiVersion })
)}
</p>
</div>
<div className={"wails-no-draggable"}>
<Button variant={"primary"} size={"xs"} onClick={() => openUrl(downloadUrl)}>
<Button variant={"primary"} size={"xs"} onClick={() => openUrl(RELEASES_URL)}>
<DownloadIcon size={14} />
{t("daemon.outdated.download")}
{t("update.card.getInstaller")}
</Button>
</div>
</div>

View File

@@ -28,7 +28,6 @@ type ProfileContextValue = {
loaded: boolean;
refresh: () => Promise<void>;
switchProfile: (id: string) => Promise<void>;
switchProfileNoConnect: (id: string) => Promise<void>;
addProfile: (name: string) => Promise<string>;
removeProfile: (id: string) => Promise<void>;
renameProfile: (id: string, newName: string) => Promise<void>;
@@ -113,16 +112,6 @@ export const ProfileProvider = ({ children }: { children: ReactNode }) => {
[username, refresh],
);
// Manage-profiles variant: switches without connecting, so the user can
// still adjust the management URL before bringing the connection up.
const switchProfileNoConnect = useCallback(
async (id: string) => {
await ProfileSwitcher.SwitchActiveNoConnect({ profileName: id, username });
await refresh();
},
[username, refresh],
);
// addProfile creates a profile by display name and returns the
// daemon-generated ID, so the caller can immediately address it by ID.
const addProfile = useCallback(
@@ -169,7 +158,6 @@ export const ProfileProvider = ({ children }: { children: ReactNode }) => {
loaded,
refresh,
switchProfile,
switchProfileNoConnect,
addProfile,
removeProfile,
renameProfile,
@@ -183,7 +171,6 @@ export const ProfileProvider = ({ children }: { children: ReactNode }) => {
loaded,
refresh,
switchProfile,
switchProfileNoConnect,
addProfile,
removeProfile,
renameProfile,

View File

@@ -45,7 +45,7 @@ export function ProfilesTab() {
activeProfileId,
loaded,
username,
switchProfileNoConnect,
switchProfile,
addProfile,
removeProfile,
renameProfile,
@@ -100,7 +100,7 @@ export function ProfilesTab() {
confirmLabel: t("profile.switch.confirm"),
});
if (!ok) return;
await guarded(i18next.t("profile.error.switchTitle"), () => switchProfileNoConnect(id));
await guarded(i18next.t("profile.error.switchTitle"), () => switchProfile(id));
};
const handleDeregister = async (id: string, name: string) => {
@@ -129,13 +129,14 @@ export function ProfilesTab() {
await guarded(i18next.t("profile.error.createTitle"), async () => {
const id = await addProfile(name);
// SetConfig is keyed by the new profile's ID, so it writes the
// not-yet-active profile before the switch makes it current.
// not-yet-active profile. Write before switching so any reconnect
// targets the right deployment.
if (!isNetbirdCloud(managementUrl)) {
await SettingsSvc.SetConfig(
new SetConfigParams({ profileName: id, username, managementUrl }),
);
}
await switchProfileNoConnect(id);
await switchProfile(id);
});
};

View File

@@ -73,13 +73,6 @@ export default function SessionExpirationDialog() {
let offCancel: (() => void) | undefined;
// Return the dialog to its interactive state and dismiss the browser popup
const resetDialog = () => {
offCancel?.();
WindowManager.CloseBrowserLogin().catch(console.error);
setBusy(false);
};
try {
const start = await Session.RequestExtend({ hint: "" });
const uri = start.verificationUriComplete || start.verificationUri;
@@ -112,22 +105,25 @@ export default function SessionExpirationDialog() {
if (outcome.kind === "cancel") {
waitPromise.cancel?.();
waitPromise.catch(() => {});
resetDialog();
return;
}
// Another surface owns this flow; keep the dialog open to retry.
if (outcome.result.preempted) {
resetDialog();
return;
}
WindowManager.CloseRenewFlow().catch(console.error);
// Close before the popup so the restore can't flash this window back.
WindowManager.CloseSessionExpiration().catch(console.error);
} catch (e) {
resetDialog();
await errorDialog({
Title: t("sessionExpiration.extendFailedTitle"),
Message: formatErrorMessage(e),
});
} finally {
offCancel?.();
WindowManager.CloseBrowserLogin().catch(console.error);
setBusy(false);
}
}, [busy, t]);
@@ -143,11 +139,12 @@ export default function SessionExpirationDialog() {
});
WindowManager.CloseSessionExpiration().catch(console.error);
} catch (e) {
setBusy(false);
await errorDialog({
Title: t("sessionExpiration.logoutFailedTitle"),
Message: formatErrorMessage(e),
});
} finally {
setBusy(false);
}
}, [busy, t]);

View File

@@ -22,9 +22,6 @@ type WelcomeStepTrayProps = {
export function WelcomeStepTray({ onContinue }: Readonly<WelcomeStepTrayProps>) {
const { t } = useTranslation();
const trayScreenshot = trayScreenshotForOS();
// macOS has no tray — the icon sits in the menu bar, so the copy says so.
const titleKey = isMacOS() ? "welcome.titleMac" : "welcome.title";
const descriptionKey = isMacOS() ? "welcome.descriptionMac" : "welcome.description";
return (
<>
@@ -39,9 +36,9 @@ export function WelcomeStepTray({ onContinue }: Readonly<WelcomeStepTrayProps>)
<div className={"flex w-full flex-col gap-1"}>
<DialogHeading id={"nb-welcome-title"} align={"left"}>
{t(titleKey)}
{t("welcome.title")}
</DialogHeading>
<DialogDescription align={"left"}>{t(descriptionKey)}</DialogDescription>
<DialogDescription align={"left"}>{t("welcome.description")}</DialogDescription>
</div>
<DialogActions>

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "Suchen Sie NetBird in der Taskleiste"
},
"welcome.titleMac": {
"message": "Suchen Sie NetBird in der Menüleiste"
},
"welcome.description": {
"message": "NetBird läuft in Ihrer Taskleiste. Klicken Sie auf das Symbol, um sich zu verbinden, Profile zu wechseln oder die Einstellungen zu öffnen."
},
"welcome.descriptionMac": {
"message": "NetBird läuft in Ihrer Menüleiste. Klicken Sie auf das Symbol, um sich zu verbinden, Profile zu wechseln oder die Einstellungen zu öffnen."
},
"welcome.continue": {
"message": "Weiter"
},
@@ -1299,13 +1293,10 @@
"message": "Dokumentation"
},
"daemon.outdated.title": {
"message": "NetBird Client ist veraltet"
"message": "NetBird-Dienst ist veraltet"
},
"daemon.outdated.description": {
"message": "Die neue GUI ist nicht mit Ihrem älteren Client kompatibel. Aktualisieren Sie Ihren Client, um die neue Anwendung zu verwenden."
},
"daemon.outdated.download": {
"message": "Neueste Version herunterladen"
"message": "Aktualisieren Sie den NetBird-Dienst, um diese App zu verwenden."
},
"error.jwt_clock_skew": {
"message": "Anmeldung fehlgeschlagen: Die Uhr dieses Geräts ist nicht mit dem Server synchron. Bitte synchronisieren Sie die Systemuhr und versuchen Sie es erneut."

View File

@@ -1377,19 +1377,11 @@
},
"welcome.title": {
"message": "Look for NetBird in your tray",
"description": "Heading on the first onboarding step, pointing the user to the tray icon. Shown on Windows and Linux; macOS uses welcome.titleMac."
},
"welcome.titleMac": {
"message": "Look for NetBird in your menu bar",
"description": "Heading on the first onboarding step on macOS, pointing the user to the menu bar icon. Use your language's Apple term for the macOS menu bar."
"description": "Heading on the first onboarding step, pointing the user to the tray icon. 'tray' = system tray / menu bar."
},
"welcome.description": {
"message": "NetBird lives in your tray. Click the icon to connect, switch profiles, or open settings.",
"description": "Body of the first onboarding step explaining the tray icon. Shown on Windows and Linux; macOS uses welcome.descriptionMac."
},
"welcome.descriptionMac": {
"message": "NetBird lives in your menu bar. Click the icon to connect, switch profiles, or open settings.",
"description": "Body of the first onboarding step on macOS explaining the menu bar icon. Use your language's Apple term for the macOS menu bar."
"description": "Body of the first onboarding step explaining the tray icon."
},
"welcome.continue": {
"message": "Continue",
@@ -1732,16 +1724,12 @@
"description": "Documentation link on the daemon-unavailable overlay."
},
"daemon.outdated.title": {
"message": "NetBird Client Is Outdated",
"description": "Title of the overlay shown when the NetBird client (daemon) is too old to drive this UI."
"message": "NetBird Service Is Outdated",
"description": "Title of the overlay shown when the NetBird background service is too old to drive this UI."
},
"daemon.outdated.description": {
"message": "The new GUI isn't compatible with the older NetBird client. Update your client to use the new application.",
"description": "Body of the daemon-outdated overlay explaining that the GUI is newer than the client and the client must be updated."
},
"daemon.outdated.download": {
"message": "Download Latest",
"description": "Button on the daemon-outdated overlay that opens the download page for the latest release."
"message": "Update the NetBird service to use this app.",
"description": "Body of the daemon-outdated overlay telling the user to upgrade the service."
},
"error.jwt_clock_skew": {
"message": "Sign-in failed: this device's clock is out of sync with the server. Please sync your system clock and try again.",

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "Busque NetBird en su bandeja del sistema"
},
"welcome.titleMac": {
"message": "Busque NetBird en su barra de menús"
},
"welcome.description": {
"message": "NetBird reside en su bandeja del sistema. Haga clic en el icono para conectarse, cambiar de perfil o abrir la configuración."
},
"welcome.descriptionMac": {
"message": "NetBird reside en su barra de menús. Haga clic en el icono para conectarse, cambiar de perfil o abrir la configuración."
},
"welcome.continue": {
"message": "Continuar"
},
@@ -1299,13 +1293,10 @@
"message": "Documentación"
},
"daemon.outdated.title": {
"message": "NetBird Client está desactualizado"
"message": "El servicio de NetBird está desactualizado"
},
"daemon.outdated.description": {
"message": "La nueva GUI no es compatible con su cliente anterior. Actualice su cliente para usar la nueva aplicación."
},
"daemon.outdated.download": {
"message": "Descargar la última versión"
"message": "Actualice el servicio de NetBird para usar esta aplicación."
},
"error.jwt_clock_skew": {
"message": "Error al iniciar sesión: el reloj de este dispositivo no está sincronizado con el servidor. Sincronice el reloj del sistema e inténtelo de nuevo."

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "Cherchez NetBird dans votre barre détat système"
},
"welcome.titleMac": {
"message": "Cherchez NetBird dans votre barre des menus"
},
"welcome.description": {
"message": "NetBird se trouve dans votre barre détat système. Cliquez sur licône pour vous connecter, changer de profil ou ouvrir les paramètres."
},
"welcome.descriptionMac": {
"message": "NetBird se trouve dans votre barre des menus. Cliquez sur licône pour vous connecter, changer de profil ou ouvrir les paramètres."
},
"welcome.continue": {
"message": "Continuer"
},
@@ -1299,13 +1293,10 @@
"message": "Documentation"
},
"daemon.outdated.title": {
"message": "Le Client NetBird est obsolète"
"message": "Le service NetBird est obsolète"
},
"daemon.outdated.description": {
"message": "La nouvelle GUI n'est pas compatible avec votre ancien client. Mettez à jour votre client pour utiliser la nouvelle application."
},
"daemon.outdated.download": {
"message": "Télécharger la dernière version"
"message": "Mettez à jour le service NetBird pour utiliser cette application."
},
"error.jwt_clock_skew": {
"message": "Échec de la connexion : lhorloge de cet appareil nest pas synchronisée avec le serveur. Veuillez synchroniser lhorloge de votre système et réessayer."

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "Keresse a NetBirdöt a tálcán"
},
"welcome.titleMac": {
"message": "Keresse a NetBirdöt a menüsorban"
},
"welcome.description": {
"message": "A NetBird a tálcán fut. Kattintson az ikonra a csatlakozáshoz, profilváltáshoz vagy a beállítások megnyitásához."
},
"welcome.descriptionMac": {
"message": "A NetBird a menüsorban fut. Kattintson az ikonra a csatlakozáshoz, profilváltáshoz vagy a beállítások megnyitásához."
},
"welcome.continue": {
"message": "Folytatás"
},
@@ -1299,13 +1293,10 @@
"message": "Dokumentáció"
},
"daemon.outdated.title": {
"message": "A NetBird Kliens elavult"
"message": "A NetBird szolgáltatás elavult"
},
"daemon.outdated.description": {
"message": "Az új GUI nem kompatibilis a régebbi klienseddel. Frissítsd a klienst az új alkalmazás használatához."
},
"daemon.outdated.download": {
"message": "Legújabb letöltése"
"message": "Frissítsd a NetBird szolgáltatást az alkalmazás használatához."
},
"error.jwt_clock_skew": {
"message": "A bejelentkezés sikertelen: az eszköz órája eltér a szerverétől. Kérjük, szinkronizálja a rendszer óráját, majd próbálja újra."

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "Cerchi NetBird nella tray"
},
"welcome.titleMac": {
"message": "Cerchi NetBird nella barra dei menu"
},
"welcome.description": {
"message": "NetBird risiede nella tray. Clicchi sull'icona per connettersi, cambiare profilo o aprire le impostazioni."
},
"welcome.descriptionMac": {
"message": "NetBird risiede nella barra dei menu. Clicchi sull'icona per connettersi, cambiare profilo o aprire le impostazioni."
},
"welcome.continue": {
"message": "Continua"
},
@@ -1299,13 +1293,10 @@
"message": "Documentazione"
},
"daemon.outdated.title": {
"message": "NetBird Client è obsoleto"
"message": "Il servizio NetBird è obsoleto"
},
"daemon.outdated.description": {
"message": "La nuova GUI non è compatibile con il tuo client precedente. Aggiorna il client per usare la nuova applicazione."
},
"daemon.outdated.download": {
"message": "Scarica l'ultima versione"
"message": "Aggiorna il servizio NetBird per usare questa app."
},
"error.jwt_clock_skew": {
"message": "Accesso non riuscito: l'orologio di questo dispositivo non è sincronizzato con il server. Sincronizzi l'orologio di sistema e riprovi."

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "トレイの NetBird を確認してください"
},
"welcome.titleMac": {
"message": "メニューバーの NetBird を確認してください"
},
"welcome.description": {
"message": "NetBird はトレイに常駐します。アイコンをクリックして、接続、プロファイルの切り替え、設定を開くことができます。"
},
"welcome.descriptionMac": {
"message": "NetBird はメニューバーに常駐します。アイコンをクリックして、接続、プロファイルの切り替え、設定を開くことができます。"
},
"welcome.continue": {
"message": "続ける"
},

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "Procure o NetBird na sua bandeja"
},
"welcome.titleMac": {
"message": "Procure o NetBird na sua barra de menus"
},
"welcome.description": {
"message": "O NetBird fica na sua bandeja. Clique no ícone para conectar, alternar perfis ou abrir as configurações."
},
"welcome.descriptionMac": {
"message": "O NetBird fica na sua barra de menus. Clique no ícone para conectar, alternar perfis ou abrir as configurações."
},
"welcome.continue": {
"message": "Continuar"
},
@@ -1299,13 +1293,10 @@
"message": "Documentação"
},
"daemon.outdated.title": {
"message": "O NetBird Client está desatualizado"
"message": "O serviço NetBird está desatualizado"
},
"daemon.outdated.description": {
"message": "A nova GUI não é compatível com o seu cliente mais antigo. Atualize o seu cliente para usar o novo aplicativo."
},
"daemon.outdated.download": {
"message": "Baixar a versão mais recente"
"message": "Atualize o serviço NetBird para usar este aplicativo."
},
"error.jwt_clock_skew": {
"message": "Falha no login: o relógio deste dispositivo está fora de sincronia com o servidor. Sincronize o relógio do sistema e tente novamente."

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "Найдите NetBird в системном трее"
},
"welcome.titleMac": {
"message": "Найдите NetBird в строке меню"
},
"welcome.description": {
"message": "NetBird находится в системном трее. Нажмите на значок, чтобы подключиться, переключить профиль или открыть настройки."
},
"welcome.descriptionMac": {
"message": "NetBird находится в строке меню. Нажмите на значок, чтобы подключиться, переключить профиль или открыть настройки."
},
"welcome.continue": {
"message": "Продолжить"
},
@@ -1299,13 +1293,10 @@
"message": "Документация"
},
"daemon.outdated.title": {
"message": "Клиент NetBird устарел"
"message": "Служба NetBird устарела"
},
"daemon.outdated.description": {
"message": "Новый GUI несовместим с вашим более старым клиентом. Обновите клиент, чтобы использовать новое приложение."
},
"daemon.outdated.download": {
"message": "Скачать последнюю версию"
"message": "Обновите службу NetBird, чтобы использовать это приложение."
},
"error.jwt_clock_skew": {
"message": "Не удалось войти: часы этого устройства рассинхронизированы с сервером. Синхронизируйте системные часы и повторите попытку."

View File

@@ -1034,15 +1034,9 @@
"welcome.title": {
"message": "在托盘中查找 NetBird"
},
"welcome.titleMac": {
"message": "在菜单栏中查找 NetBird"
},
"welcome.description": {
"message": "NetBird 驻留在您的托盘中。点击图标即可连接、切换配置文件或打开设置。"
},
"welcome.descriptionMac": {
"message": "NetBird 驻留在您的菜单栏中。点击图标即可连接、切换配置文件或打开设置。"
},
"welcome.continue": {
"message": "继续"
},
@@ -1299,13 +1293,10 @@
"message": "文档"
},
"daemon.outdated.title": {
"message": "NetBird 客户端版本过旧"
"message": "NetBird 服务版本过旧"
},
"daemon.outdated.description": {
"message": "新版 GUI 与您较旧的客户端不兼容。请更新客户端以使用应用。"
},
"daemon.outdated.download": {
"message": "下载最新版本"
"message": "请更新 NetBird 服务以使用应用。"
},
"error.jwt_clock_skew": {
"message": "登录失败:此设备的时钟与服务器不同步。请同步您的系统时钟后重试。"

View File

@@ -12,15 +12,13 @@ import (
"github.com/netbirdio/netbird/client/internal/profilemanager"
)
// ProfileSwitcher holds the switch policy shared by the tray and React
// frontend so both flip profiles identically. SwitchActive (plain selection:
// header dropdown, tray submenu) always connects after the switch;
// SwitchActiveNoConnect (manage-profiles screen) never does, so the user can
// still adjust the management URL before connecting. prevStatus from
// DaemonFeed.Get at entry only decides the teardown:
// ProfileSwitcher holds the reconnect policy shared by the tray and React
// frontend so both flip profiles identically. The policy keys off prevStatus
// from DaemonFeed.Get at SwitchActive entry:
//
// Connected/Connecting/NeedsLogin/LoginFailed/SessionExpired → Down first.
// Idle → no Down.
// Connected/Connecting → Switch + Down + Up; optimistic Connecting paint.
// NeedsLogin/LoginFailed/SessionExpired → Switch + Down; clear stale error for re-login.
// Idle → Switch only.
type ProfileSwitcher struct {
profiles *Profiles
connection *Connection
@@ -31,40 +29,29 @@ func NewProfileSwitcher(profiles *Profiles, connection *Connection, feed *Daemon
return &ProfileSwitcher{profiles: profiles, connection: connection, feed: feed}
}
// SwitchActive switches to the named profile and always connects afterwards.
// SwitchActive switches to the named profile applying the reconnect policy.
func (s *ProfileSwitcher) SwitchActive(ctx context.Context, p ProfileRef) error {
return s.switchActive(ctx, p, true)
}
// SwitchActiveNoConnect switches to the named profile without connecting,
// tearing down any existing connection first.
func (s *ProfileSwitcher) SwitchActiveNoConnect(ctx context.Context, p ProfileRef) error {
return s.switchActive(ctx, p, false)
}
func (s *ProfileSwitcher) switchActive(ctx context.Context, p ProfileRef, connect bool) error {
prevStatus := ""
if s.feed != nil {
if st, err := s.feed.Get(ctx); err == nil {
prevStatus = st.Status
} else {
log.Warnf("profileswitcher: get status: %v", err)
}
if st, err := s.feed.Get(ctx); err == nil {
prevStatus = st.Status
} else {
log.Warnf("profileswitcher: get status: %v", err)
}
needsDown := strings.EqualFold(prevStatus, StatusConnected) ||
strings.EqualFold(prevStatus, StatusConnecting) ||
wasActive := strings.EqualFold(prevStatus, StatusConnected) ||
strings.EqualFold(prevStatus, StatusConnecting)
needsDown := wasActive ||
strings.EqualFold(prevStatus, StatusNeedsLogin) ||
strings.EqualFold(prevStatus, StatusLoginFailed) ||
strings.EqualFold(prevStatus, StatusSessionExpired)
log.Infof("profileswitcher: switch profile=%q prevStatus=%q connect=%v needsDown=%v",
p.ProfileName, prevStatus, connect, needsDown)
log.Infof("profileswitcher: switch profile=%q prevStatus=%q wasActive=%v needsDown=%v",
p.ProfileName, prevStatus, wasActive, needsDown)
// Optimistic Connecting paint plus stale-push suppression during Down (see
// DaemonFeed suppression table); also arms the login-watch that pops
// browser-login when the new profile turns out to need SSO.
if connect && s.feed != nil {
// Optimistic Connecting paint only when wasActive: those prevStatuses emit
// stale Connected + transient Idle pushes during Down that must be
// suppressed until Up resumes the stream (see DaemonFeed suppression table).
if wasActive {
s.feed.BeginProfileSwitch()
}
@@ -89,9 +76,9 @@ func (s *ProfileSwitcher) switchActive(ctx context.Context, p ProfileRef, connec
}
}
if connect {
if wasActive {
if err := s.connection.Up(ctx, UpParams(p)); err != nil {
return fmt.Errorf("connect %q: %w", p.ProfileName, err)
return fmt.Errorf("reconnect %q: %w", p.ProfileName, err)
}
}

View File

@@ -185,38 +185,37 @@ func (s *WindowManager) OpenBrowserLogin(uri string) {
startURL = "/#/dialog/browser-login?uri=" + url.QueryEscape(uri)
}
s.hideOtherWindowsLocked("browser-login")
// Prefer the main window's screen (multi-monitor); falls back to OS-default centering.
var screen *application.Screen
if s.mainWindow != nil {
if sc, err := s.mainWindow.GetScreen(); err == nil {
screen = sc
}
}
opts := DialogWindowOptions("browser-login", s.title("window.title.signIn"), startURL, s.linuxIcon)
// Not always-on-top: it would obscure the browser tab the user logs in through.
opts.AlwaysOnTop = false
opts.InitialPosition = application.WindowCentered
// Open on the active (where users cursor is) display, like the session-expiration dialog.
opts.Screen = s.getScreenBasedOnCursorPosition()
opts.Screen = screen
s.browserLogin = s.app.Window.NewWithOptions(opts)
bl := s.browserLogin
// Red-X close means cancel: emit the event so startLogin() tears down the SSO wait.
bl.OnWindowEvent(events.Common.WindowClosing, func(_ *application.WindowEvent) {
s.app.Event.Emit(EventBrowserLoginCancel)
s.mu.Lock()
// Only a live user red-X still has this registered; programmatic closers
// nil s.browserLogin first and clean up themselves. Guarding here stops a
// stale close event from wiping a replacement popup's state.
userClosed := s.browserLogin == bl
if userClosed {
s.browserLogin = nil
s.restoreHiddenWindowsLocked()
}
s.browserLogin = nil
s.restoreHiddenWindowsLocked()
s.mu.Unlock()
if userClosed {
s.app.Event.Emit(EventBrowserLoginCancel)
}
})
s.centerOnCursorScreen(s.browserLogin)
s.centerWhenReady(s.browserLogin)
return
}
if uri != "" {
s.browserLogin.SetURL("/#/dialog/browser-login?uri=" + url.QueryEscape(uri))
}
s.centerOnCursorScreen(s.browserLogin)
s.browserLogin.Show()
s.browserLogin.Focus()
s.centerWhenReady(s.browserLogin)
}
// BrowserLoginWindow returns the live SSO popup, or nil. While non-nil it is the
@@ -239,15 +238,6 @@ func (s *WindowManager) CloseBrowserLogin() {
s.mu.Lock()
w := s.browserLogin
s.browserLogin = nil
// The WindowClosing hook no-ops on a programmatic close, so restore here —
// but only if a popup was actually open. The frontend calls this even when no
// popup was ever shown (e.g. resetDialog() after an early RequestExtend failure,
// or connection.ts's catch path), and hiddenForLogin is shared with
// OpenInstallProgress, so an unconditional restore could re-show windows a
// still-running install-progress is hiding.
if w != nil {
s.restoreHiddenWindowsLocked()
}
s.mu.Unlock()
if w != nil {
w.Close()
@@ -289,35 +279,6 @@ func (s *WindowManager) CloseSessionExpiration() {
}
}
// CloseRenewFlow tears down the SSO session-renewal UI in a single call: it
// closes the browser-login popup and the session-expiration window together.
func (s *WindowManager) CloseRenewFlow() {
s.mu.Lock()
bl := s.browserLogin
se := s.sessionExpiration
s.browserLogin = nil
s.sessionExpiration = nil
if se != nil {
kept := s.hiddenForLogin[:0]
for _, w := range s.hiddenForLogin {
if w != se {
kept = append(kept, w)
}
}
s.hiddenForLogin = kept
}
s.restoreHiddenWindowsLocked()
s.mu.Unlock()
// Close after unlock so the re-entrant handlers can take s.mu.
if bl != nil {
bl.Close()
}
if se != nil {
se.Close()
}
}
// OpenInstallProgress shows the install-progress window and hides the rest for the duration
// (restored on close). It owns its own result polling since the daemon restarts mid-install.
func (s *WindowManager) OpenInstallProgress(version string) {

View File

@@ -30,8 +30,6 @@ const (
statusError = "Error"
quitDownTimeout = 5 * time.Second
urlGitHubRepo = "https://github.com/netbirdio/netbird"
urlGitHubReleases = "https://github.com/netbirdio/netbird/releases/latest"
urlDocs = "https://docs.netbird.io"
@@ -448,28 +446,11 @@ func (t *Tray) buildMenu() *application.Menu {
menu.AddSeparator()
menu.Add(t.loc.T("tray.menu.quit")).
SetAccelerator("CmdOrCtrl+Q").
OnClick(func(*application.Context) { t.handleQuit() })
OnClick(func(*application.Context) { t.app.Quit() })
return menu
}
func (t *Tray) handleQuit() {
t.profileMu.Lock()
if t.switchCancel != nil {
t.switchCancel()
t.switchCancel = nil
}
t.profileMu.Unlock()
t.svc.DaemonFeed.CancelProfileSwitch()
ctx, cancel := context.WithTimeout(context.Background(), quitDownTimeout)
defer cancel()
if err := t.svc.Connection.Down(ctx); err != nil {
log.Errorf("disconnect on quit: %v", err)
}
t.app.Quit()
}
// handleConnect receives the clicked item from the buildMenu closure —
// t.upItem is menuMu-guarded and must not be read here.
func (t *Tray) handleConnect(upItem *application.MenuItem) {

View File

@@ -91,22 +91,14 @@ func availableProviders() []providerCase {
if region == "" {
region = "us-east-1"
}
// A valid Bedrock inference-profile id (region prefix + date + version),
// overridable per account. `global.` profiles can be invoked from any
// region; set AWS_BEDROCK_MODEL to match the enabled profile for the token.
model := os.Getenv("AWS_BEDROCK_MODEL")
if model == "" {
model = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
}
ps = append(ps, providerCase{name: "bedrock", catalogID: "bedrock_api", upstream: "https://bedrock-runtime." + region + ".amazonaws.com", apiKey: k, model: model, kind: harness.WireBedrock})
ps = append(ps, providerCase{name: "bedrock", catalogID: "bedrock_api", upstream: "https://bedrock-runtime." + region + ".amazonaws.com", apiKey: k, model: "us.anthropic.claude-haiku-4-5", kind: harness.WireBedrock})
}
return ps
}
// providerRequest builds a create request for a matrix provider: enabled, with
// a uniquely-priced model registered under the id an operator would paste —
// the normalized catalog id for Bedrock, the raw form elsewhere (including the
// "@version" Vertex id, which the router must normalize to route).
// a uniquely-priced model for body-routed providers and none for the
// path-routed Vertex (whose model lives in the request path).
func providerRequest(pc providerCase) api.AgentNetworkProviderRequest {
req := api.AgentNetworkProviderRequest{
Name: pc.name,
@@ -115,19 +107,10 @@ func providerRequest(pc providerCase) api.AgentNetworkProviderRequest {
ApiKey: &pc.apiKey,
Enabled: ptr(true),
}
// The router matches the normalized catalog id. Bedrock's request model
// travels as a region-prefixed inference-profile id in the URL path
// (us.anthropic...), which the router strips before matching, so register
// the normalized form here or routing fails as model_not_routable. Vertex is
// registered with the raw "@version" id the Google console documents — the
// router normalizes the registered id, so the operator can paste either
// form; keeping the raw form here guards that normalization end to end.
modelID := pc.model
if pc.kind == harness.WireBedrock {
modelID = catalogModel(pc)
}
req.Models = &[]api.AgentNetworkProviderModel{
{Id: modelID, InputPer1k: 0.001, OutputPer1k: 0.002},
if pc.kind != harness.WireVertex {
req.Models = &[]api.AgentNetworkProviderModel{
{Id: pc.model, InputPer1k: 0.001, OutputPer1k: 0.002},
}
}
return req
}
@@ -218,13 +201,11 @@ func TestProvidersMatrix(t *testing.T) {
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
// the proxy peer so WaitProxyPeer then observes it connected.
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve agent-network endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve agent-network endpoint to proxy IP")
for _, pc := range matrix {
pc := pc

View File

@@ -4,7 +4,6 @@ package agentnetwork
import (
"context"
"regexp"
"strings"
"testing"
"time"
@@ -16,29 +15,13 @@ import (
"github.com/netbirdio/netbird/shared/management/http/api"
)
// bedrockRegionPrefixes and bedrockVersionSuffix mirror the proxy's Bedrock
// model normalization (region/inference-profile prefix + version suffix) so the
// provider is registered under the same catalog key the router matches against.
var (
bedrockRegionPrefixes = []string{"us.", "eu.", "apac.", "global."}
bedrockVersionSuffix = regexp.MustCompile(`-(\d{8}-)?v\d+(:\d+)?$`)
)
// catalogModel returns the normalized catalog id the proxy stamps for a
// path-routed provider's configured model — the form the router and guardrail
// allowlist compare against (Bedrock region prefix + version stripped, Vertex
// @version stripped).
// path-routed provider's configured model — the form the guardrail allowlist is
// compared against (region prefix / @version stripped).
func catalogModel(pc providerCase) string {
switch pc.kind {
case harness.WireBedrock:
m := pc.model
for _, p := range bedrockRegionPrefixes {
if strings.HasPrefix(m, p) {
m = m[len(p):]
break
}
}
return bedrockVersionSuffix.ReplaceAllString(m, "")
return strings.TrimPrefix(pc.model, "us.")
case harness.WireVertex:
return strings.SplitN(pc.model, "@", 2)[0]
default:
@@ -125,15 +108,7 @@ func TestModelAllowlistEnforced(t *testing.T) {
require.NoError(t, perr, "create provider %s", pc.name)
id := prov.Id
ids = append(ids, id)
// Vertex allowlists the raw "@version" id an operator copies from the
// Google console; the guardrail must normalize the entry to match the
// version-stripped request model. Other providers allowlist the
// normalized catalog id.
if pc.kind == harness.WireVertex {
allowed = append(allowed, pc.model)
} else {
allowed = append(allowed, catalogModel(pc))
}
allowed = append(allowed, catalogModel(pc))
t.Cleanup(func() { _ = srv.DeleteProvider(context.Background(), id) })
}
@@ -172,13 +147,11 @@ func TestModelAllowlistEnforced(t *testing.T) {
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
// the proxy peer so WaitProxyPeer then observes it connected.
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve agent-network endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve agent-network endpoint to proxy IP")
for _, pc := range providers {
pc := pc
@@ -190,25 +163,6 @@ func TestModelAllowlistEnforced(t *testing.T) {
// the upstream), regardless of whether it is a real catalog model.
assert.Equal(t, 403, sendModel(ctx, t, cl, settings.Endpoint, proxyIP, pc, disallowedModel(pc)),
"model outside the allowlist must be denied for %s", pc.name)
if pc.kind != harness.WireVertex {
return
}
// The unversioned model id — the shape an Anthropic SDK client sends
// when configured without an "@version" suffix (customer-reported) —
// must pass the same allowlist entry and resolve upstream.
assert.Equal(t, 200, sendModel(ctx, t, cl, settings.Endpoint, proxyIP, pc, catalogModel(pc)),
"unversioned model id must be permitted for %s", pc.name)
// Token counting: the URL carries the literal "count-tokens"
// pseudo-model and the real model travels in the body (Claude Code
// sends one such call per request). Allowed body model → served.
code, _, err := cl.VertexCountTokens(ctx, settings.Endpoint, proxyIP, pc.project, pc.region, pc.model)
require.NoError(t, err, "count-tokens request must reach the proxy for %s", pc.name)
assert.Equal(t, 200, code, "count-tokens with an allowlisted body model must be permitted for %s", pc.name)
// Disallowed body model → denied before the upstream.
code, _, err = cl.VertexCountTokens(ctx, settings.Endpoint, proxyIP, pc.project, pc.region, disallowedModel(pc))
require.NoError(t, err, "count-tokens request must reach the proxy for %s", pc.name)
assert.Equal(t, 403, code, "count-tokens with a body model outside the allowlist must be denied for %s", pc.name)
})
}
}

View File

@@ -104,13 +104,11 @@ func TestProviderSkipTLSVerification(t *testing.T) {
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
// the proxy peer so WaitProxyPeer then observes it connected.
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
// Positive: skip=true reaches the self-signed upstream. Retry to absorb
// tunnel/DNS jitter on the first call; success also proves the path works.

View File

@@ -106,13 +106,11 @@ func TestVLLMProvider(t *testing.T) {
t.Cleanup(func() { _ = cl.Terminate(context.Background()) })
require.NoError(t, cl.WaitConnected(ctx, 90*time.Second), "client must connect to management")
// Resolve first: the DNS lookup triggers the lazy-connection warm-up, waking
// the proxy peer so WaitProxyPeer then observes it connected.
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
if err := cl.WaitProxyPeer(ctx, 180*time.Second); err != nil {
t.Fatalf("client did not see the proxy peer: %v\n=== proxy logs ===\n%s", err, px.Logs(context.Background()))
}
proxyIP, err := cl.ResolveProxyIP(ctx, settings.Endpoint)
require.NoError(t, err, "resolve endpoint to proxy IP")
before, _ := srv.ListAccessLogs(ctx)
sessionID := "e2e-session-vllm"

View File

@@ -231,17 +231,6 @@ func (cl *Client) Vertex(ctx context.Context, endpoint, proxyIP, project, region
return cl.post(ctx, endpoint, proxyIP, path, body, withSessionID(nil, sessionID))
}
// VertexCountTokens issues the Anthropic-on-Vertex token-count POST over the
// tunnel. The URL carries the literal "count-tokens" pseudo-model and the real
// model travels in the body — the one Vertex shape whose model is body-routed,
// sent by Claude Code before each request. The proxy must resolve the body
// model for the guardrail allowlist and routing.
func (cl *Client) VertexCountTokens(ctx context.Context, endpoint, proxyIP, project, region, model string) (int, string, error) {
path := fmt.Sprintf("/v1/projects/%s/locations/%s/publishers/anthropic/models/count-tokens:rawPredict", project, region)
body := fmt.Sprintf(`{"model":%q,"messages":[{"role":"user","content":"hi"}]}`, model)
return cl.post(ctx, endpoint, proxyIP, path, body, nil)
}
// Bedrock issues a native AWS Bedrock InvokeModel POST over the tunnel. The
// model id is carried in the request path (/model/{id}/invoke), so the proxy
// routes by path; the body uses the bedrock anthropic_version rather than a

View File

@@ -14,7 +14,6 @@ COPY proxy ./proxy
COPY route ./route
COPY shared ./shared
COPY sharedsock ./sharedsock
COPY trustedproxy ./trustedproxy
COPY upload-server ./upload-server
COPY util ./util
COPY version ./version

View File

@@ -1,17 +0,0 @@
package llm
import "strings"
// NormalizeVertexModel strips the "@version" suffix from a Google Vertex AI
// publisher model id so it matches the catalog/pricing key, e.g.
// "claude-sonnet-4-5@20250929" -> "claude-sonnet-4-5". It is the single source
// of truth shared by the request parser (which normalizes the request model
// from the URL path), the router (which normalizes the operator's registered
// Vertex model ids), and the guardrail (which normalizes allowlist entries) so
// both spellings of the same model compare equal.
func NormalizeVertexModel(modelID string) string {
if at := strings.Index(modelID, "@"); at >= 0 {
return modelID[:at]
}
return modelID
}

View File

@@ -1,23 +0,0 @@
package llm
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestNormalizeVertexModel(t *testing.T) {
cases := map[string]string{
"claude-sonnet-4-5@20250929": "claude-sonnet-4-5",
"claude-opus-4-6@20250514": "claude-opus-4-6",
// Bare ids (the form Vertex resolves to the default version) pass through.
"claude-opus-4-6": "claude-opus-4-6",
// Numeric version aliases used by Google publisher models.
"text-embedding-005@001": "text-embedding-005",
// A leading "@" yields the empty string; callers treat that as "no model".
"@cf/meta/llama-3-8b": "",
}
for in, want := range cases {
require.Equal(t, want, NormalizeVertexModel(in), "normalize %q", in)
}
}

View File

@@ -13,7 +13,6 @@ import (
"context"
"unicode/utf8"
"github.com/netbirdio/netbird/proxy/internal/llm"
"github.com/netbirdio/netbird/proxy/internal/middleware"
)
@@ -163,15 +162,6 @@ func (m *Middleware) modelInAllowlist(model string) bool {
if allowed == normalised {
return true
}
// Vertex model ids are often stored with their "@version" suffix (e.g.
// "claude-opus-4-6@20250514") — the id the Google console documents —
// while the request parser strips the suffix before the model reaches
// the guardrail. Accept an entry whose version-stripped form matches so
// both spellings of the same model pass one allowlist. Entries without
// an "@" are untouched, so non-Vertex matching stays exact.
if v := llm.NormalizeVertexModel(allowed); v != "" && v != allowed && v == normalised {
return true
}
}
return false
}

View File

@@ -70,27 +70,6 @@ func TestAllowlistMatchAllows(t *testing.T) {
assert.Equal(t, middleware.DecisionAllow, out.Decision, "model in allowlist must be allowed")
}
// A Vertex allowlist entry stored with its "@version" suffix must match the
// request model, which reaches the guardrail with the suffix already stripped
// by the request parser.
func TestAllowlistVertexVersionedEntryMatchesStrippedModel(t *testing.T) {
mw := New(Config{ModelAllowlist: []string{"claude-opus-4-6@20250514"}})
out, err := mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "claude-opus-4-6"},
))
require.NoError(t, err)
assert.Equal(t, middleware.DecisionAllow, out.Decision,
"@version allowlist entry must match the version-stripped request model")
// The stripped form must not over-match a different model.
out, err = mw.Invoke(context.Background(), newInput(
middleware.KV{Key: middleware.KeyLLMModel, Value: "claude-opus-4-8"},
))
require.NoError(t, err)
assert.Equal(t, middleware.DecisionDeny, out.Decision,
"a different model must stay denied")
}
func TestAllowlistMissDenies(t *testing.T) {
mw := New(Config{ModelAllowlist: []string{"gpt-4o"}})
out, err := mw.Invoke(context.Background(), newInput(

View File

@@ -104,96 +104,3 @@ func TestModelAllowlist_URLRoutedProviders(t *testing.T) {
})
}
}
// TestModelAllowlist_VertexRequestShapes replays the Vertex request shapes an
// Anthropic SDK client (Claude Code) actually sends — the model travels in the
// URL path, optionally without an "@version" suffix, at the global location —
// against allowlists holding either the bare catalog id or the raw versioned
// id. The URLs mirror the shape of a customer-reported request (unversioned
// model, global location, a project id that itself contains "claude").
func TestModelAllowlist_VertexRequestShapes(t *testing.T) {
const (
opusBare = "/v1/projects/corp-gcp-it-all-claude/locations/global/publishers/anthropic/models/claude-opus-4-6:rawPredict"
opusBareSSE = "/v1/projects/corp-gcp-it-all-claude/locations/global/publishers/anthropic/models/claude-opus-4-6:streamRawPredict"
countTokens = "/v1/projects/corp-gcp-it-all-claude/locations/global/publishers/anthropic/models/count-tokens:rawPredict"
messagesBody = `{"anthropic_version":"vertex-2023-10-16","messages":[{"role":"user","content":"hi"}]}`
countOpusBody = `{"model":"claude-opus-4-6","messages":[{"role":"user","content":"hi"}]}`
)
tests := []struct {
name string
url string
body string
allowlist []string
decision middleware.Decision
denyCode string
}{
{
name: "unversioned model allowed by bare catalog entry",
url: opusBare,
body: messagesBody,
allowlist: []string{"claude-opus-4-6"},
decision: middleware.DecisionAllow,
},
{
name: "unversioned model allowed by @version allowlist entry",
url: opusBare,
body: messagesBody,
allowlist: []string{"claude-opus-4-6@20250514"},
decision: middleware.DecisionAllow,
},
{
name: "streaming action allowed the same as rawPredict",
url: opusBareSSE,
body: messagesBody,
allowlist: []string{"claude-opus-4-6"},
decision: middleware.DecisionAllow,
},
{
// The original customer report: a Sonnet-only allowlist must block
// an Opus request regardless of the id carrying a version suffix.
name: "unversioned model outside the allowlist denied",
url: opusBare,
body: messagesBody,
allowlist: []string{"claude-sonnet-4-5"},
decision: middleware.DecisionDeny,
denyCode: "llm_policy.model_blocked",
},
{
name: "count-tokens resolves the body model and passes when allowed",
url: countTokens,
body: countOpusBody,
allowlist: []string{"claude-opus-4-6"},
decision: middleware.DecisionAllow,
},
{
name: "count-tokens with a disallowed body model denied",
url: countTokens,
body: countOpusBody,
allowlist: []string{"claude-sonnet-4-5"},
decision: middleware.DecisionDeny,
denyCode: "llm_policy.model_blocked",
},
{
// No model in the body: the pseudo-model stays and fails closed.
name: "count-tokens without a body model fails closed",
url: countTokens,
body: messagesBody,
allowlist: []string{"claude-opus-4-6"},
decision: middleware.DecisionDeny,
denyCode: "llm_policy.model_blocked",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
out := runParserGuardrail(t, tt.url, []byte(tt.body), tt.allowlist)
assert.Equal(t, tt.decision, out.Decision, "unexpected decision for %s", tt.name)
if tt.decision == middleware.DecisionDeny {
require.NotNil(t, out.DenyReason, "deny reason must be set for %s", tt.name)
assert.Equal(t, 403, out.DenyStatus, "deny status must be 403 for %s", tt.name)
assert.Equal(t, tt.denyCode, out.DenyReason.Code, "deny code for %s", tt.name)
}
})
}
}

View File

@@ -253,7 +253,9 @@ func parseVertexPath(reqPath string) (vertexRequest, bool) {
if c := strings.LastIndex(rest, ":"); c >= 0 {
model, action = rest[:c], rest[c+1:]
}
model = llm.NormalizeVertexModel(model)
if at := strings.Index(model, "@"); at >= 0 {
model = model[:at]
}
if model == "" {
return vertexRequest{}, false
}
@@ -274,44 +276,24 @@ func vertexPublisherVendor(publisher string) string {
}
}
// vertexCountTokensModel is the pseudo-model id of the Vertex token-count
// endpoint (…/models/count-tokens:rawPredict). It is the one Vertex request
// shape whose real model travels in the JSON body rather than the URL path.
const vertexCountTokensModel = "count-tokens"
// invokeVertex emits the model/vendor/session/prompt for a Vertex publisher
// request, using the publisher's parser to read the (vendor-native) body.
func (m middlewareImpl) invokeVertex(in *middleware.Input, vx vertexRequest) *middleware.Output {
out := &middleware.Output{Decision: middleware.DecisionAllow}
vendor := vertexPublisherVendor(vx.publisher)
var parser llm.Parser
if vendor != "" {
parser, _ = llm.ParserByName(vendor)
}
model := vx.model
// Token counting carries the literal "count-tokens" pseudo-model in the URL
// and the real model in the body (Claude Code issues one such call per
// request). Resolve the body model so the guardrail allowlist and the
// router evaluate the actual model — otherwise every count-tokens call
// under an allowlist denies as model_blocked. A body the parser cannot
// read keeps the pseudo-model, which fails closed downstream.
if model == vertexCountTokensModel && parser != nil {
if facts, err := parser.ParseRequest(in.Body); err == nil && facts.Model != "" {
if bodyModel := llm.NormalizeVertexModel(facts.Model); bodyModel != "" {
model = bodyModel
}
}
}
md := []middleware.KV{}
if vendor != "" {
md = append(md, middleware.KV{Key: middleware.KeyLLMProvider, Value: vendor})
}
md = append(md, middleware.KV{Key: middleware.KeyLLMModel, Value: model})
md = append(md, middleware.KV{Key: middleware.KeyLLMModel, Value: vx.model})
md = append(md, middleware.KV{Key: middleware.KeyLLMStream, Value: strconv.FormatBool(vx.stream)})
var parser llm.Parser
if vendor != "" {
parser, _ = llm.ParserByName(vendor)
}
sessionID := sessionIDFromHeaders(in.Headers)
if sessionID == "" && parser != nil {
sessionID = parser.ExtractSessionID(in.Body)

View File

@@ -564,14 +564,6 @@ func routeClaimsModel(route ProviderRoute, model string) bool {
if route.Bedrock && llm.NormalizeBedrockModel(candidate) == model {
return true
}
// Vertex request models likewise reach the router with the "@version"
// suffix already stripped by the parser, while the operator may register
// the raw versioned id the Google console documents (e.g.
// "claude-opus-4-6@20250514"). Normalize the candidate so both spellings
// of the same model match.
if route.Vertex && llm.NormalizeVertexModel(candidate) == model {
return true
}
}
return false
}

View File

@@ -1,64 +0,0 @@
package llm_router
import (
"context"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/proxy/internal/middleware"
)
// TestRouteClaimsModel_VertexNormalizesCandidate guards the Vertex analog of
// the native-Bedrock routing gap: the request model reaches the router already
// normalized (the parser strips the "@version" suffix from the URL path), so a
// provider registered with the raw versioned id must still match.
func TestRouteClaimsModel_VertexNormalizesCandidate(t *testing.T) {
route := ProviderRoute{Vertex: true, Models: []string{"claude-opus-4-6@20250514"}}
assert.True(t, routeClaimsModel(route, "claude-opus-4-6"),
"raw @version Vertex model must match the normalized request model")
assert.False(t, routeClaimsModel(route, "claude-haiku-4-5"),
"a model outside the provider's list must not match")
bare := ProviderRoute{Vertex: true, Models: []string{"claude-opus-4-6"}}
assert.True(t, routeClaimsModel(bare, "claude-opus-4-6"),
"a bare catalog id keeps matching exactly")
// Non-Vertex routes keep exact matching for "@"-suffixed candidates.
direct := ProviderRoute{Models: []string{"claude-opus-4-6@20250514"}}
assert.False(t, routeClaimsModel(direct, "claude-opus-4-6"),
"non-Vertex routes must not normalize @version candidates")
}
// TestRouter_VertexUnversionedModelRoutes replays the customer-reported request
// end to end through the router: an unversioned model id at the global location
// (…/models/claude-opus-4-6:rawPredict) must route on a Vertex provider whose
// models were registered with the raw "@version" ids.
func TestRouter_VertexUnversionedModelRoutes(t *testing.T) {
route := vertexRoute()
route.Models = []string{"claude-opus-4-6@20250514", "claude-sonnet-4-5@20250929"}
mw := New(Config{Providers: []ProviderRoute{route}})
in := pathRoutedInput(
"/v1/projects/corp-gcp-it-all-claude/locations/global/publishers/anthropic/models/claude-opus-4-6:rawPredict",
"anthropic",
"claude-opus-4-6",
)
out, err := mw.Invoke(context.Background(), in)
require.NoError(t, err)
assert.Equal(t, middleware.DecisionAllow, out.Decision,
"unversioned request model must route on a provider registered with @version ids")
denied := pathRoutedInput(
"/v1/projects/corp-gcp-it-all-claude/locations/global/publishers/anthropic/models/claude-haiku-4-5:rawPredict",
"anthropic",
"claude-haiku-4-5",
)
out, err = mw.Invoke(context.Background(), denied)
require.NoError(t, err)
assert.Equal(t, middleware.DecisionDeny, out.Decision,
"a model outside the provider's registered list must still deny")
require.NotNil(t, out.DenyReason)
assert.Equal(t, denyCodeNotRoutable, out.DenyReason.Code)
}

View File

@@ -1,291 +0,0 @@
package proxy_test
import (
"context"
"net"
"net/http"
"net/http/httptest"
"net/url"
"runtime"
"strings"
"sync/atomic"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"google.golang.org/grpc"
"google.golang.org/grpc/credentials/insecure"
"google.golang.org/grpc/test/bufconn"
"github.com/netbirdio/netbird/management/internals/modules/agentnetwork"
agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types"
rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
mgmtgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
"github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/proxy/internal/middleware"
"github.com/netbirdio/netbird/proxy/internal/middleware/bodytap"
mwbuiltin "github.com/netbirdio/netbird/proxy/internal/middleware/builtin"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/cost_meter"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_guardrail"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_identity_inject"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_limit_check"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_limit_record"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_request_parser"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_response_parser"
_ "github.com/netbirdio/netbird/proxy/internal/middleware/builtin/llm_router"
"github.com/netbirdio/netbird/proxy/internal/proxy"
nbproxytypes "github.com/netbirdio/netbird/proxy/internal/types"
"github.com/netbirdio/netbird/shared/management/proto"
log "github.com/sirupsen/logrus"
)
// TestReverseProxy_VertexGuardrail_ModelAllowlist drives Anthropic-on-Vertex
// requests (model in the URL path, not the body — the shape Claude Code sends)
// through the full synthesized middleware chain against an in-process
// management stack and a fake Vertex upstream. It reproduces the customer
// report end to end: with a guardrail allowlisting only Sonnet, an Opus request
// must be denied by the guardrail (model_blocked) before it reaches the
// upstream, and the allowed Sonnet request must reach it.
//
// Two provider shapes are covered because they exercise different code:
//
// - "catch_all": provider registered with no models (the historical Vertex
// default). The router waves every model through, so the guardrail is the
// only thing that can block Opus — the direct regression guard for "the
// disallowed model is never blocked".
// - "versioned_models": provider registered with the raw "@version" model ids
// the Google console documents. The router must normalize the registered id
// to route the version-stripped request model (otherwise even Sonnet denies
// as model_not_routable), and the guardrail must still block Opus.
//
// The upstream-hit counter proves whether a denied request actually
// short-circuited or leaked through to Google.
func TestReverseProxy_VertexGuardrail_ModelAllowlist(t *testing.T) {
if runtime.GOOS == "windows" {
t.Skip("sqlite store not supported on Windows")
}
cases := []struct {
name string
providerModels []agentNetworkTypes.ProviderModel
}{
{name: "catch_all", providerModels: nil},
{name: "versioned_models", providerModels: []agentNetworkTypes.ProviderModel{
{ID: "claude-sonnet-4-5@20250929"},
{ID: "claude-opus-4-6@20250514"},
}},
}
for _, tc := range cases {
tc := tc
t.Run(tc.name, func(t *testing.T) {
runVertexGuardrailCase(t, tc.providerModels)
})
}
}
func runVertexGuardrailCase(t *testing.T, providerModels []agentNetworkTypes.ProviderModel) {
t.Helper()
const (
testAccountID = "acct-vertex-guard-1"
testAdminUser = "user-admin-1"
adminGroupID = "grp-admins"
providerID = "prov-vertex-test"
guardrailID = "ainguard-sonnet-only"
cluster = "test.proxy.local"
subdomain = "vertexguard"
)
testLogger := log.New()
testLogger.SetLevel(log.PanicLevel)
ctx := context.Background()
// Fake Vertex upstream: any request that reaches it is a guardrail miss for
// the disallowed model.
var upstreamHits atomic.Int64
upstreamBody := []byte(`{"id":"msg_x","type":"message","role":"assistant","model":"claude","content":[{"type":"text","text":"pong"}],"usage":{"input_tokens":5,"output_tokens":2}}`)
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
upstreamHits.Add(1)
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write(upstreamBody)
}))
t.Cleanup(upstream.Close)
// In-process management gRPC (bufconn) over a real sqlite store + manager.
st, cleanup, err := store.NewTestStoreFromSQL(ctx, "", t.TempDir())
require.NoError(t, err, "real sqlite test store must come up")
t.Cleanup(cleanup)
anMgr := agentnetwork.NewManager(st, nil, nil, nil)
server := &mgmtgrpc.ProxyServiceServer{}
server.SetAgentNetworkLimitsService(anMgr)
lis := bufconn.Listen(1024 * 1024)
srv := grpc.NewServer()
proto.RegisterProxyServiceServer(srv, server)
go func() { _ = srv.Serve(lis) }()
t.Cleanup(srv.Stop)
conn, err := grpc.NewClient("passthrough:///bufnet",
grpc.WithContextDialer(func(_ context.Context, _ string) (net.Conn, error) { return lis.Dial() }),
grpc.WithTransportCredentials(insecure.NewCredentials()),
)
require.NoError(t, err)
t.Cleanup(func() { _ = conn.Close() })
mgmtClient := proto.NewProxyServiceClient(conn)
require.NoError(t, st.SaveAgentNetworkSettings(ctx, &agentNetworkTypes.Settings{
AccountID: testAccountID,
Cluster: cluster,
Subdomain: subdomain,
}))
require.NoError(t, st.SaveAgentNetworkProvider(ctx, &agentNetworkTypes.Provider{
ID: providerID,
AccountID: testAccountID,
ProviderID: "vertex_ai_api",
Name: "vertex-guard-test",
UpstreamURL: upstream.URL,
// A static bearer (not "keyfile::…") so the router injects a static auth
// header instead of minting a GCP OAuth token at request time — the mint
// needs network egress and would deny before the guardrail runs, masking
// the guardrail decision this test exists to observe.
APIKey: "static-vertex-token",
Enabled: true,
Models: providerModels,
SessionPrivateKey: "priv",
SessionPublicKey: "pub",
}))
// Guardrail allowlisting ONLY Sonnet.
require.NoError(t, st.SaveAgentNetworkGuardrail(ctx, &agentNetworkTypes.Guardrail{
ID: guardrailID,
AccountID: testAccountID,
Name: "sonnet-only",
Checks: agentNetworkTypes.GuardrailChecks{
ModelAllowlist: agentNetworkTypes.GuardrailModelAllowlist{
Enabled: true,
Models: []string{"claude-sonnet-4-5"},
},
},
}))
require.NoError(t, st.SaveAgentNetworkPolicy(ctx, &agentNetworkTypes.Policy{
ID: "ainpol-vertex-guard",
AccountID: testAccountID,
Name: "admins-vertex",
Enabled: true,
SourceGroups: []string{adminGroupID},
DestinationProviderIDs: []string{providerID},
GuardrailIDs: []string{guardrailID},
}))
services, err := agentnetwork.SynthesizeServices(ctx, st, testAccountID)
require.NoError(t, err)
require.Len(t, services, 1, "exactly one synth service expected")
synthSvc := services[0]
require.NotEmpty(t, synthSvc.Targets, "synth target must exist")
mwbuiltin.Configure(ctx, t.TempDir(), nil, testLogger, mgmtClient)
registry := mwbuiltin.DefaultRegistry()
mwMetrics, err := middleware.NewMetrics(nil)
require.NoError(t, err)
mwMgr := middleware.NewManager(0, mwMetrics, testLogger)
mwMgr.SetResolver(middleware.NewResolver(registry))
specs := make([]middleware.Spec, 0, len(synthSvc.Targets[0].Options.Middlewares))
for _, mw := range synthSvc.Targets[0].Options.Middlewares {
var slot middleware.Slot
switch mw.Slot {
case rpservice.MiddlewareSlotOnRequest:
slot = middleware.SlotOnRequest
case rpservice.MiddlewareSlotOnResponse:
slot = middleware.SlotOnResponse
case rpservice.MiddlewareSlotTerminal:
slot = middleware.SlotTerminal
default:
t.Fatalf("unknown middleware slot %q on %s", mw.Slot, mw.ID)
}
specs = append(specs, middleware.Spec{
ID: mw.ID,
Slot: slot,
Enabled: mw.Enabled,
FailMode: middleware.FailOpen,
Timeout: middleware.DefaultTimeout,
RawConfig: append([]byte(nil), mw.ConfigJSON...),
CanMutate: mw.CanMutate,
})
}
serviceIDStr := synthSvc.ID
require.NoError(t, mwMgr.Rebuild(serviceIDStr, []middleware.PathTargetBinding{{
ServiceID: serviceIDStr,
PathID: "/",
Specs: specs,
}}))
upstreamURL, err := url.Parse(upstream.URL)
require.NoError(t, err)
rp := proxy.NewReverseProxy(http.DefaultTransport, "auto", nil, testLogger, proxy.WithMiddlewareManager(mwMgr))
rp.AddMapping(proxy.Mapping{
ID: nbproxytypes.ServiceID(serviceIDStr),
AccountID: nbproxytypes.AccountID(testAccountID),
Host: synthSvc.Domain,
Paths: map[string]*proxy.PathTarget{
"/": {
URL: upstreamURL,
DirectUpstream: true,
AgentNetwork: true,
Middlewares: specs,
CaptureConfig: &bodytap.Config{
MaxRequestBytes: 1 << 20,
MaxResponseBytes: 1 << 20,
ContentTypes: []string{"application/json", "text/event-stream"},
},
},
},
})
// vertexBody carries no "model" field — the model lives in the URL path,
// exactly as an Anthropic SDK client (Claude Code) posts to Vertex.
const vertexBody = `{"anthropic_version":"vertex-2023-10-16","max_tokens":64,"messages":[{"role":"user","content":"Reply with exactly: pong"}]}`
send := func(t *testing.T, model string) (int, int64, string) {
t.Helper()
before := upstreamHits.Load()
path := "/v1/projects/corp-gcp-it-all-claude/locations/global/publishers/anthropic/models/" + model + ":rawPredict"
req := httptest.NewRequest("POST", "https://"+synthSvc.Domain+path, strings.NewReader(vertexBody))
req.Host = synthSvc.Domain
req.Header.Set("Content-Type", "application/json")
cd := proxy.NewCapturedData("req-" + model)
cd.SetServiceID(nbproxytypes.ServiceID(serviceIDStr))
cd.SetAccountID(nbproxytypes.AccountID(testAccountID))
cd.SetUserID(testAdminUser)
cd.SetUserGroups([]string{adminGroupID})
cd.SetAuthMethod("tunnel_peer")
req = req.WithContext(proxy.WithCapturedData(req.Context(), cd))
w := httptest.NewRecorder()
rp.ServeHTTP(w, req)
return w.Code, upstreamHits.Load() - before, w.Body.String()
}
// The disallowed model (Opus) MUST be denied by the guardrail (model_blocked)
// before the upstream — this is the customer-reported bug (Opus let through
// under a Sonnet-only allowlist). The deny code distinguishes a guardrail
// block from a router model_not_routable, which would deny for the wrong
// reason.
t.Run("opus_denied_by_guardrail", func(t *testing.T) {
code, hits, body := send(t, "claude-opus-4-6")
assert.Equal(t, http.StatusForbidden, code, "Opus must be denied under a Sonnet-only allowlist; body=%s", body)
assert.Contains(t, body, "llm_policy.model_blocked", "denial must come from the guardrail allowlist, not routing; body=%s", body)
assert.Equal(t, int64(0), hits, "a denied request must never reach the Vertex upstream")
})
// The allowed model (Sonnet) passes the guardrail and reaches the upstream.
t.Run("sonnet_allowed", func(t *testing.T) {
code, hits, body := send(t, "claude-sonnet-4-5")
assert.Equal(t, http.StatusOK, code, "Sonnet is allowlisted and must be served; body=%s", body)
assert.Equal(t, int64(1), hits, "the allowed request must reach the Vertex upstream exactly once")
})
}

View File

@@ -0,0 +1,176 @@
package client
import (
"context"
"errors"
"net"
"time"
log "github.com/sirupsen/logrus"
)
const (
raceTotalTimeout = 40 * time.Second
raceFallbackDelay = 10 * time.Second
)
type raceAttempt struct {
conn net.Conn
err error
}
type raceOutcome struct {
conn net.Conn
err error
done bool
}
type connRace struct {
opener *FallbackOpener
peerKey string
remoteRelayServer RelayServer
preferForeign bool
raceCtx context.Context
otherCtx context.Context
cancelPreferred context.CancelFunc
cancelOther context.CancelFunc
results chan raceAttempt
fallbackTimer *time.Timer
otherStarted bool
settled int
lastErr error
}
type FallbackOpener struct {
home *Client
foreignStore *ForeignRelaysStore
fallbackDelay time.Duration
totalTimeout time.Duration
// openFn performs a single attempt. It is overridable in tests; when nil the
// real home/foreign dispatch in open is used.
openFn func(ctx context.Context, peerKey string, remoteRelayServer RelayServer, foreign bool) raceAttempt
}
func NewFallbackOpener(home *Client, foreignStore *ForeignRelaysStore) *FallbackOpener {
return &FallbackOpener{
home: home,
foreignStore: foreignStore,
fallbackDelay: raceFallbackDelay,
totalTimeout: raceTotalTimeout,
}
}
func (r *FallbackOpener) Run(ctx context.Context, peerKey string, remoteRelayServer RelayServer, preferForeign bool) (net.Conn, error) {
raceCtx, cancel := context.WithTimeout(ctx, r.totalTimeout)
defer cancel()
preferredCtx, cancelPreferred := context.WithCancel(raceCtx)
otherCtx, cancelOther := context.WithCancel(raceCtx)
race := &connRace{
opener: r,
peerKey: peerKey,
remoteRelayServer: remoteRelayServer,
preferForeign: preferForeign,
raceCtx: raceCtx,
otherCtx: otherCtx,
cancelPreferred: cancelPreferred,
cancelOther: cancelOther,
results: make(chan raceAttempt, 2),
fallbackTimer: time.NewTimer(r.fallbackDelay),
}
defer race.fallbackTimer.Stop()
go func() {
race.results <- r.open(preferredCtx, peerKey, remoteRelayServer, preferForeign)
}()
for {
select {
case <-race.fallbackTimer.C:
race.startOther()
case res := <-race.results:
if o := race.handleResult(res); o.done {
return o.conn, o.err
}
case <-raceCtx.Done():
return race.onTimeout()
}
}
}
func (c *connRace) startOther() {
if c.otherStarted {
return
}
c.otherStarted = true
c.fallbackTimer.Stop()
go func() {
c.results <- c.opener.open(c.otherCtx, c.peerKey, c.remoteRelayServer, !c.preferForeign)
}()
}
func (c *connRace) handleResult(res raceAttempt) raceOutcome {
if (res.err == nil && res.conn != nil) || errors.Is(res.err, ErrConnAlreadyExists) {
c.settled++
c.stop()
return raceOutcome{conn: res.conn, err: res.err, done: true}
}
c.lastErr = res.err
c.settled++
if !c.otherStarted {
c.startOther()
return raceOutcome{}
}
if c.settled == 2 {
c.cancelPreferred()
c.cancelOther()
return raceOutcome{err: c.lastErr, done: true}
}
return raceOutcome{}
}
func (c *connRace) onTimeout() (net.Conn, error) {
c.stop()
if c.lastErr != nil {
return nil, c.lastErr
}
return nil, c.raceCtx.Err()
}
func (c *connRace) stop() {
c.cancelPreferred()
c.cancelOther()
go c.opener.drainLoser(c.results, c.settled, c.otherStarted)
}
func (r *FallbackOpener) open(ctx context.Context, peerKey string, remoteRelayServer RelayServer, foreign bool) raceAttempt {
if r.openFn != nil {
return r.openFn(ctx, peerKey, remoteRelayServer, foreign)
}
if foreign {
conn, err := r.foreignStore.OpenConn(ctx, peerKey, remoteRelayServer)
return raceAttempt{conn: conn, err: err}
}
conn, err := r.home.OpenConn(ctx, peerKey)
return raceAttempt{conn: conn, err: err}
}
func (r *FallbackOpener) drainLoser(results chan raceAttempt, settled int, otherStarted bool) {
started := 1
if otherStarted {
started = 2
}
for i := settled; i < started; i++ {
res := <-results
if res.conn != nil {
if err := res.conn.Close(); err != nil {
log.Debugf("failed to close losing relay connection: %v", err)
}
}
}
}

View File

@@ -0,0 +1,450 @@
package client
import (
"context"
"errors"
"fmt"
"net"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// The FallbackOpener race is driven by a single goroutine (Run's select loop)
// with worker goroutines that communicate only through the buffered results
// channel and the two cancel contexts. These tests exercise that state machine
// in isolation via an injected openFn, so no relay server or network is needed.
// Timing is scaled down through the fallbackDelay/totalTimeout fields.
// raceFakeConn tracks whether Close was called. Only Close is exercised by the
// race logic (drainLoser closes losers; Run returns the winner untouched).
type raceFakeConn struct {
net.Conn
label string
closed atomic.Bool
}
func (c *raceFakeConn) Close() error {
c.closed.Store(true)
return nil
}
// raceAttemptScript describes how a single scripted attempt behaves.
type raceAttemptScript struct {
delay time.Duration
conn *raceFakeConn // non-nil => the attempt succeeds and returns this conn
err error // returned when conn is nil
// ignoreCtx makes the attempt complete after delay even if its context is
// cancelled. It models an OpenConn that produced a real connection right as
// the race cancelled it - exactly the case drainLoser must clean up.
ignoreCtx bool
}
// fakeOpener replaces FallbackOpener.open. Scripts are keyed by the foreign
// flag, so which script is "preferred" depends on the preferForeign argument
// passed to Run.
type fakeOpener struct {
mu sync.Mutex
scripts map[bool]raceAttemptScript
calls []bool // foreign flag of each open() invocation, in order
}
func (f *fakeOpener) open(ctx context.Context, _ string, _ RelayServer, foreign bool) raceAttempt {
f.mu.Lock()
f.calls = append(f.calls, foreign)
s, ok := f.scripts[foreign]
f.mu.Unlock()
if !ok {
return raceAttempt{err: fmt.Errorf("no script for foreign=%v", foreign)}
}
if s.delay > 0 {
timer := time.NewTimer(s.delay)
defer timer.Stop()
if s.ignoreCtx {
<-timer.C
} else {
select {
case <-timer.C:
case <-ctx.Done():
return raceAttempt{err: ctx.Err()}
}
}
}
if s.conn != nil {
return raceAttempt{conn: s.conn}
}
return raceAttempt{err: s.err}
}
func (f *fakeOpener) callCount() int {
f.mu.Lock()
defer f.mu.Unlock()
return len(f.calls)
}
func (f *fakeOpener) firstCallForeign(t *testing.T) bool {
t.Helper()
f.mu.Lock()
defer f.mu.Unlock()
require.NotEmpty(t, f.calls, "expected at least one open attempt")
return f.calls[0]
}
func newTestOpener(f *fakeOpener, fallbackDelay, totalTimeout time.Duration) *FallbackOpener {
o := NewFallbackOpener(nil, nil)
o.openFn = f.open
o.fallbackDelay = fallbackDelay
o.totalTimeout = totalTimeout
return o
}
const (
// controller prefers the home relay, i.e. preferForeign == false.
preferHome = false
preferForeign = true
)
var errAttempt = errors.New("attempt failed")
// The preferred attempt wins before the fallback timer fires, so the other
// attempt is never started.
func TestFallbackOpener_PreferredWinsImmediately(t *testing.T) {
homeConn := &raceFakeConn{label: "home"}
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {conn: homeConn}, // preferred (home): instant success
true: {delay: 5 * time.Second, conn: foreignConn}, // would never finish in time
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.NoError(t, err)
require.Same(t, homeConn, conn)
assert.Equal(t, 1, f.callCount(), "other attempt must not start when preferred wins first")
assert.False(t, f.firstCallForeign(t), "home must be tried first when preferring home")
assert.False(t, foreignConn.closed.Load())
}
// preferForeign flips which relay is tried first.
func TestFallbackOpener_PreferForeignRoutesForeignFirst(t *testing.T) {
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
true: {conn: foreignConn}, // preferred (foreign): instant success
false: {delay: 5 * time.Second, conn: &raceFakeConn{}},
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferForeign)
require.NoError(t, err)
require.Same(t, foreignConn, conn)
assert.Equal(t, 1, f.callCount())
assert.True(t, f.firstCallForeign(t), "foreign must be tried first when preferring foreign")
}
// ErrConnAlreadyExists counts as success: Run returns it and does not start the
// other attempt.
func TestFallbackOpener_ErrConnAlreadyExistsIsSuccess(t *testing.T) {
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {err: ErrConnAlreadyExists},
true: {delay: 5 * time.Second, conn: &raceFakeConn{}},
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.ErrorIs(t, err, ErrConnAlreadyExists)
assert.Nil(t, conn)
assert.Equal(t, 1, f.callCount(), "other attempt must not start on ErrConnAlreadyExists")
}
// A preferred failure starts the other attempt immediately, without waiting for
// the fallback timer.
func TestFallbackOpener_PreferredFailsStartsOtherBeforeTimer(t *testing.T) {
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {err: errAttempt}, // preferred fails instantly
true: {delay: 5 * time.Millisecond, conn: foreignConn},
}}
fallbackDelay := 500 * time.Millisecond
o := newTestOpener(f, fallbackDelay, 2*time.Second)
start := time.Now()
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
elapsed := time.Since(start)
require.NoError(t, err)
require.Same(t, foreignConn, conn)
assert.Equal(t, 2, f.callCount())
assert.Less(t, elapsed, fallbackDelay/2, "fallback must not wait for the timer after a preferred failure")
}
// When the preferred attempt is slow, the fallback timer starts the other
// attempt and its success wins.
func TestFallbackOpener_TimerStartsOtherWhenPreferredSlow(t *testing.T) {
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 5 * time.Second}, // preferred hangs until cancelled
true: {delay: 5 * time.Millisecond, conn: foreignConn},
}}
fallbackDelay := 40 * time.Millisecond
o := newTestOpener(f, fallbackDelay, 2*time.Second)
start := time.Now()
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
elapsed := time.Since(start)
require.NoError(t, err)
require.Same(t, foreignConn, conn)
assert.Equal(t, 2, f.callCount())
assert.GreaterOrEqual(t, elapsed, fallbackDelay, "other must not start before the fallback timer fires")
}
// Both attempts fail: Run returns the last error and tries both relays.
func TestFallbackOpener_BothFail(t *testing.T) {
errOther := errors.New("other failed")
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {err: errAttempt},
true: {err: errOther},
}}
o := newTestOpener(f, 40*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.Error(t, err)
assert.Nil(t, conn)
assert.ErrorIs(t, err, errOther, "the most recent error should be surfaced")
assert.Equal(t, 2, f.callCount())
}
// When both attempts succeed, drainLoser must close the losing connection so it
// is not leaked. Here the preferred attempt wins and the foreign loser - which
// produced a real conn despite being cancelled - is closed.
func TestFallbackOpener_DoubleSuccessClosesLoser(t *testing.T) {
homeConn := &raceFakeConn{label: "home"}
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 30 * time.Millisecond, conn: homeConn}, // preferred wins
true: {delay: 80 * time.Millisecond, conn: foreignConn, ignoreCtx: true}, // loser yields a conn after cancel
}}
o := newTestOpener(f, 15*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.NoError(t, err)
require.Same(t, homeConn, conn)
assert.Equal(t, 2, f.callCount())
assert.False(t, homeConn.closed.Load(), "the winning connection must not be closed")
require.Eventually(t, foreignConn.closed.Load, time.Second, 5*time.Millisecond,
"the losing connection must be closed by drainLoser")
}
// Winner selection is purely by result arrival order, not by preference: when
// the non-preferred attempt returns first it wins even though home was
// preferred. This is the mechanism behind the split-relay concern - two peers
// racing independently have no shared tie-break, so under adversarial timing
// they can settle on different relays. Documented here as current behavior.
func TestFallbackOpener_FasterOtherWinsDespitePreference(t *testing.T) {
homeConn := &raceFakeConn{label: "home"}
foreignConn := &raceFakeConn{label: "foreign"}
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 60 * time.Millisecond, conn: homeConn, ignoreCtx: true}, // preferred but slower
true: {delay: 5 * time.Millisecond, conn: foreignConn}, // other is faster
}}
o := newTestOpener(f, 15*time.Millisecond, 2*time.Second)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.NoError(t, err)
require.Same(t, foreignConn, conn, "the first successful attempt wins regardless of preference")
require.Eventually(t, homeConn.closed.Load, time.Second, 5*time.Millisecond,
"the slower preferred attempt becomes the loser and is closed")
}
// The whole race is bounded by totalTimeout. With no attempt succeeding or
// failing, Run returns the deadline error.
func TestFallbackOpener_TotalTimeout(t *testing.T) {
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 5 * time.Second},
true: {delay: 5 * time.Second},
}}
o := newTestOpener(f, 20*time.Millisecond, 80*time.Millisecond)
conn, err := o.Run(context.Background(), "peer", RelayServer{Addr: "srv"}, preferHome)
require.Error(t, err)
assert.Nil(t, conn)
assert.ErrorIs(t, err, context.DeadlineExceeded)
}
// Cancelling the caller's context aborts the race promptly with the cancel
// error, even before the fallback timer would fire.
func TestFallbackOpener_ParentContextCanceled(t *testing.T) {
f := &fakeOpener{scripts: map[bool]raceAttemptScript{
false: {delay: 5 * time.Second},
true: {delay: 5 * time.Second},
}}
// fallbackDelay large so the timer never fires; only the parent cancel ends the race.
o := newTestOpener(f, 5*time.Second, 5*time.Second)
ctx, cancel := context.WithCancel(context.Background())
go func() {
time.Sleep(20 * time.Millisecond)
cancel()
}()
start := time.Now()
conn, err := o.Run(ctx, "peer", RelayServer{Addr: "srv"}, preferHome)
require.Error(t, err)
assert.Nil(t, conn)
assert.ErrorIs(t, err, context.Canceled)
assert.Less(t, time.Since(start), time.Second, "must return shortly after the parent context is cancelled")
assert.Equal(t, 1, f.callCount(), "the other attempt must not start")
}
// rendezvous models the relay-level requirement that a relayed connection is
// established only once BOTH peers subscribe to the same relay server. arrive
// records a peer's presence on a relay and returns a channel that closes when
// the second peer arrives, so an attempt can only complete after a real
// rendezvous - the same coupling the production code depends on.
type rendezvous struct {
mu sync.Mutex
arrivals map[string]int
gates map[string]chan struct{}
}
func newRendezvous() *rendezvous {
return &rendezvous{arrivals: map[string]int{}, gates: map[string]chan struct{}{}}
}
func (r *rendezvous) arrive(relay string) <-chan struct{} {
r.mu.Lock()
defer r.mu.Unlock()
g, ok := r.gates[relay]
if !ok {
g = make(chan struct{})
r.gates[relay] = g
}
r.arrivals[relay]++
if r.arrivals[relay] == 2 {
close(g)
}
return g
}
// splitPeer is one peer's view of the two relays. relayFor maps the foreign
// flag to a relay name; postDelay is how long after the rendezvous that peer's
// OpenConn takes to return (its per-relay subscribe latency). Different values
// per peer model the asymmetric timing that triggers finding #1.
type splitPeer struct {
rv *rendezvous
relayFor map[bool]string
postDelay map[string]time.Duration
}
func (p *splitPeer) open(ctx context.Context, _ string, _ RelayServer, foreign bool) raceAttempt {
relay := p.relayFor[foreign]
select {
case <-p.rv.arrive(relay):
case <-ctx.Done():
return raceAttempt{err: ctx.Err()}
}
timer := time.NewTimer(p.postDelay[relay])
defer timer.Stop()
select {
case <-timer.C:
case <-ctx.Done():
return raceAttempt{err: ctx.Err()}
}
return raceAttempt{conn: &raceFakeConn{label: relay}}
}
// TestFallbackOpener_SplitRelaySelection reproduces finding #1: the two peers
// run FallbackOpener.Run independently with no shared tie-break, so the winner
// is chosen purely by local result-arrival order. Under an adversarial - but
// self-consistent - timing profile they settle on DIFFERENT relays.
//
// Both peers prefer relayA (the controller's home). The split needs each peer's
// preferred relayA to be slow enough that both start their fallback (so both
// relays actually rendezvous), and then each peer's fast path to be a different
// relay:
// - peerA: relayA slow (abandoned), relayB fast -> peerA wins relayB
// - peerB: relayA fast (wins), relayB slow -> peerB wins relayA
//
// Each winner then cancels its attempt on the relay the OTHER peer actually
// kept, leaving two half-open relayed connections that were both reported as
// successful. When a deterministic cross-peer tie-break is added to fix this,
// invert the assertion below to require convergence.
func TestFallbackOpener_SplitRelaySelection(t *testing.T) {
const (
relayA = "relayA" // controller's home relay; both peers prefer it
relayB = "relayB" // non-controller's home relay
)
rv := newRendezvous()
peerA := &splitPeer{
rv: rv,
relayFor: map[bool]string{false: relayA, true: relayB}, // home=relayA
postDelay: map[string]time.Duration{
relayA: 500 * time.Millisecond, // preferred but slow -> abandoned
relayB: 10 * time.Millisecond, // fallback is fast -> peerA wins relayB
},
}
peerB := &splitPeer{
rv: rv,
relayFor: map[bool]string{false: relayB, true: relayA}, // home=relayB
postDelay: map[string]time.Duration{
relayA: 80 * time.Millisecond, // preferred, wins - but only after starting fallback
relayB: 500 * time.Millisecond, // fallback (home) is slow -> abandoned
},
}
fallbackDelay := 30 * time.Millisecond
newPeerOpener := func(p *splitPeer) *FallbackOpener {
o := NewFallbackOpener(nil, nil)
o.openFn = p.open
o.fallbackDelay = fallbackDelay
o.totalTimeout = 5 * time.Second
return o
}
oA := newPeerOpener(peerA)
oB := newPeerOpener(peerB)
type result struct {
conn net.Conn
err error
}
var ra, rb result
var wg sync.WaitGroup
wg.Add(2)
go func() {
defer wg.Done()
ra.conn, ra.err = oA.Run(context.Background(), "peerB", RelayServer{Addr: relayB}, preferHome)
}()
go func() {
defer wg.Done()
rb.conn, rb.err = oB.Run(context.Background(), "peerA", RelayServer{Addr: relayA}, preferForeign)
}()
wg.Wait()
require.NoError(t, ra.err)
require.NoError(t, rb.err)
aRelay := ra.conn.(*raceFakeConn).label
bRelay := rb.conn.(*raceFakeConn).label
t.Logf("peerA settled on %s, peerB settled on %s", aRelay, bRelay)
assert.Equal(t, aRelay, bRelay,
"peers selected different relays with no cross-peer tie-break")
assert.Equal(t, relayB, aRelay, "peerA abandoned its slow preferred relay and won the fallback")
assert.Equal(t, relayA, bRelay, "peerB won its preferred relay after starting the fallback")
}

View File

@@ -0,0 +1,155 @@
package client
import (
"context"
"net"
"sync"
"time"
log "github.com/sirupsen/logrus"
"golang.org/x/sync/singleflight"
relayAuth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
)
type foreignRelay struct {
client *Client
created time.Time
inUse int
}
type ForeignRelaysStore struct {
mu sync.RWMutex
clients map[string]*foreignRelay
group singleflight.Group
ctx context.Context
tokenStore *relayAuth.TokenStore
peerID string
mtu uint16
transportFallback *transportFallback
onDisconnect func(string)
keepUnusedServerTime time.Duration
}
func NewForeignRelaysStore(ctx context.Context, tokenStore *relayAuth.TokenStore, peerID string, mtu uint16, transportFallback *transportFallback, onDisconnect func(string), keepUnusedServerTime time.Duration) *ForeignRelaysStore {
return &ForeignRelaysStore{
clients: make(map[string]*foreignRelay),
ctx: ctx,
tokenStore: tokenStore,
peerID: peerID,
mtu: mtu,
transportFallback: transportFallback,
onDisconnect: onDisconnect,
keepUnusedServerTime: keepUnusedServerTime,
}
}
func (f *ForeignRelaysStore) OpenConn(ctx context.Context, peerKey string, remoteRelayServer RelayServer) (net.Conn, error) {
fr, err := f.acquire(remoteRelayServer)
if err != nil {
return nil, err
}
defer f.release(fr)
return fr.client.OpenConn(ctx, peerKey)
}
func (f *ForeignRelaysStore) acquire(remoteRelayServer RelayServer) (*foreignRelay, error) {
f.mu.Lock()
if fr, ok := f.clients[remoteRelayServer.Addr]; ok {
fr.inUse++
f.mu.Unlock()
return fr, nil
}
f.mu.Unlock()
v, err, _ := f.group.Do(remoteRelayServer.Addr, func() (any, error) {
f.mu.RLock()
fr, ok := f.clients[remoteRelayServer.Addr]
f.mu.RUnlock()
if ok {
return fr, nil
}
relayClient := NewClientWithServerIP(remoteRelayServer.Addr, remoteRelayServer.IP, f.tokenStore, f.peerID, f.mtu)
relayClient.SetTransportFallback(f.transportFallback)
if err := relayClient.Connect(f.ctx); err != nil {
return nil, err
}
relayClient.SetOnDisconnectListener(f.onDisconnect)
f.mu.Lock()
fr = &foreignRelay{client: relayClient, created: time.Now()}
f.clients[remoteRelayServer.Addr] = fr
f.mu.Unlock()
return fr, nil
})
if err != nil {
return nil, err
}
fr := v.(*foreignRelay)
f.mu.Lock()
if cur, ok := f.clients[remoteRelayServer.Addr]; !ok || cur != fr {
f.mu.Unlock()
return f.acquire(remoteRelayServer)
}
fr.inUse++
f.mu.Unlock()
return fr, nil
}
func (f *ForeignRelaysStore) release(fr *foreignRelay) {
f.mu.Lock()
fr.inUse--
f.mu.Unlock()
}
func (f *ForeignRelaysStore) evict(serverAddress string) {
f.mu.Lock()
defer f.mu.Unlock()
if _, ok := f.clients[serverAddress]; ok {
delete(f.clients, serverAddress)
log.Debugf("evicted disconnected foreign relay client: %s", serverAddress)
}
}
func (f *ForeignRelaysStore) cleanupUnused() {
f.mu.Lock()
defer f.mu.Unlock()
for addr, fr := range f.clients {
if time.Since(fr.created) <= f.keepUnusedServerTime {
continue
}
if fr.inUse > 0 {
continue
}
if fr.client.HasConns() {
continue
}
fr.client.SetOnDisconnectListener(nil)
go func() {
_ = fr.client.Close()
}()
log.Debugf("clean up unused relay server connection: %s", addr)
delete(f.clients, addr)
}
}
func (f *ForeignRelaysStore) states() []RelayConnState {
f.mu.RLock()
clients := make([]*Client, 0, len(f.clients))
for _, fr := range f.clients {
clients = append(clients, fr.client)
}
f.mu.RUnlock()
states := make([]RelayConnState, 0, len(clients))
for _, c := range clients {
states = append(states, relayConnState(c))
}
return states
}

View File

@@ -22,27 +22,6 @@ var (
ErrRelayClientNotConnected = fmt.Errorf("relay client not connected")
)
// RelayTrack hold the relay clients for the foreign relay servers.
// With the mutex can ensure we can open new connection in case the relay connection has been established with
// the relay server.
type RelayTrack struct {
sync.RWMutex
relayClient *Client
err error
created time.Time
// ready is closed once the dial started by openConnVia finishes (relayClient
// or err is set). Callers reusing a track wait on this instead of the track
// lock, so the dial never runs under rt.Lock.
ready chan struct{}
}
func NewRelayTrack() *RelayTrack {
return &RelayTrack{
created: time.Now(),
ready: make(chan struct{}),
}
}
type OnServerCloseListener func()
// ManagerOption configures a Manager at construction time.
@@ -59,6 +38,11 @@ type RelayConnState struct {
Err error
}
type RelayServer struct {
Addr string
IP netip.Addr
}
// WithMaxBackoffInterval caps the exponential backoff between reconnect
// attempts to the home relay. A non-positive value keeps the default.
func WithMaxBackoffInterval(d time.Duration) ManagerOption {
@@ -83,8 +67,7 @@ type Manager struct {
relayClientMu sync.RWMutex
reconnectGuard *Guard
relayClients map[string]*RelayTrack
relayClientsMutex sync.RWMutex
foreign *ForeignRelaysStore
onDisconnectedListeners map[string]*list.List
onReconnectedListenerFn func()
@@ -120,7 +103,6 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
ConnectionTimeout: defaultConnectionTimeout,
TransportFallback: tf,
},
relayClients: make(map[string]*RelayTrack),
onDisconnectedListeners: make(map[string]*list.List),
cleanupInterval: relayCleanupInterval,
keepUnusedServerTime: keepUnusedServerTime,
@@ -128,6 +110,7 @@ func NewManager(ctx context.Context, serverURLs []string, peerID string, mtu uin
for _, opt := range opts {
opt(m)
}
m.foreign = NewForeignRelaysStore(ctx, tokenStore, peerID, mtu, tf, m.onServerDisconnected, m.keepUnusedServerTime)
m.serverPicker.ServerURLs.Store(serverURLs)
m.reconnectGuard = NewGuard(m.serverPicker, m.maxBackoffInterval)
return m
@@ -159,40 +142,26 @@ func (m *Manager) Serve() error {
return err
}
// OpenConn opens a connection to the given peer key. If the peer is on the same relay server, the connection will be
// established via the relay server. If the peer is on a different relay server, the manager will establish a new
// connection to the relay server. It returns back with a net.Conn what represent the remote peer connection.
//
// serverIP, when valid and serverAddress is foreign, is used as a dial target if the FQDN-based dial fails.
// Ignored for the local home-server path. TLS verification still uses the FQDN via SNI.
func (m *Manager) OpenConn(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
func (m *Manager) OpenConn(ctx context.Context, remoteRelayServer RelayServer, peerKey string, preferForeign bool) (net.Conn, error) {
m.relayClientMu.RLock()
defer m.relayClientMu.RUnlock()
relayClient := m.relayClient
m.relayClientMu.RUnlock()
if m.relayClient == nil {
if relayClient == nil {
return nil, ErrRelayClientNotConnected
}
foreign, err := m.isForeignServer(serverAddress)
foreign, err := m.isForeignServer(relayClient, remoteRelayServer.Addr)
if err != nil {
return nil, err
}
var (
netConn net.Conn
)
if !foreign {
log.Debugf("open peer connection via permanent server: %s", peerKey)
netConn, err = m.relayClient.OpenConn(ctx, peerKey)
} else {
log.Debugf("open peer connection via foreign server: %s", serverAddress)
netConn, err = m.openConnVia(ctx, serverAddress, peerKey, serverIP)
}
if err != nil {
return nil, err
return relayClient.OpenConn(ctx, peerKey)
}
return netConn, err
opener := NewFallbackOpener(relayClient, m.foreign)
return opener.Run(ctx, peerKey, remoteRelayServer, preferForeign)
}
// Ready returns true if the home Relay client is connected to the relay server.
@@ -223,7 +192,7 @@ func (m *Manager) AddCloseListener(serverAddress string, onClosedListener OnServ
return ErrRelayClientNotConnected
}
foreign, err := m.isForeignServer(serverAddress)
foreign, err := m.isForeignServer(m.relayClient, serverAddress)
if err != nil {
return err
}
@@ -287,26 +256,7 @@ func (m *Manager) RelayStates() []RelayConnState {
states = append(states, st)
}
// Snapshot the tracks, then query each outside the map lock: a track can be
// held by an in-progress Connect, and blocking on it must not stall other
// relay operations.
m.relayClientsMutex.RLock()
tracks := make([]*RelayTrack, 0, len(m.relayClients))
for _, rt := range m.relayClients {
tracks = append(tracks, rt)
}
m.relayClientsMutex.RUnlock()
// Only connected foreign relays carry state; a failed connect is evicted
// immediately (openConnVia), so there is no error state to surface.
for _, rt := range tracks {
rt.RLock()
rc := rt.relayClient
rt.RUnlock()
if rc != nil {
states = append(states, relayConnState(rc))
}
}
states = append(states, m.foreign.states()...)
return states
}
@@ -327,76 +277,6 @@ func (m *Manager) UpdateToken(token *relayAuth.Token) error {
return m.tokenStore.UpdateToken(token)
}
func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string, serverIP netip.Addr) (net.Conn, error) {
// check if already has a connection to the desired relay server
m.relayClientsMutex.RLock()
rt, ok := m.relayClients[serverAddress]
m.relayClientsMutex.RUnlock()
if ok {
return m.openConnOnTrack(ctx, rt, peerKey)
}
// if not, establish a new connection but check it again (because changed the lock type) before starting the
// connection
m.relayClientsMutex.Lock()
rt, ok = m.relayClients[serverAddress]
if ok {
m.relayClientsMutex.Unlock()
return m.openConnOnTrack(ctx, rt, peerKey)
}
// Publish the track and release the map lock BEFORE dialing, so the dial does
// not run under rt.Lock (which would block RelayStates and the cleanup loop
// for the full dial). Concurrent callers find this track and wait on rt.ready.
rt = NewRelayTrack()
m.relayClients[serverAddress] = rt
m.relayClientsMutex.Unlock()
relayClient := NewClientWithServerIP(serverAddress, serverIP, m.tokenStore, m.peerID, m.mtu)
relayClient.SetTransportFallback(m.transportFallback)
err := relayClient.Connect(m.ctx)
if err != nil {
rt.Lock()
rt.err = err
rt.Unlock()
close(rt.ready)
m.relayClientsMutex.Lock()
delete(m.relayClients, serverAddress)
m.relayClientsMutex.Unlock()
return nil, err
}
// if connection closed then delete the relay client from the list
relayClient.SetOnDisconnectListener(m.onServerDisconnected)
rt.Lock()
rt.relayClient = relayClient
rt.Unlock()
close(rt.ready)
return relayClient.OpenConn(ctx, peerKey)
}
// openConnOnTrack opens a peer connection through an existing relay track,
// waiting for the dial started by another openConnVia call to finish. It waits
// on rt.ready rather than the track lock, so it neither holds nor contends the
// track lock across the dial.
func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (net.Conn, error) {
select {
case <-rt.ready:
case <-ctx.Done():
return nil, ctx.Err()
}
rt.RLock()
defer rt.RUnlock()
if rt.err != nil {
return nil, rt.err
}
if rt.relayClient == nil {
return nil, ErrRelayClientNotConnected
}
return rt.relayClient.OpenConn(ctx, peerKey)
}
func (m *Manager) onServerConnected() {
m.listenerLock.Lock()
defer m.listenerLock.Unlock()
@@ -422,21 +302,12 @@ func (m *Manager) onServerDisconnected(serverAddress string) {
m.relayClientMu.Unlock()
if !isHome {
m.evictForeignRelay(serverAddress)
m.foreign.evict(serverAddress)
}
m.notifyOnDisconnectListeners(serverAddress)
}
func (m *Manager) evictForeignRelay(serverAddress string) {
m.relayClientsMutex.Lock()
defer m.relayClientsMutex.Unlock()
if _, ok := m.relayClients[serverAddress]; ok {
delete(m.relayClients, serverAddress)
log.Debugf("evicted disconnected foreign relay client: %s", serverAddress)
}
}
func (m *Manager) listenGuardEvent(ctx context.Context) {
for {
select {
@@ -459,8 +330,8 @@ func (m *Manager) storeClient(client *Client) {
m.relayClient.SetOnDisconnectListener(m.onServerDisconnected)
}
func (m *Manager) isForeignServer(address string) (bool, error) {
rAddr, err := m.relayClient.ServerInstanceURL()
func (m *Manager) isForeignServer(relayClient *Client, address string) (bool, error) {
rAddr, err := relayClient.ServerInstanceURL()
if err != nil {
return false, fmt.Errorf("relay client not connected")
}
@@ -475,50 +346,11 @@ func (m *Manager) startCleanupLoop() {
case <-m.ctx.Done():
return
case <-ticker.C:
m.cleanUpUnusedRelays()
m.foreign.cleanupUnused()
}
}
}
func (m *Manager) cleanUpUnusedRelays() {
m.relayClientsMutex.Lock()
defer m.relayClientsMutex.Unlock()
for addr, rt := range m.relayClients {
rt.Lock()
// if the connection failed to the server the relay client will be nil
// but the instance will be kept in the relayClients until the next locking
if rt.err != nil {
rt.Unlock()
continue
}
// dial still in progress (openConnVia publishes the track before Connect
// completes and no longer holds rt.Lock during it), nothing to clean up.
if rt.relayClient == nil {
rt.Unlock()
continue
}
if time.Since(rt.created) <= m.keepUnusedServerTime {
rt.Unlock()
continue
}
if rt.relayClient.HasConns() {
rt.Unlock()
continue
}
rt.relayClient.SetOnDisconnectListener(nil)
go func() {
_ = rt.relayClient.Close()
}()
log.Debugf("clean up unused relay server connection: %s", addr)
delete(m.relayClients, addr)
rt.Unlock()
}
}
func (m *Manager) addListener(serverAddress string, onClosedListener OnServerCloseListener) {
m.listenerLock.Lock()
defer m.listenerLock.Unlock()

View File

@@ -2,17 +2,14 @@ package client
import (
"context"
"net/netip"
"testing"
"time"
"github.com/stretchr/testify/require"
)
// TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial drives a real, hanging foreign
// relay dial and asserts cleanUpUnusedRelays does not stall behind it.
// relay dial and asserts the foreign store cleanup does not stall behind it.
func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
serverAddr := stallingRelayListener(t)
serverAddr, accepted := stallingRelayListener(t)
mCtx, mCancel := context.WithCancel(context.Background())
t.Cleanup(mCancel)
@@ -22,39 +19,32 @@ func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
dialDone := make(chan struct{})
go func() {
defer close(dialDone)
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
}()
// The track appears in the map once the dial is in flight.
require.Eventually(t, func() bool {
m.relayClientsMutex.RLock()
defer m.relayClientsMutex.RUnlock()
_, ok := m.relayClients[serverAddr]
return ok
}, 5*time.Second, 5*time.Millisecond, "relay dial did not start")
select {
case <-accepted:
case <-time.After(5 * time.Second):
t.Fatal("relay dial did not reach the listener")
}
cleanupDone := make(chan struct{})
go func() {
defer close(cleanupDone)
m.cleanUpUnusedRelays()
m.foreign.cleanupUnused()
}()
select {
case <-cleanupDone:
case <-time.After(2 * time.Second):
t.Fatal("cleanUpUnusedRelays blocked on an in-progress relay dial while holding the relay map lock")
t.Fatal("cleanupUnused blocked on an in-progress relay dial")
}
m.relayClientsMutex.RLock()
_, stillTracked := m.relayClients[serverAddr]
m.relayClientsMutex.RUnlock()
require.True(t, stillTracked, "an in-progress relay dial must not be evicted by cleanup")
// Release the hanging dial so the goroutine can exit cleanly.
mCancel()
select {
case <-dialDone:
case <-time.After(5 * time.Second):
t.Fatal("openConnVia did not return after context cancellation")
t.Fatal("foreign OpenConn did not return after context cancellation")
}
}

View File

@@ -3,7 +3,6 @@ package client
import (
"context"
"net"
"net/netip"
"sync"
"testing"
"time"
@@ -13,13 +12,16 @@ import (
// stallingRelayListener accepts TCP connections and holds them open without ever
// responding, so a relay handshake dialed against it blocks until its context is
// cancelled. It returns the "rel://host:port" URL to dial.
func stallingRelayListener(t *testing.T) string {
// cancelled. accepted is signalled once per incoming connection so a caller can
// wait until a dial has actually reached the listener. It returns the
// "rel://host:port" URL to dial.
func stallingRelayListener(t *testing.T) (string, <-chan struct{}) {
t.Helper()
ln, err := net.Listen("tcp", "127.0.0.1:0")
require.NoError(t, err)
accepted := make(chan struct{}, 1)
var mu sync.Mutex
var conns []net.Conn
go func() {
@@ -31,6 +33,10 @@ func stallingRelayListener(t *testing.T) string {
mu.Lock()
conns = append(conns, c)
mu.Unlock()
select {
case accepted <- struct{}{}:
default:
}
}
}()
t.Cleanup(func() {
@@ -42,14 +48,14 @@ func stallingRelayListener(t *testing.T) string {
mu.Unlock()
})
return "rel://" + ln.Addr().String()
return "rel://" + ln.Addr().String(), accepted
}
// TestRelayStates_DoesNotBlockOnRealHangingDial is a regression test for
// RelayStates() called by a "status -d command" hanging behind an in-progress
// relay dial.
// foreign relay dial.
func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
serverAddr := stallingRelayListener(t)
serverAddr, accepted := stallingRelayListener(t)
mCtx, mCancel := context.WithCancel(context.Background())
t.Cleanup(mCancel)
@@ -59,15 +65,14 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
dialDone := make(chan struct{})
go func() {
defer close(dialDone)
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
}()
require.Eventually(t, func() bool {
m.relayClientsMutex.RLock()
defer m.relayClientsMutex.RUnlock()
_, ok := m.relayClients[serverAddr]
return ok
}, 5*time.Second, 5*time.Millisecond, "relay dial did not start")
select {
case <-accepted:
case <-time.After(5 * time.Second):
t.Fatal("relay dial did not reach the listener")
}
done := make(chan []RelayConnState, 1)
go func() {
@@ -86,6 +91,6 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
select {
case <-dialDone:
case <-time.After(5 * time.Second):
t.Fatal("openConnVia did not return after context cancellation")
t.Fatal("foreign OpenConn did not return after context cancellation")
}
}

View File

@@ -3,7 +3,6 @@ package client
import (
"context"
"io"
"net/netip"
"testing"
"time"
@@ -85,7 +84,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
t.Run("no server IP, dial fails", func(t *testing.T) {
dialCtx, dialCancel := context.WithTimeout(ctx, 5*time.Second)
defer dialCancel()
_, err := mgrAlice.OpenConn(dialCtx, brokenFQDN, "bob", netip.Addr{})
_, err := mgrAlice.OpenConn(dialCtx, RelayServer{Addr: brokenFQDN}, "bob", true)
if err == nil {
t.Fatalf("expected OpenConn to fail without server IP, got success")
}
@@ -95,7 +94,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
// Bob waits for Alice's incoming peer connection on his side.
bobSideCh := make(chan error, 1)
go func() {
conn, err := mgrBob.OpenConn(ctx, bobRealAddr, "alice", netip.Addr{})
conn, err := mgrBob.OpenConn(ctx, RelayServer{Addr: bobRealAddr}, "alice", false)
if err != nil {
bobSideCh <- err
return
@@ -113,7 +112,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
bobSideCh <- nil
}()
aliceConn, err := mgrAlice.OpenConn(ctx, brokenFQDN, "bob", bobAdvertisedIP)
aliceConn, err := mgrAlice.OpenConn(ctx, RelayServer{Addr: brokenFQDN, IP: bobAdvertisedIP}, "bob", true)
if err != nil {
t.Fatalf("alice OpenConn with server IP: %s", err)
}

View File

@@ -3,7 +3,6 @@ package client
import (
"context"
"fmt"
"net/netip"
"testing"
"time"
@@ -106,11 +105,11 @@ func TestForeignConn(t *testing.T) {
if err != nil {
t.Fatalf("failed to get relay address: %s", err)
}
connAliceToBob, err := clientAlice.OpenConn(ctx, bobsSrvAddr, "bob", netip.Addr{})
connAliceToBob, err := clientAlice.OpenConn(ctx, RelayServer{Addr: bobsSrvAddr}, "bob", true)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}
connBobToAlice, err := clientBob.OpenConn(ctx, bobsSrvAddr, "alice", netip.Addr{})
connBobToAlice, err := clientBob.OpenConn(ctx, RelayServer{Addr: bobsSrvAddr}, "alice", false)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}
@@ -210,7 +209,7 @@ func TestForeginConnClose(t *testing.T) {
if err != nil {
t.Fatalf("failed to serve manager: %s", err)
}
conn, err := mgr.OpenConn(ctx, toURL(srvCfg2)[0], "bob", netip.Addr{})
conn, err := mgr.OpenConn(ctx, RelayServer{Addr: toURL(srvCfg2)[0]}, "bob", true)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}
@@ -302,7 +301,7 @@ func TestForeignAutoClose(t *testing.T) {
}
t.Log("open connection to another peer")
if _, err = mgr.OpenConn(ctx, foreignServerURL, "anotherpeer", netip.Addr{}); err == nil {
if _, err = mgr.OpenConn(ctx, RelayServer{Addr: foreignServerURL}, "anotherpeer", true); err == nil {
t.Fatalf("should have failed to open connection to another peer")
}
@@ -372,7 +371,7 @@ func TestAutoReconnect(t *testing.T) {
if err != nil {
t.Errorf("failed to get relay address: %s", err)
}
conn, err := clientAlice.OpenConn(ctx, ra, "bob", netip.Addr{})
conn, err := clientAlice.OpenConn(ctx, RelayServer{Addr: ra}, "bob", false)
if err != nil {
t.Errorf("failed to bind channel: %s", err)
}
@@ -392,7 +391,7 @@ func TestAutoReconnect(t *testing.T) {
}
log.Infof("reopent the connection")
_, err = clientAlice.OpenConn(ctx, ra, "bob", netip.Addr{})
_, err = clientAlice.OpenConn(ctx, RelayServer{Addr: ra}, "bob", false)
if err != nil {
t.Errorf("failed to open channel: %s", err)
}
@@ -454,7 +453,7 @@ func TestNotifierDoubleAdd(t *testing.T) {
t.Fatalf("failed to serve manager: %s", err)
}
conn1, err := clientAlice.OpenConn(ctx, clientAlice.ServerURLs()[0], "bob", netip.Addr{})
conn1, err := clientAlice.OpenConn(ctx, RelayServer{Addr: clientAlice.ServerURLs()[0]}, "bob", false)
if err != nil {
t.Fatalf("failed to bind channel: %s", err)
}