From e9b3b6210d13b60978a7cca0ca0a0fee1bbe634b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Fri, 20 Dec 2024 12:10:39 +0100 Subject: [PATCH 01/26] Improve WireGuard handshake success rate The controller peer sends WireGuard handshake requests only --- client/internal/peer/conn.go | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 8bbea6a2b..3902c44fb 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -531,11 +531,18 @@ func (conn *Conn) listenGuardEvent(ctx context.Context) { } func (conn *Conn) configureWGEndpoint(addr *net.UDPAddr) error { + var endpoint *net.UDPAddr + + // Force to only one side send handshake request to avoid the handshake congestion in WireGuard connection. + // Configure up the WireGuard endpoint only on the initiator side. + if isWireGuardInitiator(conn.config) { + endpoint = addr + } return conn.config.WgConfig.WgInterface.UpdatePeer( conn.config.WgConfig.RemoteKey, conn.config.WgConfig.AllowedIps, defaultWgKeepAlive, - addr, + endpoint, conn.config.WgConfig.PreSharedKey, ) } @@ -761,6 +768,11 @@ func isController(config ConnConfig) bool { return config.LocalKey > config.Key } +// isWireGuardInitiator returns true if the local peer is the initiator of the WireGuard connection +func isWireGuardInitiator(config ConnConfig) bool { + return isController(config) +} + func isRosenpassEnabled(remoteRosenpassPubKey []byte) bool { return remoteRosenpassPubKey != nil } From bfa6df13c5b9c687d948976bfbd01dfa80fbd0e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 20 Jan 2025 15:28:14 +0100 Subject: [PATCH 02/26] The non handshake initiator peer start the handshake after timeout --- client/internal/peer/conn.go | 48 +++---- client/internal/peer/endpoint.go | 86 +++++++++++++ client/internal/peer/endpoint_test.go | 178 ++++++++++++++++++++++++++ 3 files changed, 284 insertions(+), 28 deletions(-) create mode 100644 client/internal/peer/endpoint.go create mode 100644 client/internal/peer/endpoint_test.go diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 3902c44fb..3639d62ee 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -15,7 +15,6 @@ import ( log "github.com/sirupsen/logrus" "golang.zx2c4.com/wireguard/wgctrl/wgtypes" - "github.com/netbirdio/netbird/client/iface" "github.com/netbirdio/netbird/client/iface/configurer" "github.com/netbirdio/netbird/client/iface/wgproxy" "github.com/netbirdio/netbird/client/internal/peer/guard" @@ -53,10 +52,17 @@ const ( connPriorityICEP2P ConnPriority = 3 ) +type WgInterface interface { + UpdatePeer(peerKey string, allowedIps string, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error + RemovePeer(publicKey string) error + GetProxy() wgproxy.Proxy + GetStats(peerKey string) (configurer.WGStats, error) +} + type WgConfig struct { WgListenPort int RemoteKey string - WgInterface iface.IWGIface + WgInterface WgInterface AllowedIps string PreSharedKey *wgtypes.Key } @@ -116,6 +122,8 @@ type Conn struct { guard *guard.Guard semaphore *semaphoregroup.SemaphoreGroup + + endpointUpdater *endpointUpdater } // NewConn creates a new not opened Conn to the remote peer. @@ -142,6 +150,11 @@ func NewConn(engineCtx context.Context, config ConnConfig, statusRecorder *Statu statusRelay: NewAtomicConnStatus(), statusICE: NewAtomicConnStatus(), semaphore: semaphore, + endpointUpdater: &endpointUpdater{ + log: connLog, + wgConfig: config.WgConfig, + initiator: isWireGuardInitiator(config), + }, } ctrl := isController(config) @@ -239,7 +252,7 @@ func (conn *Conn) Close() { conn.wgProxyICE = nil } - if err := conn.removeWgPeer(); err != nil { + if err := conn.endpointUpdater.removeWgPeer(); err != nil { conn.log.Errorf("failed to remove wg endpoint: %v", err) } @@ -364,7 +377,7 @@ func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEC wgProxy.Work() } - if err = conn.configureWGEndpoint(ep); err != nil { + if err = conn.endpointUpdater.configureWGEndpoint(ep); err != nil { conn.handleConfigurationFailure(err, wgProxy) return } @@ -396,7 +409,7 @@ func (conn *Conn) onICEStateDisconnected() { conn.log.Infof("ICE disconnected, set Relay to active connection") conn.wgProxyRelay.Work() - if err := conn.configureWGEndpoint(conn.wgProxyRelay.EndpointAddr()); err != nil { + if err := conn.endpointUpdater.configureWGEndpoint(conn.wgProxyRelay.EndpointAddr()); err != nil { conn.log.Errorf("failed to switch to relay conn: %v", err) } conn.workerRelay.EnableWgWatcher(conn.ctx) @@ -459,7 +472,7 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) { } wgProxy.Work() - if err := conn.configureWGEndpoint(wgProxy.EndpointAddr()); err != nil { + if err := conn.endpointUpdater.configureWGEndpoint(wgProxy.EndpointAddr()); err != nil { if err := wgProxy.CloseConn(); err != nil { conn.log.Warnf("Failed to close relay connection: %v", err) } @@ -489,7 +502,7 @@ func (conn *Conn) onRelayDisconnected() { if conn.currentConnPriority == connPriorityRelay { conn.log.Debugf("clean up WireGuard config") - if err := conn.removeWgPeer(); err != nil { + if err := conn.endpointUpdater.removeWgPeer(); err != nil { conn.log.Errorf("failed to remove wg endpoint: %v", err) } } @@ -530,23 +543,6 @@ func (conn *Conn) listenGuardEvent(ctx context.Context) { } } -func (conn *Conn) configureWGEndpoint(addr *net.UDPAddr) error { - var endpoint *net.UDPAddr - - // Force to only one side send handshake request to avoid the handshake congestion in WireGuard connection. - // Configure up the WireGuard endpoint only on the initiator side. - if isWireGuardInitiator(conn.config) { - endpoint = addr - } - return conn.config.WgConfig.WgInterface.UpdatePeer( - conn.config.WgConfig.RemoteKey, - conn.config.WgConfig.AllowedIps, - defaultWgKeepAlive, - endpoint, - conn.config.WgConfig.PreSharedKey, - ) -} - func (conn *Conn) updateRelayStatus(relayServerAddr string, rosenpassPubKey []byte) { peerState := State{ PubKey: conn.config.Key, @@ -726,10 +722,6 @@ func (conn *Conn) iceP2PIsActive() bool { return conn.currentConnPriority == connPriorityICEP2P && conn.statusICE.Get() == StatusConnected } -func (conn *Conn) removeWgPeer() error { - return conn.config.WgConfig.WgInterface.RemovePeer(conn.config.WgConfig.RemoteKey) -} - func (conn *Conn) handleConfigurationFailure(err error, wgProxy wgproxy.Proxy) { conn.log.Warnf("Failed to update wg peer configuration: %v", err) if wgProxy != nil { diff --git a/client/internal/peer/endpoint.go b/client/internal/peer/endpoint.go new file mode 100644 index 000000000..c2392bf59 --- /dev/null +++ b/client/internal/peer/endpoint.go @@ -0,0 +1,86 @@ +package peer + +import ( + "context" + "net" + "sync" + "time" + + "github.com/sirupsen/logrus" +) + +// fallbackDelay could be const but because of testing it is a var +var fallbackDelay = 5 * time.Second + +type endpointUpdater struct { + log *logrus.Entry + wgConfig WgConfig + initiator bool + + cancelFunc func() + configUpdateMutex sync.Mutex +} + +// configureWGEndpoint sets up the WireGuard endpoint configuration. +// The initiator immediately configures the endpoint, while the non-initiator +// waits for a fallback period before configuring to avoid handshake congestion. +func (e *endpointUpdater) configureWGEndpoint(addr *net.UDPAddr) error { + if e.initiator { + return e.updateWireGuardPeer(addr) + } + + // prevent to run new update while cancel the previous update + e.configUpdateMutex.Lock() + if e.cancelFunc != nil { + e.cancelFunc() + } + e.configUpdateMutex.Unlock() + + var ctx context.Context + ctx, e.cancelFunc = context.WithCancel(context.Background()) + go e.scheduleDelayedUpdate(ctx, addr) + + return e.updateWireGuardPeer(nil) +} + +func (e *endpointUpdater) removeWgPeer() error { + e.configUpdateMutex.Lock() + defer e.configUpdateMutex.Unlock() + + if e.cancelFunc != nil { + e.cancelFunc() + } + + return e.wgConfig.WgInterface.RemovePeer(e.wgConfig.RemoteKey) +} + +// scheduleDelayedUpdate waits for the fallback period before updating the endpoint +func (e *endpointUpdater) scheduleDelayedUpdate(ctx context.Context, addr *net.UDPAddr) { + t := time.NewTimer(fallbackDelay) + select { + case <-ctx.Done(): + t.Stop() + return + case <-t.C: + e.configUpdateMutex.Lock() + defer e.configUpdateMutex.Unlock() + + if ctx.Err() != nil { + return + } + + if err := e.updateWireGuardPeer(addr); err != nil { + e.log.Errorf("failed to update WireGuard peer, address: %s, error: %v", addr, err) + } + } +} + +func (e *endpointUpdater) updateWireGuardPeer(endpoint *net.UDPAddr) error { + return e.wgConfig.WgInterface.UpdatePeer( + e.wgConfig.RemoteKey, + e.wgConfig.AllowedIps, + defaultWgKeepAlive, + endpoint, + e.wgConfig.PreSharedKey, + ) +} diff --git a/client/internal/peer/endpoint_test.go b/client/internal/peer/endpoint_test.go new file mode 100644 index 000000000..ec980b7d7 --- /dev/null +++ b/client/internal/peer/endpoint_test.go @@ -0,0 +1,178 @@ +package peer + +import ( + "net" + "testing" + "time" + + log "github.com/sirupsen/logrus" + "github.com/stretchr/testify/mock" + "golang.zx2c4.com/wireguard/wgctrl/wgtypes" + + "github.com/netbirdio/netbird/client/iface/configurer" + "github.com/netbirdio/netbird/client/iface/wgproxy" +) + +type MockWgInterface struct { + mock.Mock + + lastSetAddr *net.UDPAddr +} + +func (m *MockWgInterface) GetStats(peerKey string) (configurer.WGStats, error) { + panic("implement me") +} + +func (m *MockWgInterface) GetProxy() wgproxy.Proxy { + panic("implement me") +} + +func (m *MockWgInterface) UpdatePeer(peerKey string, allowedIps string, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error { + args := m.Called(peerKey, allowedIps, keepAlive, endpoint, preSharedKey) + m.lastSetAddr = endpoint + return args.Error(0) +} + +func (m *MockWgInterface) RemovePeer(publicKey string) error { + args := m.Called(publicKey) + return args.Error(0) +} + +func Test_endpointUpdater_initiator(t *testing.T) { + mockWgInterface := &MockWgInterface{} + e := &endpointUpdater{ + log: log.WithField("peer", "my-peer-key"), + wgConfig: WgConfig{ + WgListenPort: 51820, + RemoteKey: "secret-remote-key", + WgInterface: mockWgInterface, + AllowedIps: "172.16.254.1", + }, + initiator: true, + } + addr := &net.UDPAddr{ + IP: net.ParseIP("127.0.0.1"), + Port: 1234, + } + + mockWgInterface.On( + "UpdatePeer", + e.wgConfig.RemoteKey, + e.wgConfig.AllowedIps, + defaultWgKeepAlive, + addr, + (*wgtypes.Key)(nil), + ).Return(nil) + + if err := e.configureWGEndpoint(addr); err != nil { + t.Fatalf("updateWireGuardPeer() failed: %v", err) + } + + mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, addr, (*wgtypes.Key)(nil)) +} + +func Test_endpointUpdater_nonInitiator(t *testing.T) { + fallbackDelay = 1 * time.Second + mockWgInterface := &MockWgInterface{} + e := &endpointUpdater{ + log: log.WithField("peer", "my-peer-key"), + wgConfig: WgConfig{ + WgListenPort: 51820, + RemoteKey: "secret-remote-key", + WgInterface: mockWgInterface, + AllowedIps: "172.16.254.1", + }, + initiator: false, + } + addr := &net.UDPAddr{ + IP: net.ParseIP("127.0.0.1"), + Port: 1234, + } + + mockWgInterface.On( + "UpdatePeer", + e.wgConfig.RemoteKey, + e.wgConfig.AllowedIps, + defaultWgKeepAlive, + (*net.UDPAddr)(nil), + (*wgtypes.Key)(nil), + ).Return(nil) + + mockWgInterface.On( + "UpdatePeer", + e.wgConfig.RemoteKey, + e.wgConfig.AllowedIps, + defaultWgKeepAlive, + addr, + (*wgtypes.Key)(nil), + ).Return(nil) + + err := e.configureWGEndpoint(addr) + if err != nil { + t.Fatalf("updateWireGuardPeer() failed: %v", err) + } + mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, (*net.UDPAddr)(nil), (*wgtypes.Key)(nil)) + + time.Sleep(fallbackDelay + time.Second) + + mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, addr, (*wgtypes.Key)(nil)) +} + +func Test_endpointUpdater_overRule(t *testing.T) { + fallbackDelay = 1 * time.Second + mockWgInterface := &MockWgInterface{} + e := &endpointUpdater{ + log: log.WithField("peer", "my-peer-key"), + wgConfig: WgConfig{ + WgListenPort: 51820, + RemoteKey: "secret-remote-key", + WgInterface: mockWgInterface, + AllowedIps: "172.16.254.1", + }, + initiator: false, + } + addr1 := &net.UDPAddr{ + IP: net.ParseIP("127.0.0.1"), + Port: 1000, + } + + addr2 := &net.UDPAddr{ + IP: net.ParseIP("127.0.0.1"), + Port: 1001, + } + + mockWgInterface.On( + "UpdatePeer", + e.wgConfig.RemoteKey, + e.wgConfig.AllowedIps, + defaultWgKeepAlive, + (*net.UDPAddr)(nil), + (*wgtypes.Key)(nil), + ).Return(nil) + + mockWgInterface.On( + "UpdatePeer", + e.wgConfig.RemoteKey, + e.wgConfig.AllowedIps, + defaultWgKeepAlive, + addr2, + (*wgtypes.Key)(nil), + ).Return(nil) + + if err := e.configureWGEndpoint(addr1); err != nil { + t.Fatalf("updateWireGuardPeer() failed: %v", err) + } + mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, (*net.UDPAddr)(nil), (*wgtypes.Key)(nil)) + + if err := e.configureWGEndpoint(addr2); err != nil { + t.Fatalf("updateWireGuardPeer() failed: %v", err) + } + + time.Sleep(fallbackDelay + time.Second) + + mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, addr2, (*wgtypes.Key)(nil)) + + if mockWgInterface.lastSetAddr != addr2 { + t.Fatalf("lastSetAddr is not equal to addr2") + } +} From 6a0f6efc18fd8b703ec0d245d97e4418d2fba42a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 20 Jan 2025 16:28:16 +0100 Subject: [PATCH 03/26] Always stop timer --- client/internal/peer/endpoint.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/client/internal/peer/endpoint.go b/client/internal/peer/endpoint.go index c2392bf59..98c71af90 100644 --- a/client/internal/peer/endpoint.go +++ b/client/internal/peer/endpoint.go @@ -57,9 +57,10 @@ func (e *endpointUpdater) removeWgPeer() error { // scheduleDelayedUpdate waits for the fallback period before updating the endpoint func (e *endpointUpdater) scheduleDelayedUpdate(ctx context.Context, addr *net.UDPAddr) { t := time.NewTimer(fallbackDelay) + defer t.Stop() + select { case <-ctx.Done(): - t.Stop() return case <-t.C: e.configUpdateMutex.Lock() From ffe74365a83cd922797296791fe091c6c3adad18 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Fri, 14 Feb 2025 15:54:40 +0100 Subject: [PATCH 04/26] Code cleaning --- client/iface/bind/endpoint.go | 13 ++++++++++++- client/iface/wgproxy/bind/proxy.go | 18 +++++++++++------- 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/client/iface/bind/endpoint.go b/client/iface/bind/endpoint.go index 1926ff88f..bce2460de 100644 --- a/client/iface/bind/endpoint.go +++ b/client/iface/bind/endpoint.go @@ -1,5 +1,16 @@ package bind -import wgConn "golang.zx2c4.com/wireguard/conn" +import ( + "net" + + wgConn "golang.zx2c4.com/wireguard/conn" +) type Endpoint = wgConn.StdNetEndpoint + +func EndpointToUDPAddr(e Endpoint) *net.UDPAddr { + return &net.UDPAddr{ + IP: e.Addr().AsSlice(), + Port: int(e.Port()), + } +} diff --git a/client/iface/wgproxy/bind/proxy.go b/client/iface/wgproxy/bind/proxy.go index 8a2e65382..1d5390a57 100644 --- a/client/iface/wgproxy/bind/proxy.go +++ b/client/iface/wgproxy/bind/proxy.go @@ -16,7 +16,7 @@ import ( type ProxyBind struct { Bind *bind.ICEBind - wgAddr *net.UDPAddr + // wgEndpoint is a fake address that generated by the Bind.SetEndpoint based on the remote NetBird peer address wgEndpoint *bind.Endpoint remoteConn net.Conn ctx context.Context @@ -32,21 +32,25 @@ type ProxyBind struct { // AddTurnConn adds a new connection to the bind. // endpoint is the NetBird address of the remote peer. The SetEndpoint return with the address what will be used in the // WireGuard configuration. +// +// Parameters: +// - ctx: Context is used for proxyToLocal to avoid unnecessary error messages +// - nbAddr: The NetBird UDP address of the remote peer, it required to generate fake address +// - remoteConn: The established TURN connection to the remote peer func (p *ProxyBind) AddTurnConn(ctx context.Context, nbAddr *net.UDPAddr, remoteConn net.Conn) error { - addr, err := p.Bind.SetEndpoint(nbAddr, remoteConn) + fakeAddr, err := p.Bind.SetEndpoint(nbAddr, remoteConn) if err != nil { return err } - p.wgAddr = addr - p.wgEndpoint = addrToEndpoint(addr) + p.wgEndpoint = addrToEndpoint(fakeAddr) p.remoteConn = remoteConn p.ctx, p.cancel = context.WithCancel(ctx) return err - } + func (p *ProxyBind) EndpointAddr() *net.UDPAddr { - return p.wgAddr + return bind.EndpointToUDPAddr(*p.wgEndpoint) } func (p *ProxyBind) Work() { @@ -93,7 +97,7 @@ func (p *ProxyBind) close() error { p.cancel() - p.Bind.RemoveEndpoint(p.wgAddr) + p.Bind.RemoveEndpoint(bind.EndpointToUDPAddr(*p.wgEndpoint)) if rErr := p.remoteConn.Close(); rErr != nil && !errors.Is(rErr, net.ErrClosed) { return rErr From 1f088b7e6974ca65b192c1486e9d05a899f60a82 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Sun, 16 Feb 2025 21:50:35 +0100 Subject: [PATCH 05/26] Extend the proxy interface with RedirectTo function and implement it in Bind proxy --- client/iface/wgproxy/bind/proxy.go | 85 ++++++++++++++++++++-------- client/iface/wgproxy/ebpf/wrapper.go | 4 ++ client/iface/wgproxy/factory_usp.go | 4 +- client/iface/wgproxy/proxy.go | 1 + client/iface/wgproxy/udp/proxy.go | 4 ++ client/internal/peer/conn.go | 11 +++- 6 files changed, 80 insertions(+), 29 deletions(-) diff --git a/client/iface/wgproxy/bind/proxy.go b/client/iface/wgproxy/bind/proxy.go index 1d5390a57..af5606511 100644 --- a/client/iface/wgproxy/bind/proxy.go +++ b/client/iface/wgproxy/bind/proxy.go @@ -14,19 +14,27 @@ import ( ) type ProxyBind struct { - Bind *bind.ICEBind + bind *bind.ICEBind // wgEndpoint is a fake address that generated by the Bind.SetEndpoint based on the remote NetBird peer address - wgEndpoint *bind.Endpoint - remoteConn net.Conn - ctx context.Context - cancel context.CancelFunc - closeMu sync.Mutex - closed bool + wgRelayedEndpoint *bind.Endpoint + wgCurrentUsed *bind.Endpoint + remoteConn net.Conn + ctx context.Context + cancel context.CancelFunc + closeMu sync.Mutex + closed bool - pausedMu sync.Mutex - paused bool - isStarted bool + paused bool + pausedCond *sync.Cond + isStarted bool +} + +func NewProxyBind(bind *bind.ICEBind) *ProxyBind { + return &ProxyBind{ + bind: bind, + pausedCond: sync.NewCond(&sync.Mutex{}), + } } // AddTurnConn adds a new connection to the bind. @@ -38,19 +46,19 @@ type ProxyBind struct { // - nbAddr: The NetBird UDP address of the remote peer, it required to generate fake address // - remoteConn: The established TURN connection to the remote peer func (p *ProxyBind) AddTurnConn(ctx context.Context, nbAddr *net.UDPAddr, remoteConn net.Conn) error { - fakeAddr, err := p.Bind.SetEndpoint(nbAddr, remoteConn) + fakeAddr, err := p.bind.SetEndpoint(nbAddr, remoteConn) if err != nil { return err } - p.wgEndpoint = addrToEndpoint(fakeAddr) + p.wgRelayedEndpoint = addrToEndpoint(fakeAddr) p.remoteConn = remoteConn p.ctx, p.cancel = context.WithCancel(ctx) return err } func (p *ProxyBind) EndpointAddr() *net.UDPAddr { - return bind.EndpointToUDPAddr(*p.wgEndpoint) + return bind.EndpointToUDPAddr(*p.wgRelayedEndpoint) } func (p *ProxyBind) Work() { @@ -58,15 +66,20 @@ func (p *ProxyBind) Work() { return } - p.pausedMu.Lock() + p.pausedCond.L.Lock() p.paused = false - p.pausedMu.Unlock() + + p.wgCurrentUsed = p.wgRelayedEndpoint // Start the proxy only once if !p.isStarted { p.isStarted = true go p.proxyToLocal(p.ctx) } + + p.pausedCond.L.Unlock() + // todo: review to should be inside the lock scope + p.pausedCond.Signal() } func (p *ProxyBind) Pause() { @@ -74,9 +87,19 @@ func (p *ProxyBind) Pause() { return } - p.pausedMu.Lock() + p.pausedCond.L.Lock() p.paused = true - p.pausedMu.Unlock() + p.pausedCond.L.Unlock() +} + +func (p *ProxyBind) RedirectTo(endpoint *net.UDPAddr) { + p.pausedCond.L.Lock() + p.paused = false + + p.wgCurrentUsed = addrToEndpoint(endpoint) + + p.pausedCond.L.Unlock() + p.pausedCond.Signal() } func (p *ProxyBind) CloseConn() error { @@ -97,7 +120,12 @@ func (p *ProxyBind) close() error { p.cancel() - p.Bind.RemoveEndpoint(bind.EndpointToUDPAddr(*p.wgEndpoint)) + p.pausedCond.L.Lock() + p.paused = false + p.pausedCond.L.Unlock() + p.pausedCond.Signal() + + p.bind.RemoveEndpoint(bind.EndpointToUDPAddr(*p.wgCurrentUsed)) if rErr := p.remoteConn.Close(); rErr != nil && !errors.Is(rErr, net.ErrClosed) { return rErr @@ -123,18 +151,25 @@ func (p *ProxyBind) proxyToLocal(ctx context.Context) { return } - p.pausedMu.Lock() - if p.paused { - p.pausedMu.Unlock() - continue + for { + p.pausedCond.L.Lock() + if p.paused { + p.pausedCond.Wait() + if !p.paused { + break + } + p.pausedCond.L.Unlock() + continue + } + break } msg := bind.RecvMessage{ - Endpoint: p.wgEndpoint, + Endpoint: p.wgCurrentUsed, Buffer: buf[:n], } - p.Bind.RecvChan <- msg - p.pausedMu.Unlock() + p.bind.RecvChan <- msg + p.pausedCond.L.Unlock() } } diff --git a/client/iface/wgproxy/ebpf/wrapper.go b/client/iface/wgproxy/ebpf/wrapper.go index 54cab4e1b..412afdee1 100644 --- a/client/iface/wgproxy/ebpf/wrapper.go +++ b/client/iface/wgproxy/ebpf/wrapper.go @@ -69,6 +69,10 @@ func (p *ProxyWrapper) Pause() { p.pausedMu.Unlock() } +func (p *ProxyWrapper) RedirectTo(endpoint *net.UDPAddr) { + // todo implement me +} + // CloseConn close the remoteConn and automatically remove the conn instance from the map func (e *ProxyWrapper) CloseConn() error { if e.cancel == nil { diff --git a/client/iface/wgproxy/factory_usp.go b/client/iface/wgproxy/factory_usp.go index e2d479331..141b4c1f9 100644 --- a/client/iface/wgproxy/factory_usp.go +++ b/client/iface/wgproxy/factory_usp.go @@ -20,9 +20,7 @@ func NewUSPFactory(iceBind *bind.ICEBind) *USPFactory { } func (w *USPFactory) GetProxy() Proxy { - return &proxyBind.ProxyBind{ - Bind: w.bind, - } + return proxyBind.NewProxyBind(w.bind) } func (w *USPFactory) Free() error { diff --git a/client/iface/wgproxy/proxy.go b/client/iface/wgproxy/proxy.go index 243aa2bd2..53a5ca7b9 100644 --- a/client/iface/wgproxy/proxy.go +++ b/client/iface/wgproxy/proxy.go @@ -12,4 +12,5 @@ type Proxy interface { Work() // Work start or resume the proxy Pause() // Pause to forward the packages from remote connection to WireGuard. The opposite way still works. CloseConn() error + RedirectTo(endpoint *net.UDPAddr) } diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index ba0004b8a..93bb293fc 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/client/iface/wgproxy/udp/proxy.go @@ -95,6 +95,10 @@ func (p *WGUDPProxy) Pause() { p.pausedMu.Unlock() } +func (p *WGUDPProxy) RedirectTo(endpoint *net.UDPAddr) { + // todo implement me +} + // CloseConn close the localConn func (p *WGUDPProxy) CloseConn() error { if p.cancel == nil { diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 3639d62ee..927000647 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -370,18 +370,26 @@ func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEC conn.workerRelay.DisableWgWatcher() if conn.wgProxyRelay != nil { + conn.log.Debugf("pause Relayed proxy") conn.wgProxyRelay.Pause() } if wgProxy != nil { + conn.log.Debugf("run ICE proxy") wgProxy.Work() } + conn.log.Infof("configure WireGuard endpoint to: %s", ep.String()) if err = conn.endpointUpdater.configureWGEndpoint(ep); err != nil { conn.handleConfigurationFailure(err, wgProxy) return } wgConfigWorkaround() + + if conn.wgProxyRelay != nil { + conn.wgProxyRelay.RedirectTo(ep) + } + conn.currentConnPriority = priority conn.statusICE.Set(StatusConnected) conn.updateIceState(iceConnInfo) @@ -407,11 +415,12 @@ func (conn *Conn) onICEStateDisconnected() { // switch back to relay connection if conn.isReadyToUpgrade() { conn.log.Infof("ICE disconnected, set Relay to active connection") - conn.wgProxyRelay.Work() if err := conn.endpointUpdater.configureWGEndpoint(conn.wgProxyRelay.EndpointAddr()); err != nil { conn.log.Errorf("failed to switch to relay conn: %v", err) } + + conn.wgProxyRelay.Work() conn.workerRelay.EnableWgWatcher(conn.ctx) conn.currentConnPriority = connPriorityRelay } else { From 06a17f0eee9b3cffc261e65616624eb6124733a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Sun, 16 Feb 2025 22:16:42 +0100 Subject: [PATCH 06/26] Implement redirect to in eBPF proxy --- client/iface/wgproxy/ebpf/proxy.go | 16 ++--- client/iface/wgproxy/ebpf/wrapper.go | 80 ++++++++++++++++-------- client/iface/wgproxy/factory_kernel.go | 4 +- client/iface/wgproxy/proxy_linux_test.go | 6 +- client/iface/wgproxy/proxy_test.go | 4 +- client/internal/peer/conn.go | 1 + 6 files changed, 69 insertions(+), 42 deletions(-) diff --git a/client/iface/wgproxy/ebpf/proxy.go b/client/iface/wgproxy/ebpf/proxy.go index e21fc35d4..6462fbadb 100644 --- a/client/iface/wgproxy/ebpf/proxy.go +++ b/client/iface/wgproxy/ebpf/proxy.go @@ -26,6 +26,10 @@ const ( loopbackAddr = "127.0.0.1" ) +var ( + localHostNetIP = net.ParseIP("127.0.0.1") +) + // WGEBPFProxy definition for proxy with EBPF support type WGEBPFProxy struct { localWGListenPort int @@ -249,19 +253,17 @@ func (p *WGEBPFProxy) prepareSenderRawSocket() (net.PacketConn, error) { return packetConn, nil } -func (p *WGEBPFProxy) sendPkg(data []byte, port int) error { - localhost := net.ParseIP("127.0.0.1") - +func (p *WGEBPFProxy) sendPkg(data []byte, endpointAddr *net.UDPAddr) error { payload := gopacket.Payload(data) ipH := &layers.IPv4{ - DstIP: localhost, - SrcIP: localhost, + DstIP: localHostNetIP, + SrcIP: endpointAddr.IP, Version: 4, TTL: 64, Protocol: layers.IPProtocolUDP, } udpH := &layers.UDP{ - SrcPort: layers.UDPPort(port), + SrcPort: layers.UDPPort(endpointAddr.Port), DstPort: layers.UDPPort(p.localWGListenPort), } @@ -276,7 +278,7 @@ func (p *WGEBPFProxy) sendPkg(data []byte, port int) error { if err != nil { return fmt.Errorf("serialize layers: %w", err) } - if _, err = p.rawConn.WriteTo(layerBuffer.Bytes(), &net.IPAddr{IP: localhost}); err != nil { + if _, err = p.rawConn.WriteTo(layerBuffer.Bytes(), &net.IPAddr{IP: localHostNetIP}); err != nil { return fmt.Errorf("write to raw conn: %w", err) } return nil diff --git a/client/iface/wgproxy/ebpf/wrapper.go b/client/iface/wgproxy/ebpf/wrapper.go index 412afdee1..c30cc03f6 100644 --- a/client/iface/wgproxy/ebpf/wrapper.go +++ b/client/iface/wgproxy/ebpf/wrapper.go @@ -15,32 +15,39 @@ import ( // ProxyWrapper help to keep the remoteConn instance for net.Conn.Close function call type ProxyWrapper struct { - WgeBPFProxy *WGEBPFProxy + wgeBPFProxy *WGEBPFProxy remoteConn net.Conn ctx context.Context cancel context.CancelFunc - wgEndpointAddr *net.UDPAddr + wgRelayedEndpointAddr *net.UDPAddr + wgEndpointCurrentUsedAddr *net.UDPAddr - pausedMu sync.Mutex - paused bool - isStarted bool + paused bool + pausedCond *sync.Cond + isStarted bool } +func NewProxyWrapper(proxy *WGEBPFProxy) *ProxyWrapper { + return &ProxyWrapper{ + wgeBPFProxy: proxy, + pausedCond: sync.NewCond(&sync.Mutex{}), + } +} func (p *ProxyWrapper) AddTurnConn(ctx context.Context, endpoint *net.UDPAddr, remoteConn net.Conn) error { - addr, err := p.WgeBPFProxy.AddTurnConn(remoteConn) + addr, err := p.wgeBPFProxy.AddTurnConn(remoteConn) if err != nil { return fmt.Errorf("add turn conn: %w", err) } p.remoteConn = remoteConn p.ctx, p.cancel = context.WithCancel(ctx) - p.wgEndpointAddr = addr + p.wgRelayedEndpointAddr = addr return err } func (p *ProxyWrapper) EndpointAddr() *net.UDPAddr { - return p.wgEndpointAddr + return p.wgRelayedEndpointAddr } func (p *ProxyWrapper) Work() { @@ -48,14 +55,19 @@ func (p *ProxyWrapper) Work() { return } - p.pausedMu.Lock() + p.pausedCond.L.Lock() p.paused = false - p.pausedMu.Unlock() + + p.wgEndpointCurrentUsedAddr = p.wgRelayedEndpointAddr if !p.isStarted { p.isStarted = true go p.proxyToLocal(p.ctx) } + + p.pausedCond.L.Unlock() + // todo: review to should be inside the lock scope + p.pausedCond.Signal() } func (p *ProxyWrapper) Pause() { @@ -64,31 +76,42 @@ func (p *ProxyWrapper) Pause() { } log.Tracef("pause proxy reading from: %s", p.remoteConn.RemoteAddr()) - p.pausedMu.Lock() + p.pausedCond.L.Lock() p.paused = true - p.pausedMu.Unlock() + p.pausedCond.L.Unlock() } func (p *ProxyWrapper) RedirectTo(endpoint *net.UDPAddr) { - // todo implement me + p.pausedCond.L.Lock() + p.paused = false + + p.wgEndpointCurrentUsedAddr = endpoint + + p.pausedCond.L.Unlock() + p.pausedCond.Signal() } // CloseConn close the remoteConn and automatically remove the conn instance from the map -func (e *ProxyWrapper) CloseConn() error { - if e.cancel == nil { +func (p *ProxyWrapper) CloseConn() error { + if p.cancel == nil { return fmt.Errorf("proxy not started") } - e.cancel() + p.cancel() - if err := e.remoteConn.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + p.pausedCond.L.Lock() + p.paused = false + p.pausedCond.L.Unlock() + p.pausedCond.Signal() + + if err := p.remoteConn.Close(); err != nil && !errors.Is(err, net.ErrClosed) { return fmt.Errorf("failed to close remote conn: %w", err) } return nil } func (p *ProxyWrapper) proxyToLocal(ctx context.Context) { - defer p.WgeBPFProxy.removeTurnConn(uint16(p.wgEndpointAddr.Port)) + defer p.wgeBPFProxy.removeTurnConn(uint16(p.wgRelayedEndpointAddr.Port)) buf := make([]byte, 1500) for { @@ -97,14 +120,21 @@ func (p *ProxyWrapper) proxyToLocal(ctx context.Context) { return } - p.pausedMu.Lock() - if p.paused { - p.pausedMu.Unlock() - continue + for { + p.pausedCond.L.Lock() + if p.paused { + p.pausedCond.Wait() + if !p.paused { + break + } + p.pausedCond.L.Unlock() + continue + } + break } - err = p.WgeBPFProxy.sendPkg(buf[:n], p.wgEndpointAddr.Port) - p.pausedMu.Unlock() + err = p.wgeBPFProxy.sendPkg(buf[:n], p.wgEndpointCurrentUsedAddr) + p.pausedCond.L.Unlock() if err != nil { if ctx.Err() != nil { @@ -122,7 +152,7 @@ func (p *ProxyWrapper) readFromRemote(ctx context.Context, buf []byte) (int, err return 0, ctx.Err() } if !errors.Is(err, io.EOF) { - log.Errorf("failed to read from turn conn (endpoint: :%d): %s", p.wgEndpointAddr.Port, err) + log.Errorf("failed to read from turn conn (endpoint: :%d): %s", p.wgRelayedEndpointAddr.Port, err) } return 0, err } diff --git a/client/iface/wgproxy/factory_kernel.go b/client/iface/wgproxy/factory_kernel.go index 3ad7dc59d..c5e0b290d 100644 --- a/client/iface/wgproxy/factory_kernel.go +++ b/client/iface/wgproxy/factory_kernel.go @@ -36,9 +36,7 @@ func (w *KernelFactory) GetProxy() Proxy { return udpProxy.NewWGUDPProxy(w.wgPort) } - return &ebpf.ProxyWrapper{ - WgeBPFProxy: w.ebpfProxy, - } + return ebpf.NewProxyWrapper(w.ebpfProxy) } func (w *KernelFactory) Free() error { diff --git a/client/iface/wgproxy/proxy_linux_test.go b/client/iface/wgproxy/proxy_linux_test.go index 298c98cc0..4d83b39bd 100644 --- a/client/iface/wgproxy/proxy_linux_test.go +++ b/client/iface/wgproxy/proxy_linux_test.go @@ -32,10 +32,8 @@ func TestProxyCloseByRemoteConnEBPF(t *testing.T) { proxy Proxy }{ { - name: "ebpf proxy", - proxy: &ebpf.ProxyWrapper{ - WgeBPFProxy: ebpfProxy, - }, + name: "ebpf proxy", + proxy: ebpf.NewProxyWrapper(ebpfProxy), }, } diff --git a/client/iface/wgproxy/proxy_test.go b/client/iface/wgproxy/proxy_test.go index 64b617621..2165b8aba 100644 --- a/client/iface/wgproxy/proxy_test.go +++ b/client/iface/wgproxy/proxy_test.go @@ -98,9 +98,7 @@ func TestProxyCloseByRemoteConn(t *testing.T) { t.Errorf("failed to free ebpf proxy: %s", err) } }() - proxyWrapper := &ebpf.ProxyWrapper{ - WgeBPFProxy: ebpfProxy, - } + proxyWrapper := ebpf.NewProxyWrapper(ebpfProxy) tests = append(tests, struct { name string diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 927000647..b46dd33cb 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -387,6 +387,7 @@ func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEC wgConfigWorkaround() if conn.wgProxyRelay != nil { + conn.log.Debugf("redirect packages from relayed conn to WireGuard") conn.wgProxyRelay.RedirectTo(ep) } From 4db73a13d7da23176ebc9dce35a2392b2f071a20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 14:58:01 +0100 Subject: [PATCH 07/26] Implement redirect logic in UDP proxy --- client/iface/wgproxy/bind/proxy.go | 2 +- client/iface/wgproxy/ebpf/wrapper.go | 2 +- client/iface/wgproxy/proxy.go | 7 +- client/iface/wgproxy/udp/proxy.go | 90 ++++++++++++----- client/iface/wgproxy/udp/rawsocket.go | 139 ++++++++++++++++++++++++++ 5 files changed, 213 insertions(+), 27 deletions(-) create mode 100644 client/iface/wgproxy/udp/rawsocket.go diff --git a/client/iface/wgproxy/bind/proxy.go b/client/iface/wgproxy/bind/proxy.go index af5606511..20804175e 100644 --- a/client/iface/wgproxy/bind/proxy.go +++ b/client/iface/wgproxy/bind/proxy.go @@ -92,7 +92,7 @@ func (p *ProxyBind) Pause() { p.pausedCond.L.Unlock() } -func (p *ProxyBind) RedirectTo(endpoint *net.UDPAddr) { +func (p *ProxyBind) RedirectAs(endpoint *net.UDPAddr) { p.pausedCond.L.Lock() p.paused = false diff --git a/client/iface/wgproxy/ebpf/wrapper.go b/client/iface/wgproxy/ebpf/wrapper.go index c30cc03f6..a3ee8ac8f 100644 --- a/client/iface/wgproxy/ebpf/wrapper.go +++ b/client/iface/wgproxy/ebpf/wrapper.go @@ -81,7 +81,7 @@ func (p *ProxyWrapper) Pause() { p.pausedCond.L.Unlock() } -func (p *ProxyWrapper) RedirectTo(endpoint *net.UDPAddr) { +func (p *ProxyWrapper) RedirectAs(endpoint *net.UDPAddr) { p.pausedCond.L.Lock() p.paused = false diff --git a/client/iface/wgproxy/proxy.go b/client/iface/wgproxy/proxy.go index 53a5ca7b9..470144abb 100644 --- a/client/iface/wgproxy/proxy.go +++ b/client/iface/wgproxy/proxy.go @@ -11,6 +11,11 @@ type Proxy interface { EndpointAddr() *net.UDPAddr // EndpointAddr returns the address of the WireGuard peer endpoint Work() // Work start or resume the proxy Pause() // Pause to forward the packages from remote connection to WireGuard. The opposite way still works. + /* + RedirectAs resume the forwarding the packages from relayed connection to WireGuard interface if it was paused + and rewrite the src address to the endpoint address. + With this logic can avoid the package loss from relayed connections. + */ + RedirectAs(endpoint *net.UDPAddr) CloseConn() error - RedirectTo(endpoint *net.UDPAddr) } diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index 93bb293fc..e447f5eb9 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/client/iface/wgproxy/udp/proxy.go @@ -18,16 +18,18 @@ import ( type WGUDPProxy struct { localWGListenPort int - remoteConn net.Conn - localConn net.Conn - ctx context.Context - cancel context.CancelFunc - closeMu sync.Mutex - closed bool + remoteConn net.Conn + localConn net.Conn + srcFakerConn *SrcFaker + sendPkg func(data []byte) (int, error) + ctx context.Context + cancel context.CancelFunc + closeMu sync.Mutex + closed bool - pausedMu sync.Mutex - paused bool - isStarted bool + paused bool + pausedCond *sync.Cond + isStarted bool } // NewWGUDPProxy instantiate a UDP based WireGuard proxy. This is not a thread safe implementation @@ -35,6 +37,7 @@ func NewWGUDPProxy(wgPort int) *WGUDPProxy { log.Debugf("Initializing new user space proxy with port %d", wgPort) p := &WGUDPProxy{ localWGListenPort: wgPort, + pausedCond: sync.NewCond(&sync.Mutex{}), } return p } @@ -54,6 +57,7 @@ func (p *WGUDPProxy) AddTurnConn(ctx context.Context, endpoint *net.UDPAddr, rem p.ctx, p.cancel = context.WithCancel(ctx) p.localConn = localConn + p.sendPkg = p.localConn.Write p.remoteConn = remoteConn return err @@ -73,15 +77,17 @@ func (p *WGUDPProxy) Work() { return } - p.pausedMu.Lock() + p.pausedCond.L.Lock() p.paused = false - p.pausedMu.Unlock() + p.sendPkg = p.localConn.Write if !p.isStarted { p.isStarted = true go p.proxyToRemote(p.ctx) go p.proxyToLocal(p.ctx) } + p.pausedCond.L.Unlock() + p.pausedCond.Signal() } // Pause pauses the proxy from receiving data from the remote peer @@ -90,13 +96,33 @@ func (p *WGUDPProxy) Pause() { return } - p.pausedMu.Lock() + p.pausedCond.L.Lock() p.paused = true - p.pausedMu.Unlock() + p.pausedCond.L.Unlock() } -func (p *WGUDPProxy) RedirectTo(endpoint *net.UDPAddr) { - // todo implement me +// RedirectAs start to use the fake sourced raw socket as package sender +func (p *WGUDPProxy) RedirectAs(endpoint *net.UDPAddr) { + p.pausedCond.L.Lock() + defer func() { + p.pausedCond.L.Unlock() + p.pausedCond.Signal() + }() + + p.paused = false + if p.srcFakerConn != nil { + if err := p.srcFakerConn.Close(); err != nil { + log.Errorf("failed to close src faker conn: %s", err) + } + p.srcFakerConn = nil + } + srcFakerConn, err := NewSrcFaker(p.localWGListenPort, endpoint) + if err != nil { + log.Errorf("failed to create src faker conn: %s", err) + return + } + p.srcFakerConn = srcFakerConn + p.sendPkg = p.srcFakerConn.SendPkg } // CloseConn close the localConn @@ -108,6 +134,8 @@ func (p *WGUDPProxy) CloseConn() error { } func (p *WGUDPProxy) close() error { + var result *multierror.Error + p.closeMu.Lock() defer p.closeMu.Unlock() @@ -115,11 +143,14 @@ func (p *WGUDPProxy) close() error { if p.closed { return nil } - p.closed = true p.cancel() - var result *multierror.Error + p.pausedCond.L.Lock() + p.paused = false + p.pausedCond.L.Unlock() + p.pausedCond.Signal() + if err := p.remoteConn.Close(); err != nil && !errors.Is(err, net.ErrClosed) { result = multierror.Append(result, fmt.Errorf("remote conn: %s", err)) } @@ -127,6 +158,11 @@ func (p *WGUDPProxy) close() error { if err := p.localConn.Close(); err != nil { result = multierror.Append(result, fmt.Errorf("local conn: %s", err)) } + + if err := p.srcFakerConn.Close(); err != nil { + result = multierror.Append(result, fmt.Errorf("src faker raw conn: %s", err)) + } + return cerrors.FormatErrorOrNil(result) } @@ -179,14 +215,20 @@ func (p *WGUDPProxy) proxyToLocal(ctx context.Context) { return } - p.pausedMu.Lock() - if p.paused { - p.pausedMu.Unlock() - continue + for { + p.pausedCond.L.Lock() + if p.paused { + p.pausedCond.Wait() + if !p.paused { + break + } + p.pausedCond.L.Unlock() + continue + } + break } - - _, err = p.localConn.Write(buf[:n]) - p.pausedMu.Unlock() + _, err = p.sendPkg(buf[:n]) + p.pausedCond.L.Unlock() if err != nil { if ctx.Err() != nil { diff --git a/client/iface/wgproxy/udp/rawsocket.go b/client/iface/wgproxy/udp/rawsocket.go new file mode 100644 index 000000000..f7d292d44 --- /dev/null +++ b/client/iface/wgproxy/udp/rawsocket.go @@ -0,0 +1,139 @@ +package udp + +import ( + "fmt" + "net" + "os" + "syscall" + + "github.com/google/gopacket" + "github.com/google/gopacket/layers" + log "github.com/sirupsen/logrus" + + nbnet "github.com/netbirdio/netbird/util/net" +) + +var ( + serializeOpts = gopacket.SerializeOptions{ + ComputeChecksums: true, + FixLengths: true, + } + + localHostNetIPAddr = &net.IPAddr{ + IP: net.ParseIP("127.0.0.1"), + } +) + +type SrcFaker struct { + srcAddr *net.UDPAddr + + rawSocket net.PacketConn + ipH gopacket.SerializableLayer + udpH gopacket.SerializableLayer + layerBuffer gopacket.SerializeBuffer +} + +func NewSrcFaker(dstPort int, srcAddr *net.UDPAddr) (*SrcFaker, error) { + rawSocket, err := prepareSenderRawSocket() + if err != nil { + return nil, err + } + + ipH, udpH, err := prepareHeaders(dstPort, srcAddr) + if err != nil { + return nil, err + } + + f := &SrcFaker{ + srcAddr: srcAddr, + rawSocket: rawSocket, + ipH: ipH, + udpH: udpH, + layerBuffer: gopacket.NewSerializeBuffer(), + } + + return f, nil +} + +func (f *SrcFaker) Close() error { + return f.rawSocket.Close() +} + +func (f *SrcFaker) SendPkg(data []byte) (int, error) { + defer func() { + if err := f.layerBuffer.Clear(); err != nil { + log.Errorf("failed to clear layer buffer: %s", err) + } + }() + + payload := gopacket.Payload(data) + + err := gopacket.SerializeLayers(f.layerBuffer, serializeOpts, f.ipH, f.udpH, payload) + if err != nil { + return 0, fmt.Errorf("serialize layers: %w", err) + } + n, err := f.rawSocket.WriteTo(f.layerBuffer.Bytes(), localHostNetIPAddr) + if err != nil { + return 0, fmt.Errorf("write to raw conn: %w", err) + } + return n, nil +} + +func prepareHeaders(dstPort int, srcAddr *net.UDPAddr) (gopacket.SerializableLayer, gopacket.SerializableLayer, error) { + ipH := &layers.IPv4{ + DstIP: net.ParseIP("127.0.0.1"), + SrcIP: srcAddr.IP, + Version: 4, + TTL: 64, + Protocol: layers.IPProtocolUDP, + } + udpH := &layers.UDP{ + SrcPort: layers.UDPPort(srcAddr.Port), + DstPort: layers.UDPPort(dstPort), // dst is the localhost WireGuard port + } + + err := udpH.SetNetworkLayerForChecksum(ipH) + if err != nil { + return nil, nil, fmt.Errorf("set network layer for checksum: %w", err) + } + + return ipH, udpH, nil +} + +func prepareSenderRawSocket() (net.PacketConn, error) { + // Create a raw socket. + fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_RAW, syscall.IPPROTO_RAW) + if err != nil { + return nil, fmt.Errorf("creating raw socket failed: %w", err) + } + + // Set the IP_HDRINCL option on the socket to tell the kernel that headers are included in the packet. + err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_HDRINCL, 1) + if err != nil { + return nil, fmt.Errorf("setting IP_HDRINCL failed: %w", err) + } + + // Bind the socket to the "lo" interface. + err = syscall.SetsockoptString(fd, syscall.SOL_SOCKET, syscall.SO_BINDTODEVICE, "lo") + if err != nil { + return nil, fmt.Errorf("binding to lo interface failed: %w", err) + } + + // Set the fwmark on the socket. + err = nbnet.SetSocketOpt(fd) + if err != nil { + return nil, fmt.Errorf("setting fwmark failed: %w", err) + } + + // Convert the file descriptor to a PacketConn. + file := os.NewFile(uintptr(fd), fmt.Sprintf("fd %d", fd)) + if file == nil { + return nil, fmt.Errorf("converting fd to file failed") + } + packetConn, err := net.FilePacketConn(file) + if err != nil { + return nil, fmt.Errorf("converting file to packet conn failed: %w", err) + } + + return packetConn, nil +} From d5042f688f9f17bc0d2bf45938bbf186b91bd0ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 15:14:11 +0100 Subject: [PATCH 08/26] Fix interface changes --- client/internal/peer/conn.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index b46dd33cb..d56efdecb 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -388,7 +388,7 @@ func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEC if conn.wgProxyRelay != nil { conn.log.Debugf("redirect packages from relayed conn to WireGuard") - conn.wgProxyRelay.RedirectTo(ep) + conn.wgProxyRelay.RedirectAs(ep) } conn.currentConnPriority = priority From 2d5b5f59c2f5ec40bf1831016bda7e632525f751 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 15:49:04 +0100 Subject: [PATCH 09/26] Remove log line --- client/internal/peer/conn.go | 1 - 1 file changed, 1 deletion(-) diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index d56efdecb..4e2ab0043 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -379,7 +379,6 @@ func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEC wgProxy.Work() } - conn.log.Infof("configure WireGuard endpoint to: %s", ep.String()) if err = conn.endpointUpdater.configureWGEndpoint(ep); err != nil { conn.handleConfigurationFailure(err, wgProxy) return From b17c1d96a52abf164cd905afe2a8ebc016d8cd7f Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Mon, 17 Feb 2025 15:50:15 +0100 Subject: [PATCH 10/26] [client] Improve WireGuard handshake success rate (#3092) The controller peer sends WireGuard handshake requests only From 082452eb5f3a9d704ea2becdb84acac8b548ac1c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 15:50:49 +0100 Subject: [PATCH 11/26] Add info log line --- client/internal/peer/conn.go | 1 + 1 file changed, 1 insertion(+) diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 4e2ab0043..d56efdecb 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -379,6 +379,7 @@ func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEC wgProxy.Work() } + conn.log.Infof("configure WireGuard endpoint to: %s", ep.String()) if err = conn.endpointUpdater.configureWGEndpoint(ep); err != nil { conn.handleConfigurationFailure(err, wgProxy) return From 335866ac60ca409bcfcf5c6e9e2a124b4e7bd59a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 15:56:48 +0100 Subject: [PATCH 12/26] Close unused rawsocket --- client/iface/wgproxy/udp/proxy.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index e447f5eb9..502182393 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/client/iface/wgproxy/udp/proxy.go @@ -81,6 +81,13 @@ func (p *WGUDPProxy) Work() { p.paused = false p.sendPkg = p.localConn.Write + if p.srcFakerConn != nil { + if err := p.srcFakerConn.Close(); err != nil { + log.Errorf("failed to close src faker conn: %s", err) + } + p.srcFakerConn = nil + } + if !p.isStarted { p.isStarted = true go p.proxyToRemote(p.ctx) From aca443bdec9efbcc348d82054a2846b8ccda5b8c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 19:26:40 +0100 Subject: [PATCH 13/26] Build UDP proxy on Linux only --- client/iface/wgproxy/udp/proxy.go | 2 ++ client/iface/wgproxy/udp/rawsocket.go | 2 ++ 2 files changed, 4 insertions(+) diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index 502182393..aacbd692d 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/client/iface/wgproxy/udp/proxy.go @@ -1,3 +1,5 @@ +//go:build linux && !android + package udp import ( diff --git a/client/iface/wgproxy/udp/rawsocket.go b/client/iface/wgproxy/udp/rawsocket.go index f7d292d44..d611fa59b 100644 --- a/client/iface/wgproxy/udp/rawsocket.go +++ b/client/iface/wgproxy/udp/rawsocket.go @@ -1,3 +1,5 @@ +//go:build linux && !android + package udp import ( From 775b4feb7ed2aa2c93c6d5cb5e2a35accccdefb4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 20:17:47 +0100 Subject: [PATCH 14/26] Fix close operation --- client/iface/wgproxy/udp/proxy.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index aacbd692d..1e2274ca7 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/client/iface/wgproxy/udp/proxy.go @@ -168,8 +168,10 @@ func (p *WGUDPProxy) close() error { result = multierror.Append(result, fmt.Errorf("local conn: %s", err)) } - if err := p.srcFakerConn.Close(); err != nil { - result = multierror.Append(result, fmt.Errorf("src faker raw conn: %s", err)) + if p.srcFakerConn != nil { + if err := p.srcFakerConn.Close(); err != nil { + result = multierror.Append(result, fmt.Errorf("src faker raw conn: %s", err)) + } } return cerrors.FormatErrorOrNil(result) From 360c7134f711ee034e799951775f2458f7a953b5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 21:26:37 +0100 Subject: [PATCH 15/26] Add unit test --- client/iface/wgproxy/proxy_linux_test.go | 54 ------ client/iface/wgproxy/proxy_test.go | 208 ++++++++++++++--------- 2 files changed, 132 insertions(+), 130 deletions(-) delete mode 100644 client/iface/wgproxy/proxy_linux_test.go diff --git a/client/iface/wgproxy/proxy_linux_test.go b/client/iface/wgproxy/proxy_linux_test.go deleted file mode 100644 index 4d83b39bd..000000000 --- a/client/iface/wgproxy/proxy_linux_test.go +++ /dev/null @@ -1,54 +0,0 @@ -//go:build linux && !android - -package wgproxy - -import ( - "context" - "os" - "testing" - - "github.com/netbirdio/netbird/client/iface/wgproxy/ebpf" -) - -func TestProxyCloseByRemoteConnEBPF(t *testing.T) { - if os.Getenv("GITHUB_ACTIONS") != "true" { - t.Skip("Skipping test as it requires root privileges") - } - ctx := context.Background() - - ebpfProxy := ebpf.NewWGEBPFProxy(51831) - if err := ebpfProxy.Listen(); err != nil { - t.Fatalf("failed to initialize ebpf proxy: %s", err) - } - - defer func() { - if err := ebpfProxy.Free(); err != nil { - t.Errorf("failed to free ebpf proxy: %s", err) - } - }() - - tests := []struct { - name string - proxy Proxy - }{ - { - name: "ebpf proxy", - proxy: ebpf.NewProxyWrapper(ebpfProxy), - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - relayedConn := newMockConn() - err := tt.proxy.AddTurnConn(ctx, nil, relayedConn) - if err != nil { - t.Errorf("error: %v", err) - } - - _ = relayedConn.Close() - if err := tt.proxy.CloseConn(); err != nil { - t.Errorf("error: %v", err) - } - }) - } -} diff --git a/client/iface/wgproxy/proxy_test.go b/client/iface/wgproxy/proxy_test.go index 2165b8aba..7c4faa6b7 100644 --- a/client/iface/wgproxy/proxy_test.go +++ b/client/iface/wgproxy/proxy_test.go @@ -1,117 +1,173 @@ -//go:build linux +//go:build linux && !android package wgproxy import ( "context" - "io" "net" - "os" - "runtime" "testing" - "time" "github.com/netbirdio/netbird/client/iface/wgproxy/ebpf" - udpProxy "github.com/netbirdio/netbird/client/iface/wgproxy/udp" + "github.com/netbirdio/netbird/client/iface/wgproxy/udp" "github.com/netbirdio/netbird/util" ) -func TestMain(m *testing.M) { - _ = util.InitLog("trace", "console") - code := m.Run() - os.Exit(code) +func init() { + _ = util.InitLog("debug", "console") } +func TestProxyRedirect(t *testing.T) { + ebpfProxy := ebpf.NewWGEBPFProxy(51831) + if err := ebpfProxy.Listen(); err != nil { + t.Fatalf("failed to initialize ebpf proxy: %s", err) + } -type mocConn struct { - closeChan chan struct{} - closed bool -} + defer func() { + if err := ebpfProxy.Free(); err != nil { + t.Errorf("failed to free ebpf proxy: %s", err) + } + }() -func newMockConn() *mocConn { - return &mocConn{ - closeChan: make(chan struct{}), + tests := []struct { + name string + proxy Proxy + wgPort int + }{ + { + name: "ebpf kernel proxy", + proxy: ebpf.NewProxyWrapper(ebpfProxy), + wgPort: 51831, + }, + { + name: "udp kernel proxy", + proxy: udp.NewWGUDPProxy(51832), + wgPort: 51832, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + redirectTraffic(t, tt.proxy, tt.wgPort) + }) } } -func (m *mocConn) Read(b []byte) (n int, err error) { - <-m.closeChan - return 0, io.EOF -} +func redirectTraffic(t *testing.T, proxy Proxy, wgPort int) { + t.Helper() -func (m *mocConn) Write(b []byte) (n int, err error) { - <-m.closeChan - return 0, io.EOF -} - -func (m *mocConn) Close() error { - if m.closed == true { - return nil + msgHelloFromRelay := []byte("hello from relay") + msgRedirected := [][]byte{ + []byte("hello 1. to p2p"), + []byte("hello 2. to p2p"), + []byte("hello 3. to p2p"), } - m.closed = true - close(m.closeChan) - return nil -} + dummyWgListener, err := net.ListenUDP("udp", &net.UDPAddr{ + IP: net.IPv4(127, 0, 0, 1), + Port: wgPort}) + if err != nil { + t.Fatalf("failed to listen on udp port: %s", err) + } -func (m *mocConn) LocalAddr() net.Addr { - panic("implement me") -} + relayedServer, err := net.ListenUDP("udp", + &net.UDPAddr{ + IP: net.IPv4(127, 0, 0, 1), + Port: 1234, + }, + ) -func (m *mocConn) RemoteAddr() net.Addr { - return &net.UDPAddr{ - IP: net.ParseIP("172.16.254.1"), + relayedConn, err := net.Dial("udp", "127.0.0.1:1234") + + defer func() { + _ = dummyWgListener.Close() + _ = relayedConn.Close() + _ = relayedServer.Close() + }() + + if err := proxy.AddTurnConn(context.Background(), nil, relayedConn); err != nil { + t.Errorf("error: %v", err) + } + defer func() { + if err := proxy.CloseConn(); err != nil { + t.Errorf("error: %v", err) + } + }() + + proxy.Work() + + if _, err := relayedServer.WriteTo(msgHelloFromRelay, relayedConn.LocalAddr()); err != nil { + t.Errorf("error relayedServer.Write(msgHelloFromRelay): %v", err) + } + + n, err := dummyWgListener.Read(make([]byte, 1024)) + if err != nil { + t.Errorf("error: %v", err) + } + + if n != len(msgHelloFromRelay) { + t.Errorf("expected %d bytes, got %d", len(msgHelloFromRelay), n) + } + + p2pEndpointAddr := &net.UDPAddr{ + IP: net.IPv4(192, 168, 0, 56), + Port: 1234, + } + proxy.RedirectAs(p2pEndpointAddr) + + for _, msg := range msgRedirected { + if _, err := relayedServer.WriteTo(msg, relayedConn.LocalAddr()); err != nil { + t.Errorf("error: %v", err) + } + } + + for i := 0; i < len(msgRedirected); i++ { + buf := make([]byte, 1024) + n, rAddr, err := dummyWgListener.ReadFrom(buf) + if err != nil { + t.Errorf("error: %v", err) + } + + if rAddr.String() != p2pEndpointAddr.String() { + t.Errorf("expected %s, got %s", p2pEndpointAddr.String(), rAddr.String()) + } + if string(buf[:n]) != string(msgRedirected[i]) { + t.Errorf("expected %s, got %s", string(msgRedirected[i]), string(buf[:n])) + } } } -func (m *mocConn) SetDeadline(t time.Time) error { - panic("implement me") -} - -func (m *mocConn) SetReadDeadline(t time.Time) error { - panic("implement me") -} - -func (m *mocConn) SetWriteDeadline(t time.Time) error { - panic("implement me") -} - -func TestProxyCloseByRemoteConn(t *testing.T) { +func TestProxyCloseByRemoteConnEBPF(t *testing.T) { ctx := context.Background() + ebpfProxy := ebpf.NewWGEBPFProxy(51831) + if err := ebpfProxy.Listen(); err != nil { + t.Fatalf("failed to initialize ebpf proxy: %s", err) + } + + defer func() { + if err := ebpfProxy.Free(); err != nil { + t.Errorf("failed to free ebpf proxy: %s", err) + } + }() + tests := []struct { name string proxy Proxy }{ { - name: "userspace proxy", - proxy: udpProxy.NewWGUDPProxy(51830), + name: "ebpf proxy", + proxy: ebpf.NewProxyWrapper(ebpfProxy), + }, + { + name: "udp proxy", + proxy: udp.NewWGUDPProxy(51832), }, } - if runtime.GOOS == "linux" && os.Getenv("GITHUB_ACTIONS") != "true" { - ebpfProxy := ebpf.NewWGEBPFProxy(51831) - if err := ebpfProxy.Listen(); err != nil { - t.Fatalf("failed to initialize ebpf proxy: %s", err) - } - defer func() { - if err := ebpfProxy.Free(); err != nil { - t.Errorf("failed to free ebpf proxy: %s", err) - } - }() - proxyWrapper := ebpf.NewProxyWrapper(ebpfProxy) - - tests = append(tests, struct { - name string - proxy Proxy - }{ - name: "ebpf proxy", - proxy: proxyWrapper, - }) - } - + relayedConn, _ := net.Dial("udp", "127.0.0.1:1234") + defer func() { + _ = relayedConn.Close() + }() for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - relayedConn := newMockConn() err := tt.proxy.AddTurnConn(ctx, nil, relayedConn) if err != nil { t.Errorf("error: %v", err) From 1963644c9909aab6810e63ced652248214e9f034 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 21:47:34 +0100 Subject: [PATCH 16/26] Add close test for all implementation --- client/iface/wgproxy/bind/proxy.go | 6 ++++- client/iface/wgproxy/proxy_test.go | 36 +++++++++++++++++++++--------- 2 files changed, 31 insertions(+), 11 deletions(-) diff --git a/client/iface/wgproxy/bind/proxy.go b/client/iface/wgproxy/bind/proxy.go index 20804175e..c1cc8dd30 100644 --- a/client/iface/wgproxy/bind/proxy.go +++ b/client/iface/wgproxy/bind/proxy.go @@ -110,6 +110,10 @@ func (p *ProxyBind) CloseConn() error { } func (p *ProxyBind) close() error { + if p.remoteConn == nil { + return nil + } + p.closeMu.Lock() defer p.closeMu.Unlock() @@ -125,7 +129,7 @@ func (p *ProxyBind) close() error { p.pausedCond.L.Unlock() p.pausedCond.Signal() - p.bind.RemoveEndpoint(bind.EndpointToUDPAddr(*p.wgCurrentUsed)) + p.bind.RemoveEndpoint(bind.EndpointToUDPAddr(*p.wgRelayedEndpoint)) if rErr := p.remoteConn.Close(); rErr != nil && !errors.Is(rErr, net.ErrClosed) { return rErr diff --git a/client/iface/wgproxy/proxy_test.go b/client/iface/wgproxy/proxy_test.go index 7c4faa6b7..80ca57564 100644 --- a/client/iface/wgproxy/proxy_test.go +++ b/client/iface/wgproxy/proxy_test.go @@ -7,14 +7,18 @@ import ( "net" "testing" + "github.com/netbirdio/netbird/client/iface/bind" + bindproxy "github.com/netbirdio/netbird/client/iface/wgproxy/bind" "github.com/netbirdio/netbird/client/iface/wgproxy/ebpf" "github.com/netbirdio/netbird/client/iface/wgproxy/udp" + "github.com/netbirdio/netbird/util" ) func init() { _ = util.InitLog("debug", "console") } + func TestProxyRedirect(t *testing.T) { ebpfProxy := ebpf.NewWGEBPFProxy(51831) if err := ebpfProxy.Listen(); err != nil { @@ -28,9 +32,10 @@ func TestProxyRedirect(t *testing.T) { }() tests := []struct { - name string - proxy Proxy - wgPort int + name string + proxy Proxy + wgPort int + endpointAddr *net.UDPAddr }{ { name: "ebpf kernel proxy", @@ -45,12 +50,12 @@ func TestProxyRedirect(t *testing.T) { } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - redirectTraffic(t, tt.proxy, tt.wgPort) + redirectTraffic(t, tt.proxy, tt.wgPort, tt.endpointAddr) }) } } -func redirectTraffic(t *testing.T, proxy Proxy, wgPort int) { +func redirectTraffic(t *testing.T, proxy Proxy, wgPort int, endPointAddr *net.UDPAddr) { t.Helper() msgHelloFromRelay := []byte("hello from relay") @@ -82,7 +87,7 @@ func redirectTraffic(t *testing.T, proxy Proxy, wgPort int) { _ = relayedServer.Close() }() - if err := proxy.AddTurnConn(context.Background(), nil, relayedConn); err != nil { + if err := proxy.AddTurnConn(context.Background(), endPointAddr, relayedConn); err != nil { t.Errorf("error: %v", err) } defer func() { @@ -134,7 +139,7 @@ func redirectTraffic(t *testing.T, proxy Proxy, wgPort int) { } } -func TestProxyCloseByRemoteConnEBPF(t *testing.T) { +func TestProxyCloseByRemoteConn(t *testing.T) { ctx := context.Background() ebpfProxy := ebpf.NewWGEBPFProxy(51831) @@ -148,9 +153,15 @@ func TestProxyCloseByRemoteConnEBPF(t *testing.T) { } }() + iceBind := bind.NewICEBind(nil, nil) + endpointAddress := &net.UDPAddr{ + IP: net.IPv4(10, 0, 0, 1), + Port: 1234, + } tests := []struct { - name string - proxy Proxy + name string + proxy Proxy + endpointAddress *net.UDPAddr }{ { name: "ebpf proxy", @@ -160,6 +171,11 @@ func TestProxyCloseByRemoteConnEBPF(t *testing.T) { name: "udp proxy", proxy: udp.NewWGUDPProxy(51832), }, + { + name: "bind proxy", + proxy: bindproxy.NewProxyBind(iceBind), + endpointAddress: endpointAddress, + }, } relayedConn, _ := net.Dial("udp", "127.0.0.1:1234") @@ -168,7 +184,7 @@ func TestProxyCloseByRemoteConnEBPF(t *testing.T) { }() for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - err := tt.proxy.AddTurnConn(ctx, nil, relayedConn) + err := tt.proxy.AddTurnConn(ctx, endpointAddress, relayedConn) if err != nil { t.Errorf("error: %v", err) } From 3d80a25b4d19be78975bf8df2033444c30c0c17b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 22:03:15 +0100 Subject: [PATCH 17/26] Fix possible blocker if the bind will be closed earlier then proxy --- client/iface/bind/ice_bind.go | 15 ++++++++++++--- client/iface/wgproxy/bind/proxy.go | 12 +++++++++--- 2 files changed, 21 insertions(+), 6 deletions(-) diff --git a/client/iface/bind/ice_bind.go b/client/iface/bind/ice_bind.go index 41f415af7..c203b5bfc 100644 --- a/client/iface/bind/ice_bind.go +++ b/client/iface/bind/ice_bind.go @@ -1,6 +1,7 @@ package bind import ( + "context" "fmt" "net" "net/netip" @@ -38,7 +39,7 @@ func (rc receiverCreator) CreateIPv4ReceiverFn(pc *ipv4.PacketConn, conn *net.UD // use the port because in the Send function the wgConn.Endpoint the port info is not exported. type ICEBind struct { *wgConn.StdNetBind - RecvChan chan RecvMessage + recvChan chan RecvMessage transportNet transport.Net filterFn FilterFn @@ -58,7 +59,7 @@ func NewICEBind(transportNet transport.Net, filterFn FilterFn) *ICEBind { b, _ := wgConn.NewStdNetBind().(*wgConn.StdNetBind) ib := &ICEBind{ StdNetBind: b, - RecvChan: make(chan RecvMessage, 1), + recvChan: make(chan RecvMessage, 1), transportNet: transportNet, filterFn: filterFn, endpoints: make(map[netip.Addr]net.Conn), @@ -155,6 +156,14 @@ func (b *ICEBind) Send(bufs [][]byte, ep wgConn.Endpoint) error { return nil } +func (b *ICEBind) Recv(ctx context.Context, msg RecvMessage) { + select { + case <-ctx.Done(): + return + case b.recvChan <- msg: + } +} + func (s *ICEBind) createIPv4ReceiverFn(pc *ipv4.PacketConn, conn *net.UDPConn, rxOffload bool, msgsPool *sync.Pool) wgConn.ReceiveFunc { s.muUDPMux.Lock() defer s.muUDPMux.Unlock() @@ -264,7 +273,7 @@ func (c *ICEBind) receiveRelayed(buffs [][]byte, sizes []int, eps []wgConn.Endpo select { case <-c.closedChan: return 0, net.ErrClosed - case msg, ok := <-c.RecvChan: + case msg, ok := <-c.recvChan: if !ok { return 0, net.ErrClosed } diff --git a/client/iface/wgproxy/bind/proxy.go b/client/iface/wgproxy/bind/proxy.go index c1cc8dd30..b0dfef06b 100644 --- a/client/iface/wgproxy/bind/proxy.go +++ b/client/iface/wgproxy/bind/proxy.go @@ -13,8 +13,14 @@ import ( "github.com/netbirdio/netbird/client/iface/bind" ) +type IceBind interface { + SetEndpoint(addr *net.UDPAddr, conn net.Conn) (*net.UDPAddr, error) + RemoveEndpoint(addr *net.UDPAddr) + Recv(ctx context.Context, msg bind.RecvMessage) +} + type ProxyBind struct { - bind *bind.ICEBind + bind IceBind // wgEndpoint is a fake address that generated by the Bind.SetEndpoint based on the remote NetBird peer address wgRelayedEndpoint *bind.Endpoint @@ -30,7 +36,7 @@ type ProxyBind struct { isStarted bool } -func NewProxyBind(bind *bind.ICEBind) *ProxyBind { +func NewProxyBind(bind IceBind) *ProxyBind { return &ProxyBind{ bind: bind, pausedCond: sync.NewCond(&sync.Mutex{}), @@ -172,7 +178,7 @@ func (p *ProxyBind) proxyToLocal(ctx context.Context) { Endpoint: p.wgCurrentUsed, Buffer: buf[:n], } - p.bind.RecvChan <- msg + p.bind.Recv(ctx, msg) p.pausedCond.L.Unlock() } } From 1f83ba4563a0549378f8faece7ffca9115a0b651 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 17 Feb 2025 22:06:04 +0100 Subject: [PATCH 18/26] Ignore err in tests --- client/iface/wgproxy/proxy_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/client/iface/wgproxy/proxy_test.go b/client/iface/wgproxy/proxy_test.go index 80ca57564..64a46d84e 100644 --- a/client/iface/wgproxy/proxy_test.go +++ b/client/iface/wgproxy/proxy_test.go @@ -72,14 +72,14 @@ func redirectTraffic(t *testing.T, proxy Proxy, wgPort int, endPointAddr *net.UD t.Fatalf("failed to listen on udp port: %s", err) } - relayedServer, err := net.ListenUDP("udp", + relayedServer, _ := net.ListenUDP("udp", &net.UDPAddr{ IP: net.IPv4(127, 0, 0, 1), Port: 1234, }, ) - relayedConn, err := net.Dial("udp", "127.0.0.1:1234") + relayedConn, _ := net.Dial("udp", "127.0.0.1:1234") defer func() { _ = dummyWgListener.Close() From 1eacff250e351af346e79c49efaa6a30956f3de0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Tue, 18 Feb 2025 13:47:42 +0100 Subject: [PATCH 19/26] Remove WireGuard kernel code from FreeBSD --- client/iface/iface_new_freebsd.go | 40 +++++++++ .../{iface_new_unix.go => iface_new_linux.go} | 2 +- .../iface/wgproxy/factory_kernel_freebsd.go | 29 ------ client/iface/wgproxy/proxy_linux_test.go | 80 +++++++++++++++++ client/iface/wgproxy/proxy_seed_test.go | 34 +++++++ client/iface/wgproxy/proxy_test.go | 88 +++++-------------- 6 files changed, 175 insertions(+), 98 deletions(-) create mode 100644 client/iface/iface_new_freebsd.go rename client/iface/{iface_new_unix.go => iface_new_linux.go} (97%) delete mode 100644 client/iface/wgproxy/factory_kernel_freebsd.go create mode 100644 client/iface/wgproxy/proxy_linux_test.go create mode 100644 client/iface/wgproxy/proxy_seed_test.go diff --git a/client/iface/iface_new_freebsd.go b/client/iface/iface_new_freebsd.go new file mode 100644 index 000000000..e8e26aee5 --- /dev/null +++ b/client/iface/iface_new_freebsd.go @@ -0,0 +1,40 @@ +//go:build freebsd + +package iface + +import ( + "fmt" + + "github.com/netbirdio/netbird/client/iface/bind" + "github.com/netbirdio/netbird/client/iface/device" + "github.com/netbirdio/netbird/client/iface/netstack" + "github.com/netbirdio/netbird/client/iface/wgproxy" +) + +// NewWGIFace Creates a new WireGuard interface instance +func NewWGIFace(opts WGIFaceOpts) (*WGIface, error) { + wgAddress, err := device.ParseWGAddress(opts.Address) + if err != nil { + return nil, err + } + + wgIFace := &WGIface{} + + if netstack.IsEnabled() { + iceBind := bind.NewICEBind(opts.TransportNet, opts.FilterFn) + wgIFace.tun = device.NewNetstackDevice(opts.IFaceName, wgAddress, opts.WGPort, opts.WGPrivKey, opts.MTU, iceBind, netstack.ListenAddr()) + wgIFace.userspaceBind = true + wgIFace.wgProxyFactory = wgproxy.NewUSPFactory(iceBind) + return wgIFace, nil + } + + if device.ModuleTunIsLoaded() { + iceBind := bind.NewICEBind(opts.TransportNet, opts.FilterFn) + wgIFace.tun = device.NewUSPDevice(opts.IFaceName, wgAddress, opts.WGPort, opts.WGPrivKey, opts.MTU, iceBind) + wgIFace.userspaceBind = true + wgIFace.wgProxyFactory = wgproxy.NewUSPFactory(iceBind) + return wgIFace, nil + } + + return nil, fmt.Errorf("couldn't check or load tun module") +} diff --git a/client/iface/iface_new_unix.go b/client/iface/iface_new_linux.go similarity index 97% rename from client/iface/iface_new_unix.go rename to client/iface/iface_new_linux.go index f10b17c9a..89b598027 100644 --- a/client/iface/iface_new_unix.go +++ b/client/iface/iface_new_linux.go @@ -1,4 +1,4 @@ -//go:build (linux && !android) || freebsd +//go:build linux && !android package iface diff --git a/client/iface/wgproxy/factory_kernel_freebsd.go b/client/iface/wgproxy/factory_kernel_freebsd.go deleted file mode 100644 index 736944229..000000000 --- a/client/iface/wgproxy/factory_kernel_freebsd.go +++ /dev/null @@ -1,29 +0,0 @@ -package wgproxy - -import ( - log "github.com/sirupsen/logrus" - - udpProxy "github.com/netbirdio/netbird/client/iface/wgproxy/udp" -) - -// KernelFactory todo: check eBPF support on FreeBSD -type KernelFactory struct { - wgPort int -} - -func NewKernelFactory(wgPort int) *KernelFactory { - log.Infof("WireGuard Proxy Factory will produce UDP proxy") - f := &KernelFactory{ - wgPort: wgPort, - } - - return f -} - -func (w *KernelFactory) GetProxy() Proxy { - return udpProxy.NewWGUDPProxy(w.wgPort) -} - -func (w *KernelFactory) Free() error { - return nil -} diff --git a/client/iface/wgproxy/proxy_linux_test.go b/client/iface/wgproxy/proxy_linux_test.go new file mode 100644 index 000000000..947d29c40 --- /dev/null +++ b/client/iface/wgproxy/proxy_linux_test.go @@ -0,0 +1,80 @@ +//go:build linux && !android + +package wgproxy + +import ( + "fmt" + "net" + + "github.com/netbirdio/netbird/client/iface/bind" + bindproxy "github.com/netbirdio/netbird/client/iface/wgproxy/bind" + "github.com/netbirdio/netbird/client/iface/wgproxy/ebpf" + "github.com/netbirdio/netbird/client/iface/wgproxy/udp" +) + +func seedProxies() ([]proxyInstance, error) { + pl := make([]proxyInstance, 0) + + ebpfProxy := ebpf.NewWGEBPFProxy(51831) + if err := ebpfProxy.Listen(); err != nil { + return nil, fmt.Errorf("failed to initialize ebpf proxy: %s", err) + } + + pEbpf := proxyInstance{ + name: "ebpf kernel proxy", + proxy: ebpf.NewProxyWrapper(ebpfProxy), + wgPort: 51831, + closeFn: ebpfProxy.Free, + } + pl = append(pl, pEbpf) + + pUDP := proxyInstance{ + name: "udp kernel proxy", + proxy: udp.NewWGUDPProxy(51832), + wgPort: 51832, + closeFn: func() error { return nil }, + } + pl = append(pl, pUDP) + return pl, nil +} + +func seedProxyForProxyCloseByRemoteConn() ([]proxyInstance, error) { + pl := make([]proxyInstance, 0) + + ebpfProxy := ebpf.NewWGEBPFProxy(51831) + if err := ebpfProxy.Listen(); err != nil { + return nil, fmt.Errorf("failed to initialize ebpf proxy: %s", err) + } + + pEbpf := proxyInstance{ + name: "ebpf kernel proxy", + proxy: ebpf.NewProxyWrapper(ebpfProxy), + wgPort: 51831, + closeFn: ebpfProxy.Free, + } + pl = append(pl, pEbpf) + + pUDP := proxyInstance{ + name: "udp kernel proxy", + proxy: udp.NewWGUDPProxy(51832), + wgPort: 51832, + closeFn: func() error { return nil }, + } + pl = append(pl, pUDP) + + iceBind := bind.NewICEBind(nil, nil) + endpointAddress := &net.UDPAddr{ + IP: net.IPv4(10, 0, 0, 1), + Port: 1234, + } + + pBind := proxyInstance{ + name: "bind proxy", + proxy: bindproxy.NewProxyBind(iceBind), + endpointAddr: endpointAddress, + closeFn: func() error { return nil }, + } + pl = append(pl, pBind) + + return pl, nil +} diff --git a/client/iface/wgproxy/proxy_seed_test.go b/client/iface/wgproxy/proxy_seed_test.go new file mode 100644 index 000000000..c52672a9f --- /dev/null +++ b/client/iface/wgproxy/proxy_seed_test.go @@ -0,0 +1,34 @@ +//go:build !linux + +package wgproxy + +import ( + "net" + + "github.com/netbirdio/netbird/client/iface/bind" + bindproxy "github.com/netbirdio/netbird/client/iface/wgproxy/bind" +) + +func seedProxies() ([]proxyInstance, error) { + // todo extend with Bind proxy + pl := make([]proxyInstance, 0) + return pl, nil +} + +func seedProxyForProxyCloseByRemoteConn() ([]proxyInstance, error) { + pl := make([]proxyInstance, 0) + iceBind := bind.NewICEBind(nil, nil) + endpointAddress := &net.UDPAddr{ + IP: net.IPv4(10, 0, 0, 1), + Port: 1234, + } + + pBind := proxyInstance{ + name: "bind proxy", + proxy: bindproxy.NewProxyBind(iceBind), + endpointAddr: endpointAddress, + closeFn: func() error { return nil }, + } + pl = append(pl, pBind) + return pl, nil +} diff --git a/client/iface/wgproxy/proxy_test.go b/client/iface/wgproxy/proxy_test.go index 64a46d84e..0bb0638a0 100644 --- a/client/iface/wgproxy/proxy_test.go +++ b/client/iface/wgproxy/proxy_test.go @@ -1,5 +1,3 @@ -//go:build linux && !android - package wgproxy import ( @@ -7,11 +5,6 @@ import ( "net" "testing" - "github.com/netbirdio/netbird/client/iface/bind" - bindproxy "github.com/netbirdio/netbird/client/iface/wgproxy/bind" - "github.com/netbirdio/netbird/client/iface/wgproxy/ebpf" - "github.com/netbirdio/netbird/client/iface/wgproxy/udp" - "github.com/netbirdio/netbird/util" ) @@ -19,38 +12,27 @@ func init() { _ = util.InitLog("debug", "console") } +type proxyInstance struct { + name string + proxy Proxy + wgPort int + endpointAddr *net.UDPAddr + closeFn func() error +} + +// TestProxyRedirect todo extend the proxies with Bind proxy func TestProxyRedirect(t *testing.T) { - ebpfProxy := ebpf.NewWGEBPFProxy(51831) - if err := ebpfProxy.Listen(); err != nil { - t.Fatalf("failed to initialize ebpf proxy: %s", err) + tests, err := seedProxies() + if err != nil { + t.Fatalf("error: %v", err) } - defer func() { - if err := ebpfProxy.Free(); err != nil { - t.Errorf("failed to free ebpf proxy: %s", err) - } - }() - - tests := []struct { - name string - proxy Proxy - wgPort int - endpointAddr *net.UDPAddr - }{ - { - name: "ebpf kernel proxy", - proxy: ebpf.NewProxyWrapper(ebpfProxy), - wgPort: 51831, - }, - { - name: "udp kernel proxy", - proxy: udp.NewWGUDPProxy(51832), - wgPort: 51832, - }, - } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { redirectTraffic(t, tt.proxy, tt.wgPort, tt.endpointAddr) + if err := tt.closeFn(); err != nil { + t.Errorf("error: %v", err) + } }) } } @@ -142,49 +124,19 @@ func redirectTraffic(t *testing.T, proxy Proxy, wgPort int, endPointAddr *net.UD func TestProxyCloseByRemoteConn(t *testing.T) { ctx := context.Background() - ebpfProxy := ebpf.NewWGEBPFProxy(51831) - if err := ebpfProxy.Listen(); err != nil { - t.Fatalf("failed to initialize ebpf proxy: %s", err) - } - - defer func() { - if err := ebpfProxy.Free(); err != nil { - t.Errorf("failed to free ebpf proxy: %s", err) - } - }() - - iceBind := bind.NewICEBind(nil, nil) - endpointAddress := &net.UDPAddr{ - IP: net.IPv4(10, 0, 0, 1), - Port: 1234, - } - tests := []struct { - name string - proxy Proxy - endpointAddress *net.UDPAddr - }{ - { - name: "ebpf proxy", - proxy: ebpf.NewProxyWrapper(ebpfProxy), - }, - { - name: "udp proxy", - proxy: udp.NewWGUDPProxy(51832), - }, - { - name: "bind proxy", - proxy: bindproxy.NewProxyBind(iceBind), - endpointAddress: endpointAddress, - }, + tests, err := seedProxyForProxyCloseByRemoteConn() + if err != nil { + t.Fatalf("error: %v", err) } relayedConn, _ := net.Dial("udp", "127.0.0.1:1234") defer func() { _ = relayedConn.Close() }() + for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - err := tt.proxy.AddTurnConn(ctx, endpointAddress, relayedConn) + err := tt.proxy.AddTurnConn(ctx, tt.endpointAddr, relayedConn) if err != nil { t.Errorf("error: %v", err) } From f0020ad4ce73a4a48d88945198c68d9fac378e57 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Tue, 18 Feb 2025 13:58:20 +0100 Subject: [PATCH 20/26] Fallback to package loss solution if the raw socket does not work. --- client/iface/wgproxy/udp/proxy.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index 1e2274ca7..e480346c0 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/client/iface/wgproxy/udp/proxy.go @@ -128,6 +128,8 @@ func (p *WGUDPProxy) RedirectAs(endpoint *net.UDPAddr) { srcFakerConn, err := NewSrcFaker(p.localWGListenPort, endpoint) if err != nil { log.Errorf("failed to create src faker conn: %s", err) + // fallback to continue without redirecting + p.paused = true return } p.srcFakerConn = srcFakerConn From 648b4cdf729552d479d2bd54b4fa7f1f285c0f27 Mon Sep 17 00:00:00 2001 From: Zoltan Papp Date: Fri, 21 Feb 2025 14:50:29 +0100 Subject: [PATCH 21/26] Update client/iface/wgproxy/udp/proxy.go Co-authored-by: Viktor Liu <17948409+lixmal@users.noreply.github.com> --- client/iface/wgproxy/udp/proxy.go | 14 +++----------- 1 file changed, 3 insertions(+), 11 deletions(-) diff --git a/client/iface/wgproxy/udp/proxy.go b/client/iface/wgproxy/udp/proxy.go index e480346c0..3b32def25 100644 --- a/client/iface/wgproxy/udp/proxy.go +++ b/client/iface/wgproxy/udp/proxy.go @@ -228,17 +228,9 @@ func (p *WGUDPProxy) proxyToLocal(ctx context.Context) { return } - for { - p.pausedCond.L.Lock() - if p.paused { - p.pausedCond.Wait() - if !p.paused { - break - } - p.pausedCond.L.Unlock() - continue - } - break + p.pausedCond.L.Lock() + for p.paused { + p.pausedCond.Wait() } _, err = p.sendPkg(buf[:n]) p.pausedCond.L.Unlock() From d496d216935ce614860e9a0e9721ef45955025f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Fri, 21 Feb 2025 14:55:56 +0100 Subject: [PATCH 22/26] Apply same pausedCond logic on all implementation --- client/iface/wgproxy/bind/proxy.go | 14 +++----------- client/iface/wgproxy/ebpf/wrapper.go | 14 +++----------- 2 files changed, 6 insertions(+), 22 deletions(-) diff --git a/client/iface/wgproxy/bind/proxy.go b/client/iface/wgproxy/bind/proxy.go index b0dfef06b..6340b2d4f 100644 --- a/client/iface/wgproxy/bind/proxy.go +++ b/client/iface/wgproxy/bind/proxy.go @@ -161,17 +161,9 @@ func (p *ProxyBind) proxyToLocal(ctx context.Context) { return } - for { - p.pausedCond.L.Lock() - if p.paused { - p.pausedCond.Wait() - if !p.paused { - break - } - p.pausedCond.L.Unlock() - continue - } - break + p.pausedCond.L.Lock() + for p.paused { + p.pausedCond.Wait() } msg := bind.RecvMessage{ diff --git a/client/iface/wgproxy/ebpf/wrapper.go b/client/iface/wgproxy/ebpf/wrapper.go index a3ee8ac8f..98d14e80c 100644 --- a/client/iface/wgproxy/ebpf/wrapper.go +++ b/client/iface/wgproxy/ebpf/wrapper.go @@ -120,17 +120,9 @@ func (p *ProxyWrapper) proxyToLocal(ctx context.Context) { return } - for { - p.pausedCond.L.Lock() - if p.paused { - p.pausedCond.Wait() - if !p.paused { - break - } - p.pausedCond.L.Unlock() - continue - } - break + p.pausedCond.L.Lock() + for p.paused { + p.pausedCond.Wait() } err = p.wgeBPFProxy.sendPkg(buf[:n], p.wgEndpointCurrentUsedAddr) From 651e88d61199d0e84dc8518cf02ba17710fa13f8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Fri, 21 Feb 2025 14:59:42 +0100 Subject: [PATCH 23/26] Eliminate code duplication --- client/iface/wgproxy/ebpf/proxy.go | 43 +----------------- client/iface/wgproxy/rawsocket/rawsocket.go | 48 +++++++++++++++++++++ client/iface/wgproxy/udp/rawsocket.go | 44 +------------------ 3 files changed, 52 insertions(+), 83 deletions(-) create mode 100644 client/iface/wgproxy/rawsocket/rawsocket.go diff --git a/client/iface/wgproxy/ebpf/proxy.go b/client/iface/wgproxy/ebpf/proxy.go index 6462fbadb..0201e37e8 100644 --- a/client/iface/wgproxy/ebpf/proxy.go +++ b/client/iface/wgproxy/ebpf/proxy.go @@ -6,9 +6,7 @@ import ( "context" "fmt" "net" - "os" "sync" - "syscall" "github.com/google/gopacket" "github.com/google/gopacket/layers" @@ -17,6 +15,7 @@ import ( log "github.com/sirupsen/logrus" nberrors "github.com/netbirdio/netbird/client/errors" + "github.com/netbirdio/netbird/client/iface/wgproxy/rawsocket" "github.com/netbirdio/netbird/client/internal/ebpf" ebpfMgr "github.com/netbirdio/netbird/client/internal/ebpf/manager" nbnet "github.com/netbirdio/netbird/util/net" @@ -65,7 +64,7 @@ func (p *WGEBPFProxy) Listen() error { return err } - p.rawConn, err = p.prepareSenderRawSocket() + p.rawConn, err = rawsocket.PrepareSenderRawSocket() if err != nil { return err } @@ -215,44 +214,6 @@ generatePort: return p.lastUsedPort, nil } -func (p *WGEBPFProxy) prepareSenderRawSocket() (net.PacketConn, error) { - // Create a raw socket. - fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_RAW, syscall.IPPROTO_RAW) - if err != nil { - return nil, fmt.Errorf("creating raw socket failed: %w", err) - } - - // Set the IP_HDRINCL option on the socket to tell the kernel that headers are included in the packet. - err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_HDRINCL, 1) - if err != nil { - return nil, fmt.Errorf("setting IP_HDRINCL failed: %w", err) - } - - // Bind the socket to the "lo" interface. - err = syscall.SetsockoptString(fd, syscall.SOL_SOCKET, syscall.SO_BINDTODEVICE, "lo") - if err != nil { - return nil, fmt.Errorf("binding to lo interface failed: %w", err) - } - - // Set the fwmark on the socket. - err = nbnet.SetSocketOpt(fd) - if err != nil { - return nil, fmt.Errorf("setting fwmark failed: %w", err) - } - - // Convert the file descriptor to a PacketConn. - file := os.NewFile(uintptr(fd), fmt.Sprintf("fd %d", fd)) - if file == nil { - return nil, fmt.Errorf("converting fd to file failed") - } - packetConn, err := net.FilePacketConn(file) - if err != nil { - return nil, fmt.Errorf("converting file to packet conn failed: %w", err) - } - - return packetConn, nil -} - func (p *WGEBPFProxy) sendPkg(data []byte, endpointAddr *net.UDPAddr) error { payload := gopacket.Payload(data) ipH := &layers.IPv4{ diff --git a/client/iface/wgproxy/rawsocket/rawsocket.go b/client/iface/wgproxy/rawsocket/rawsocket.go new file mode 100644 index 000000000..a0da99334 --- /dev/null +++ b/client/iface/wgproxy/rawsocket/rawsocket.go @@ -0,0 +1,48 @@ +package rawsocket + +import ( + "fmt" + "net" + "os" + "syscall" + + nbnet "github.com/netbirdio/netbird/util/net" +) + +func PrepareSenderRawSocket() (net.PacketConn, error) { + // Create a raw socket. + fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_RAW, syscall.IPPROTO_RAW) + if err != nil { + return nil, fmt.Errorf("creating raw socket failed: %w", err) + } + + // Set the IP_HDRINCL option on the socket to tell the kernel that headers are included in the packet. + err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_HDRINCL, 1) + if err != nil { + return nil, fmt.Errorf("setting IP_HDRINCL failed: %w", err) + } + + // Bind the socket to the "lo" interface. + err = syscall.SetsockoptString(fd, syscall.SOL_SOCKET, syscall.SO_BINDTODEVICE, "lo") + if err != nil { + return nil, fmt.Errorf("binding to lo interface failed: %w", err) + } + + // Set the fwmark on the socket. + err = nbnet.SetSocketOpt(fd) + if err != nil { + return nil, fmt.Errorf("setting fwmark failed: %w", err) + } + + // Convert the file descriptor to a PacketConn. + file := os.NewFile(uintptr(fd), fmt.Sprintf("fd %d", fd)) + if file == nil { + return nil, fmt.Errorf("converting fd to file failed") + } + packetConn, err := net.FilePacketConn(file) + if err != nil { + return nil, fmt.Errorf("converting file to packet conn failed: %w", err) + } + + return packetConn, nil +} diff --git a/client/iface/wgproxy/udp/rawsocket.go b/client/iface/wgproxy/udp/rawsocket.go index d611fa59b..fdc911463 100644 --- a/client/iface/wgproxy/udp/rawsocket.go +++ b/client/iface/wgproxy/udp/rawsocket.go @@ -5,14 +5,12 @@ package udp import ( "fmt" "net" - "os" - "syscall" "github.com/google/gopacket" "github.com/google/gopacket/layers" log "github.com/sirupsen/logrus" - nbnet "github.com/netbirdio/netbird/util/net" + "github.com/netbirdio/netbird/client/iface/wgproxy/rawsocket" ) var ( @@ -36,7 +34,7 @@ type SrcFaker struct { } func NewSrcFaker(dstPort int, srcAddr *net.UDPAddr) (*SrcFaker, error) { - rawSocket, err := prepareSenderRawSocket() + rawSocket, err := rawsocket.PrepareSenderRawSocket() if err != nil { return nil, err } @@ -101,41 +99,3 @@ func prepareHeaders(dstPort int, srcAddr *net.UDPAddr) (gopacket.SerializableLay return ipH, udpH, nil } - -func prepareSenderRawSocket() (net.PacketConn, error) { - // Create a raw socket. - fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_RAW, syscall.IPPROTO_RAW) - if err != nil { - return nil, fmt.Errorf("creating raw socket failed: %w", err) - } - - // Set the IP_HDRINCL option on the socket to tell the kernel that headers are included in the packet. - err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_HDRINCL, 1) - if err != nil { - return nil, fmt.Errorf("setting IP_HDRINCL failed: %w", err) - } - - // Bind the socket to the "lo" interface. - err = syscall.SetsockoptString(fd, syscall.SOL_SOCKET, syscall.SO_BINDTODEVICE, "lo") - if err != nil { - return nil, fmt.Errorf("binding to lo interface failed: %w", err) - } - - // Set the fwmark on the socket. - err = nbnet.SetSocketOpt(fd) - if err != nil { - return nil, fmt.Errorf("setting fwmark failed: %w", err) - } - - // Convert the file descriptor to a PacketConn. - file := os.NewFile(uintptr(fd), fmt.Sprintf("fd %d", fd)) - if file == nil { - return nil, fmt.Errorf("converting fd to file failed") - } - packetConn, err := net.FilePacketConn(file) - if err != nil { - return nil, fmt.Errorf("converting file to packet conn failed: %w", err) - } - - return packetConn, nil -} From 06d71257b4d0871d0e70fe2902383263390b9d9b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Fri, 21 Feb 2025 16:00:31 +0100 Subject: [PATCH 24/26] Build rawsocket code on linux only. --- client/iface/wgproxy/rawsocket/rawsocket.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/client/iface/wgproxy/rawsocket/rawsocket.go b/client/iface/wgproxy/rawsocket/rawsocket.go index a0da99334..df68124c2 100644 --- a/client/iface/wgproxy/rawsocket/rawsocket.go +++ b/client/iface/wgproxy/rawsocket/rawsocket.go @@ -1,3 +1,5 @@ +//go:build linux && !android + package rawsocket import ( From 559d34758853aba5bd9eb9fa1b183e7c3e66ad99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Fri, 21 Feb 2025 16:19:28 +0100 Subject: [PATCH 25/26] Revert async WireGuard handshake --- client/internal/peer/conn.go | 31 +++++++++++++++++++------------ 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 514c7bf30..bc694d306 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -115,8 +115,6 @@ type Conn struct { guard *guard.Guard semaphore *semaphoregroup.SemaphoreGroup - - endpointUpdater *endpointUpdater } // NewConn creates a new not opened Conn to the remote peer. @@ -143,11 +141,6 @@ func NewConn(engineCtx context.Context, config ConnConfig, statusRecorder *Statu statusRelay: NewAtomicConnStatus(), statusICE: NewAtomicConnStatus(), semaphore: semaphore, - endpointUpdater: &endpointUpdater{ - log: connLog, - wgConfig: config.WgConfig, - initiator: isWireGuardInitiator(config), - }, } ctrl := isController(config) @@ -245,7 +238,7 @@ func (conn *Conn) Close() { conn.wgProxyICE = nil } - if err := conn.endpointUpdater.removeWgPeer(); err != nil { + if err := conn.removeWgPeer(); err != nil { conn.log.Errorf("failed to remove wg endpoint: %v", err) } @@ -373,7 +366,7 @@ func (conn *Conn) onICEConnectionIsReady(priority ConnPriority, iceConnInfo ICEC } conn.log.Infof("configure WireGuard endpoint to: %s", ep.String()) - if err = conn.endpointUpdater.configureWGEndpoint(ep); err != nil { + if err = conn.configureWGEndpoint(ep); err != nil { conn.handleConfigurationFailure(err, wgProxy) return } @@ -410,7 +403,7 @@ func (conn *Conn) onICEStateDisconnected() { if conn.isReadyToUpgrade() { conn.log.Infof("ICE disconnected, set Relay to active connection") - if err := conn.endpointUpdater.configureWGEndpoint(conn.wgProxyRelay.EndpointAddr()); err != nil { + if err := conn.configureWGEndpoint(conn.wgProxyRelay.EndpointAddr()); err != nil { conn.log.Errorf("failed to switch to relay conn: %v", err) } @@ -475,7 +468,7 @@ func (conn *Conn) onRelayConnectionIsReady(rci RelayConnInfo) { } wgProxy.Work() - if err := conn.endpointUpdater.configureWGEndpoint(wgProxy.EndpointAddr()); err != nil { + if err := conn.configureWGEndpoint(wgProxy.EndpointAddr()); err != nil { if err := wgProxy.CloseConn(); err != nil { conn.log.Warnf("Failed to close relay connection: %v", err) } @@ -505,7 +498,7 @@ func (conn *Conn) onRelayDisconnected() { if conn.currentConnPriority == connPriorityRelay { conn.log.Debugf("clean up WireGuard config") - if err := conn.endpointUpdater.removeWgPeer(); err != nil { + if err := conn.removeWgPeer(); err != nil { conn.log.Errorf("failed to remove wg endpoint: %v", err) } } @@ -546,6 +539,16 @@ func (conn *Conn) listenGuardEvent(ctx context.Context) { } } +func (conn *Conn) configureWGEndpoint(addr *net.UDPAddr) error { + return conn.config.WgConfig.WgInterface.UpdatePeer( + conn.config.WgConfig.RemoteKey, + conn.config.WgConfig.AllowedIps, + defaultWgKeepAlive, + addr, + conn.config.WgConfig.PreSharedKey, + ) +} + func (conn *Conn) updateRelayStatus(relayServerAddr string, rosenpassPubKey []byte) { peerState := State{ PubKey: conn.config.Key, @@ -725,6 +728,10 @@ func (conn *Conn) iceP2PIsActive() bool { return conn.currentConnPriority == connPriorityICEP2P && conn.statusICE.Get() == StatusConnected } +func (conn *Conn) removeWgPeer() error { + return conn.config.WgConfig.WgInterface.RemovePeer(conn.config.WgConfig.RemoteKey) +} + func (conn *Conn) handleConfigurationFailure(err error, wgProxy wgproxy.Proxy) { conn.log.Warnf("Failed to update wg peer configuration: %v", err) if wgProxy != nil { From 990aa8f7cd370f51f20f88e71b6cbd616f6d8684 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Zolt=C3=A1n=20Papp?= Date: Mon, 24 Feb 2025 10:45:59 +0100 Subject: [PATCH 26/26] Remove wg initiator logic --- client/internal/peer/conn.go | 5 - client/internal/peer/endpoint.go | 87 ------------- client/internal/peer/endpoint_test.go | 178 -------------------------- 3 files changed, 270 deletions(-) delete mode 100644 client/internal/peer/endpoint.go delete mode 100644 client/internal/peer/endpoint_test.go diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index bc694d306..785903e1b 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -770,11 +770,6 @@ func isController(config ConnConfig) bool { return config.LocalKey > config.Key } -// isWireGuardInitiator returns true if the local peer is the initiator of the WireGuard connection -func isWireGuardInitiator(config ConnConfig) bool { - return isController(config) -} - func isRosenpassEnabled(remoteRosenpassPubKey []byte) bool { return remoteRosenpassPubKey != nil } diff --git a/client/internal/peer/endpoint.go b/client/internal/peer/endpoint.go deleted file mode 100644 index 98c71af90..000000000 --- a/client/internal/peer/endpoint.go +++ /dev/null @@ -1,87 +0,0 @@ -package peer - -import ( - "context" - "net" - "sync" - "time" - - "github.com/sirupsen/logrus" -) - -// fallbackDelay could be const but because of testing it is a var -var fallbackDelay = 5 * time.Second - -type endpointUpdater struct { - log *logrus.Entry - wgConfig WgConfig - initiator bool - - cancelFunc func() - configUpdateMutex sync.Mutex -} - -// configureWGEndpoint sets up the WireGuard endpoint configuration. -// The initiator immediately configures the endpoint, while the non-initiator -// waits for a fallback period before configuring to avoid handshake congestion. -func (e *endpointUpdater) configureWGEndpoint(addr *net.UDPAddr) error { - if e.initiator { - return e.updateWireGuardPeer(addr) - } - - // prevent to run new update while cancel the previous update - e.configUpdateMutex.Lock() - if e.cancelFunc != nil { - e.cancelFunc() - } - e.configUpdateMutex.Unlock() - - var ctx context.Context - ctx, e.cancelFunc = context.WithCancel(context.Background()) - go e.scheduleDelayedUpdate(ctx, addr) - - return e.updateWireGuardPeer(nil) -} - -func (e *endpointUpdater) removeWgPeer() error { - e.configUpdateMutex.Lock() - defer e.configUpdateMutex.Unlock() - - if e.cancelFunc != nil { - e.cancelFunc() - } - - return e.wgConfig.WgInterface.RemovePeer(e.wgConfig.RemoteKey) -} - -// scheduleDelayedUpdate waits for the fallback period before updating the endpoint -func (e *endpointUpdater) scheduleDelayedUpdate(ctx context.Context, addr *net.UDPAddr) { - t := time.NewTimer(fallbackDelay) - defer t.Stop() - - select { - case <-ctx.Done(): - return - case <-t.C: - e.configUpdateMutex.Lock() - defer e.configUpdateMutex.Unlock() - - if ctx.Err() != nil { - return - } - - if err := e.updateWireGuardPeer(addr); err != nil { - e.log.Errorf("failed to update WireGuard peer, address: %s, error: %v", addr, err) - } - } -} - -func (e *endpointUpdater) updateWireGuardPeer(endpoint *net.UDPAddr) error { - return e.wgConfig.WgInterface.UpdatePeer( - e.wgConfig.RemoteKey, - e.wgConfig.AllowedIps, - defaultWgKeepAlive, - endpoint, - e.wgConfig.PreSharedKey, - ) -} diff --git a/client/internal/peer/endpoint_test.go b/client/internal/peer/endpoint_test.go deleted file mode 100644 index ec980b7d7..000000000 --- a/client/internal/peer/endpoint_test.go +++ /dev/null @@ -1,178 +0,0 @@ -package peer - -import ( - "net" - "testing" - "time" - - log "github.com/sirupsen/logrus" - "github.com/stretchr/testify/mock" - "golang.zx2c4.com/wireguard/wgctrl/wgtypes" - - "github.com/netbirdio/netbird/client/iface/configurer" - "github.com/netbirdio/netbird/client/iface/wgproxy" -) - -type MockWgInterface struct { - mock.Mock - - lastSetAddr *net.UDPAddr -} - -func (m *MockWgInterface) GetStats(peerKey string) (configurer.WGStats, error) { - panic("implement me") -} - -func (m *MockWgInterface) GetProxy() wgproxy.Proxy { - panic("implement me") -} - -func (m *MockWgInterface) UpdatePeer(peerKey string, allowedIps string, keepAlive time.Duration, endpoint *net.UDPAddr, preSharedKey *wgtypes.Key) error { - args := m.Called(peerKey, allowedIps, keepAlive, endpoint, preSharedKey) - m.lastSetAddr = endpoint - return args.Error(0) -} - -func (m *MockWgInterface) RemovePeer(publicKey string) error { - args := m.Called(publicKey) - return args.Error(0) -} - -func Test_endpointUpdater_initiator(t *testing.T) { - mockWgInterface := &MockWgInterface{} - e := &endpointUpdater{ - log: log.WithField("peer", "my-peer-key"), - wgConfig: WgConfig{ - WgListenPort: 51820, - RemoteKey: "secret-remote-key", - WgInterface: mockWgInterface, - AllowedIps: "172.16.254.1", - }, - initiator: true, - } - addr := &net.UDPAddr{ - IP: net.ParseIP("127.0.0.1"), - Port: 1234, - } - - mockWgInterface.On( - "UpdatePeer", - e.wgConfig.RemoteKey, - e.wgConfig.AllowedIps, - defaultWgKeepAlive, - addr, - (*wgtypes.Key)(nil), - ).Return(nil) - - if err := e.configureWGEndpoint(addr); err != nil { - t.Fatalf("updateWireGuardPeer() failed: %v", err) - } - - mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, addr, (*wgtypes.Key)(nil)) -} - -func Test_endpointUpdater_nonInitiator(t *testing.T) { - fallbackDelay = 1 * time.Second - mockWgInterface := &MockWgInterface{} - e := &endpointUpdater{ - log: log.WithField("peer", "my-peer-key"), - wgConfig: WgConfig{ - WgListenPort: 51820, - RemoteKey: "secret-remote-key", - WgInterface: mockWgInterface, - AllowedIps: "172.16.254.1", - }, - initiator: false, - } - addr := &net.UDPAddr{ - IP: net.ParseIP("127.0.0.1"), - Port: 1234, - } - - mockWgInterface.On( - "UpdatePeer", - e.wgConfig.RemoteKey, - e.wgConfig.AllowedIps, - defaultWgKeepAlive, - (*net.UDPAddr)(nil), - (*wgtypes.Key)(nil), - ).Return(nil) - - mockWgInterface.On( - "UpdatePeer", - e.wgConfig.RemoteKey, - e.wgConfig.AllowedIps, - defaultWgKeepAlive, - addr, - (*wgtypes.Key)(nil), - ).Return(nil) - - err := e.configureWGEndpoint(addr) - if err != nil { - t.Fatalf("updateWireGuardPeer() failed: %v", err) - } - mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, (*net.UDPAddr)(nil), (*wgtypes.Key)(nil)) - - time.Sleep(fallbackDelay + time.Second) - - mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, addr, (*wgtypes.Key)(nil)) -} - -func Test_endpointUpdater_overRule(t *testing.T) { - fallbackDelay = 1 * time.Second - mockWgInterface := &MockWgInterface{} - e := &endpointUpdater{ - log: log.WithField("peer", "my-peer-key"), - wgConfig: WgConfig{ - WgListenPort: 51820, - RemoteKey: "secret-remote-key", - WgInterface: mockWgInterface, - AllowedIps: "172.16.254.1", - }, - initiator: false, - } - addr1 := &net.UDPAddr{ - IP: net.ParseIP("127.0.0.1"), - Port: 1000, - } - - addr2 := &net.UDPAddr{ - IP: net.ParseIP("127.0.0.1"), - Port: 1001, - } - - mockWgInterface.On( - "UpdatePeer", - e.wgConfig.RemoteKey, - e.wgConfig.AllowedIps, - defaultWgKeepAlive, - (*net.UDPAddr)(nil), - (*wgtypes.Key)(nil), - ).Return(nil) - - mockWgInterface.On( - "UpdatePeer", - e.wgConfig.RemoteKey, - e.wgConfig.AllowedIps, - defaultWgKeepAlive, - addr2, - (*wgtypes.Key)(nil), - ).Return(nil) - - if err := e.configureWGEndpoint(addr1); err != nil { - t.Fatalf("updateWireGuardPeer() failed: %v", err) - } - mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, (*net.UDPAddr)(nil), (*wgtypes.Key)(nil)) - - if err := e.configureWGEndpoint(addr2); err != nil { - t.Fatalf("updateWireGuardPeer() failed: %v", err) - } - - time.Sleep(fallbackDelay + time.Second) - - mockWgInterface.AssertCalled(t, "UpdatePeer", e.wgConfig.RemoteKey, e.wgConfig.AllowedIps, defaultWgKeepAlive, addr2, (*wgtypes.Key)(nil)) - - if mockWgInterface.lastSetAddr != addr2 { - t.Fatalf("lastSetAddr is not equal to addr2") - } -}