mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-09 23:19:11 +02:00
[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.
This commit is contained in:
@@ -228,7 +228,7 @@ func (e *ConnMgr) RemovePeerConn(peerKey string) {
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
defer conn.Close(false)
|
||||
defer conn.Close()
|
||||
|
||||
if !e.isStartedWithLazyMgr() {
|
||||
return
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -0,0 +1,99 @@
|
||||
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 nil, 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")
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user