mirror of
https://github.com/netbirdio/netbird.git
synced 2026-09-14 18:59:08 +02:00
Zoltán Papp
4a72c444f3
The relayed conn watcher compared the conn pointer under relayLock, released it, and only then tore the connection down. A new offer could install its replacement in that window, so a watcher that validated the old pointer went on to close the proxy of the connection that had already replaced it and report the peer as disconnected while it was up. Move the decision to where the teardown happens. Conn records which relayed connection the current proxy was built from, and onRelayDisconnected takes the connection the signal belongs to and drops it under conn.mu when it is no longer the current one. Check and effect are now in the same critical section, so the verdict cannot go stale before it is acted on. This also covers the proxy read loops, whose disconnect listener took no argument and had the same defect: it now names the connection it belongs to. The WG timeout path keeps passing nil, since it deliberately tears down whatever is current.
138 lines
3.3 KiB
Go
138 lines
3.3 KiB
Go
package peer
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"net/netip"
|
|
"sync"
|
|
"sync/atomic"
|
|
|
|
log "github.com/sirupsen/logrus"
|
|
|
|
relayClient "github.com/netbirdio/netbird/shared/relay/client"
|
|
)
|
|
|
|
type RelayConnInfo struct {
|
|
relayedConn *relayClient.Conn
|
|
rosenpassPubKey []byte
|
|
rosenpassAddr string
|
|
}
|
|
|
|
type WorkerRelay struct {
|
|
peerCtx context.Context
|
|
log *log.Entry
|
|
isController bool
|
|
config ConnConfig
|
|
conn *Conn
|
|
relayManager *relayClient.Manager
|
|
|
|
relayedConn *relayClient.Conn
|
|
relayLock sync.Mutex
|
|
|
|
relaySupportedOnRemotePeer atomic.Bool
|
|
}
|
|
|
|
func NewWorkerRelay(ctx context.Context, log *log.Entry, ctrl bool, config ConnConfig, conn *Conn, relayManager *relayClient.Manager) *WorkerRelay {
|
|
r := &WorkerRelay{
|
|
peerCtx: ctx,
|
|
log: log,
|
|
isController: ctrl,
|
|
config: config,
|
|
conn: conn,
|
|
relayManager: relayManager,
|
|
}
|
|
return r
|
|
}
|
|
|
|
func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
|
|
if !w.isRelaySupported(remoteOfferAnswer) {
|
|
w.log.Infof("Relay is not supported by remote peer")
|
|
w.relaySupportedOnRemotePeer.Store(false)
|
|
return
|
|
}
|
|
w.relaySupportedOnRemotePeer.Store(true)
|
|
|
|
// the relayManager will return with error in case if the connection has lost with relay server
|
|
currentRelayAddress, _, err := w.relayManager.RelayInstanceAddress()
|
|
if err != nil {
|
|
w.log.Errorf("failed to handle new offer: %s", err)
|
|
return
|
|
}
|
|
|
|
srv := w.preferredRelayServer(currentRelayAddress, remoteOfferAnswer.RelaySrvAddress)
|
|
var serverIP netip.Addr
|
|
if srv == remoteOfferAnswer.RelaySrvAddress {
|
|
serverIP = remoteOfferAnswer.RelaySrvIP
|
|
}
|
|
|
|
relayedConn, err := w.relayManager.OpenConn(w.peerCtx, srv, w.config.Key, serverIP)
|
|
if err != nil {
|
|
if errors.Is(err, relayClient.ErrConnAlreadyExists) {
|
|
w.log.Debugf("handled offer by reusing existing relay connection")
|
|
return
|
|
}
|
|
w.log.Errorf("failed to open connection via Relay: %s", err)
|
|
return
|
|
}
|
|
|
|
w.relayLock.Lock()
|
|
w.relayedConn = relayedConn
|
|
w.relayLock.Unlock()
|
|
|
|
go w.watchRelayedConn(relayedConn)
|
|
|
|
w.log.Debugf("peer conn opened via Relay: %s", srv)
|
|
go w.conn.onRelayConnectionIsReady(RelayConnInfo{
|
|
relayedConn: relayedConn,
|
|
rosenpassPubKey: remoteOfferAnswer.RosenpassPubKey,
|
|
rosenpassAddr: remoteOfferAnswer.RosenpassAddr,
|
|
})
|
|
}
|
|
|
|
func (w *WorkerRelay) RelayInstanceAddress() (string, netip.Addr, error) {
|
|
return w.relayManager.RelayInstanceAddress()
|
|
}
|
|
|
|
func (w *WorkerRelay) IsRelayConnectionSupportedWithPeer() bool {
|
|
return w.relaySupportedOnRemotePeer.Load() && w.RelayIsSupportedLocally()
|
|
}
|
|
|
|
func (w *WorkerRelay) RelayIsSupportedLocally() bool {
|
|
return w.relayManager.HasRelayAddress()
|
|
}
|
|
|
|
func (w *WorkerRelay) CloseConn() {
|
|
w.relayLock.Lock()
|
|
conn := w.relayedConn
|
|
w.relayedConn = nil
|
|
w.relayLock.Unlock()
|
|
|
|
if conn == nil {
|
|
return
|
|
}
|
|
|
|
if err := conn.Close(); err != nil {
|
|
w.log.Warnf("failed to close relay connection: %v", err)
|
|
}
|
|
}
|
|
|
|
func (w *WorkerRelay) isRelaySupported(answer *OfferAnswer) bool {
|
|
if !w.relayManager.HasRelayAddress() {
|
|
return false
|
|
}
|
|
return answer.RelaySrvAddress != ""
|
|
}
|
|
|
|
func (w *WorkerRelay) preferredRelayServer(myRelayAddress, remoteRelayAddress string) string {
|
|
if w.isController {
|
|
return myRelayAddress
|
|
}
|
|
return remoteRelayAddress
|
|
}
|
|
|
|
func (w *WorkerRelay) watchRelayedConn(relayedConn *relayClient.Conn) {
|
|
<-relayedConn.Context().Done()
|
|
|
|
w.conn.onRelayDisconnected(relayedConn)
|
|
}
|