Skip to content
2 changes: 1 addition & 1 deletion client/internal/engine.go
Original file line number Diff line number Diff line change
Expand Up @@ -554,7 +554,7 @@ func (e *Engine) Start(netbirdConfig *mgmProto.NetbirdConfig, mgmtURL *url.URL)
e.connMgr.Start(e.ctx)

e.srWatcher = guard.NewSRWatcher(e.signal, e.relayManager, e.mobileDep.IFaceDiscover, iceCfg)
e.srWatcher.Start()
e.srWatcher.Start(peer.IsForceRelayed())

e.receiveSignalEvents()
e.receiveManagementEvents()
Expand Down
43 changes: 30 additions & 13 deletions client/internal/peer/conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -181,17 +181,20 @@ func (conn *Conn) Open(engineCtx context.Context) error {

conn.workerRelay = NewWorkerRelay(conn.ctx, conn.Log, isController(conn.config), conn.config, conn, conn.relayManager)

relayIsSupportedLocally := conn.workerRelay.RelayIsSupportedLocally()
workerICE, err := NewWorkerICE(conn.ctx, conn.Log, conn.config, conn, conn.signaler, conn.iFaceDiscover, conn.statusRecorder, relayIsSupportedLocally)
if err != nil {
return err
forceRelay := IsForceRelayed()
if !forceRelay {
relayIsSupportedLocally := conn.workerRelay.RelayIsSupportedLocally()
workerICE, err := NewWorkerICE(conn.ctx, conn.Log, conn.config, conn, conn.signaler, conn.iFaceDiscover, conn.statusRecorder, relayIsSupportedLocally)
if err != nil {
return err
}
conn.workerICE = workerICE
}
conn.workerICE = workerICE

conn.handshaker = NewHandshaker(conn.Log, conn.config, conn.signaler, conn.workerICE, conn.workerRelay, conn.metricsStages)

conn.handshaker.AddRelayListener(conn.workerRelay.OnNewOffer)
if !isForceRelayed() {
if !forceRelay {
conn.handshaker.AddICEListener(conn.workerICE.OnNewOffer)
}

Expand Down Expand Up @@ -247,7 +250,9 @@ func (conn *Conn) Close(signalToRemote bool) {
conn.wgWatcherCancel()
}
conn.workerRelay.CloseConn()
conn.workerICE.Close()
if conn.workerICE != nil {
conn.workerICE.Close()
}

if conn.wgProxyRelay != nil {
err := conn.wgProxyRelay.CloseConn()
Expand Down Expand Up @@ -290,7 +295,9 @@ func (conn *Conn) OnRemoteAnswer(answer OfferAnswer) {
// OnRemoteCandidate Handles ICE connection Candidate provided by the remote peer.
func (conn *Conn) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HAMap) {
conn.dumpState.RemoteCandidate()
conn.workerICE.OnRemoteCandidate(candidate, haRoutes)
if conn.workerICE != nil {
conn.workerICE.OnRemoteCandidate(candidate, haRoutes)
}
}

// SetOnConnected sets a handler function to be triggered by Conn when a new connection to a remote peer established
Expand Down Expand Up @@ -717,14 +724,24 @@ func (conn *Conn) isConnectedOnAllWay() (connected bool) {
}
}()

// For JS platform: only relay connection is supported
if runtime.GOOS == "js" {
// For force-relayed connections (JS or NB_FORCE_RELAY): only relay status matters
if IsForceRelayed() {
if !conn.workerRelay.IsRelayConnectionSupportedWithPeer() {
return false
}
return conn.statusRelay.Get() == worker.StatusConnected
}

// For non-JS platforms: check ICE connection status
if conn.statusICE.Get() == worker.StatusDisconnected && !conn.workerICE.InProgress() {
return false
// For non-forced platforms: check ICE connection status only if remote peer supports ICE
if conn.handshaker.RemoteICESupported() {
if conn.statusICE.Get() == worker.StatusDisconnected && !conn.workerICE.InProgress() {
return false
}
} else {
// ICE is not available, so relay is the only possible transport
if !conn.workerRelay.IsRelayConnectionSupportedWithPeer() {
return false
}
}

// If relay is supported with peer, it must also be connected
Expand Down
2 changes: 1 addition & 1 deletion client/internal/peer/env.go
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,7 @@ const (
EnvKeyNBForceRelay = "NB_FORCE_RELAY"
)

func isForceRelayed() bool {
func IsForceRelayed() bool {
if runtime.GOOS == "js" {
return true
}
Expand Down
8 changes: 5 additions & 3 deletions client/internal/peer/guard/sr_watcher.go
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ func NewSRWatcher(signalClient chNotifier, relayManager chNotifier, iFaceDiscove
return srw
}

func (w *SRWatcher) Start() {
func (w *SRWatcher) Start(disableICEMonitor bool) {
w.mu.Lock()
defer w.mu.Unlock()

Expand All @@ -50,8 +50,10 @@ func (w *SRWatcher) Start() {
ctx, cancel := context.WithCancel(context.Background())
w.cancelIceMonitor = cancel

iceMonitor := NewICEMonitor(w.iFaceDiscover, w.iceConfig, GetICEMonitorPeriod())
go iceMonitor.Start(ctx, w.onICEChanged)
if !disableICEMonitor {
iceMonitor := NewICEMonitor(w.iFaceDiscover, w.iceConfig, GetICEMonitorPeriod())
go iceMonitor.Start(ctx, w.onICEChanged)
}
w.signalClient.SetOnReconnectedListener(w.onReconnected)
w.relayManager.SetOnReconnectedListener(w.onReconnected)

Expand Down
48 changes: 41 additions & 7 deletions client/internal/peer/handshaker.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"context"
"errors"
"sync"
"sync/atomic"

log "github.com/sirupsen/logrus"

Expand Down Expand Up @@ -59,14 +60,18 @@ type Handshaker struct {
relayListener *AsyncOfferListener
iceListener func(remoteOfferAnswer *OfferAnswer)

// remoteICESupported tracks whether the remote peer includes ICE credentials in its offers/answers.
// When false, the local side skips ICE listener dispatch and suppresses ICE credentials in responses.
remoteICESupported atomic.Bool

// remoteOffersCh is a channel used to wait for remote credentials to proceed with the connection
remoteOffersCh chan OfferAnswer
// remoteAnswerCh is a channel used to wait for remote credentials answer (confirmation of our offer) to proceed with the connection
remoteAnswerCh chan OfferAnswer
}

func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *WorkerICE, relay *WorkerRelay, metricsStages *MetricsStages) *Handshaker {
return &Handshaker{
h := &Handshaker{
log: log,
config: config,
signaler: signaler,
Expand All @@ -76,6 +81,13 @@ func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *W
remoteOffersCh: make(chan OfferAnswer),
remoteAnswerCh: make(chan OfferAnswer),
}
// assume remote supports ICE until we learn otherwise from received offers
h.remoteICESupported.Store(ice != nil)
return h
}

func (h *Handshaker) RemoteICESupported() bool {
return h.remoteICESupported.Load()
}

func (h *Handshaker) AddRelayListener(offer func(remoteOfferAnswer *OfferAnswer)) {
Expand All @@ -97,11 +109,13 @@ func (h *Handshaker) Listen(ctx context.Context) {
h.metricsStages.RecordSignalingReceived()
}

h.updateRemoteICEState(&remoteOfferAnswer)

if h.relayListener != nil {
h.relayListener.Notify(&remoteOfferAnswer)
}

if h.iceListener != nil {
if h.iceListener != nil && h.RemoteICESupported() {
h.iceListener(&remoteOfferAnswer)
}

Expand All @@ -117,11 +131,13 @@ func (h *Handshaker) Listen(ctx context.Context) {
h.metricsStages.RecordSignalingReceived()
}

h.updateRemoteICEState(&remoteOfferAnswer)

if h.relayListener != nil {
h.relayListener.Notify(&remoteOfferAnswer)
}

if h.iceListener != nil {
if h.iceListener != nil && h.RemoteICESupported() {
h.iceListener(&remoteOfferAnswer)
}
case <-ctx.Done():
Expand Down Expand Up @@ -183,15 +199,18 @@ func (h *Handshaker) sendAnswer() error {
}

func (h *Handshaker) buildOfferAnswer() OfferAnswer {
uFrag, pwd := h.ice.GetLocalUserCredentials()
sid := h.ice.SessionID()
answer := OfferAnswer{
IceCredentials: IceCredentials{uFrag, pwd},
WgListenPort: h.config.LocalWgPort,
Version: version.NetbirdVersion(),
RosenpassPubKey: h.config.RosenpassConfig.PubKey,
RosenpassAddr: h.config.RosenpassConfig.Addr,
SessionID: &sid,
}

if h.ice != nil && h.RemoteICESupported() {
uFrag, pwd := h.ice.GetLocalUserCredentials()
sid := h.ice.SessionID()
answer.IceCredentials = IceCredentials{uFrag, pwd}
answer.SessionID = &sid
}

if addr, err := h.relay.RelayInstanceAddress(); err == nil {
Expand All @@ -200,3 +219,18 @@ func (h *Handshaker) buildOfferAnswer() OfferAnswer {

return answer
}

func (h *Handshaker) updateRemoteICEState(offer *OfferAnswer) {
hasICE := offer.IceCredentials.UFrag != "" && offer.IceCredentials.Pwd != ""
prev := h.remoteICESupported.Swap(hasICE)
if prev != hasICE {
if hasICE {
h.log.Infof("remote peer started sending ICE credentials")
} else {
h.log.Infof("remote peer stopped sending ICE credentials")
if h.ice != nil {
h.ice.Close()
}
}
}
}
10 changes: 7 additions & 3 deletions client/internal/peer/signaler.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,9 +46,13 @@ func (s *Signaler) Ready() bool {

// SignalOfferAnswer signals either an offer or an answer to remote peer
func (s *Signaler) signalOfferAnswer(offerAnswer OfferAnswer, remoteKey string, bodyType sProto.Body_Type) error {
sessionIDBytes, err := offerAnswer.SessionID.Bytes()
if err != nil {
log.Warnf("failed to get session ID bytes: %v", err)
var sessionIDBytes []byte
if offerAnswer.SessionID != nil {
var err error
sessionIDBytes, err = offerAnswer.SessionID.Bytes()
if err != nil {
log.Warnf("failed to get session ID bytes: %v", err)
}
}
msg, err := signal.MarshalCredential(
s.wgPrivateKey,
Expand Down
Loading