mirror of
https://github.com/netbirdio/netbird.git
synced 2026-09-12 17:59:06 +02:00
[client] Fix relay instance address race (#7498)
Read the relay instance URL and IP atomically to prevent reconnects from mixing values from different connections. Extend existing connection and offer/answer logs with relay URLs and IPs to help trace mismatched advertisements.
This commit is contained in:
@@ -116,7 +116,7 @@ func (h *Handshaker) Listen(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case remoteOfferAnswer := <-h.remoteOffersCh:
|
||||
h.log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials())
|
||||
h.log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t, relay server: %s, relay IP: %s", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials(), remoteOfferAnswer.RelaySrvAddress, remoteOfferAnswer.RelaySrvIP)
|
||||
|
||||
// Record signaling received for reconnection attempts
|
||||
if h.metricsStages != nil {
|
||||
@@ -138,7 +138,7 @@ func (h *Handshaker) Listen(ctx context.Context) {
|
||||
continue
|
||||
}
|
||||
case remoteOfferAnswer := <-h.remoteAnswerCh:
|
||||
h.log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials())
|
||||
h.log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t, relay server: %s, relay IP: %s", remoteOfferAnswer.Version, remoteOfferAnswer.WgListenPort, remoteOfferAnswer.SessionIDString(), remoteOfferAnswer.hasICECredentials(), remoteOfferAnswer.RelaySrvAddress, remoteOfferAnswer.RelaySrvIP)
|
||||
|
||||
// Record signaling received for reconnection attempts
|
||||
if h.metricsStages != nil {
|
||||
@@ -209,14 +209,14 @@ func (h *Handshaker) sendOffer() error {
|
||||
}
|
||||
|
||||
offer := h.buildOfferAnswer()
|
||||
h.log.Debugf("sending offer with serial: %s", offer.SessionIDString())
|
||||
h.log.Debugf("sending offer with serial: %s, relay server: %s, relay IP: %s", offer.SessionIDString(), offer.RelaySrvAddress, offer.RelaySrvIP)
|
||||
|
||||
return h.signaler.SignalOffer(offer, h.config.Key)
|
||||
}
|
||||
|
||||
func (h *Handshaker) sendAnswer() error {
|
||||
answer := h.buildOfferAnswer()
|
||||
h.log.Debugf("sending answer with serial: %s", answer.SessionIDString())
|
||||
h.log.Debugf("sending answer with serial: %s, relay server: %s, relay IP: %s", answer.SessionIDString(), answer.RelaySrvAddress, answer.RelaySrvIP)
|
||||
|
||||
return h.signaler.SignalAnswer(answer, h.config.Key)
|
||||
}
|
||||
|
||||
@@ -279,7 +279,7 @@ func (c *Client) Connect(ctx context.Context) error {
|
||||
c.stateSubscription = NewPeersStateSubscription(c.log, c.relayConn, c.closeConnsByPeerID)
|
||||
|
||||
c.log = c.log.WithField("relay", instanceURL.String())
|
||||
c.log.Infof("relay connection established")
|
||||
c.log.Infof("relay connection established, server IP: %s", connectedIP(c.relayConn))
|
||||
|
||||
c.serviceIsRunning = true
|
||||
|
||||
@@ -364,23 +364,6 @@ func (c *Client) ServerInstanceURL() (string, error) {
|
||||
return c.instanceURL.String(), nil
|
||||
}
|
||||
|
||||
// ConnectedIP returns the IP address of the live relay-server connection,
|
||||
// extracted from the underlying socket's RemoteAddr. Zero value if not
|
||||
// connected or if the address is not an IP literal.
|
||||
func (c *Client) ConnectedIP() netip.Addr {
|
||||
c.mu.Lock()
|
||||
conn := c.relayConn
|
||||
c.mu.Unlock()
|
||||
if conn == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
addr := conn.RemoteAddr()
|
||||
if addr == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
return extractIPLiteral(addr.String())
|
||||
}
|
||||
|
||||
// SetOnDisconnectListener sets a function that will be called when the connection to the relay server is closed.
|
||||
func (c *Client) SetOnDisconnectListener(fn func(string)) {
|
||||
c.listenerMutex.Lock()
|
||||
@@ -777,6 +760,17 @@ func (c *Client) listenForStopEvents(ctx context.Context, hc *healthcheck.Receiv
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) serverInstanceAddress() (string, netip.Addr, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
addr, err := c.ServerInstanceURL()
|
||||
if err != nil {
|
||||
return "", netip.Addr{}, err
|
||||
}
|
||||
return addr, connectedIP(c.relayConn), nil
|
||||
}
|
||||
|
||||
func (c *Client) closeAllConns() {
|
||||
for _, container := range c.conns {
|
||||
container.close()
|
||||
@@ -923,6 +917,17 @@ func (c *Client) handlePeersWentOfflineMsg(buf []byte) {
|
||||
c.stateSubscription.OnPeersWentOffline(peersID)
|
||||
}
|
||||
|
||||
func connectedIP(conn net.Conn) netip.Addr {
|
||||
if conn == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
addr := conn.RemoteAddr()
|
||||
if addr == nil {
|
||||
return netip.Addr{}
|
||||
}
|
||||
return extractIPLiteral(addr.String())
|
||||
}
|
||||
|
||||
// extractIPLiteral returns the IP from address forms produced by the relay
|
||||
// dialers (URL or host:port). Zero value if the host is not an IP.
|
||||
func extractIPLiteral(s string) netip.Addr {
|
||||
|
||||
@@ -8,6 +8,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.opentelemetry.io/otel"
|
||||
|
||||
"github.com/netbirdio/netbird/client/iface"
|
||||
@@ -68,18 +70,17 @@ func TestClient_ServerIPRecoversFromUnresolvableFQDN(t *testing.T) {
|
||||
if !c.Ready() {
|
||||
t.Fatalf("client not ready after connect")
|
||||
}
|
||||
if got := c.ConnectedIP(); got.String() != "127.0.0.1" {
|
||||
t.Fatalf("ConnectedIP = %q, want 127.0.0.1", got)
|
||||
}
|
||||
url, ip, err := c.serverInstanceAddress()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, srvCfg.ExposedAddress, url, "relay URL must come from the handshake")
|
||||
assert.Equal(t, netip.MustParseAddr("127.0.0.1"), ip, "relay IP must come from the connection")
|
||||
})
|
||||
}
|
||||
|
||||
// TestClient_ConnectedIPAfterFQDNDial verifies ConnectedIP returns the
|
||||
// resolved IP after a successful FQDN-based dial. The underlying socket's
|
||||
// RemoteAddr must be exposed through the dialer wrappers; if it returns
|
||||
// the dial-time URL instead, ConnectedIP returns empty and the dial
|
||||
// IP we advertise to peers is empty too.
|
||||
func TestClient_ConnectedIPAfterFQDNDial(t *testing.T) {
|
||||
// TestClient_ServerInstanceAddressAfterFQDNDial verifies the relay address
|
||||
// includes the resolved IP after an FQDN dial. The dialer wrappers must expose
|
||||
// the socket's RemoteAddr; returning the dial-time URL would lose the IP.
|
||||
func TestClient_ServerInstanceAddressAfterFQDNDial(t *testing.T) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
|
||||
defer cancel()
|
||||
|
||||
@@ -111,10 +112,10 @@ func TestClient_ConnectedIPAfterFQDNDial(t *testing.T) {
|
||||
}
|
||||
t.Cleanup(func() { _ = c.Close() })
|
||||
|
||||
got := c.ConnectedIP().String()
|
||||
if got != "127.0.0.1" && got != "::1" {
|
||||
t.Fatalf("ConnectedIP after FQDN dial = %q, want 127.0.0.1 or ::1", got)
|
||||
}
|
||||
url, ip, err := c.serverInstanceAddress()
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, srvCfg.ExposedAddress, url, "relay URL must come from the handshake")
|
||||
assert.Contains(t, []string{"127.0.0.1", "::1"}, ip.String(), "relay IP must resolve to localhost")
|
||||
}
|
||||
|
||||
func TestSubstituteHost(t *testing.T) {
|
||||
@@ -214,15 +215,12 @@ func TestSubstituteHost(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestClient_ConnectedIPEmptyWhenNotConnected(t *testing.T) {
|
||||
c := NewClient("rel://example.invalid:80", hmacTokenStore, "x", iface.DefaultMTU)
|
||||
if got := c.ConnectedIP(); got.IsValid() {
|
||||
t.Fatalf("ConnectedIP on disconnected client = %q, want zero", got)
|
||||
}
|
||||
func TestConnectedIPNilConnection(t *testing.T) {
|
||||
assert.False(t, connectedIP(nil).IsValid(), "missing connection must not provide an IP")
|
||||
}
|
||||
|
||||
// staticAddr is a net.Addr that returns a fixed string. Used to verify
|
||||
// ConnectedIP parses RemoteAddr correctly.
|
||||
// connectedIP parses RemoteAddr correctly.
|
||||
type staticAddr struct{ s string }
|
||||
|
||||
func (a staticAddr) Network() string { return "tcp" }
|
||||
@@ -235,7 +233,7 @@ type stubConn struct {
|
||||
|
||||
func (s stubConn) RemoteAddr() net.Addr { return s.remote }
|
||||
|
||||
func TestClient_ConnectedIPParsesRemoteAddr(t *testing.T) {
|
||||
func TestConnectedIPParsesRemoteAddr(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
s string
|
||||
@@ -252,15 +250,12 @@ func TestClient_ConnectedIPParsesRemoteAddr(t *testing.T) {
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
c := &Client{relayConn: stubConn{remote: staticAddr{s: tt.s}}}
|
||||
got := c.ConnectedIP()
|
||||
got := connectedIP(stubConn{remote: staticAddr{s: tt.s}})
|
||||
var gotStr string
|
||||
if got.IsValid() {
|
||||
gotStr = got.String()
|
||||
}
|
||||
if gotStr != tt.want {
|
||||
t.Errorf("ConnectedIP(%q) = %q, want %q", tt.s, gotStr, tt.want)
|
||||
}
|
||||
assert.Equal(t, tt.want, gotStr, "IP extracted from RemoteAddr %q", tt.s)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -256,11 +256,7 @@ func (m *Manager) RelayInstanceAddress() (string, netip.Addr, error) {
|
||||
if m.relayClient == nil {
|
||||
return "", netip.Addr{}, ErrRelayClientNotConnected
|
||||
}
|
||||
addr, err := m.relayClient.ServerInstanceURL()
|
||||
if err != nil {
|
||||
return "", netip.Addr{}, err
|
||||
}
|
||||
return addr, m.relayClient.ConnectedIP(), nil
|
||||
return m.relayClient.serverInstanceAddress()
|
||||
}
|
||||
|
||||
// ServerURLs returns the addresses of the relay servers.
|
||||
|
||||
@@ -0,0 +1,103 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"runtime"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestManager_RelayInstanceAddressAcrossReconnect(t *testing.T) {
|
||||
relays := []struct {
|
||||
url *RelayAddr
|
||||
conn stubConn
|
||||
ip netip.Addr
|
||||
}{
|
||||
{
|
||||
url: &RelayAddr{addr: "rels://relay-a.example:443"},
|
||||
conn: stubConn{remote: staticAddr{s: "192.0.2.1:443"}},
|
||||
ip: netip.MustParseAddr("192.0.2.1"),
|
||||
},
|
||||
{
|
||||
url: &RelayAddr{addr: "rels://relay-b.example:443"},
|
||||
conn: stubConn{remote: staticAddr{s: "192.0.2.2:443"}},
|
||||
ip: netip.MustParseAddr("192.0.2.2"),
|
||||
},
|
||||
}
|
||||
c := &Client{
|
||||
instanceURL: relays[0].url,
|
||||
relayConn: relays[0].conn,
|
||||
serviceIsRunning: true,
|
||||
}
|
||||
m := &Manager{relayClient: c}
|
||||
started := make(chan struct{})
|
||||
stop := make(chan struct{})
|
||||
done := make(chan struct{})
|
||||
t.Cleanup(func() {
|
||||
close(stop)
|
||||
<-done
|
||||
})
|
||||
go func() {
|
||||
defer close(done)
|
||||
for i := 0; ; i++ {
|
||||
select {
|
||||
case <-stop:
|
||||
return
|
||||
default:
|
||||
}
|
||||
// Publish successive connection states using the lifecycle locks.
|
||||
// Yield before publication so a getter using only muInstanceURL
|
||||
// can read the old URL while waiting for the new connection's IP.
|
||||
c.mu.Lock()
|
||||
runtime.Gosched()
|
||||
relay := relays[i%len(relays)]
|
||||
c.muInstanceURL.Lock()
|
||||
c.instanceURL = relay.url
|
||||
c.muInstanceURL.Unlock()
|
||||
c.relayConn = relay.conn
|
||||
c.mu.Unlock()
|
||||
if i == 0 {
|
||||
close(started)
|
||||
}
|
||||
}
|
||||
}()
|
||||
<-started
|
||||
|
||||
for range 1000 {
|
||||
url, ip, err := m.RelayInstanceAddress()
|
||||
require.NoError(t, err)
|
||||
wantIP := relays[0].ip
|
||||
if url == relays[1].url.String() {
|
||||
wantIP = relays[1].ip
|
||||
}
|
||||
if !assert.Equal(t, wantIP, ip, "advertised IP must belong to relay %s", url) {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_RelayInstanceAddressDisconnected(t *testing.T) {
|
||||
for _, tt := range []struct {
|
||||
name string
|
||||
client *Client
|
||||
}{
|
||||
{name: "no client"},
|
||||
{name: "not connected", client: &Client{}},
|
||||
{
|
||||
name: "closed connection",
|
||||
client: &Client{
|
||||
relayConn: stubConn{remote: staticAddr{s: "192.0.2.1:443"}},
|
||||
},
|
||||
},
|
||||
} {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
m := &Manager{relayClient: tt.client}
|
||||
url, ip, err := m.RelayInstanceAddress()
|
||||
assert.Error(t, err)
|
||||
assert.Empty(t, url, "disconnected relay must not advertise a URL")
|
||||
assert.False(t, ip.IsValid(), "disconnected relay must not advertise a stale IP")
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -63,7 +63,12 @@ func (sp *ServerPicker) PickServer(parentCtx context.Context) (*Client, error) {
|
||||
if !ok {
|
||||
return nil, <-errChan
|
||||
}
|
||||
log.Infof("chosen home Relay server: %s", cr.Url)
|
||||
instanceURL, serverIP, err := cr.RelayClient.serverInstanceAddress()
|
||||
if err != nil {
|
||||
log.Infof("chosen home Relay server: %s, instance address unavailable: %v", cr.Url, err)
|
||||
return cr.RelayClient, nil
|
||||
}
|
||||
log.Infof("chosen home Relay server: %s, instance URL: %s, server IP: %s", cr.Url, instanceURL, serverIP)
|
||||
return cr.RelayClient, nil
|
||||
case <-ctx.Done():
|
||||
return nil, fmt.Errorf("connect to relay server: %w", ctx.Err())
|
||||
|
||||
Reference in New Issue
Block a user