Compare commits

..

2 Commits

Author SHA1 Message Date
Zoltan Papp
f1c06ef1a4 [client] Return empty map from test mock GetStats to satisfy nilnil linter 2026-07-19 21:48:53 +02:00
Zoltan Papp
13433af4a5 [client] Preserve routed allowed IPs across lazy connection idle transitions
When a peer went idle under lazy connections, the WireGuard peer was
removed and the wake endpoint re-armed with only the overlay allowed
IPs, dropping routed prefixes installed by the route manager. The
asynchronous route-watcher re-add uses update_only, which is a silent
no-op while the peer is absent, so routed subnets stayed black-holed
until the peer was woken by other means.

Split the idle transition from the full close: Conn.Idle tears down
transports, cancels pending delayed endpoint updates and updates the
status without touching the WireGuard peer. The activity listener then
arms the wake endpoint via the new IdlePeerEndpoint operation, which
removes and re-creates the peer in a single transaction: handshake
state is dropped (keeping the handshake-first wake semantics of packet
staging and first-packet capture/reinjection) while the currently
installed allowed IPs are preserved. The read-modify-write runs under
the interface mutex, serializing it against concurrent allowed IP
updates from the route manager.

Conn.Close no longer takes signalToRemote: the idle transition is the
only path that signals GOAWAY to the remote peer.
2026-07-19 21:37:15 +02:00
78 changed files with 1422 additions and 2940 deletions

View File

@@ -247,9 +247,6 @@ func (c *Client) DebugBundle(platformFiles PlatformFiles, anonymize bool) (strin
deps.SyncResponse = resp
if e := cc.Engine(); e != nil {
deps.RefreshStatus = func() {
e.RunHealthProbes(context.Background(), true)
}
if cm := e.GetClientMetrics(); cm != nil {
deps.ClientMetrics = cm
}

View File

@@ -3,6 +3,7 @@ package configurer
import (
"net"
"net/netip"
"time"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
)
@@ -19,6 +20,43 @@ func buildPresharedKeyConfig(peerKey wgtypes.Key, psk wgtypes.Key, updateOnly bo
}
}
// buildIdlePeerEndpointConfig creates a config that removes and re-creates a peer in a
// single transaction with the given allowed IPs, endpoint and disabled keepalive.
func buildIdlePeerEndpointConfig(peerKey wgtypes.Key, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) wgtypes.Config {
keepAlive := time.Duration(0)
return wgtypes.Config{
Peers: []wgtypes.PeerConfig{
{
PublicKey: peerKey,
Remove: true,
},
{
PublicKey: peerKey,
AllowedIPs: prefixesToIPNets(allowedIPs),
Endpoint: endpoint,
PersistentKeepaliveInterval: &keepAlive,
},
},
}
}
// mergePrefixes returns the union of the two prefix lists, keeping the original order and
// dropping duplicates.
func mergePrefixes(current, base []netip.Prefix) []netip.Prefix {
merged := make([]netip.Prefix, 0, len(current)+len(base))
seen := make(map[netip.Prefix]struct{}, len(current)+len(base))
for _, group := range [][]netip.Prefix{current, base} {
for _, prefix := range group {
if _, ok := seen[prefix]; ok {
continue
}
seen[prefix] = struct{}{}
merged = append(merged, prefix)
}
}
return merged
}
func prefixesToIPNets(prefixes []netip.Prefix) []net.IPNet {
ipNets := make([]net.IPNet, len(prefixes))
for i, prefix := range prefixes {

View File

@@ -3,6 +3,7 @@
package configurer
import (
"errors"
"fmt"
"net"
"net/netip"
@@ -145,6 +146,39 @@ func (c *KernelConfigurer) RemovePeer(peerKey string) error {
return nil
}
// IdlePeerEndpoint re-creates the peer pointing at the lazy wake endpoint, dropping
// handshake state while preserving the peer's currently installed allowed IPs.
func (c *KernelConfigurer) IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error {
peerKeyParsed, err := wgtypes.ParseKey(peerKey)
if err != nil {
return err
}
var current []netip.Prefix
existing, err := c.getPeer(c.deviceName, peerKey)
switch {
case errors.Is(err, ErrPeerNotFound):
case err != nil:
return fmt.Errorf("get peer: %w", err)
default:
for _, ipNet := range existing.AllowedIPs {
addr, ok := netip.AddrFromSlice(ipNet.IP)
if !ok {
log.Warnf("failed to convert allowed IP %s of peer %s", ipNet.String(), peerKey)
continue
}
ones, _ := ipNet.Mask.Size()
current = append(current, netip.PrefixFrom(addr.Unmap(), ones))
}
}
config := buildIdlePeerEndpointConfig(peerKeyParsed, mergePrefixes(current, allowedIPs), endpoint)
if err := c.configure(config); err != nil {
return fmt.Errorf(`received error "%w" while setting idle endpoint for peer %s on interface %s`, err, peerKey, c.deviceName)
}
return nil
}
func (c *KernelConfigurer) AddAllowedIP(peerKey string, allowedIP netip.Prefix) error {
ipNet := net.IPNet{
IP: allowedIP.Addr().AsSlice(),

View File

@@ -210,6 +210,64 @@ func (c *WGUSPConfigurer) RemovePeer(peerKey string) error {
return ipcErr
}
// IdlePeerEndpoint re-creates the peer pointing at the lazy wake endpoint, dropping
// handshake state while preserving the peer's currently installed allowed IPs.
func (c *WGUSPConfigurer) IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error {
peerKeyParsed, err := wgtypes.ParseKey(peerKey)
if err != nil {
return err
}
current, err := c.currentAllowedIPs(hex.EncodeToString(peerKeyParsed[:]))
if err != nil {
return fmt.Errorf("get current allowed IPs: %w", err)
}
config := buildIdlePeerEndpointConfig(peerKeyParsed, mergePrefixes(current, allowedIPs), endpoint)
if err := c.device.IpcSet(toWgUserspaceString(config)); err != nil {
return fmt.Errorf("set idle peer endpoint: %w", err)
}
if endpoint != nil {
addr, err := netip.ParseAddr(endpoint.IP.String())
if err != nil {
return fmt.Errorf("parse endpoint address: %w", err)
}
c.activityRecorder.UpsertAddress(peerKey, netip.AddrPortFrom(addr.Unmap(), uint16(endpoint.Port)))
}
return nil
}
// currentAllowedIPs returns the allowed IPs currently installed for the peer identified by
// its hex-encoded public key. It returns an empty list when the peer does not exist.
func (c *WGUSPConfigurer) currentAllowedIPs(hexKey string) ([]netip.Prefix, error) {
ipc, err := c.device.IpcGet()
if err != nil {
return nil, err
}
var prefixes []netip.Prefix
foundPeer := false
for _, line := range strings.Split(ipc, "\n") {
line = strings.TrimSpace(line)
if strings.HasPrefix(line, "public_key=") {
foundPeer = line == "public_key="+hexKey
continue
}
if !foundPeer || !strings.HasPrefix(line, "allowed_ip=") {
continue
}
prefix, err := netip.ParsePrefix(strings.TrimPrefix(line, "allowed_ip="))
if err != nil {
log.Warnf("failed to parse allowed IP %q: %v", line, err)
continue
}
prefixes = append(prefixes, prefix)
}
return prefixes, nil
}
func (c *WGUSPConfigurer) AddAllowedIP(peerKey string, allowedIP netip.Prefix) error {
ipNet := net.IPNet{
IP: allowedIP.Addr().AsSlice(),

View File

@@ -0,0 +1,158 @@
package configurer
import (
"encoding/hex"
"net"
"net/netip"
"strings"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
wgconn "golang.zx2c4.com/wireguard/conn"
"golang.zx2c4.com/wireguard/device"
"golang.zx2c4.com/wireguard/tun/netstack"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface/bind"
)
// newTestConfigurer creates a configurer backed by an in-memory wireguard-go device.
func newTestConfigurer(t *testing.T) *WGUSPConfigurer {
t.Helper()
tunDev, _, err := netstack.CreateNetTUN([]netip.Addr{netip.MustParseAddr("100.64.0.1")}, []netip.Addr{}, 1280)
require.NoError(t, err)
wgDev := device.NewDevice(tunDev, wgconn.NewDefaultBind(), device.NewLogger(device.LogLevelSilent, "[test] "))
t.Cleanup(wgDev.Close)
c := NewUSPConfigurerNoUAPI(wgDev, "utun-test", bind.NewActivityRecorder())
key, err := wgtypes.GeneratePrivateKey()
require.NoError(t, err)
require.NoError(t, c.ConfigureInterface(key.String(), 0))
return c
}
// peerAllowedIPs returns the allowed IPs currently installed for the given peer.
func peerAllowedIPs(t *testing.T, c *WGUSPConfigurer, pubKey string) []string {
t.Helper()
var ips []string
for _, line := range peerIpcLines(t, c, pubKey) {
if strings.HasPrefix(line, "allowed_ip=") {
ips = append(ips, strings.TrimPrefix(line, "allowed_ip="))
}
}
return ips
}
// peerEndpoint returns the endpoint currently installed for the given peer.
func peerEndpoint(t *testing.T, c *WGUSPConfigurer, pubKey string) string {
t.Helper()
for _, line := range peerIpcLines(t, c, pubKey) {
if strings.HasPrefix(line, "endpoint=") {
return strings.TrimPrefix(line, "endpoint=")
}
}
return ""
}
// peerIpcLines returns the uapi config lines belonging to the given peer.
func peerIpcLines(t *testing.T, c *WGUSPConfigurer, pubKey string) []string {
t.Helper()
key, err := wgtypes.ParseKey(pubKey)
require.NoError(t, err)
hexKey := hex.EncodeToString(key[:])
ipc, err := c.device.IpcGet()
require.NoError(t, err)
var lines []string
inPeer := false
for _, line := range strings.Split(ipc, "\n") {
line = strings.TrimSpace(line)
if strings.HasPrefix(line, "public_key=") {
inPeer = line == "public_key="+hexKey
continue
}
if inPeer && line != "" {
lines = append(lines, line)
}
}
return lines
}
// TestUSPConfigurer_IdlePeerEndpointPreservesAllowedIPs verifies the invariant the lazy idle
// transition relies on: IdlePeerEndpoint re-creates the peer (dropping handshake state via the
// remove+add transaction) while keeping every installed allowed IP, including routed
// prefixes added later by the route manager, and points the endpoint at the wake listener.
func TestUSPConfigurer_IdlePeerEndpointPreservesAllowedIPs(t *testing.T) {
c := newTestConfigurer(t)
peerKey, err := wgtypes.GeneratePrivateKey()
require.NoError(t, err)
pubKey := peerKey.PublicKey().String()
overlay := netip.MustParsePrefix("100.64.0.5/32")
routed := netip.MustParsePrefix("10.99.0.0/24")
realEndpoint := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 51821}
require.NoError(t, c.UpdatePeer(pubKey, []netip.Prefix{overlay}, 25*time.Second, realEndpoint, nil))
require.NoError(t, c.AddAllowedIP(pubKey, routed))
ips := peerAllowedIPs(t, c, pubKey)
require.Contains(t, ips, overlay.String(), "overlay prefix must be installed before the idle endpoint swap")
require.Contains(t, ips, routed.String(), "routed prefix must be installed before the idle endpoint swap")
wakeEndpoint := &net.UDPAddr{IP: net.ParseIP("127.2.0.5"), Port: 17473}
require.NoError(t, c.IdlePeerEndpoint(pubKey, []netip.Prefix{overlay}, wakeEndpoint))
ips = peerAllowedIPs(t, c, pubKey)
assert.Contains(t, ips, routed.String(), "routed prefix must survive the idle endpoint swap")
assert.Contains(t, ips, overlay.String(), "overlay prefix must survive the idle endpoint swap")
assert.Equal(t, "127.2.0.5:17473", peerEndpoint(t, c, pubKey), "endpoint must point at the wake listener after the idle endpoint swap")
}
// TestUSPConfigurer_IdlePeerEndpointCreatesMissingPeer verifies the cold-start arm path: when the
// peer does not exist yet (never connected), IdlePeerEndpoint creates it with the given base
// allowed IPs and the wake endpoint.
func TestUSPConfigurer_IdlePeerEndpointCreatesMissingPeer(t *testing.T) {
c := newTestConfigurer(t)
peerKey, err := wgtypes.GeneratePrivateKey()
require.NoError(t, err)
pubKey := peerKey.PublicKey().String()
overlay := netip.MustParsePrefix("100.64.0.5/32")
wakeEndpoint := &net.UDPAddr{IP: net.ParseIP("127.2.0.5"), Port: 17473}
require.NoError(t, c.IdlePeerEndpoint(pubKey, []netip.Prefix{overlay}, wakeEndpoint))
assert.Equal(t, []string{overlay.String()}, peerAllowedIPs(t, c, pubKey), "missing peer must be created with the base allowed IPs")
assert.Equal(t, "127.2.0.5:17473", peerEndpoint(t, c, pubKey), "missing peer must be created with the wake endpoint")
}
// TestUSPConfigurer_AddAllowedIPOnMissingPeerIsSilentNoOp documents the wireguard-go
// behavior the removed-peer idle flow raced against: AddAllowedIP uses update_only,
// which is a silent no-op when the peer does not exist. The idle transition must
// therefore keep the WireGuard peer (Conn.Idle + IdlePeerEndpoint) instead of leaving a
// window where the peer is absent.
func TestUSPConfigurer_AddAllowedIPOnMissingPeerIsSilentNoOp(t *testing.T) {
c := newTestConfigurer(t)
peerKey, err := wgtypes.GeneratePrivateKey()
require.NoError(t, err)
pubKey := peerKey.PublicKey().String()
routed := netip.MustParsePrefix("10.99.0.0/24")
require.NoError(t, c.AddAllowedIP(pubKey, routed), "update-only on a missing peer must not return an error")
assert.Empty(t, peerIpcLines(t, c, pubKey), "update-only call must not create the peer")
}

View File

@@ -15,6 +15,7 @@ type WGConfigurer interface {
ConfigureInterface(privateKey string, port int) error
UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
RemovePeer(peerKey string) error
IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error
AddAllowedIP(peerKey string, allowedIP netip.Prefix) error
RemoveAllowedIP(peerKey string, allowedIP netip.Prefix) error
SetPresharedKey(peerKey string, psk wgtypes.Key, updateOnly bool) error

View File

@@ -184,6 +184,19 @@ func (w *WGIface) RemovePeer(peerKey string) error {
return w.configurer.RemovePeer(peerKey)
}
// IdlePeerEndpoint re-creates the peer pointing at the lazy wake endpoint, dropping
// handshake state while preserving the peer's currently installed allowed IPs.
func (w *WGIface) IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error {
w.mu.Lock()
defer w.mu.Unlock()
if w.configurer == nil {
return ErrIfaceNotFound
}
log.Debugf("Resetting peer on interface %s: %s, endpoint %s", w.tun.DeviceName(), peerKey, endpoint)
return w.configurer.IdlePeerEndpoint(peerKey, allowedIPs, endpoint)
}
// AddAllowedIP adds a prefix to the allowed IPs list of peer
func (w *WGIface) AddAllowedIP(peerKey string, allowedIP netip.Prefix) error {
w.mu.Lock()

View File

@@ -228,7 +228,7 @@ func (e *ConnMgr) RemovePeerConn(peerKey string) {
if !ok {
return
}
defer conn.Close(false)
defer conn.Close()
if !e.isStartedWithLazyMgr() {
return

View File

@@ -1778,7 +1778,7 @@ func (e *Engine) addNewPeer(peerConfig *mgmProto.RemotePeerConfig) error {
}
if exists := e.connMgr.AddPeerConn(e.ctx, peerKey, conn); exists {
conn.Close(false)
conn.Close()
return fmt.Errorf("peer already exists: %s", peerKey)
}

View File

@@ -53,6 +53,7 @@ type MockWGIface struct {
UpdateAddrFunc func(newAddr wgaddr.Address) error
UpdatePeerFunc func(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
RemovePeerFunc func(peerKey string) error
IdlePeerEndpointFunc func(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error
AddAllowedIPFunc func(peerKey string, allowedIP netip.Prefix) error
RemoveAllowedIPFunc func(peerKey string, allowedIP netip.Prefix) error
CloseFunc func() error
@@ -124,6 +125,13 @@ func (m *MockWGIface) RemovePeer(peerKey string) error {
return m.RemovePeerFunc(peerKey)
}
func (m *MockWGIface) IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error {
if m.IdlePeerEndpointFunc == nil {
return nil
}
return m.IdlePeerEndpointFunc(peerKey, allowedIPs, endpoint)
}
func (m *MockWGIface) AddAllowedIP(peerKey string, allowedIP netip.Prefix) error {
return m.AddAllowedIPFunc(peerKey, allowedIP)
}

View File

@@ -32,6 +32,7 @@ type wgIfaceBase interface {
UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
RemoveEndpointAddress(key string) error
RemovePeer(peerKey string) error
IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error
AddAllowedIP(peerKey string, allowedIP netip.Prefix) error
RemoveAllowedIP(peerKey string, allowedIP netip.Prefix) error
Close() error

View File

@@ -106,7 +106,7 @@ func (d *BindListener) setupLazyConn() error {
IP: d.fakeIP.AsSlice(),
Port: lazyBindPort,
}
return d.wgIface.UpdatePeer(d.peerCfg.PublicKey, d.peerCfg.AllowedIPs, 0, endpoint, nil)
return d.wgIface.IdlePeerEndpoint(d.peerCfg.PublicKey, d.peerCfg.AllowedIPs, endpoint)
}
// ReadPackets blocks until activity is detected on the LazyConn or the listener is closed.

View File

@@ -9,7 +9,6 @@ import (
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface/device"
"github.com/netbirdio/netbird/client/iface/wgaddr"
@@ -45,7 +44,7 @@ type MockWGIfaceBind struct {
endpointMgr *mockEndpointManager
}
func (m *MockWGIfaceBind) UpdatePeer(string, []netip.Prefix, time.Duration, *net.UDPAddr, *wgtypes.Key) error {
func (m *MockWGIfaceBind) IdlePeerEndpoint(string, []netip.Prefix, *net.UDPAddr) error {
return nil
}

View File

@@ -101,7 +101,7 @@ func (d *UDPListener) Close() {
func (d *UDPListener) createEndpoint() error {
d.peerCfg.Log.Debugf("creating lazy endpoint: %s", d.endpoint.String())
return d.wgIface.UpdatePeer(d.peerCfg.PublicKey, d.peerCfg.AllowedIPs, 0, d.endpoint, nil)
return d.wgIface.IdlePeerEndpoint(d.peerCfg.PublicKey, d.peerCfg.AllowedIPs, d.endpoint)
}
func (d *UDPListener) newConn() (*net.UDPConn, error) {

View File

@@ -5,10 +5,8 @@ import (
"net"
"net/netip"
"sync"
"time"
log "github.com/sirupsen/logrus"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface/wgaddr"
"github.com/netbirdio/netbird/client/internal/lazyconn"
@@ -30,7 +28,7 @@ type Event struct {
}
type WgInterface interface {
UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error
IsUserspaceBind() bool
Address() wgaddr.Address
MTU() uint16

View File

@@ -8,7 +8,6 @@ import (
"time"
log "github.com/sirupsen/logrus"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface/wgaddr"
"github.com/netbirdio/netbird/client/internal/lazyconn"
@@ -26,7 +25,7 @@ func (m *MocPeer) ConnID() peerid.ConnID {
type MocWGIface struct {
}
func (m MocWGIface) UpdatePeer(string, []netip.Prefix, time.Duration, *net.UDPAddr, *wgtypes.Key) error {
func (m MocWGIface) IdlePeerEndpoint(string, []netip.Prefix, *net.UDPAddr) error {
return nil
}

View File

@@ -280,7 +280,7 @@ func (m *Manager) DeactivatePeer(peerID peerid.ConnID) {
return
}
m.peerStore.PeerConnClose(mp.peerCfg.PublicKey)
m.peerStore.PeerConnIdle(mp.peerCfg.PublicKey, false)
mp.peerCfg.Log.Infof("start activity monitor")
@@ -569,7 +569,7 @@ func (m *Manager) onPeerInactivityTimedOut(peerIDs map[string]struct{}) {
mp.peerCfg.Log.Infof("connection timed out")
// this is blocking operation, potentially can be optimized
m.peerStore.PeerConnIdle(mp.peerCfg.PublicKey)
m.peerStore.PeerConnIdle(mp.peerCfg.PublicKey, true)
mp.expectedWatcher = watcherActivity

View File

@@ -3,9 +3,6 @@ package lazyconn
import (
"net"
"net/netip"
"time"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface/wgaddr"
"github.com/netbirdio/netbird/monotime"
@@ -13,7 +10,7 @@ import (
type WGIface interface {
RemovePeer(peerKey string) error
UpdatePeer(peerKey string, allowedIps []netip.Prefix, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error
IdlePeerEndpoint(peerKey string, allowedIPs []netip.Prefix, endpoint *net.UDPAddr) error
IsUserspaceBind() bool
Address() wgaddr.Address
LastActivities() map[string]monotime.Time

View File

@@ -285,8 +285,20 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
return nil
}
// Close closes this peer Conn issuing a close event to the Conn closeCh
func (conn *Conn) Close(signalToRemote bool) {
// Close closes this peer Conn issuing a close event to the Conn closeCh and removes the WireGuard peer
func (conn *Conn) Close() {
conn.close(false, true)
}
// Idle tears down the connection for the lazy idle state: transports and proxies are
// closed but the WireGuard peer is kept, so its AllowedIPs (including routed prefixes
// installed by the route manager) survive until the activity listener re-points the
// endpoint at the wake listener.
func (conn *Conn) Idle(signalToRemote bool) {
conn.close(signalToRemote, false)
}
func (conn *Conn) close(signalToRemote bool, removeWgPeer bool) {
conn.mu.Lock()
defer conn.wgWatcherWg.Wait()
defer conn.mu.Unlock()
@@ -329,8 +341,12 @@ func (conn *Conn) Close(signalToRemote bool) {
conn.wgProxyICE = nil
}
if err := conn.endpointUpdater.RemoveWgPeer(); err != nil {
conn.Log.Errorf("failed to remove wg endpoint: %v", err)
if removeWgPeer {
if err := conn.endpointUpdater.RemoveWgPeer(); err != nil {
conn.Log.Errorf("failed to remove wg endpoint: %v", err)
}
} else {
conn.endpointUpdater.CancelPendingUpdates()
}
if conn.evalStatus() == StatusConnected && conn.onDisconnected != nil {

View File

@@ -65,6 +65,16 @@ func (e *EndpointUpdater) RemoveWgPeer() error {
return e.wgConfig.WgInterface.RemovePeer(e.wgConfig.RemoteKey)
}
// CancelPendingUpdates stops a scheduled delayed endpoint update without touching the
// WireGuard peer. Used on the idle transition where the peer is kept so a pending
// responder-side update cannot overwrite the wake endpoint later.
func (e *EndpointUpdater) CancelPendingUpdates() {
e.mu.Lock()
defer e.mu.Unlock()
e.waitForCloseTheDelayedUpdate()
}
func (e *EndpointUpdater) RemoveEndpointAddress() error {
e.mu.Lock()
defer e.mu.Unlock()

View File

@@ -0,0 +1,101 @@
package peer
import (
"net"
"net/netip"
"sync"
"testing"
"time"
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
"github.com/netbirdio/netbird/client/iface/configurer"
"github.com/netbirdio/netbird/client/iface/wgaddr"
"github.com/netbirdio/netbird/client/iface/wgproxy"
)
type endpointTestWGIface struct {
mu sync.Mutex
updateCalls int
removeCalls int
}
func (m *endpointTestWGIface) UpdatePeer(string, []netip.Prefix, time.Duration, *net.UDPAddr, *wgtypes.Key) error {
m.mu.Lock()
defer m.mu.Unlock()
m.updateCalls++
return nil
}
func (m *endpointTestWGIface) RemovePeer(string) error {
m.mu.Lock()
defer m.mu.Unlock()
m.removeCalls++
return nil
}
func (m *endpointTestWGIface) GetStats() (map[string]configurer.WGStats, error) {
return map[string]configurer.WGStats{}, nil
}
func (m *endpointTestWGIface) GetProxy() wgproxy.Proxy { return nil }
func (m *endpointTestWGIface) Address() wgaddr.Address { return wgaddr.Address{} }
func (m *endpointTestWGIface) RemoveEndpointAddress(string) error { return nil }
func (m *endpointTestWGIface) counts() (updates, removes int) {
m.mu.Lock()
defer m.mu.Unlock()
return m.updateCalls, m.removeCalls
}
func newTestEndpointUpdater(iface *endpointTestWGIface, initiator bool) *EndpointUpdater {
cfg := WgConfig{
RemoteKey: "remoteKey",
WgInterface: iface,
AllowedIps: []netip.Prefix{netip.MustParsePrefix("100.64.0.5/32")},
}
return NewEndpointUpdater(log.WithField("peer", "test"), cfg, initiator)
}
// TestEndpointUpdater_CancelPendingUpdates ensures a scheduled responder-side delayed
// update is stopped without removing the WireGuard peer, so an idle transition cannot
// be overwritten later by a stale endpoint update.
func TestEndpointUpdater_CancelPendingUpdates(t *testing.T) {
iface := &endpointTestWGIface{}
e := newTestEndpointUpdater(iface, false)
addr := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 51820}
require.NoError(t, e.ConfigureWGEndpoint(addr, nil))
updates, _ := iface.counts()
require.Equal(t, 1, updates, "responder must apply the immediate nil-endpoint update")
// CancelPendingUpdates waits for the delayed-update goroutine to exit, so the
// call counts below are final: the 5s fallback update can never fire anymore.
e.CancelPendingUpdates()
updates, removes := iface.counts()
assert.Equal(t, 1, updates, "delayed endpoint update must not fire after cancellation")
assert.Equal(t, 0, removes, "cancellation must not remove the WireGuard peer")
}
// TestEndpointUpdater_CancelPendingUpdatesNoPending ensures cancellation is a safe no-op
// when no delayed update is scheduled (initiator path).
func TestEndpointUpdater_CancelPendingUpdatesNoPending(t *testing.T) {
iface := &endpointTestWGIface{}
e := newTestEndpointUpdater(iface, true)
addr := &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 51820}
require.NoError(t, e.ConfigureWGEndpoint(addr, nil))
e.CancelPendingUpdates()
updates, removes := iface.counts()
assert.Equal(t, 1, updates, "initiator applies exactly one direct update")
assert.Equal(t, 0, removes, "cancellation must not remove the WireGuard peer")
}

View File

@@ -54,15 +54,19 @@ 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
_, _, err := w.relayManager.RelayInstanceAddress()
currentRelayAddress, _, err := w.relayManager.RelayInstanceAddress()
if err != nil {
w.log.Errorf("failed to handle new offer: %s", err)
return
}
preferForeign := !w.isController
remoteRelayServer := relayClient.RelayServer{Addr: remoteOfferAnswer.RelaySrvAddress, IP: remoteOfferAnswer.RelaySrvIP}
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, remoteRelayServer, w.config.Key, preferForeign)
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)
if err != nil {
if errors.Is(err, relayClient.ErrConnAlreadyExists) {
w.log.Debugf("handled offer by reusing existing relay connection")
@@ -76,13 +80,14 @@ func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
w.relayedConn = relayedConn
w.relayLock.Unlock()
if err := w.relayManager.AddCloseListener(relayedConn.RemoteAddr().String(), w.onRelayClientDisconnected); err != nil {
w.log.Errorf("failed to add close listener: %s", err)
err = w.relayManager.AddCloseListener(srv, w.onRelayClientDisconnected)
if err != nil {
log.Errorf("failed to add close listener: %s", err)
_ = relayedConn.Close()
return
}
w.log.Debugf("peer conn opened via Relay: %s", relayedConn.RemoteAddr())
w.log.Debugf("peer conn opened via Relay: %s", srv)
go w.conn.onRelayConnectionIsReady(RelayConnInfo{
relayedConn: relayedConn,
rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
@@ -121,6 +126,13 @@ 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

@@ -108,7 +108,10 @@ func (s *Store) PeerConnOpenWithFirstPacket(ctx context.Context, pubKey string,
}
}
func (s *Store) PeerConnIdle(pubKey string) {
// PeerConnIdle transitions the peer connection to the lazy idle state, keeping the
// WireGuard peer and its AllowedIPs in place. signalToRemote indicates whether the
// remote peer should be notified (false when the remote initiated the idle via GOAWAY).
func (s *Store) PeerConnIdle(pubKey string, signalToRemote bool) {
s.peerConnsMu.RLock()
defer s.peerConnsMu.RUnlock()
@@ -116,18 +119,7 @@ func (s *Store) PeerConnIdle(pubKey string) {
if !ok {
return
}
p.Close(true)
}
func (s *Store) PeerConnClose(pubKey string) {
s.peerConnsMu.RLock()
defer s.peerConnsMu.RUnlock()
p, ok := s.peerConns[pubKey]
if !ok {
return
}
p.Close(false)
p.Idle(signalToRemote)
}
func (s *Store) PeersPubKey() []string {

View File

@@ -233,9 +233,6 @@ func (c *Client) DebugBundle(anonymize bool) (string, error) {
deps.SyncResponse = resp
if e := cc.Engine(); e != nil {
deps.RefreshStatus = func() {
e.RunHealthProbes(context.Background(), true)
}
if cm := e.GetClientMetrics(); cm != nil {
deps.ClientMetrics = cm
}

View File

@@ -8,7 +8,6 @@
{"code": "fr", "displayName": "Français", "englishName": "French"},
{"code": "it", "displayName": "Italiano", "englishName": "Italian"},
{"code": "pt", "displayName": "Português", "englishName": "Portuguese"},
{"code": "zh-CN", "displayName": "简体中文", "englishName": "Simplified Chinese"},
{"code": "ja", "displayName": "日本語", "englishName": "Japanese"}
{"code": "zh-CN", "displayName": "简体中文", "englishName": "Simplified Chinese"}
]
}

File diff suppressed because it is too large Load Diff

View File

@@ -226,7 +226,7 @@ func (s *serverInstances) createRelayServer(cfg *CombinedConfig, tlsSupport bool
}
hashedSecret := sha256.Sum256([]byte(cfg.Relay.AuthSecret))
authenticator := auth.NewTimedHMACValidator(hashedSecret[:])
authenticator := auth.NewTimedHMACValidator(hashedSecret[:], 24*time.Hour)
relayCfg := relayServer.Config{
Meter: s.metricsServer.Meter,

3
go.mod
View File

@@ -113,7 +113,7 @@ require (
github.com/ti-mo/conntrack v0.5.1
github.com/ti-mo/netfilter v0.5.2
github.com/vmihailenco/msgpack/v5 v5.4.1
github.com/wailsapp/wails/v3 v3.0.0-alpha2.117
github.com/wailsapp/wails/v3 v3.0.0-alpha2.111
github.com/yusufpapurcu/wmi v1.2.4
github.com/zcalusic/sysinfo v1.1.3
go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0
@@ -303,6 +303,7 @@ require (
github.com/tklauser/numcpus v0.10.0 // indirect
github.com/vishvananda/netns v0.0.5 // indirect
github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
github.com/wailsapp/wails/webview2 v1.0.27 // indirect
github.com/wlynxg/anet v0.0.5 // indirect
github.com/x448/float16 v0.8.4 // indirect
github.com/zeebo/blake3 v0.2.3 // indirect

6
go.sum
View File

@@ -660,8 +660,10 @@ github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IU
github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.117 h1:udyjqPG3AIgkod5QDR/WblCkpV8R86BFPSrsWxSyt5Y=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.117/go.mod h1:74WH2FScMsgucZvHHvv7eOefDXCm/CjuIxqhhZgPhKg=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.111 h1:MKx1nOnhnDuEGrRBmtxLOJq1NERwailu2cI4BvzWhi4=
github.com/wailsapp/wails/v3 v3.0.0-alpha2.111/go.mod h1:wrdvmyeCsB/K3YqJDoH8E3MwcN8NXAMnEFaDTW46w60=
github.com/wailsapp/wails/webview2 v1.0.27 h1:wjgAi/I8BBZ7kUGU8um3XF3ILEfzr96Q2Q1G4GPjMns=
github.com/wailsapp/wails/webview2 v1.0.27/go.mod h1:zdM4jcO1IaC61RiJL5F1BzgoqBHFIdacz8gPr5exr0o=
github.com/wlynxg/anet v0.0.5 h1:J3VJGi1gvo0JwZ/P1/Yc/8p63SoW98B5dHkYDmpgvvU=
github.com/wlynxg/anet v0.0.5/go.mod h1:eay5PRQr7fIVAMbTbchTnO9gG65Hg/uYGdc7mguHxoA=
github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=

View File

@@ -197,12 +197,6 @@ type JSONMetadataInjection struct {
// enforces a 128-char limit per value; oversized values are
// truncated rather than failing the request. 0 disables the cap.
MaxValueLength int
// Sanitize, when true, replaces characters outside the destination's
// accepted set with '_' before emitting each value. AWS Bedrock's
// X-Amzn-Bedrock-Request-Metadata restricts values to a limited character
// class, so unsanitized group display names (e.g. containing spaces) would
// make Bedrock reject the request with 400.
Sanitize bool
}
// providers is the canonical list of supported Agent Network providers.
@@ -335,18 +329,6 @@ var providers = []Provider{
{ID: "amazon.nova-lite", Label: "Amazon Nova Lite (Bedrock)", InputPer1k: 0.00006, OutputPer1k: 0.00024, ContextWindow: 300000},
{ID: "amazon.nova-micro", Label: "Amazon Nova Micro (Bedrock)", InputPer1k: 0.000035, OutputPer1k: 0.00014, ContextWindow: 128000},
},
// Bedrock accepts a cost-allocation metadata header; stamp the caller's
// user + authorizing group so spend can be attributed in AWS Cost
// Management. Sanitized because Bedrock restricts the value character set.
IdentityInjection: &IdentityInjection{
JSONMetadata: &JSONMetadataInjection{
Header: "X-Amzn-Bedrock-Request-Metadata",
UserKey: "user",
GroupsKey: "group",
MaxValueLength: 256,
Sanitize: true,
},
},
},
{
ID: "vertex_ai_api",

View File

@@ -540,7 +540,6 @@ type identityInjectJSONMetadata struct {
UserKey string `json:"user_key,omitempty"`
GroupsKey string `json:"groups_key,omitempty"`
MaxValueLength int `json:"max_value_length,omitempty"`
Sanitize bool `json:"sanitize,omitempty"`
}
// buildIdentityInjectConfigJSON walks the enabled providers and emits
@@ -584,11 +583,9 @@ func buildIdentityInjectConfigJSON(providers []*types.Provider, groupIndex map[s
func buildIdentityInjectRule(p *types.Provider, entry catalog.Provider) (identityInjectProvider, bool) {
rule := identityInjectProvider{ProviderID: p.ID}
// Identity-stamping shape (one of HeaderPair / JSONMetadata). Skip the
// shape silently when the catalog entry doesn't declare one, or when the
// operator disabled metadata for this provider — extras can still apply,
// see below. MetadataDisabled suppresses only the identity dimensions
// (user + authorizing group), not the catalog's routing ExtraHeaders.
if !p.MetadataDisabled && entry.IdentityInjection != nil {
// shape silently when the catalog entry doesn't declare one — extras
// can still apply, see below.
if entry.IdentityInjection != nil {
switch {
case entry.IdentityInjection.HeaderPair != nil:
rule.HeaderPair = buildIdentityHeaderPair(p, entry.IdentityInjection.HeaderPair)
@@ -654,7 +651,6 @@ func buildIdentityJSONMetadata(p *types.Provider, jm *catalog.JSONMetadataInject
UserKey: userKey,
GroupsKey: groupsKey,
MaxValueLength: jm.MaxValueLength,
Sanitize: jm.Sanitize,
}
}

View File

@@ -698,94 +698,6 @@ func TestSynthesizeServices_IdentityInject_Portkey_NotCustomizable(t *testing.T)
"same fixed-schema guarantee for the groups dimension")
}
// TestSynthesizeServices_IdentityInject_Bedrock pins Bedrock's cost-allocation
// metadata: a JSONMetadata shape emitting X-Amzn-Bedrock-Request-Metadata with
// the reserved user/group keys, sanitized to Bedrock's accepted charset.
func TestSynthesizeServices_IdentityInject_Bedrock(t *testing.T) {
ctx := context.Background()
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockStore := store.NewMockStore(ctrl)
br := newSynthTestProvider()
br.ID = "prov-bedrock"
br.ProviderID = "bedrock_api"
br.UpstreamURL = "https://bedrock-runtime.us-east-1.amazonaws.com"
br.APIKey = "bedrock-bearer"
br.CreatedAt = time.Date(2026, 4, 2, 0, 0, 0, 0, time.UTC)
policy := newSynthTestPolicy(br.ID, "grp-eng", "")
policy.ID = "pol-bedrock"
expectSynthBaseInputs(mockStore, ctx, newSynthTestSettings(),
[]*types.Provider{br},
[]*types.Policy{policy},
[]*types.Guardrail{})
services, err := SynthesizeServices(ctx, mockStore, testAccountID)
require.NoError(t, err)
require.Len(t, services, 1)
var injectCfg identityInjectConfig
for _, m := range services[0].Targets[0].Options.Middlewares {
if m.ID == middlewareIDLLMIdentityInject {
require.NoError(t, json.Unmarshal(m.ConfigJSON, &injectCfg))
break
}
}
require.Len(t, injectCfg.Providers, 1)
entry := injectCfg.Providers[0]
require.NotNil(t, entry.JSONMetadata, "Bedrock uses the JSONMetadata shape for cost-allocation metadata")
assert.Nil(t, entry.HeaderPair, "shapes are mutually exclusive")
assert.Equal(t, "X-Amzn-Bedrock-Request-Metadata", entry.JSONMetadata.Header,
"the caller identity lands in Bedrock's cost-allocation metadata header")
assert.Equal(t, "user", entry.JSONMetadata.UserKey)
assert.Equal(t, "group", entry.JSONMetadata.GroupsKey)
assert.True(t, entry.JSONMetadata.Sanitize,
"Bedrock restricts the metadata value charset, so values must be sanitized")
}
// TestSynthesizeServices_MetadataDisabled_SuppressesInjection verifies the
// per-provider opt-out: a provider with MetadataDisabled set emits no
// identity-inject entry (Bedrock has no catalog ExtraHeaders, so the whole
// entry is dropped).
func TestSynthesizeServices_MetadataDisabled_SuppressesInjection(t *testing.T) {
ctx := context.Background()
ctrl := gomock.NewController(t)
defer ctrl.Finish()
mockStore := store.NewMockStore(ctrl)
br := newSynthTestProvider()
br.ID = "prov-bedrock"
br.ProviderID = "bedrock_api"
br.UpstreamURL = "https://bedrock-runtime.us-east-1.amazonaws.com"
br.APIKey = "bedrock-bearer"
br.MetadataDisabled = true
br.CreatedAt = time.Date(2026, 4, 2, 0, 0, 0, 0, time.UTC)
policy := newSynthTestPolicy(br.ID, "grp-eng", "")
policy.ID = "pol-bedrock"
expectSynthBaseInputs(mockStore, ctx, newSynthTestSettings(),
[]*types.Provider{br},
[]*types.Policy{policy},
[]*types.Guardrail{})
services, err := SynthesizeServices(ctx, mockStore, testAccountID)
require.NoError(t, err)
require.Len(t, services, 1)
var injectCfg identityInjectConfig
for _, m := range services[0].Targets[0].Options.Middlewares {
if m.ID == middlewareIDLLMIdentityInject {
require.NoError(t, json.Unmarshal(m.ConfigJSON, &injectCfg))
break
}
}
assert.Empty(t, injectCfg.Providers,
"metadata_disabled must drop the provider's identity-inject entry")
}
// TestSynthesizeServices_IdentityInject_Vercel pins Vercel AI
// Gateway's wiring: HeaderPair shape with fixed wire names dictated
// by Vercel's Custom Reporting API (ai-reporting-user /

View File

@@ -51,12 +51,6 @@ type Provider struct {
// private or self-signed certificate. The synthesiser propagates it into
// the router route so the proxy dials that provider's upstream insecurely.
SkipTLSVerification bool `gorm:"column:skip_tls_verification"`
// MetadataDisabled suppresses identity metadata injection for this provider.
// Metadata (the caller's user + authorizing group) is injected by default;
// when true the synthesiser omits the provider's identity-inject shape, so no
// user/group headers (e.g. Bedrock's X-Amzn-Bedrock-Request-Metadata) are
// stamped. Catalog ExtraHeaders (routing config) are unaffected.
MetadataDisabled bool `gorm:"column:metadata_disabled"`
// SessionPrivateKey + SessionPublicKey are the ed25519 keypair the
// synthesised reverse-proxy service uses to sign / verify session
// JWTs after a successful OIDC handshake. Generated once on
@@ -143,9 +137,6 @@ func (p *Provider) FromAPIRequest(req *api.AgentNetworkProviderRequest) {
if req.SkipTlsVerification != nil {
p.SkipTLSVerification = *req.SkipTlsVerification
}
if req.MetadataDisabled != nil {
p.MetadataDisabled = *req.MetadataDisabled
}
// Identity-header overrides for catalogs flagged Customizable.
// nil pointer = "field omitted on the wire" → leave the stored
// value untouched (per the openapi description). Empty string is
@@ -179,7 +170,6 @@ func (p *Provider) ToAPIResponse() *api.AgentNetworkProvider {
Models: models,
Enabled: p.Enabled,
SkipTlsVerification: p.SkipTLSVerification,
MetadataDisabled: p.MetadataDisabled,
CreatedAt: &created,
UpdatedAt: &updated,
}

View File

@@ -42,38 +42,3 @@ func TestProvider_SkipTLSVerification_RoundTrip(t *testing.T) {
assert.False(t, p.SkipTLSVerification, "explicit false must clear skip_tls_verification")
assert.False(t, p.ToAPIResponse().SkipTlsVerification, "response must reflect the cleared value")
}
// TestProvider_MetadataDisabled_RoundTrip covers the request→provider→response
// mapping of metadata_disabled, with the same update semantics: nil preserves,
// explicit false clears.
func TestProvider_MetadataDisabled_RoundTrip(t *testing.T) {
enable := true
disable := false
base := func() *api.AgentNetworkProviderRequest {
return &api.AgentNetworkProviderRequest{
ProviderId: "bedrock_api",
Name: "bedrock",
UpstreamUrl: "https://bedrock-runtime.us-east-1.amazonaws.com",
}
}
p := NewProvider("acc-1")
req := base()
req.MetadataDisabled = &enable
p.FromAPIRequest(req)
assert.True(t, p.MetadataDisabled, "create with metadata_disabled=true must set the field")
assert.True(t, p.ToAPIResponse().MetadataDisabled, "response must surface metadata_disabled")
// Omitting the field on update leaves the stored value untouched.
p.FromAPIRequest(base())
assert.True(t, p.MetadataDisabled, "omitting metadata_disabled on update must preserve it")
// Explicit false clears it (re-enables metadata).
req = base()
req.MetadataDisabled = &disable
p.FromAPIRequest(req)
assert.False(t, p.MetadataDisabled, "explicit false must clear metadata_disabled")
assert.False(t, p.ToAPIResponse().MetadataDisabled, "response must reflect the cleared value")
}

View File

@@ -18,7 +18,6 @@ import (
"github.com/netbirdio/netbird/client/embed"
"github.com/netbirdio/netbird/proxy"
nbacme "github.com/netbirdio/netbird/proxy/internal/acme"
"github.com/netbirdio/netbird/trustedproxy"
"github.com/netbirdio/netbird/util"
)
@@ -210,7 +209,7 @@ func runServer(cmd *cobra.Command, args []string) error {
return fmt.Errorf("invalid domain value %q: %w", proxyDomain, err)
}
parsedTrustedProxies, err := trustedproxy.Parse(trustedProxies)
parsedTrustedProxies, err := proxy.ParseTrustedProxies(trustedProxies)
if err != nil {
return fmt.Errorf("invalid --trusted-proxies: %w", err)
}

View File

@@ -16,7 +16,6 @@ import (
"github.com/netbirdio/netbird/proxy/auth"
"github.com/netbirdio/netbird/proxy/internal/types"
"github.com/netbirdio/netbird/shared/management/proto"
"github.com/netbirdio/netbird/trustedproxy"
)
const (
@@ -67,7 +66,7 @@ type denyBucket struct {
type Logger struct {
client gRPCClient
logger *log.Logger
trustedProxies *trustedproxy.List
trustedProxies []netip.Prefix
usageMux sync.Mutex
domainUsage map[string]*domainUsage
@@ -83,7 +82,7 @@ type Logger struct {
// NewLogger creates a new access log Logger. The trustedProxies parameter
// configures which upstream proxy IP ranges are trusted for extracting
// the real client IP from X-Forwarded-For headers.
func NewLogger(client gRPCClient, logger *log.Logger, trustedProxies *trustedproxy.List) *Logger {
func NewLogger(client gRPCClient, logger *log.Logger, trustedProxies []netip.Prefix) *Logger {
if logger == nil {
logger = log.StandardLogger()
}

View File

@@ -4,13 +4,13 @@ import (
"net/http"
"net/netip"
"github.com/netbirdio/netbird/trustedproxy"
"github.com/netbirdio/netbird/proxy/internal/proxy"
)
// extractSourceIP resolves the real client IP from the request using trusted
// proxy configuration. When trustedProxies is non-empty and the direct
// connection is from a trusted source, it walks X-Forwarded-For right-to-left
// skipping trusted IPs. Otherwise it returns RemoteAddr directly.
func extractSourceIP(r *http.Request, trustedProxies *trustedproxy.List) netip.Addr {
return trustedProxies.ResolveClientIP(r.RemoteAddr, r.Header.Get("X-Forwarded-For"))
func extractSourceIP(r *http.Request, trustedProxies []netip.Prefix) netip.Addr {
return proxy.ResolveClientIP(r.RemoteAddr, r.Header.Get("X-Forwarded-For"), trustedProxies)
}

View File

@@ -1,38 +0,0 @@
package llm
import (
"regexp"
"strings"
)
// bedrockRegionPrefixes are the cross-region inference-profile prefixes that
// front a Bedrock model id (e.g. "eu.anthropic.claude-...").
var bedrockRegionPrefixes = []string{"us.", "eu.", "apac.", "global."}
// bedrockVersionSuffix matches the trailing "-vN[:N]" or "-YYYYMMDD-vN[:N]"
// version/throughput suffix of a Bedrock model id.
var bedrockVersionSuffix = regexp.MustCompile(`-(\d{8}-)?v\d+(:\d+)?$`)
// NormalizeBedrockModel strips an ARN wrapper, a cross-region inference-profile
// prefix, and the version/throughput suffix from a Bedrock model id so it
// matches the catalog/pricing key, e.g.
// "eu.anthropic.claude-sonnet-4-5-20250929-v1:0" -> "anthropic.claude-sonnet-4-5"
// and the inference-profile ARN's last segment likewise. It is the single
// source of truth shared by the request parser (which normalizes the request
// model from the URL path) and the router (which normalizes the operator's
// registered Bedrock model ids so both sides compare equal).
func NormalizeBedrockModel(modelID string) string {
m := modelID
if strings.HasPrefix(m, "arn:") {
if i := strings.LastIndex(m, "/"); i >= 0 {
m = m[i+1:]
}
}
for _, p := range bedrockRegionPrefixes {
if strings.HasPrefix(m, p) {
m = m[len(p):]
break
}
}
return bedrockVersionSuffix.ReplaceAllString(m, "")
}

View File

@@ -1,23 +0,0 @@
package llm
import (
"testing"
"github.com/stretchr/testify/require"
)
func TestNormalizeBedrockModel(t *testing.T) {
cases := map[string]string{
"eu.anthropic.claude-sonnet-4-5-20250929-v1:0": "anthropic.claude-sonnet-4-5",
"us.anthropic.claude-haiku-4-5": "anthropic.claude-haiku-4-5",
"us.anthropic.claude-opus-4-8-20250101-v1:0": "anthropic.claude-opus-4-8",
"anthropic.claude-sonnet-4-5-20250929-v1:0": "anthropic.claude-sonnet-4-5",
"meta.llama3-3-70b-instruct-v1:0": "meta.llama3-3-70b-instruct",
"amazon.nova-pro-v1:0": "amazon.nova-pro",
// Inference-profile ARN — model id lives in the last path segment.
"arn:aws:bedrock:eu-central-1:123456789012:inference-profile/eu.anthropic.claude-sonnet-4-5-20250929-v1:0": "anthropic.claude-sonnet-4-5",
}
for in, want := range cases {
require.Equal(t, want, NormalizeBedrockModel(in), "normalize %q", in)
}
}

View File

@@ -64,11 +64,6 @@ type JSONMetadataRule struct {
UserKey string `json:"user_key,omitempty"`
GroupsKey string `json:"groups_key,omitempty"`
MaxValueLength int `json:"max_value_length,omitempty"`
// Sanitize replaces characters outside the destination provider's accepted
// set with '_' before emitting each value. AWS Bedrock's
// X-Amzn-Bedrock-Request-Metadata restricts values to [A-Za-z0-9 +-=._:/@];
// group display names with other characters would otherwise 400.
Sanitize bool `json:"sanitize,omitempty"`
}
// Config is the on-wire configuration accepted by the factory. An

View File

@@ -292,21 +292,15 @@ func applyJSONMetadata(rule *JSONMetadataRule, in *middleware.Input) *middleware
mutations := &middleware.Mutations{}
mutations.HeadersRemove = append(mutations.HeadersRemove, rule.Header)
emit := func(v string) string {
if rule.Sanitize {
v = sanitizeMetadataValue(v)
}
return truncate(v, rule.MaxValueLength)
}
payload := map[string]string{}
if rule.UserKey != "" {
if identity := identityFor(in); identity != "" {
payload[rule.UserKey] = emit(identity)
payload[rule.UserKey] = truncate(identity, rule.MaxValueLength)
}
}
if rule.GroupsKey != "" {
if csv := authorisingTagsCSV(in); csv != "" {
payload[rule.GroupsKey] = emit(csv)
payload[rule.GroupsKey] = truncate(csv, rule.MaxValueLength)
}
}
if len(payload) == 0 {
@@ -365,36 +359,6 @@ func truncate(s string, maxBytes int) string {
return s[:maxBytes]
}
// sanitizeMetadataValue replaces any character outside AWS Bedrock's accepted
// request-metadata class — letters, digits, space, and + - = . _ : / @ — with
// '_'. This keeps values (notably the groups CSV, whose commas are rejected, and
// group display names with arbitrary characters) from making Bedrock reject the
// request with 400. The result stays opaque to the gateway.
func sanitizeMetadataValue(s string) string {
var b strings.Builder
b.Grow(len(s))
for _, r := range s {
if metadataCharAllowed(r) {
b.WriteRune(r)
} else {
b.WriteByte('_')
}
}
return b.String()
}
func metadataCharAllowed(r rune) bool {
switch {
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
return true
}
switch r {
case ' ', '+', '-', '=', '.', '_', ':', '/', '@':
return true
}
return false
}
// tagsIDsFromAuthorising reads llm_router's authorising-groups metadata
// (a CSV of group ids) and returns the parsed slice. Returns nil when
// the key is absent or empty so the caller can fall back to the full

View File

@@ -304,46 +304,6 @@ func TestInject_JSONMetadata_TruncatesValues(t *testing.T) {
"per-value byte length must be capped at MaxValueLength")
}
// TestInject_JSONMetadata_Sanitize pins the AWS-Bedrock sanitization path: when
// Sanitize is set, characters outside Bedrock's accepted metadata class
// (notably the groups CSV comma and arbitrary characters in group display
// names) are replaced with '_' so Bedrock doesn't reject the request. Allowed
// characters (letters, digits, spaces, and @ . _ : / + - =) pass through.
func TestInject_JSONMetadata_Sanitize(t *testing.T) {
rule := ProviderInjection{
ProviderID: portkeyProvider,
JSONMetadata: &JSONMetadataRule{
Header: "X-Amzn-Bedrock-Request-Metadata",
UserKey: "user",
GroupsKey: "group",
MaxValueLength: 256,
Sanitize: true,
},
}
mw := New(Config{Providers: []ProviderInjection{rule}})
in := newInput(portkeyProvider, "alice", []string{"g1", "g2"})
in.UserEmail = "alice@example.com"
// Group display names carry characters Bedrock rejects (comma, '#'); the CSV
// join adds another comma between the two groups.
in.UserGroupNames = []string{"Eng,Team", "Ops#1"}
out, err := mw.Invoke(context.Background(), in)
require.NoError(t, err)
require.NotNil(t, out.Mutations)
require.Len(t, out.Mutations.HeadersAdd, 1)
added := out.Mutations.HeadersAdd[0]
assert.Equal(t, "X-Amzn-Bedrock-Request-Metadata", added.Key,
"the Bedrock cost-allocation header carries the metadata JSON")
var payload map[string]string
require.NoError(t, json.Unmarshal([]byte(added.Value), &payload))
assert.Equal(t, "alice@example.com", payload["user"],
"'@' and '.' are in Bedrock's accepted set and must be preserved")
assert.NotContains(t, payload["group"], ",", "commas must be sanitized — Bedrock rejects them")
assert.NotContains(t, payload["group"], "#", "disallowed characters must be sanitized")
assert.Contains(t, payload["group"], "Eng", "allowed characters must be preserved")
}
// TestInject_JSONMetadata_EmptyIdentity_StripsButDoesNotAdd verifies the
// anti-spoof Remove still fires when there's nothing to stamp.
func TestInject_JSONMetadata_EmptyIdentity_StripsButDoesNotAdd(t *testing.T) {

View File

@@ -1,30 +0,0 @@
package llm_router
import (
"testing"
"github.com/stretchr/testify/assert"
)
// TestRouteClaimsModel_BedrockNormalizesCandidate guards the fix for the native
// Bedrock routing gap: the request model reaches the router already normalized
// (the parser strips the region/inference-profile prefix and version suffix),
// so a provider registered with the raw inference-profile id must still match.
func TestRouteClaimsModel_BedrockNormalizesCandidate(t *testing.T) {
route := ProviderRoute{Bedrock: true, Models: []string{"us.anthropic.claude-haiku-4-5"}}
assert.True(t, routeClaimsModel(route, "anthropic.claude-haiku-4-5"),
"raw region-prefixed Bedrock model must match the normalized request model")
assert.False(t, routeClaimsModel(route, "anthropic.claude-opus-4-8"),
"a model outside the provider's list must not match")
// A provider registered with the already-normalized id also matches.
normalized := ProviderRoute{Bedrock: true, Models: []string{"anthropic.claude-haiku-4-5"}}
assert.True(t, routeClaimsModel(normalized, "anthropic.claude-haiku-4-5"),
"normalized Bedrock model must match")
// Non-Bedrock routes keep exact matching (no prefix stripping).
openai := ProviderRoute{Models: []string{"gpt-4o"}}
assert.True(t, routeClaimsModel(openai, "gpt-4o"), "exact model must match")
assert.False(t, routeClaimsModel(openai, "us.gpt-4o"),
"non-Bedrock routes must not strip a us. prefix")
}

View File

@@ -23,7 +23,6 @@ import (
"golang.org/x/oauth2"
"golang.org/x/oauth2/google"
"github.com/netbirdio/netbird/proxy/internal/llm"
"github.com/netbirdio/netbird/proxy/internal/middleware"
)
@@ -556,14 +555,6 @@ func routeClaimsModel(route ProviderRoute, model string) bool {
if candidate == model {
return true
}
// Bedrock request models reach the router already normalized (the parser
// strips the region / inference-profile prefix and version suffix), but
// the operator may register the raw inference-profile id (e.g.
// "us.anthropic.claude-haiku-4-5"). Normalize the candidate so both sides
// compare equal; otherwise a native Bedrock request denies as not-routable.
if route.Bedrock && llm.NormalizeBedrockModel(candidate) == model {
return true
}
}
return false
}

View File

@@ -22,7 +22,6 @@ import (
"github.com/netbirdio/netbird/proxy/internal/roundtrip"
"github.com/netbirdio/netbird/proxy/internal/types"
"github.com/netbirdio/netbird/proxy/web"
"github.com/netbirdio/netbird/trustedproxy"
)
type ReverseProxy struct {
@@ -30,10 +29,10 @@ type ReverseProxy struct {
// forwardedProto overrides the X-Forwarded-Proto header value.
// Valid values: "auto" (detect from TLS), "http", "https".
forwardedProto string
// trustedProxies is the set of trusted upstream proxies. When the direct
// connection comes from a trusted proxy, forwarding headers are preserved
// and appended to instead of being stripped.
trustedProxies *trustedproxy.List
// trustedProxies is a list of IP prefixes for trusted upstream proxies.
// When the direct connection comes from a trusted proxy, forwarding
// headers are preserved and appended to instead of being stripped.
trustedProxies []netip.Prefix
mappingsMux sync.RWMutex
mappings map[string]Mapping
logger *log.Logger
@@ -64,7 +63,7 @@ func WithMiddlewareManager(m *middleware.Manager) Option {
// between requested URLs and targets.
// The internal mappings can be modified using the AddMapping
// and RemoveMapping functions.
func NewReverseProxy(transport http.RoundTripper, forwardedProto string, trustedProxies *trustedproxy.List, logger *log.Logger, opts ...Option) *ReverseProxy {
func NewReverseProxy(transport http.RoundTripper, forwardedProto string, trustedProxies []netip.Prefix, logger *log.Logger, opts ...Option) *ReverseProxy {
if logger == nil {
logger = log.StandardLogger()
}
@@ -528,7 +527,7 @@ func (p *ReverseProxy) isSelfTargetLoop(r *http.Request, target *url.URL) bool {
if !types.IsOverlayOrigin(r.Context()) {
return false
}
srcIP := trustedproxy.ExtractHostIP(r.RemoteAddr)
srcIP := extractHostIP(r.RemoteAddr)
if !srcIP.IsValid() {
return false
}
@@ -579,9 +578,9 @@ func (p *ReverseProxy) rewriteFunc(target *url.URL, matchedPath string, passHost
stampNetBirdIdentity(r)
clientIP := trustedproxy.ExtractHostIP(r.In.RemoteAddr)
clientIP := extractHostIP(r.In.RemoteAddr)
if p.trustedProxies.Contains(clientIP) {
if isTrustedAddr(clientIP, p.trustedProxies) {
p.setTrustedForwardingHeaders(r, clientIP)
} else {
p.setUntrustedForwardingHeaders(r, clientIP)
@@ -665,7 +664,7 @@ func (p *ReverseProxy) setTrustedForwardingHeaders(r *httputil.ProxyRequest, cli
if realIP := r.In.Header.Get("X-Real-IP"); realIP != "" {
r.Out.Header.Set("X-Real-IP", realIP)
} else {
resolved := p.trustedProxies.ResolveClientIP(r.In.RemoteAddr, r.In.Header.Get("X-Forwarded-For"))
resolved := ResolveClientIP(r.In.RemoteAddr, r.In.Header.Get("X-Forwarded-For"), p.trustedProxies)
r.Out.Header.Set("X-Real-IP", resolved.String())
}

View File

@@ -23,7 +23,6 @@ import (
"github.com/netbirdio/netbird/proxy/internal/roundtrip"
"github.com/netbirdio/netbird/proxy/internal/types"
"github.com/netbirdio/netbird/proxy/web"
"github.com/netbirdio/netbird/trustedproxy"
)
func TestRewriteFunc_HostRewriting(t *testing.T) {
@@ -303,7 +302,7 @@ func TestExtractHostIP(t *testing.T) {
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.expected, trustedproxy.ExtractHostIP(tt.remoteAddr))
assert.Equal(t, tt.expected, extractHostIP(tt.remoteAddr))
})
}
}
@@ -331,7 +330,7 @@ func TestExtractForwardedPort(t *testing.T) {
func TestRewriteFunc_TrustedProxy(t *testing.T) {
target, _ := url.Parse("http://backend.internal:8080")
trusted := trustedproxy.FromPrefixes([]netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")})
trusted := []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")}
t.Run("appends to X-Forwarded-For", func(t *testing.T) {
p := &ReverseProxy{forwardedProto: "auto", trustedProxies: trusted}

View File

@@ -0,0 +1,81 @@
package proxy
import (
"net/netip"
"strings"
)
// IsTrustedProxy checks if the given IP string falls within any of the trusted prefixes.
func IsTrustedProxy(ipStr string, trusted []netip.Prefix) bool {
addr, err := netip.ParseAddr(ipStr)
if err != nil || len(trusted) == 0 {
return false
}
return isTrustedAddr(addr.Unmap(), trusted)
}
// ResolveClientIP extracts the real client IP from X-Forwarded-For using the trusted proxy list.
// It walks the XFF chain right-to-left, skipping IPs that match trusted prefixes.
// The first untrusted IP is the real client.
//
// If the trusted list is empty or remoteAddr is not trusted, it returns the
// remoteAddr IP directly (ignoring any forwarding headers).
func ResolveClientIP(remoteAddr, xff string, trusted []netip.Prefix) netip.Addr {
remoteIP := extractHostIP(remoteAddr)
if len(trusted) == 0 || !isTrustedAddr(remoteIP, trusted) {
return remoteIP
}
if xff == "" {
return remoteIP
}
parts := strings.Split(xff, ",")
for i := len(parts) - 1; i >= 0; i-- {
ip := strings.TrimSpace(parts[i])
if ip == "" {
continue
}
addr, err := netip.ParseAddr(ip)
if err != nil {
continue
}
addr = addr.Unmap()
if !isTrustedAddr(addr, trusted) {
return addr
}
}
// All IPs in XFF are trusted; return the leftmost as best guess.
if first := strings.TrimSpace(parts[0]); first != "" {
if addr, err := netip.ParseAddr(first); err == nil {
return addr.Unmap()
}
}
return remoteIP
}
// extractHostIP parses the IP from a host:port string and returns it unmapped.
func extractHostIP(hostPort string) netip.Addr {
if ap, err := netip.ParseAddrPort(hostPort); err == nil {
return ap.Addr().Unmap()
}
if addr, err := netip.ParseAddr(hostPort); err == nil {
return addr.Unmap()
}
return netip.Addr{}
}
// isTrustedAddr checks if the given address falls within any of the trusted prefixes.
func isTrustedAddr(addr netip.Addr, trusted []netip.Prefix) bool {
if !addr.IsValid() {
return false
}
for _, prefix := range trusted {
if prefix.Contains(addr) {
return true
}
}
return false
}

View File

@@ -0,0 +1,129 @@
package proxy
import (
"net/netip"
"testing"
"github.com/stretchr/testify/assert"
)
func TestIsTrustedProxy(t *testing.T) {
trusted := []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("192.168.1.0/24"),
netip.MustParsePrefix("fd00::/8"),
}
tests := []struct {
name string
ip string
trusted []netip.Prefix
want bool
}{
{"empty trusted list", "10.0.0.1", nil, false},
{"IP within /8 prefix", "10.1.2.3", trusted, true},
{"IP within /24 prefix", "192.168.1.100", trusted, true},
{"IP outside all prefixes", "203.0.113.50", trusted, false},
{"boundary IP just outside prefix", "192.168.2.1", trusted, false},
{"unparsable IP", "not-an-ip", trusted, false},
{"IPv6 in trusted range", "fd00::1", trusted, true},
{"IPv6 outside range", "2001:db8::1", trusted, false},
{"empty string", "", trusted, false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, IsTrustedProxy(tt.ip, tt.trusted))
})
}
}
func TestResolveClientIP(t *testing.T) {
trusted := []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("172.16.0.0/12"),
}
tests := []struct {
name string
remoteAddr string
xff string
trusted []netip.Prefix
want netip.Addr
}{
{
name: "empty trusted list returns RemoteAddr",
remoteAddr: "203.0.113.50:9999",
xff: "1.2.3.4",
trusted: nil,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "untrusted RemoteAddr ignores XFF",
remoteAddr: "203.0.113.50:9999",
xff: "1.2.3.4, 10.0.0.1",
trusted: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "trusted RemoteAddr with single client in XFF",
remoteAddr: "10.0.0.1:5000",
xff: "203.0.113.50",
trusted: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "trusted RemoteAddr walks past trusted entries in XFF",
remoteAddr: "10.0.0.1:5000",
xff: "203.0.113.50, 10.0.0.2, 172.16.0.5",
trusted: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "trusted RemoteAddr with empty XFF falls back to RemoteAddr",
remoteAddr: "10.0.0.1:5000",
xff: "",
trusted: trusted,
want: netip.MustParseAddr("10.0.0.1"),
},
{
name: "all XFF IPs trusted returns leftmost",
remoteAddr: "10.0.0.1:5000",
xff: "10.0.0.2, 172.16.0.1, 10.0.0.3",
trusted: trusted,
want: netip.MustParseAddr("10.0.0.2"),
},
{
name: "XFF with whitespace",
remoteAddr: "10.0.0.1:5000",
xff: " 203.0.113.50 , 10.0.0.2 ",
trusted: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "XFF with empty segments",
remoteAddr: "10.0.0.1:5000",
xff: "203.0.113.50,,10.0.0.2",
trusted: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "multi-hop with mixed trust",
remoteAddr: "10.0.0.1:5000",
xff: "8.8.8.8, 203.0.113.50, 172.16.0.1",
trusted: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "RemoteAddr without port",
remoteAddr: "10.0.0.1",
xff: "203.0.113.50",
trusted: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, ResolveClientIP(tt.remoteAddr, tt.xff, tt.trusted))
})
}
}

View File

@@ -2,13 +2,13 @@ package proxy
import (
"context"
"net/netip"
"time"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/embed"
"github.com/netbirdio/netbird/proxy/internal/acme"
"github.com/netbirdio/netbird/trustedproxy"
)
// Config bundles every knob the proxy reads at construction time. It mirrors
@@ -83,9 +83,9 @@ type Config struct {
// ForwardedProto overrides the X-Forwarded-Proto value sent to
// backends. Valid values: "auto", "http", "https".
ForwardedProto string
// TrustedProxies is the set of trusted upstream proxies that may set
// forwarding headers.
TrustedProxies *trustedproxy.List
// TrustedProxies is a list of IP prefixes for trusted upstream
// proxies that may set forwarding headers.
TrustedProxies []netip.Prefix
// WireguardPort is the UDP port for the embedded NetBird tunnel.
// Zero asks the OS for a random port.
WireguardPort uint16

View File

@@ -10,14 +10,12 @@ import (
log "github.com/sirupsen/logrus"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/trustedproxy"
)
func TestWrapProxyProtocol_OverridesRemoteAddr(t *testing.T) {
srv := &Server{
Logger: log.StandardLogger(),
TrustedProxies: trustedproxy.FromPrefixes([]netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")}),
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("127.0.0.1/32")},
ProxyProtocol: true,
}
@@ -68,7 +66,7 @@ func TestWrapProxyProtocol_OverridesRemoteAddr(t *testing.T) {
func TestProxyProtocolPolicy_TrustedRequires(t *testing.T) {
srv := &Server{
Logger: log.StandardLogger(),
TrustedProxies: trustedproxy.FromPrefixes([]netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")}),
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
}
opts := proxyproto.ConnPolicyOptions{
@@ -82,7 +80,7 @@ func TestProxyProtocolPolicy_TrustedRequires(t *testing.T) {
func TestProxyProtocolPolicy_UntrustedIgnores(t *testing.T) {
srv := &Server{
Logger: log.StandardLogger(),
TrustedProxies: trustedproxy.FromPrefixes([]netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")}),
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
}
opts := proxyproto.ConnPolicyOptions{
@@ -96,7 +94,7 @@ func TestProxyProtocolPolicy_UntrustedIgnores(t *testing.T) {
func TestProxyProtocolPolicy_InvalidIPRejects(t *testing.T) {
srv := &Server{
Logger: log.StandardLogger(),
TrustedProxies: trustedproxy.FromPrefixes([]netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")}),
TrustedProxies: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
}
opts := proxyproto.ConnPolicyOptions{

View File

@@ -67,7 +67,6 @@ import (
"github.com/netbirdio/netbird/proxy/web"
"github.com/netbirdio/netbird/shared/management/domain"
"github.com/netbirdio/netbird/shared/management/proto"
"github.com/netbirdio/netbird/trustedproxy"
"github.com/netbirdio/netbird/util/embeddedroots"
)
@@ -80,19 +79,19 @@ type portRouter struct {
type Server struct {
ctx context.Context
mgmtClient proto.ProxyServiceClient
proxy *proxy.ReverseProxy
netbird *roundtrip.NetBird
acme *acme.Manager
mgmtClient proto.ProxyServiceClient
proxy *proxy.ReverseProxy
netbird *roundtrip.NetBird
acme *acme.Manager
staticCertWatcher *certwatch.Watcher
auth *auth.Middleware
http *http.Server
https *http.Server
debug *http.Server
healthServer *health.Server
healthChecker *health.Checker
meter *proxymetrics.Metrics
accessLog *accesslog.Logger
auth *auth.Middleware
http *http.Server
https *http.Server
debug *http.Server
healthServer *health.Server
healthChecker *health.Checker
meter *proxymetrics.Metrics
accessLog *accesslog.Logger
// middlewareManager drives per-target middleware dispatch. Always
// constructed during boot; an empty registry produces empty chains and
// the reverse-proxy stays on the no-capture fast path.
@@ -100,16 +99,16 @@ type Server struct {
// middlewareRegistry is the source of registered middleware factories.
// Concrete middlewares register themselves through init().
middlewareRegistry *middleware.Registry
mainRouter *nbtcp.Router
mainPort uint16
udpMu sync.Mutex
udpRelays map[types.ServiceID]*udprelay.Relay
udpRelayWg sync.WaitGroup
portMu sync.RWMutex
portRouters map[uint16]*portRouter
svcPorts map[types.ServiceID][]uint16
lastMappings map[types.ServiceID]*proto.ProxyMapping
portRouterWg sync.WaitGroup
mainRouter *nbtcp.Router
mainPort uint16
udpMu sync.Mutex
udpRelays map[types.ServiceID]*udprelay.Relay
udpRelayWg sync.WaitGroup
portMu sync.RWMutex
portRouters map[uint16]*portRouter
svcPorts map[types.ServiceID][]uint16
lastMappings map[types.ServiceID]*proto.ProxyMapping
portRouterWg sync.WaitGroup
// hijackTracker tracks hijacked connections (e.g. WebSocket upgrades)
// so they can be closed during graceful shutdown, since http.Server.Shutdown
@@ -193,10 +192,10 @@ type Server struct {
// ForwardedProto overrides the X-Forwarded-Proto value sent to backends.
// Valid values: "auto" (detect from TLS), "http", "https".
ForwardedProto string
// TrustedProxies is the set of trusted upstream proxies. When set,
// forwarding headers from these sources are preserved and appended to
// instead of being stripped.
TrustedProxies *trustedproxy.List
// TrustedProxies is a list of IP prefixes for trusted upstream proxies.
// When set, forwarding headers from these sources are preserved and
// appended to instead of being stripped.
TrustedProxies []netip.Prefix
// WireguardPort is the port for the NetBird tunnel interface. Use 0
// for a random OS-assigned port. A fixed port only works with
// single-account deployments; multiple accounts will fail to bind
@@ -719,7 +718,7 @@ func (s *Server) wrapProxyProtocol(ln net.Listener) net.Listener {
Listener: ln,
ReadHeaderTimeout: proxyProtoHeaderTimeout,
}
if !s.TrustedProxies.Empty() {
if len(s.TrustedProxies) > 0 {
ppListener.ConnPolicy = s.proxyProtocolPolicy
} else {
s.Logger.Warn("PROXY protocol enabled without trusted proxies; any source may send PROXY headers")
@@ -743,8 +742,10 @@ func (s *Server) proxyProtocolPolicy(opts proxyproto.ConnPolicyOptions) (proxypr
addr = addr.Unmap()
// called per accept
if s.TrustedProxies.Contains(addr) {
return proxyproto.REQUIRE, nil
for _, prefix := range s.TrustedProxies {
if prefix.Contains(addr) {
return proxyproto.REQUIRE, nil
}
}
return proxyproto.IGNORE, nil
}

43
proxy/trustedproxy.go Normal file
View File

@@ -0,0 +1,43 @@
package proxy
import (
"fmt"
"net/netip"
"strings"
)
// ParseTrustedProxies parses a comma-separated list of CIDR prefixes or bare IPs
// into a slice of netip.Prefix values suitable for trusted proxy configuration.
// Bare IPs are converted to single-host prefixes (/32 or /128).
func ParseTrustedProxies(raw string) ([]netip.Prefix, error) {
if raw == "" {
return nil, nil
}
parts := strings.Split(raw, ",")
prefixes := make([]netip.Prefix, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part == "" {
continue
}
prefix, err := netip.ParsePrefix(part)
if err == nil {
prefixes = append(prefixes, prefix)
continue
}
addr, addrErr := netip.ParseAddr(part)
if addrErr != nil {
return nil, fmt.Errorf("parse trusted proxy %q: not a valid CIDR or IP: %w", part, addrErr)
}
bits := 32
if addr.Is6() {
bits = 128
}
prefixes = append(prefixes, netip.PrefixFrom(addr, bits))
}
return prefixes, nil
}

View File

@@ -0,0 +1,90 @@
package proxy
import (
"net/netip"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestParseTrustedProxies(t *testing.T) {
tests := []struct {
name string
raw string
want []netip.Prefix
wantErr bool
}{
{
name: "empty string returns nil",
raw: "",
want: nil,
},
{
name: "single CIDR",
raw: "10.0.0.0/8",
want: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
},
{
name: "single bare IPv4",
raw: "1.2.3.4",
want: []netip.Prefix{netip.MustParsePrefix("1.2.3.4/32")},
},
{
name: "single bare IPv6",
raw: "::1",
want: []netip.Prefix{netip.MustParsePrefix("::1/128")},
},
{
name: "comma-separated CIDRs",
raw: "10.0.0.0/8, 192.168.1.0/24",
want: []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("192.168.1.0/24"),
},
},
{
name: "mixed CIDRs and bare IPs",
raw: "10.0.0.0/8, 1.2.3.4, fd00::/8",
want: []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("1.2.3.4/32"),
netip.MustParsePrefix("fd00::/8"),
},
},
{
name: "whitespace around entries",
raw: " 10.0.0.0/8 , 192.168.0.0/16 ",
want: []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("192.168.0.0/16"),
},
},
{
name: "trailing comma produces no extra entry",
raw: "10.0.0.0/8,",
want: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
},
{
name: "invalid entry",
raw: "not-an-ip",
wantErr: true,
},
{
name: "partially invalid",
raw: "10.0.0.0/8, garbage",
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := ParseTrustedProxies(tt.raw)
if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, tt.want, got)
})
}
}

View File

@@ -24,7 +24,6 @@ import (
"github.com/netbirdio/netbird/shared/metrics"
"github.com/netbirdio/netbird/shared/relay/auth"
"github.com/netbirdio/netbird/stun"
"github.com/netbirdio/netbird/trustedproxy"
"github.com/netbirdio/netbird/util"
)
@@ -46,9 +45,6 @@ type Config struct {
LogLevel string
LogFile string
HealthcheckListenAddress string
// TrustedProxies is a comma-separated list of upstream proxy CIDRs/IPs whose
// X-Real-Ip/X-Real-Port headers are trusted. Empty means never trust these headers.
TrustedProxies string
// STUN server configuration
EnableSTUN bool
STUNPorts []int
@@ -120,7 +116,6 @@ func init() {
rootCmd.PersistentFlags().StringVar(&cobraConfig.LogLevel, "log-level", "info", "log level")
rootCmd.PersistentFlags().StringVar(&cobraConfig.LogFile, "log-file", "console", "log file")
rootCmd.PersistentFlags().StringVarP(&cobraConfig.HealthcheckListenAddress, "health-listen-address", "H", ":9000", "listen address of healthcheck server")
rootCmd.PersistentFlags().StringVar(&cobraConfig.TrustedProxies, "trusted-proxies", "", "comma-separated list of upstream proxy CIDRs or IPs whose X-Real-Ip/X-Real-Port headers are trusted; leave empty to always use the direct connection address")
rootCmd.PersistentFlags().BoolVar(&cobraConfig.EnableSTUN, "enable-stun", false, "enable embedded STUN server")
rootCmd.PersistentFlags().IntSliceVar(&cobraConfig.STUNPorts, "stun-ports", []int{3478}, "ports for the embedded STUN server (can be specified multiple times or comma-separated)")
rootCmd.PersistentFlags().StringVar(&cobraConfig.STUNLogLevel, "stun-log-level", "info", "log level for STUN server (panic, fatal, error, warn, info, debug, trace)")
@@ -160,15 +155,8 @@ func execute(cmd *cobra.Command, args []string) error {
return fmt.Errorf("setup metrics: %v", err)
}
trustedProxies, err := trustedproxy.Parse(cobraConfig.TrustedProxies)
if err != nil {
log.Debugf("failed to parse trusted proxies: %s", err)
return fmt.Errorf("failed to parse trusted proxies: %s", err)
}
srvListenerCfg := server.ListenerConfig{
Address: cobraConfig.ListenAddress,
TrustedProxies: trustedProxies,
Address: cobraConfig.ListenAddress,
}
tlsConfig, tlsSupport, err := handleTLSConfig(cobraConfig)
@@ -185,7 +173,7 @@ func execute(cmd *cobra.Command, args []string) error {
}
hashedSecret := sha256.Sum256([]byte(cobraConfig.AuthSecret))
authenticator := auth.NewTimedHMACValidator(hashedSecret[:])
authenticator := auth.NewTimedHMACValidator(hashedSecret[:], 24*time.Hour)
cfg := server.Config{
Meter: metricsServer.Meter,

View File

@@ -5,8 +5,14 @@ import (
"fmt"
"time"
log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/relay/server/listener"
"github.com/netbirdio/netbird/shared/relay/messages"
//nolint:staticcheck
"github.com/netbirdio/netbird/shared/relay/messages/address"
//nolint:staticcheck
authmsg "github.com/netbirdio/netbird/shared/relay/messages/auth"
)
const (
@@ -17,30 +23,55 @@ const (
type Validator interface {
Validate(any) error
// Deprecated: Use Validate instead.
ValidateHelloMsgType(any) error
}
// preparedMsg contains the marshalled success response message
// preparedMsg contains the marshalled success response messages
type preparedMsg struct {
responseAuthMsg []byte
responseHelloMsg []byte
responseAuthMsg []byte
}
func newPreparedMsg(instanceURL string) (*preparedMsg, error) {
rhm, err := marshalResponseHelloMsg(instanceURL)
if err != nil {
return nil, err
}
ram, err := messages.MarshalAuthResponse(instanceURL)
if err != nil {
return nil, fmt.Errorf("failed to marshal auth response msg: %w", err)
}
return &preparedMsg{
responseAuthMsg: ram,
responseHelloMsg: rhm,
responseAuthMsg: ram,
}, nil
}
func marshalResponseHelloMsg(instanceURL string) ([]byte, error) {
addr := &address.Address{URL: instanceURL}
addrData, err := addr.Marshal()
if err != nil {
return nil, fmt.Errorf("failed to marshal response address: %w", err)
}
//nolint:staticcheck
responseMsg, err := messages.MarshalHelloResponse(addrData)
if err != nil {
return nil, fmt.Errorf("failed to marshal hello response: %w", err)
}
return responseMsg, nil
}
type handshake struct {
conn listener.Conn
validator Validator
preparedMsg *preparedMsg
peerID *messages.PeerID
handshakeMethodAuth bool
peerID *messages.PeerID
}
func (h *handshake) handshakeReceive(ctx context.Context) (*messages.PeerID, error) {
@@ -62,11 +93,17 @@ func (h *handshake) handshakeReceive(ctx context.Context) (*messages.PeerID, err
return nil, fmt.Errorf("determine message type from %s: %w", h.conn.RemoteAddr(), err)
}
if msgType != messages.MsgTypeAuth {
var peerID *messages.PeerID
switch msgType {
//nolint:staticcheck
case messages.MsgTypeHello:
peerID, err = h.handleHelloMsg(buf)
case messages.MsgTypeAuth:
h.handshakeMethodAuth = true
peerID, err = h.handleAuthMsg(buf)
default:
return nil, fmt.Errorf("invalid message type %d from %s", msgType, h.conn.RemoteAddr())
}
peerID, err := h.handleAuthMsg(buf)
if err != nil {
return peerID, err
}
@@ -75,17 +112,46 @@ func (h *handshake) handshakeReceive(ctx context.Context) (*messages.PeerID, err
}
func (h *handshake) handshakeResponse(ctx context.Context) error {
if _, err := h.conn.Write(ctx, h.preparedMsg.responseAuthMsg); err != nil {
var responseMsg []byte
if h.handshakeMethodAuth {
responseMsg = h.preparedMsg.responseAuthMsg
} else {
responseMsg = h.preparedMsg.responseHelloMsg
}
if _, err := h.conn.Write(ctx, responseMsg); err != nil {
return fmt.Errorf("handshake response write to %s (%s): %w", h.peerID, h.conn.RemoteAddr(), err)
}
return nil
}
func (h *handshake) handleHelloMsg(buf []byte) (*messages.PeerID, error) {
//nolint:staticcheck
peerID, authData, err := messages.UnmarshalHelloMsg(buf)
if err != nil {
return nil, fmt.Errorf("unmarshal hello message: %w", err)
}
log.Warnf("peer %s (%s) is using deprecated initial message type", peerID, h.conn.RemoteAddr())
authMsg, err := authmsg.UnmarshalMsg(authData)
if err != nil {
return nil, fmt.Errorf("unmarshal auth message: %w", err)
}
//nolint:staticcheck
if err := h.validator.ValidateHelloMsgType(authMsg.AdditionalData); err != nil {
return nil, fmt.Errorf("validate %s (%s): %w", peerID, h.conn.RemoteAddr(), err)
}
return peerID, nil
}
func (h *handshake) handleAuthMsg(buf []byte) (*messages.PeerID, error) {
rawPeerID, authPayload, err := messages.UnmarshalAuthMsg(buf)
if err != nil {
return nil, fmt.Errorf("unmarshal auth message: %w", err)
return nil, fmt.Errorf("unmarshal hello message: %w", err)
}
if err := h.validator.Validate(authPayload); err != nil {

View File

@@ -15,7 +15,6 @@ import (
"github.com/netbirdio/netbird/relay/protocol"
relaylistener "github.com/netbirdio/netbird/relay/server/listener"
"github.com/netbirdio/netbird/shared/relay"
"github.com/netbirdio/netbird/trustedproxy"
)
const (
@@ -28,9 +27,6 @@ type Listener struct {
Address string
// TLSConfig is the TLS configuration for the server.
TLSConfig *tls.Config
// TrustedProxies is the set of upstream proxies whose X-Real-Ip/X-Real-Port
// headers are trusted. Headers from any other immediate peer are ignored.
TrustedProxies *trustedproxy.List
server *http.Server
acceptFn func(conn relaylistener.Conn)
@@ -79,7 +75,7 @@ func (l *Listener) Shutdown(ctx context.Context) error {
}
func (l *Listener) onAccept(w http.ResponseWriter, r *http.Request) {
connRemoteAddr := remoteAddr(r, l.TrustedProxies)
connRemoteAddr := remoteAddr(r)
acceptOptions := &websocket.AcceptOptions{
OriginPatterns: []string{"*"},
@@ -106,17 +102,9 @@ func (l *Listener) onAccept(w http.ResponseWriter, r *http.Request) {
l.acceptFn(conn)
}
func remoteAddr(r *http.Request, trustedProxies *trustedproxy.List) string {
realIP := r.Header.Get("X-Real-Ip")
realPort := r.Header.Get("X-Real-Port")
if realIP == "" || realPort == "" {
func remoteAddr(r *http.Request) string {
if r.Header.Get("X-Real-Ip") == "" || r.Header.Get("X-Real-Port") == "" {
return r.RemoteAddr
}
if !trustedProxies.IsTrusted(r.RemoteAddr) {
log.Debugf("ignoring X-Real-Ip header from untrusted peer %s", r.RemoteAddr)
return r.RemoteAddr
}
return net.JoinHostPort(realIP, realPort)
return net.JoinHostPort(r.Header.Get("X-Real-Ip"), r.Header.Get("X-Real-Port"))
}

View File

@@ -15,17 +15,14 @@ import (
"github.com/netbirdio/netbird/relay/server/listener/quic"
"github.com/netbirdio/netbird/relay/server/listener/ws"
quictls "github.com/netbirdio/netbird/shared/relay/tls"
"github.com/netbirdio/netbird/trustedproxy"
)
// ListenerConfig is the configuration for the listener.
// Address: the address to bind the listener to. It could be an address behind a reverse proxy.
// TLSConfig: the TLS configuration for the listener.
// TrustedProxies: upstream proxy prefixes whose forwarding headers (X-Real-Ip/X-Real-Port) are trusted.
type ListenerConfig struct {
Address string
TLSConfig *tls.Config
TrustedProxies *trustedproxy.List
Address string
TLSConfig *tls.Config
}
// Server is the main entry point for the relay server.
@@ -65,9 +62,8 @@ func NewServer(config Config) (*Server, error) {
// Listen starts the relay server.
func (r *Server) Listen(cfg ListenerConfig) error {
wSListener := &ws.Listener{
Address: cfg.Address,
TLSConfig: cfg.TLSConfig,
TrustedProxies: cfg.TrustedProxies,
Address: cfg.Address,
TLSConfig: cfg.TLSConfig,
}
r.listenerMux.Lock()

View File

@@ -5164,10 +5164,6 @@ components:
type: boolean
description: Whether upstream TLS certificate verification is skipped when the proxy dials this provider's URL. Intended for self-hosted / internal gateways behind a private or self-signed certificate.
example: false
metadata_disabled:
type: boolean
description: Whether identity metadata injection is disabled for this provider. When enabled (the default), the proxy stamps the caller's user and authorizing group onto upstream requests as provider-specific metadata (e.g. AWS Bedrock's X-Amzn-Bedrock-Request-Metadata header). Set true to suppress it.
example: false
created_at:
type: string
format: date-time
@@ -5188,7 +5184,6 @@ components:
- models
- enabled
- skip_tls_verification
- metadata_disabled
- created_at
- updated_at
AgentNetworkProviderRequest:
@@ -5245,10 +5240,6 @@ components:
type: boolean
description: Skip upstream TLS certificate verification when the proxy dials this provider's URL. For self-hosted / internal gateways behind a private or self-signed certificate. Defaults to false. When omitted on update, the stored value is left unchanged.
example: false
metadata_disabled:
type: boolean
description: Disable identity metadata injection (the caller's user + authorizing group) for this provider. Defaults to false (metadata is injected). When omitted on update, the stored value is left unchanged.
example: false
required:
- provider_id
- name

View File

@@ -2227,9 +2227,6 @@ type AgentNetworkProvider struct {
// IdentityHeaderUserId Wire header name the proxy stamps with the caller's display identity (user email or peer name) when the catalog entry's HeaderPair is `customizable`. Empty disables stamping for this dimension. Ignored when the catalog entry has a fixed HeaderPair (e.g. LiteLLM, Portkey). Used today by Bifrost: typical values are `x-bf-lh-netbird_user_id` (always-on log metadata) or `x-bf-dim-netbird_user_id` (Prometheus / OTEL — requires the label to be pre-declared in the gateway's `client.prometheus_labels` config).
IdentityHeaderUserId *string `json:"identity_header_user_id,omitempty"`
// MetadataDisabled Whether identity metadata injection is disabled for this provider. When enabled (the default), the proxy stamps the caller's user and authorizing group onto upstream requests as provider-specific metadata (e.g. AWS Bedrock's X-Amzn-Bedrock-Request-Metadata header). Set true to suppress it.
MetadataDisabled bool `json:"metadata_disabled"`
// Models Models exposed through this endpoint, with the operator's per-1k input/output prices. Empty means all catalog models are allowed at catalog prices.
Models []AgentNetworkProviderModel `json:"models"`
@@ -2281,9 +2278,6 @@ type AgentNetworkProviderRequest struct {
// IdentityHeaderUserId Wire header name for the caller's display identity. See AgentNetworkProvider.identity_header_user_id. When omitted on a request, the stored value is left unchanged; pass an empty string explicitly to clear it (which disables stamping for this dimension).
IdentityHeaderUserId *string `json:"identity_header_user_id,omitempty"`
// MetadataDisabled Disable identity metadata injection (the caller's user + authorizing group) for this provider. Defaults to false (metadata is injected). When omitted on update, the stored value is left unchanged.
MetadataDisabled *bool `json:"metadata_disabled,omitempty"`
// Models Models exposed through this endpoint, with the operator's per-1k input/output prices. Empty means all catalog models are allowed at catalog prices.
Models *[]AgentNetworkProviderModel `json:"models,omitempty"`

View File

@@ -8,3 +8,7 @@ type Auth struct {
func (a *Auth) Validate(any) error {
return nil
}
func (a *Auth) ValidateHelloMsgType(any) error {
return nil
}

View File

@@ -1,8 +1,10 @@
package hmac
import (
"bytes"
"crypto/hmac"
"encoding/base64"
"encoding/gob"
"fmt"
"hash"
"strconv"
@@ -16,6 +18,14 @@ type Token struct {
Signature string
}
func unmarshalToken(payload []byte) (Token, error) {
var creds Token
buffer := bytes.NewBuffer(payload)
decoder := gob.NewDecoder(buffer)
err := decoder.Decode(&creds)
return creds, err
}
// TimedHMAC generates a token with TTL and uses a pre-shared secret known to the relay server
type TimedHMAC struct {
secret string

View File

@@ -0,0 +1,33 @@
package hmac
import (
"crypto/sha256"
"fmt"
"time"
log "github.com/sirupsen/logrus"
)
type TimedHMACValidator struct {
*TimedHMAC
}
func NewTimedHMACValidator(secret string, duration time.Duration) *TimedHMACValidator {
ta := NewTimedHMAC(secret, duration)
return &TimedHMACValidator{
ta,
}
}
func (a *TimedHMACValidator) Validate(credentials any) error {
b, ok := credentials.([]byte)
if !ok {
return fmt.Errorf("invalid credentials type")
}
c, err := unmarshalToken(b)
if err != nil {
log.Debugf("failed to unmarshal token: %s", err)
return err
}
return a.TimedHMAC.Validate(sha256.New, c)
}

View File

@@ -1,19 +1,28 @@
package auth
import (
"time"
auth "github.com/netbirdio/netbird/shared/relay/auth/hmac"
authv2 "github.com/netbirdio/netbird/shared/relay/auth/hmac/v2"
)
type TimedHMACValidator struct {
authenticatorV2 *authv2.Validator
authenticator *auth.TimedHMACValidator
}
func NewTimedHMACValidator(secret []byte) *TimedHMACValidator {
func NewTimedHMACValidator(secret []byte, duration time.Duration) *TimedHMACValidator {
return &TimedHMACValidator{
authenticatorV2: authv2.NewValidator(secret),
authenticator: auth.NewTimedHMACValidator(string(secret), duration),
}
}
func (a *TimedHMACValidator) Validate(credentials any) error {
return a.authenticatorV2.Validate(credentials)
}
func (a *TimedHMACValidator) ValidateHelloMsgType(credentials any) error {
return a.authenticator.Validate(credentials)
}

View File

@@ -1,165 +0,0 @@
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
}
func NewFallbackOpener(home *Client, foreignStore *ForeignRelaysStore) *FallbackOpener {
return &FallbackOpener{
home: home,
foreignStore: foreignStore,
}
}
func (r *FallbackOpener) Run(ctx context.Context, peerKey string, remoteRelayServer RelayServer, preferForeign bool) (net.Conn, error) {
raceCtx, cancel := context.WithTimeout(ctx, raceTotalTimeout)
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(raceFallbackDelay),
}
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 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

@@ -1,175 +0,0 @@
package client
import (
"context"
"errors"
"net"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/netbirdio/netbird/relay/server"
)
type fakeConn struct {
net.Conn
closed chan struct{}
}
func newFakeConn() *fakeConn {
return &fakeConn{closed: make(chan struct{})}
}
func (c *fakeConn) Close() error {
close(c.closed)
return nil
}
func newTestConnRace(t *testing.T) *connRace {
t.Helper()
raceCtx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
_, cancelPreferred := context.WithCancel(raceCtx)
otherCtx, cancelOther := context.WithCancel(raceCtx)
timer := time.NewTimer(time.Hour)
timer.Stop()
return &connRace{
opener: &FallbackOpener{},
peerKey: "peerKey",
raceCtx: raceCtx,
otherCtx: otherCtx,
cancelPreferred: cancelPreferred,
cancelOther: cancelOther,
results: make(chan raceAttempt, 2),
fallbackTimer: timer,
}
}
func TestHandleResult_PreferredSucceeds(t *testing.T) {
c := newTestConnRace(t)
conn := newFakeConn()
o := c.handleResult(raceAttempt{conn: conn})
require.True(t, o.done)
require.NoError(t, o.err)
require.Same(t, net.Conn(conn), o.conn)
require.False(t, c.otherStarted, "fallback must not start once the preferred attempt wins")
}
func TestHandleResult_ConnAlreadyExistsIsSuccess(t *testing.T) {
c := newTestConnRace(t)
o := c.handleResult(raceAttempt{err: ErrConnAlreadyExists})
require.True(t, o.done)
require.ErrorIs(t, o.err, ErrConnAlreadyExists)
require.False(t, c.otherStarted)
}
func TestHandleResult_PreferredFailsStartsOther(t *testing.T) {
c := newTestConnRace(t)
// The fallback attempt opens against a stalling listener so startOther's
// goroutine blocks on Connect until raceCtx is cancelled by t.Cleanup.
serverAddr, _ := stallingRelayListener(t)
c.opener.foreignStore = NewForeignRelaysStore(c.raceCtx, hmacTokenStore, "alice", 1280, newTransportFallback(), func(string) {}, keepUnusedServerTime)
c.remoteRelayServer = RelayServer{Addr: serverAddr}
c.preferForeign = false
o := c.handleResult(raceAttempt{err: errors.New("boom")})
require.False(t, o.done, "a single failure must not settle the race")
require.True(t, c.otherStarted, "the fallback attempt must start after the preferred one fails")
require.EqualError(t, c.lastErr, "boom")
}
func TestHandleResult_BothFailReturnsLastErr(t *testing.T) {
c := newTestConnRace(t)
c.otherStarted = true
c.settled = 1
c.lastErr = errors.New("first")
o := c.handleResult(raceAttempt{err: errors.New("second")})
require.True(t, o.done)
require.EqualError(t, o.err, "second")
}
func TestOnTimeout_PrefersLastErr(t *testing.T) {
c := newTestConnRace(t)
c.lastErr = errors.New("dial failed")
_, err := c.onTimeout()
require.EqualError(t, err, "dial failed")
}
func TestOnTimeout_FallsBackToCtxErr(t *testing.T) {
c := newTestConnRace(t)
raceCtx, cancel := context.WithCancel(context.Background())
cancel()
c.raceCtx = raceCtx
_, err := c.onTimeout()
require.ErrorIs(t, err, context.Canceled)
}
func TestDrainLoser_ClosesLateWinner(t *testing.T) {
r := &FallbackOpener{}
results := make(chan raceAttempt, 2)
loser := newFakeConn()
results <- raceAttempt{conn: loser}
done := make(chan struct{})
go func() {
defer close(done)
r.drainLoser(results, 1, true)
}()
select {
case <-loser.closed:
case <-time.After(2 * time.Second):
t.Fatal("losing connection was not closed")
}
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("drainLoser did not return")
}
}
func TestDrainLoser_NoOtherAttempt(t *testing.T) {
r := &FallbackOpener{}
results := make(chan raceAttempt)
done := make(chan struct{})
go func() {
defer close(done)
r.drainLoser(results, 1, false)
}()
select {
case <-done:
case <-time.After(2 * time.Second):
t.Fatal("drainLoser blocked with no outstanding attempt")
}
}
func startTestRelayServer(t *testing.T, addr string) string {
t.Helper()
srv, err := server.NewServer(newManagerTestServerConfig(addr))
require.NoError(t, err)
errChan := make(chan error, 1)
go func() {
if err := srv.Listen(server.ListenerConfig{Address: addr}); err != nil {
errChan <- err
}
}()
t.Cleanup(func() { _ = srv.Shutdown(context.Background()) })
require.NoError(t, waitForServerToStart(errChan))
return addr
}

View File

@@ -1,155 +0,0 @@
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

@@ -1,184 +0,0 @@
package client
import (
"context"
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
)
func newTestForeignStore(t *testing.T, ctx context.Context) *ForeignRelaysStore {
t.Helper()
return NewForeignRelaysStore(ctx, hmacTokenStore, "alice", 1280, newTransportFallback(), func(string) {}, keepUnusedServerTime)
}
func TestForeignStore_AcquireDedupsConcurrentOpens(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
addr := startTestRelayServer(t, "127.0.0.1:52601")
server := RelayServer{Addr: "rel://" + addr}
store := newTestForeignStore(t, ctx)
const n = 8
var wg sync.WaitGroup
results := make([]*foreignRelay, n)
for i := 0; i < n; i++ {
wg.Add(1)
go func(i int) {
defer wg.Done()
fr, err := store.acquire(server)
require.NoError(t, err)
results[i] = fr
}(i)
}
wg.Wait()
first := results[0]
require.NotNil(t, first)
for _, fr := range results {
require.Same(t, first, fr, "all acquires must share the same foreign relay")
}
store.mu.RLock()
require.Len(t, store.clients, 1, "only one client entry must be stored")
require.Equal(t, n, first.inUse, "every acquire must be counted")
store.mu.RUnlock()
}
func TestForeignStore_AcquireReleaseRefcount(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
addr := startTestRelayServer(t, "127.0.0.1:52602")
server := RelayServer{Addr: "rel://" + addr}
store := newTestForeignStore(t, ctx)
fr, err := store.acquire(server)
require.NoError(t, err)
_, err = store.acquire(server)
require.NoError(t, err)
store.mu.RLock()
require.Equal(t, 2, fr.inUse)
store.mu.RUnlock()
store.release(fr)
store.mu.RLock()
require.Equal(t, 1, fr.inUse)
require.Len(t, store.clients, 1, "release must not evict the client")
store.mu.RUnlock()
}
func TestForeignStore_AcquireConnectFailure(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
t.Cleanup(cancel)
store := newTestForeignStore(t, ctx)
// Nothing is listening on this port, so Connect fails.
_, err := store.acquire(RelayServer{Addr: "rel://127.0.0.1:1"})
require.Error(t, err)
store.mu.RLock()
require.Empty(t, store.clients, "a failed connect must not leave a client behind")
store.mu.RUnlock()
}
func TestForeignStore_Evict(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
store := newTestForeignStore(t, ctx)
store.clients["rel://a"] = &foreignRelay{}
store.clients["rel://b"] = &foreignRelay{}
store.evict("rel://a")
store.evict("rel://missing")
require.NotContains(t, store.clients, "rel://a")
require.Contains(t, store.clients, "rel://b")
}
func TestForeignStore_CleanupUnused_KeepsRecent(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
addr := startTestRelayServer(t, "127.0.0.1:52603")
store := newTestForeignStore(t, ctx)
fr, err := store.acquire(RelayServer{Addr: "rel://" + addr})
require.NoError(t, err)
store.release(fr)
store.cleanupUnused()
store.mu.RLock()
require.Len(t, store.clients, 1, "a freshly created client must be kept")
store.mu.RUnlock()
}
func TestForeignStore_CleanupUnused_KeepsInUse(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
addr := startTestRelayServer(t, "127.0.0.1:52604")
store := newTestForeignStore(t, ctx)
fr, err := store.acquire(RelayServer{Addr: "rel://" + addr})
require.NoError(t, err)
store.mu.Lock()
fr.created = time.Now().Add(-2 * keepUnusedServerTime)
store.mu.Unlock()
store.cleanupUnused()
store.mu.RLock()
require.Len(t, store.clients, 1, "an in-use client must be kept even when aged")
store.mu.RUnlock()
}
func TestForeignStore_CleanupUnused_EvictsAgedIdle(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
addr := startTestRelayServer(t, "127.0.0.1:52605")
store := newTestForeignStore(t, ctx)
fr, err := store.acquire(RelayServer{Addr: "rel://" + addr})
require.NoError(t, err)
store.release(fr)
store.mu.Lock()
fr.created = time.Now().Add(-2 * keepUnusedServerTime)
store.mu.Unlock()
require.False(t, fr.client.HasConns(), "no peer connections were opened")
store.cleanupUnused()
store.mu.RLock()
require.Empty(t, store.clients, "an aged idle client must be evicted")
store.mu.RUnlock()
}
func TestForeignStore_States(t *testing.T) {
ctx, cancel := context.WithCancel(context.Background())
t.Cleanup(cancel)
addr := startTestRelayServer(t, "127.0.0.1:52606")
store := newTestForeignStore(t, ctx)
fr, err := store.acquire(RelayServer{Addr: "rel://" + addr})
require.NoError(t, err)
store.release(fr)
states := store.states()
require.Len(t, states, 1)
require.NotEmpty(t, states[0].URL)
}

View File

@@ -22,6 +22,27 @@ 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.
@@ -38,11 +59,6 @@ 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 {
@@ -67,7 +83,8 @@ type Manager struct {
relayClientMu sync.RWMutex
reconnectGuard *Guard
foreign *ForeignRelaysStore
relayClients map[string]*RelayTrack
relayClientsMutex sync.RWMutex
onDisconnectedListeners map[string]*list.List
onReconnectedListenerFn func()
@@ -103,6 +120,7 @@ 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,
@@ -110,7 +128,6 @@ 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
@@ -142,26 +159,40 @@ func (m *Manager) Serve() error {
return err
}
func (m *Manager) OpenConn(ctx context.Context, remoteRelayServer RelayServer, peerKey string, preferForeign bool) (net.Conn, error) {
// 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) {
m.relayClientMu.RLock()
relayClient := m.relayClient
m.relayClientMu.RUnlock()
defer m.relayClientMu.RUnlock()
if relayClient == nil {
if m.relayClient == nil {
return nil, ErrRelayClientNotConnected
}
foreign, err := m.isForeignServer(relayClient, remoteRelayServer.Addr)
foreign, err := m.isForeignServer(serverAddress)
if err != nil {
return nil, err
}
var (
netConn net.Conn
)
if !foreign {
return relayClient.OpenConn(ctx, peerKey)
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
}
opener := NewFallbackOpener(relayClient, m.foreign)
return opener.Run(ctx, peerKey, remoteRelayServer, preferForeign)
return netConn, err
}
// Ready returns true if the home Relay client is connected to the relay server.
@@ -192,7 +223,7 @@ func (m *Manager) AddCloseListener(serverAddress string, onClosedListener OnServ
return ErrRelayClientNotConnected
}
foreign, err := m.isForeignServer(m.relayClient, serverAddress)
foreign, err := m.isForeignServer(serverAddress)
if err != nil {
return err
}
@@ -256,7 +287,26 @@ func (m *Manager) RelayStates() []RelayConnState {
states = append(states, st)
}
states = append(states, m.foreign.states()...)
// 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))
}
}
return states
}
@@ -277,6 +327,76 @@ 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()
@@ -302,12 +422,21 @@ func (m *Manager) onServerDisconnected(serverAddress string) {
m.relayClientMu.Unlock()
if !isHome {
m.foreign.evict(serverAddress)
m.evictForeignRelay(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 {
@@ -330,8 +459,8 @@ func (m *Manager) storeClient(client *Client) {
m.relayClient.SetOnDisconnectListener(m.onServerDisconnected)
}
func (m *Manager) isForeignServer(relayClient *Client, address string) (bool, error) {
rAddr, err := relayClient.ServerInstanceURL()
func (m *Manager) isForeignServer(address string) (bool, error) {
rAddr, err := m.relayClient.ServerInstanceURL()
if err != nil {
return false, fmt.Errorf("relay client not connected")
}
@@ -346,11 +475,50 @@ func (m *Manager) startCleanupLoop() {
case <-m.ctx.Done():
return
case <-ticker.C:
m.foreign.cleanupUnused()
m.cleanUpUnusedRelays()
}
}
}
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,14 +2,17 @@ package client
import (
"context"
"net/netip"
"testing"
"time"
"github.com/stretchr/testify/require"
)
// TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial drives a real, hanging foreign
// relay dial and asserts the foreign store cleanup does not stall behind it.
// relay dial and asserts cleanUpUnusedRelays does not stall behind it.
func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
serverAddr, accepted := stallingRelayListener(t)
serverAddr := stallingRelayListener(t)
mCtx, mCancel := context.WithCancel(context.Background())
t.Cleanup(mCancel)
@@ -19,32 +22,39 @@ func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
dialDone := make(chan struct{})
go func() {
defer close(dialDone)
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
}()
select {
case <-accepted:
case <-time.After(5 * time.Second):
t.Fatal("relay dial did not reach the listener")
}
// 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")
cleanupDone := make(chan struct{})
go func() {
defer close(cleanupDone)
m.foreign.cleanupUnused()
m.cleanUpUnusedRelays()
}()
select {
case <-cleanupDone:
case <-time.After(2 * time.Second):
t.Fatal("cleanupUnused blocked on an in-progress relay dial")
t.Fatal("cleanUpUnusedRelays blocked on an in-progress relay dial while holding the relay map lock")
}
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("foreign OpenConn did not return after context cancellation")
t.Fatal("openConnVia did not return after context cancellation")
}
}

View File

@@ -3,6 +3,7 @@ package client
import (
"context"
"net"
"net/netip"
"sync"
"testing"
"time"
@@ -12,16 +13,13 @@ 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. 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{}) {
// cancelled. It returns the "rel://host:port" URL to dial.
func stallingRelayListener(t *testing.T) string {
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() {
@@ -33,10 +31,6 @@ func stallingRelayListener(t *testing.T) (string, <-chan struct{}) {
mu.Lock()
conns = append(conns, c)
mu.Unlock()
select {
case accepted <- struct{}{}:
default:
}
}
}()
t.Cleanup(func() {
@@ -48,14 +42,14 @@ func stallingRelayListener(t *testing.T) (string, <-chan struct{}) {
mu.Unlock()
})
return "rel://" + ln.Addr().String(), accepted
return "rel://" + ln.Addr().String()
}
// TestRelayStates_DoesNotBlockOnRealHangingDial is a regression test for
// RelayStates() called by a "status -d command" hanging behind an in-progress
// foreign relay dial.
// relay dial.
func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
serverAddr, accepted := stallingRelayListener(t)
serverAddr := stallingRelayListener(t)
mCtx, mCancel := context.WithCancel(context.Background())
t.Cleanup(mCancel)
@@ -65,14 +59,15 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
dialDone := make(chan struct{})
go func() {
defer close(dialDone)
_, _ = m.foreign.OpenConn(mCtx, "peerKey", RelayServer{Addr: serverAddr})
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
}()
select {
case <-accepted:
case <-time.After(5 * time.Second):
t.Fatal("relay dial did not reach the listener")
}
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")
done := make(chan []RelayConnState, 1)
go func() {
@@ -91,6 +86,6 @@ func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
select {
case <-dialDone:
case <-time.After(5 * time.Second):
t.Fatal("foreign OpenConn did not return after context cancellation")
t.Fatal("openConnVia did not return after context cancellation")
}
}

View File

@@ -3,6 +3,7 @@ package client
import (
"context"
"io"
"net/netip"
"testing"
"time"
@@ -84,7 +85,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, RelayServer{Addr: brokenFQDN}, "bob", true)
_, err := mgrAlice.OpenConn(dialCtx, brokenFQDN, "bob", netip.Addr{})
if err == nil {
t.Fatalf("expected OpenConn to fail without server IP, got success")
}
@@ -94,7 +95,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, RelayServer{Addr: bobRealAddr}, "alice", false)
conn, err := mgrBob.OpenConn(ctx, bobRealAddr, "alice", netip.Addr{})
if err != nil {
bobSideCh <- err
return
@@ -112,7 +113,7 @@ func TestManager_ForeignRelayServerIP(t *testing.T) {
bobSideCh <- nil
}()
aliceConn, err := mgrAlice.OpenConn(ctx, RelayServer{Addr: brokenFQDN, IP: bobAdvertisedIP}, "bob", true)
aliceConn, err := mgrAlice.OpenConn(ctx, brokenFQDN, "bob", bobAdvertisedIP)
if err != nil {
t.Fatalf("alice OpenConn with server IP: %s", err)
}

View File

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

View File

@@ -0,0 +1,21 @@
// Deprecated: This package is deprecated and will be removed in a future release.
package address
import (
"bytes"
"encoding/gob"
"fmt"
)
type Address struct {
URL string
}
func (addr *Address) Marshal() ([]byte, error) {
var buf bytes.Buffer
enc := gob.NewEncoder(&buf)
if err := enc.Encode(addr); err != nil {
return nil, fmt.Errorf("encode Address: %w", err)
}
return buf.Bytes(), nil
}

View File

@@ -0,0 +1,43 @@
// Deprecated: This package is deprecated and will be removed in a future release.
package auth
import (
"bytes"
"encoding/gob"
"fmt"
)
type Algorithm int
const (
AlgoUnknown Algorithm = iota
AlgoHMACSHA256
AlgoHMACSHA512
)
func (a Algorithm) String() string {
switch a {
case AlgoHMACSHA256:
return "HMAC-SHA256"
case AlgoHMACSHA512:
return "HMAC-SHA512"
default:
return "Unknown"
}
}
type Msg struct {
AuthAlgorithm Algorithm
AdditionalData []byte
}
func UnmarshalMsg(data []byte) (*Msg, error) {
var msg *Msg
buf := bytes.NewBuffer(data)
dec := gob.NewDecoder(buf)
if err := dec.Decode(&msg); err != nil {
return nil, fmt.Errorf("decode Msg: %w", err)
}
return msg, nil
}

View File

@@ -14,10 +14,9 @@ const (
CurrentProtocolVersion = 1
MsgTypeUnknown MsgType = 0
// MsgTypeHello and MsgTypeHelloResponse are the removed legacy handshake
// message types. They are retained only to reserve wire values 1 and 2 so
// the values are never reused; the server rejects both.
MsgTypeHello = 1
// Deprecated: Use MsgTypeAuth instead.
MsgTypeHello = 1
// Deprecated: Use MsgTypeAuthResponse instead.
MsgTypeHelloResponse = 2
MsgTypeTransport = 3
MsgTypeClose = 4
@@ -43,6 +42,10 @@ const (
offsetAuthPeerID = sizeOfProtoHeader + sizeOfMagicByte
headerTotalSizeAuth = sizeOfProtoHeader + headerSizeAuth
// hello message
headerSizeHello = sizeOfMagicByte + peerIDSize
headerSizeHelloResp = 0
// transport
headerSizeTransport = peerIDSize
offsetTransportID = sizeOfProtoHeader
@@ -110,6 +113,7 @@ func DetermineClientMessageType(msg []byte) (MsgType, error) {
msgType := MsgType(msg[1])
switch msgType {
case
MsgTypeHello,
MsgTypeAuth,
MsgTypeTransport,
MsgTypeClose,
@@ -131,6 +135,7 @@ func DetermineServerMessageType(msg []byte) (MsgType, error) {
msgType := MsgType(msg[1])
switch msgType {
case
MsgTypeHelloResponse,
MsgTypeAuthResponse,
MsgTypeTransport,
MsgTypeClose,
@@ -143,6 +148,67 @@ func DetermineServerMessageType(msg []byte) (MsgType, error) {
}
}
// Deprecated: Use MarshalAuthMsg instead.
// MarshalHelloMsg initial hello message
// The Hello message is the first message sent by a client after establishing a connection with the Relay server. This
// message is used to authenticate the client with the server. The authentication is done using an HMAC method.
// The protocol does not limit to use HMAC, it can be any other method. If the authentication failed the server will
// close the network connection without any response.
func MarshalHelloMsg(peerID PeerID, additions []byte) ([]byte, error) {
msg := make([]byte, sizeOfProtoHeader+sizeOfMagicByte, sizeOfProtoHeader+headerSizeHello+len(additions))
msg[0] = byte(CurrentProtocolVersion)
msg[1] = byte(MsgTypeHello)
copy(msg[sizeOfProtoHeader:sizeOfProtoHeader+sizeOfMagicByte], magicHeader)
msg = append(msg, peerID[:]...)
msg = append(msg, additions...)
return msg, nil
}
// Deprecated: Use UnmarshalAuthMsg instead.
// UnmarshalHelloMsg extracts peerID and the additional data from the hello message. The Additional data is used to
// authenticate the client with the server.
func UnmarshalHelloMsg(msg []byte) (*PeerID, []byte, error) {
if len(msg) < sizeOfProtoHeader+headerSizeHello {
return nil, nil, ErrInvalidMessageLength
}
if !bytes.Equal(msg[sizeOfProtoHeader:sizeOfProtoHeader+sizeOfMagicByte], magicHeader) {
return nil, nil, errors.New("invalid magic header")
}
peerID := PeerID(msg[sizeOfProtoHeader+sizeOfMagicByte : sizeOfProtoHeader+headerSizeHello])
return &peerID, msg[headerSizeHello:], nil
}
// Deprecated: Use MarshalAuthResponse instead.
// MarshalHelloResponse creates a response message to the hello message.
// In case of success connection the server response with a Hello Response message. This message contains the server's
// instance URL. This URL will be used by choose the common Relay server in case if the peers are in different Relay
// servers.
func MarshalHelloResponse(additionalData []byte) ([]byte, error) {
msg := make([]byte, sizeOfProtoHeader, sizeOfProtoHeader+headerSizeHelloResp+len(additionalData))
msg[0] = byte(CurrentProtocolVersion)
msg[1] = byte(MsgTypeHelloResponse)
msg = append(msg, additionalData...)
return msg, nil
}
// Deprecated: Use UnmarshalAuthResponse instead.
// UnmarshalHelloResponse extracts the additional data from the hello response message.
func UnmarshalHelloResponse(msg []byte) ([]byte, error) {
if len(msg) < sizeOfProtoHeader+headerSizeHelloResp {
return nil, ErrInvalidMessageLength
}
return msg, nil
}
// MarshalAuthMsg initial authentication message
// The Auth message is the first message sent by a client after establishing a connection with the Relay server. This
// message is used to authenticate the client with the server. The authentication is done using an HMAC method.

View File

@@ -4,11 +4,28 @@ import (
"testing"
)
func TestDetermineClientMessageTypeRejectsHello(t *testing.T) {
// The reserved legacy Hello message (type 1) must be rejected by the server.
msg := []byte{byte(CurrentProtocolVersion), byte(MsgTypeHello)}
if _, err := DetermineClientMessageType(msg); err == nil {
t.Fatalf("expected hello message type to be rejected")
func TestMarshalHelloMsg(t *testing.T) {
peerID := HashID("abdFAaBcawquEiCMzAabYosuUaGLtSNhKxz+")
msg, err := MarshalHelloMsg(peerID, nil)
if err != nil {
t.Fatalf("error: %v", err)
}
msgType, err := DetermineClientMessageType(msg)
if err != nil {
t.Fatalf("error: %v", err)
}
if msgType != MsgTypeHello {
t.Errorf("expected %d, got %d", MsgTypeHello, msgType)
}
receivedPeerID, _, err := UnmarshalHelloMsg(msg)
if err != nil {
t.Fatalf("error: %v", err)
}
if receivedPeerID.String() != peerID.String() {
t.Errorf("expected %s, got %s", peerID, receivedPeerID)
}
}

View File

@@ -1,132 +0,0 @@
package trustedproxy
import (
"fmt"
"net/netip"
"strings"
)
// List holds a parsed set of trusted upstream proxy prefixes and answers trust
// questions against it. The zero value (and a nil *List) is a valid, empty list
// that never trusts any address, so callers can use it without a nil check.
type List struct {
prefixes []netip.Prefix
}
// Parse parses a comma-separated list of CIDR prefixes or bare IPs into a List.
// Bare IPs are converted to single-host prefixes (/32 or /128). An empty input
// yields an empty List that trusts nothing.
func Parse(raw string) (*List, error) {
if raw == "" {
return &List{}, nil
}
parts := strings.Split(raw, ",")
prefixes := make([]netip.Prefix, 0, len(parts))
for _, part := range parts {
part = strings.TrimSpace(part)
if part == "" {
continue
}
prefix, err := netip.ParsePrefix(part)
if err == nil {
prefixes = append(prefixes, prefix)
continue
}
addr, addrErr := netip.ParseAddr(part)
if addrErr != nil {
return nil, fmt.Errorf("parse trusted proxy %q: not a valid CIDR or IP: %w", part, addrErr)
}
bits := 32
if addr.Is6() {
bits = 128
}
prefixes = append(prefixes, netip.PrefixFrom(addr, bits))
}
return &List{prefixes: prefixes}, nil
}
// FromPrefixes wraps an already-parsed set of prefixes in a List.
func FromPrefixes(prefixes []netip.Prefix) *List {
return &List{prefixes: prefixes}
}
// Empty reports whether the list contains no prefixes.
func (l *List) Empty() bool {
return l == nil || len(l.prefixes) == 0
}
// IsTrusted reports whether the given host:port or bare IP falls within the list.
func (l *List) IsTrusted(remoteAddr string) bool {
if l.Empty() {
return false
}
return l.Contains(ExtractHostIP(remoteAddr))
}
// Contains reports whether the given address falls within any trusted prefix.
func (l *List) Contains(addr netip.Addr) bool {
if l.Empty() || !addr.IsValid() {
return false
}
for _, prefix := range l.prefixes {
if prefix.Contains(addr) {
return true
}
}
return false
}
// ResolveClientIP extracts the real client IP from X-Forwarded-For using the
// list. It walks the XFF chain right-to-left, skipping IPs that match trusted
// prefixes; the first untrusted IP is the real client. If the list is empty or
// remoteAddr is not trusted, it returns the remoteAddr IP directly, ignoring any
// forwarding headers.
func (l *List) ResolveClientIP(remoteAddr, xff string) netip.Addr {
remoteIP := ExtractHostIP(remoteAddr)
if l.Empty() || !l.Contains(remoteIP) {
return remoteIP
}
if xff == "" {
return remoteIP
}
parts := strings.Split(xff, ",")
for i := len(parts) - 1; i >= 0; i-- {
ip := strings.TrimSpace(parts[i])
if ip == "" {
continue
}
addr, err := netip.ParseAddr(ip)
if err != nil {
continue
}
addr = addr.Unmap()
if !l.Contains(addr) {
return addr
}
}
if first := strings.TrimSpace(parts[0]); first != "" {
if addr, err := netip.ParseAddr(first); err == nil {
return addr.Unmap()
}
}
return remoteIP
}
// ExtractHostIP parses the IP from a host:port string and returns it unmapped.
func ExtractHostIP(hostPort string) netip.Addr {
if ap, err := netip.ParseAddrPort(hostPort); err == nil {
return ap.Addr().Unmap()
}
if addr, err := netip.ParseAddr(hostPort); err == nil {
return addr.Unmap()
}
return netip.Addr{}
}

View File

@@ -1,216 +0,0 @@
package trustedproxy
import (
"net/netip"
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestParse(t *testing.T) {
tests := []struct {
name string
raw string
want []netip.Prefix
wantErr bool
}{
{
name: "empty string returns empty list",
raw: "",
want: nil,
},
{
name: "single CIDR",
raw: "10.0.0.0/8",
want: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
},
{
name: "single bare IPv4",
raw: "1.2.3.4",
want: []netip.Prefix{netip.MustParsePrefix("1.2.3.4/32")},
},
{
name: "single bare IPv6",
raw: "::1",
want: []netip.Prefix{netip.MustParsePrefix("::1/128")},
},
{
name: "comma-separated CIDRs",
raw: "10.0.0.0/8, 192.168.1.0/24",
want: []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("192.168.1.0/24"),
},
},
{
name: "mixed CIDRs and bare IPs",
raw: "10.0.0.0/8, 1.2.3.4, fd00::/8",
want: []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("1.2.3.4/32"),
netip.MustParsePrefix("fd00::/8"),
},
},
{
name: "whitespace around entries",
raw: " 10.0.0.0/8 , 192.168.0.0/16 ",
want: []netip.Prefix{
netip.MustParsePrefix("10.0.0.0/8"),
netip.MustParsePrefix("192.168.0.0/16"),
},
},
{
name: "trailing comma produces no extra entry",
raw: "10.0.0.0/8,",
want: []netip.Prefix{netip.MustParsePrefix("10.0.0.0/8")},
},
{
name: "invalid entry",
raw: "not-an-ip",
wantErr: true,
},
{
name: "partially invalid",
raw: "10.0.0.0/8, garbage",
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got, err := Parse(tt.raw)
if tt.wantErr {
require.Error(t, err)
return
}
require.NoError(t, err)
assert.Equal(t, tt.want, got.prefixes)
})
}
}
func TestListIsTrusted(t *testing.T) {
list, err := Parse("10.0.0.0/8, 192.168.1.0/24, fd00::/8")
require.NoError(t, err)
tests := []struct {
name string
addr string
list *List
want bool
}{
{"nil list", "10.0.0.1", nil, false},
{"empty list", "10.0.0.1", &List{}, false},
{"IP within /8 prefix", "10.1.2.3", list, true},
{"IP within /24 prefix", "192.168.1.100", list, true},
{"IP outside all prefixes", "203.0.113.50", list, false},
{"boundary IP just outside prefix", "192.168.2.1", list, false},
{"unparsable IP", "not-an-ip", list, false},
{"IPv6 in trusted range", "fd00::1", list, true},
{"IPv6 outside range", "2001:db8::1", list, false},
{"empty string", "", list, false},
{"host:port within prefix", "10.1.2.3:9999", list, true},
{"host:port outside prefix", "203.0.113.50:9999", list, false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, tt.list.IsTrusted(tt.addr))
})
}
}
func TestListResolveClientIP(t *testing.T) {
trusted, err := Parse("10.0.0.0/8, 172.16.0.0/12")
require.NoError(t, err)
tests := []struct {
name string
remoteAddr string
xff string
list *List
want netip.Addr
}{
{
name: "empty list returns RemoteAddr",
remoteAddr: "203.0.113.50:9999",
xff: "1.2.3.4",
list: &List{},
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "nil list returns RemoteAddr",
remoteAddr: "203.0.113.50:9999",
xff: "1.2.3.4",
list: nil,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "untrusted RemoteAddr ignores XFF",
remoteAddr: "203.0.113.50:9999",
xff: "1.2.3.4, 10.0.0.1",
list: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "trusted RemoteAddr with single client in XFF",
remoteAddr: "10.0.0.1:5000",
xff: "203.0.113.50",
list: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "trusted RemoteAddr walks past trusted entries in XFF",
remoteAddr: "10.0.0.1:5000",
xff: "203.0.113.50, 10.0.0.2, 172.16.0.5",
list: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "trusted RemoteAddr with empty XFF falls back to RemoteAddr",
remoteAddr: "10.0.0.1:5000",
xff: "",
list: trusted,
want: netip.MustParseAddr("10.0.0.1"),
},
{
name: "all XFF IPs trusted returns leftmost",
remoteAddr: "10.0.0.1:5000",
xff: "10.0.0.2, 172.16.0.1, 10.0.0.3",
list: trusted,
want: netip.MustParseAddr("10.0.0.2"),
},
{
name: "XFF with whitespace",
remoteAddr: "10.0.0.1:5000",
xff: " 203.0.113.50 , 10.0.0.2 ",
list: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "XFF with empty segments",
remoteAddr: "10.0.0.1:5000",
xff: "203.0.113.50,,10.0.0.2",
list: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "multi-hop with mixed trust",
remoteAddr: "10.0.0.1:5000",
xff: "8.8.8.8, 203.0.113.50, 172.16.0.1",
list: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
{
name: "RemoteAddr without port",
remoteAddr: "10.0.0.1",
xff: "203.0.113.50",
list: trusted,
want: netip.MustParseAddr("203.0.113.50"),
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
assert.Equal(t, tt.want, tt.list.ResolveClientIP(tt.remoteAddr, tt.xff))
})
}
}