[client] Extract Handshaker and Signaler into a signaling package

Move the signaling protocol out of the peer package into
client/internal/peer/signaling: OfferAnswer, IceCredentials, Handshaker
and Signaler. Break the peer<->signaling cycle by giving the Handshaker
a plain Config and an ICEWorker interface instead of *Conn/*WorkerICE,
and by passing the relay manager directly rather than the relay worker.
Combine the ICE worker's local-credentials and session-id accessors into
a single Credentials() returning a Credentials struct.

Move the ICE session id to the ice package as ice.SessionID, since it
identifies an ICE agent session and is minted there alongside
GenerateICECredentials; signaling only carries it.
This commit is contained in:
Zoltan Papp
2026-07-11 19:09:07 +02:00
parent 9e75d5c732
commit 917e85f648
10 changed files with 129 additions and 83 deletions
+8 -7
View File
@@ -48,6 +48,7 @@ import (
"github.com/netbirdio/netbird/client/internal/peer" "github.com/netbirdio/netbird/client/internal/peer"
"github.com/netbirdio/netbird/client/internal/peer/guard" "github.com/netbirdio/netbird/client/internal/peer/guard"
icemaker "github.com/netbirdio/netbird/client/internal/peer/ice" icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
"github.com/netbirdio/netbird/client/internal/peerstore" "github.com/netbirdio/netbird/client/internal/peerstore"
"github.com/netbirdio/netbird/client/internal/portforward" "github.com/netbirdio/netbird/client/internal/portforward"
"github.com/netbirdio/netbird/client/internal/profilemanager" "github.com/netbirdio/netbird/client/internal/profilemanager"
@@ -182,7 +183,7 @@ type EngineServices struct {
type Engine struct { type Engine struct {
// signal is a Signal Service client // signal is a Signal Service client
signal signal.Client signal signal.Client
signaler *peer.Signaler signaler *signaling.Signaler
// mgmClient is a Management Service client // mgmClient is a Management Service client
mgmClient mgm.Client mgmClient mgm.Client
// peerConns is a map that holds all the peers that are known to this peer // peerConns is a map that holds all the peers that are known to this peer
@@ -318,7 +319,7 @@ func NewEngine(
ctx: ctx, ctx: ctx,
cancel: cancel, cancel: cancel,
signal: services.SignalClient, signal: services.SignalClient,
signaler: peer.NewSignaler(services.SignalClient, config.WgPrivateKey), signaler: signaling.NewSignaler(services.SignalClient, config.WgPrivateKey),
mgmClient: services.MgmClient, mgmClient: services.MgmClient,
relayManager: services.RelayManager, relayManager: services.RelayManager,
peerStore: peerstore.NewConnStore(), peerStore: peerstore.NewConnStore(),
@@ -2747,7 +2748,7 @@ func createFile(path string) error {
return file.Close() return file.Close()
} }
func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) { func convertToOfferAnswer(msg *sProto.Message) (*signaling.OfferAnswer, error) {
remoteCred, err := signal.UnMarshalCredential(msg) remoteCred, err := signal.UnMarshalCredential(msg)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -2763,9 +2764,9 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
} }
// Handle optional SessionID // Handle optional SessionID
var sessionID *peer.ICESessionID var sessionID *icemaker.SessionID
if sessionBytes := msg.GetBody().GetSessionId(); sessionBytes != nil { if sessionBytes := msg.GetBody().GetSessionId(); sessionBytes != nil {
if id, err := peer.ICESessionIDFromBytes(sessionBytes); err != nil { if id, err := icemaker.SessionIDFromBytes(sessionBytes); err != nil {
log.Warnf("Invalid session ID in message: %v", err) log.Warnf("Invalid session ID in message: %v", err)
sessionID = nil // Set to nil if conversion fails sessionID = nil // Set to nil if conversion fails
} else { } else {
@@ -2775,8 +2776,8 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) {
relayIP := decodeRelayIP(msg.GetBody().GetRelayServerIP()) relayIP := decodeRelayIP(msg.GetBody().GetRelayServerIP())
offerAnswer := peer.OfferAnswer{ offerAnswer := signaling.OfferAnswer{
IceCredentials: peer.IceCredentials{ IceCredentials: signaling.IceCredentials{
UFrag: remoteCred.UFrag, UFrag: remoteCred.UFrag,
Pwd: remoteCred.Pwd, Pwd: remoteCred.Pwd,
}, },
+25 -15
View File
@@ -21,6 +21,7 @@ import (
"github.com/netbirdio/netbird/client/internal/peer/guard" "github.com/netbirdio/netbird/client/internal/peer/guard"
icemaker "github.com/netbirdio/netbird/client/internal/peer/ice" icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
"github.com/netbirdio/netbird/client/internal/peer/id" "github.com/netbirdio/netbird/client/internal/peer/id"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
"github.com/netbirdio/netbird/client/internal/peer/state_dump" "github.com/netbirdio/netbird/client/internal/peer/state_dump"
"github.com/netbirdio/netbird/client/internal/peer/status" "github.com/netbirdio/netbird/client/internal/peer/status"
"github.com/netbirdio/netbird/client/internal/peer/wg_watcher" "github.com/netbirdio/netbird/client/internal/peer/wg_watcher"
@@ -50,7 +51,7 @@ type MetricsRecorder interface {
type ServiceDependencies struct { type ServiceDependencies struct {
StatusRecorder *status.Recorder StatusRecorder *status.Recorder
Signaler *Signaler Signaler *signaling.Signaler
IFaceDiscover stdnet.ExternalIFaceDiscover IFaceDiscover stdnet.ExternalIFaceDiscover
RelayManager *relayClient.Manager RelayManager *relayClient.Manager
SrWatcher *guard.SRWatcher SrWatcher *guard.SRWatcher
@@ -109,7 +110,7 @@ type Conn struct {
ctxCancel context.CancelFunc ctxCancel context.CancelFunc
config ConnConfig config ConnConfig
statusRecorder *status.Recorder statusRecorder *status.Recorder
signaler *Signaler signaler *signaling.Signaler
iFaceDiscover stdnet.ExternalIFaceDiscover iFaceDiscover stdnet.ExternalIFaceDiscover
relayManager *relayClient.Manager relayManager *relayClient.Manager
srWatcher *guard.SRWatcher srWatcher *guard.SRWatcher
@@ -136,7 +137,7 @@ type Conn struct {
// dials spawned by the event loop, keeping only the newest offer while a // dials spawned by the event loop, keeping only the newest offer while a
// dial is running. // dial is running.
relayDialInFlight bool relayDialInFlight bool
pendingRelayOffer *OfferAnswer pendingRelayOffer *signaling.OfferAnswer
wgWatcher *wg_watcher.WGWatcher wgWatcher *wg_watcher.WGWatcher
wgWatcherWg sync.WaitGroup wgWatcherWg sync.WaitGroup
@@ -150,7 +151,7 @@ type Conn struct {
wgProxyICE wgproxy.Proxy wgProxyICE wgproxy.Proxy
wgProxyRelay wgproxy.Proxy wgProxyRelay wgproxy.Proxy
handshaker *Handshaker handshaker *signaling.Handshaker
guard *guard.Guard guard *guard.Guard
wg sync.WaitGroup wg sync.WaitGroup
@@ -236,7 +237,16 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error {
conn.workerICE = workerICE conn.workerICE = workerICE
} }
conn.handshaker = NewHandshaker(conn.Log, conn.config, conn.signaler, conn.workerICE, conn.workerRelay) var iceWorker signaling.ICEWorker
if conn.workerICE != nil {
iceWorker = conn.workerICE
}
conn.handshaker = signaling.NewHandshaker(conn.Log, signaling.Config{
Key: conn.config.Key,
LocalWgPort: conn.config.LocalWgPort,
RosenpassPubKey: conn.config.RosenpassConfig.PubKey,
RosenpassAddr: conn.config.RosenpassConfig.Addr,
}, conn.signaler, iceWorker, conn.relayManager)
conn.guard = guard.NewGuard(conn.Log, conn.isConnectedOnAllWay, conn.config.Timeout, conn.srWatcher) conn.guard = guard.NewGuard(conn.Log, conn.isConnectedOnAllWay, conn.config.Timeout, conn.srWatcher)
@@ -296,7 +306,7 @@ func (conn *Conn) Close(signalToRemote bool) {
// OnRemoteOffer handles an offer from the remote peer. It never blocks; the // OnRemoteOffer handles an offer from the remote peer. It never blocks; the
// offer is coalesced in the mailbox and processed by the event loop, older // offer is coalesced in the mailbox and processed by the event loop, older
// unprocessed offers are replaced by the newest one. // unprocessed offers are replaced by the newest one.
func (conn *Conn) OnRemoteOffer(offer OfferAnswer) { func (conn *Conn) OnRemoteOffer(offer signaling.OfferAnswer) {
conn.dumpState.RemoteOffer() conn.dumpState.RemoteOffer()
conn.Log.Infof("OnRemoteOffer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay) conn.Log.Infof("OnRemoteOffer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay)
if !conn.post(evRemoteOffer{offer: offer}) { if !conn.post(evRemoteOffer{offer: offer}) {
@@ -306,7 +316,7 @@ func (conn *Conn) OnRemoteOffer(offer OfferAnswer) {
// OnRemoteAnswer handles an answer from the remote peer. It never blocks; the // OnRemoteAnswer handles an answer from the remote peer. It never blocks; the
// answer is coalesced in the mailbox and processed by the event loop. // answer is coalesced in the mailbox and processed by the event loop.
func (conn *Conn) OnRemoteAnswer(answer OfferAnswer) { func (conn *Conn) OnRemoteAnswer(answer signaling.OfferAnswer) {
conn.dumpState.RemoteAnswer() conn.dumpState.RemoteAnswer()
conn.Log.Infof("OnRemoteAnswer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay) conn.Log.Infof("OnRemoteAnswer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay)
if !conn.post(evRemoteAnswer{answer: answer}) { if !conn.post(evRemoteAnswer{answer: answer}) {
@@ -511,11 +521,11 @@ func (conn *Conn) releaseEvents(evs []event) {
// handleRemoteOffer applies a remote offer on the event loop: refreshes the // handleRemoteOffer applies a remote offer on the event loop: refreshes the
// remote ICE support state, dispatches the offer to the relay and ICE workers // remote ICE support state, dispatches the offer to the relay and ICE workers
// and answers it. // and answers it.
func (conn *Conn) handleRemoteOffer(offer *OfferAnswer) { func (conn *Conn) handleRemoteOffer(offer *signaling.OfferAnswer) {
conn.Log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", offer.Version, offer.WgListenPort, offer.SessionIDString(), offer.hasICECredentials()) conn.Log.Infof("received offer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", offer.Version, offer.WgListenPort, offer.SessionIDString(), offer.HasICECredentials())
conn.metricsStages.RecordSignalingReceived() conn.metricsStages.RecordSignalingReceived()
conn.handshaker.updateRemoteICEState(offer) conn.handshaker.UpdateRemoteICEState(offer)
conn.dispatchOfferToRelay(offer) conn.dispatchOfferToRelay(offer)
conn.dispatchOfferToICE(offer) conn.dispatchOfferToICE(offer)
@@ -528,11 +538,11 @@ func (conn *Conn) handleRemoteOffer(offer *OfferAnswer) {
// handleRemoteAnswer applies a remote answer on the event loop the same way // handleRemoteAnswer applies a remote answer on the event loop the same way
// as an offer, without answering it. // as an offer, without answering it.
func (conn *Conn) handleRemoteAnswer(answer *OfferAnswer) { func (conn *Conn) handleRemoteAnswer(answer *signaling.OfferAnswer) {
conn.Log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", answer.Version, answer.WgListenPort, answer.SessionIDString(), answer.hasICECredentials()) conn.Log.Infof("received answer, running version %s, remote WireGuard listen port %d, session id: %s, remote ICE supported: %t", answer.Version, answer.WgListenPort, answer.SessionIDString(), answer.HasICECredentials())
conn.metricsStages.RecordSignalingReceived() conn.metricsStages.RecordSignalingReceived()
conn.handshaker.updateRemoteICEState(answer) conn.handshaker.UpdateRemoteICEState(answer)
conn.dispatchOfferToRelay(answer) conn.dispatchOfferToRelay(answer)
conn.dispatchOfferToICE(answer) conn.dispatchOfferToICE(answer)
} }
@@ -544,7 +554,7 @@ func (conn *Conn) handleRemoteCandidate(e evRemoteCandidate) {
conn.workerICE.OnRemoteCandidate(e.candidate, e.haRoutes) conn.workerICE.OnRemoteCandidate(e.candidate, e.haRoutes)
} }
func (conn *Conn) dispatchOfferToICE(offer *OfferAnswer) { func (conn *Conn) dispatchOfferToICE(offer *signaling.OfferAnswer) {
if conn.workerICE == nil || !conn.handshaker.RemoteICESupported() { if conn.workerICE == nil || !conn.handshaker.RemoteICESupported() {
return return
} }
@@ -554,7 +564,7 @@ func (conn *Conn) dispatchOfferToICE(offer *OfferAnswer) {
// dispatchOfferToRelay runs the blocking relay dial on a helper goroutine. A // dispatchOfferToRelay runs the blocking relay dial on a helper goroutine. A
// single dial is kept in flight; newer offers replace the pending one and the // single dial is kept in flight; newer offers replace the pending one and the
// newest is dispatched once the running dial reports completion. // newest is dispatched once the running dial reports completion.
func (conn *Conn) dispatchOfferToRelay(offer *OfferAnswer) { func (conn *Conn) dispatchOfferToRelay(offer *signaling.OfferAnswer) {
if conn.relayDialInFlight { if conn.relayDialInFlight {
conn.pendingRelayOffer = offer conn.pendingRelayOffer = offer
return return
+3 -2
View File
@@ -15,6 +15,7 @@ import (
"github.com/netbirdio/netbird/client/iface" "github.com/netbirdio/netbird/client/iface"
"github.com/netbirdio/netbird/client/internal/peer/guard" "github.com/netbirdio/netbird/client/internal/peer/guard"
"github.com/netbirdio/netbird/client/internal/peer/ice" "github.com/netbirdio/netbird/client/internal/peer/ice"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
"github.com/netbirdio/netbird/client/internal/peer/status" "github.com/netbirdio/netbird/client/internal/peer/status"
"github.com/netbirdio/netbird/client/internal/stdnet" "github.com/netbirdio/netbird/client/internal/stdnet"
"github.com/netbirdio/netbird/util" "github.com/netbirdio/netbird/util"
@@ -76,8 +77,8 @@ func TestConn_DiscardMessagesWhenNotOpened(t *testing.T) {
conn, err := NewConn(connConf, sd) conn, err := NewConn(connConf, sd)
require.NoError(t, err) require.NoError(t, err)
offerAnswer := OfferAnswer{ offerAnswer := signaling.OfferAnswer{
IceCredentials: IceCredentials{ IceCredentials: signaling.IceCredentials{
UFrag: "test", UFrag: "test",
Pwd: "test", Pwd: "test",
}, },
+3 -2
View File
@@ -5,6 +5,7 @@ import (
"github.com/pion/ice/v4" "github.com/pion/ice/v4"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
"github.com/netbirdio/netbird/route" "github.com/netbirdio/netbird/route"
) )
@@ -21,11 +22,11 @@ type evClose struct {
} }
type evRemoteOffer struct { type evRemoteOffer struct {
offer OfferAnswer offer signaling.OfferAnswer
} }
type evRemoteAnswer struct { type evRemoteAnswer struct {
answer OfferAnswer answer signaling.OfferAnswer
} }
type evRemoteCandidate struct { type evRemoteCandidate struct {
@@ -1,4 +1,4 @@
package peer package ice
import ( import (
"crypto/rand" "crypto/rand"
@@ -9,26 +9,26 @@ import (
const sessionIDSize = 5 const sessionIDSize = 5
type ICESessionID string type SessionID string
// NewICESessionID generates a new session ID for distinguishing sessions // NewSessionID generates a new session ID for distinguishing sessions
func NewICESessionID() (ICESessionID, error) { func NewSessionID() (SessionID, error) {
b := make([]byte, sessionIDSize) b := make([]byte, sessionIDSize)
if _, err := io.ReadFull(rand.Reader, b); err != nil { if _, err := io.ReadFull(rand.Reader, b); err != nil {
return "", fmt.Errorf("failed to generate session ID: %w", err) return "", fmt.Errorf("failed to generate session ID: %w", err)
} }
return ICESessionID(hex.EncodeToString(b)), nil return SessionID(hex.EncodeToString(b)), nil
} }
func ICESessionIDFromBytes(b []byte) (ICESessionID, error) { func SessionIDFromBytes(b []byte) (SessionID, error) {
if len(b) != sessionIDSize { if len(b) != sessionIDSize {
return "", fmt.Errorf("invalid session ID length: %d", len(b)) return "", fmt.Errorf("invalid session ID length: %d", len(b))
} }
return ICESessionID(hex.EncodeToString(b)), nil return SessionID(hex.EncodeToString(b)), nil
} }
// Bytes returns the raw bytes of the session ID for protobuf serialization // Bytes returns the raw bytes of the session ID for protobuf serialization
func (id ICESessionID) Bytes() ([]byte, error) { func (id SessionID) Bytes() ([]byte, error) {
if len(id) == 0 { if len(id) == 0 {
return nil, fmt.Errorf("ICE session ID is empty") return nil, fmt.Errorf("ICE session ID is empty")
} }
@@ -42,6 +42,6 @@ func (id ICESessionID) Bytes() ([]byte, error) {
return b, nil return b, nil
} }
func (id ICESessionID) String() string { func (id SessionID) String() string {
return string(id) return string(id)
} }
+8 -7
View File
@@ -3,6 +3,7 @@ package peer
import ( import (
"testing" "testing"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
"github.com/stretchr/testify/assert" "github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
@@ -10,9 +11,9 @@ import (
func TestMailbox_OfferCoalescing(t *testing.T) { func TestMailbox_OfferCoalescing(t *testing.T) {
mb := newMailbox() mb := newMailbox()
require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{WgListenPort: 1}})) require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 1}}))
require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{WgListenPort: 2}})) require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 2}}))
require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{WgListenPort: 3}})) require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 3}}))
evs := mb.drain() evs := mb.drain()
require.Len(t, evs, 1, "consecutive offers must coalesce to a single event") require.Len(t, evs, 1, "consecutive offers must coalesce to a single event")
@@ -26,7 +27,7 @@ func TestMailbox_OfferFlushesCandidates(t *testing.T) {
require.True(t, mb.post(evRemoteCandidate{})) require.True(t, mb.post(evRemoteCandidate{}))
require.True(t, mb.post(evRemoteCandidate{})) require.True(t, mb.post(evRemoteCandidate{}))
require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{}})) require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
evs := mb.drain() evs := mb.drain()
require.Len(t, evs, 1, "candidates of the superseded session must be flushed") require.Len(t, evs, 1, "candidates of the superseded session must be flushed")
@@ -37,7 +38,7 @@ func TestMailbox_OfferFlushesCandidates(t *testing.T) {
func TestMailbox_CandidatesKeepOrderAfterOffer(t *testing.T) { func TestMailbox_CandidatesKeepOrderAfterOffer(t *testing.T) {
mb := newMailbox() mb := newMailbox()
require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{}})) require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
require.True(t, mb.post(evRemoteCandidate{haRoutes: nil})) require.True(t, mb.post(evRemoteCandidate{haRoutes: nil}))
require.True(t, mb.post(evRemoteCandidate{haRoutes: nil})) require.True(t, mb.post(evRemoteCandidate{haRoutes: nil}))
@@ -66,8 +67,8 @@ func TestMailbox_DrainOrder(t *testing.T) {
mb := newMailbox() mb := newMailbox()
require.True(t, mb.post(evGuardTick{})) require.True(t, mb.post(evGuardTick{}))
require.True(t, mb.post(evRemoteAnswer{answer: OfferAnswer{}})) require.True(t, mb.post(evRemoteAnswer{answer: signaling.OfferAnswer{}}))
require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{}})) require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}}))
require.True(t, mb.post(evRelayDown{})) require.True(t, mb.post(evRelayDown{}))
require.True(t, mb.post(evICEDown{sessionChanged: true})) require.True(t, mb.post(evICEDown{sessionChanged: true}))
require.True(t, mb.post(evClose{})) require.True(t, mb.post(evClose{}))
@@ -1,4 +1,4 @@
package peer package signaling
import ( import (
"errors" "errors"
@@ -8,6 +8,8 @@ import (
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
relayClient "github.com/netbirdio/netbird/shared/relay/client"
"github.com/netbirdio/netbird/version" "github.com/netbirdio/netbird/version"
) )
@@ -45,10 +47,10 @@ type OfferAnswer struct {
// fails. Zero value if the peer did not advertise an IP. // fails. Zero value if the peer did not advertise an IP.
RelaySrvIP netip.Addr RelaySrvIP netip.Addr
// SessionID is the unique identifier of the session, used to discard old messages // SessionID is the unique identifier of the session, used to discard old messages
SessionID *ICESessionID SessionID *icemaker.SessionID
} }
func (o *OfferAnswer) hasICECredentials() bool { func (o *OfferAnswer) HasICECredentials() bool {
return o.IceCredentials.UFrag != "" && o.IceCredentials.Pwd != "" return o.IceCredentials.UFrag != "" && o.IceCredentials.Pwd != ""
} }
@@ -59,29 +61,51 @@ func (o *OfferAnswer) SessionIDString() string {
return o.SessionID.String() return o.SessionID.String()
} }
// Config carries the peer-specific values the Handshaker embeds into offers
// and answers.
type Config struct {
Key string
LocalWgPort int
RosenpassPubKey []byte
RosenpassAddr string
}
// Credentials are the local ICE credentials and session id the Handshaker embeds in offers.
type Credentials struct {
UFrag string
Pwd string
SessionID icemaker.SessionID
}
// ICEWorker is the subset of the ICE worker the Handshaker needs to build offers.
type ICEWorker interface {
Credentials() Credentials
Close()
}
// Handshaker keeps the signaling protocol logic: building and sending offers // Handshaker keeps the signaling protocol logic: building and sending offers
// and answers and tracking whether the remote peer supports ICE. Incoming // and answers and tracking whether the remote peer supports ICE. Incoming
// message processing is driven by the Conn event loop. // message processing is driven by the Conn event loop.
type Handshaker struct { type Handshaker struct {
mu sync.Mutex mu sync.Mutex
log *log.Entry log *log.Entry
config ConnConfig config Config
signaler *Signaler signaler *Signaler
ice *WorkerICE ice ICEWorker
relay *WorkerRelay relayManager *relayClient.Manager
// remoteICESupported tracks whether the remote peer includes ICE credentials in its offers/answers. // remoteICESupported tracks whether the remote peer includes ICE credentials in its offers/answers.
// When false, the local side skips ICE dispatch and suppresses ICE credentials in responses. // When false, the local side skips ICE dispatch and suppresses ICE credentials in responses.
remoteICESupported atomic.Bool remoteICESupported atomic.Bool
} }
func NewHandshaker(log *log.Entry, config ConnConfig, signaler *Signaler, ice *WorkerICE, relay *WorkerRelay) *Handshaker { func NewHandshaker(log *log.Entry, config Config, signaler *Signaler, ice ICEWorker, relayManager *relayClient.Manager) *Handshaker {
h := &Handshaker{ h := &Handshaker{
log: log, log: log,
config: config, config: config,
signaler: signaler, signaler: signaler,
ice: ice, ice: ice,
relay: relay, relayManager: relayManager,
} }
// assume remote supports ICE until we learn otherwise from received offers // assume remote supports ICE until we learn otherwise from received offers
h.remoteICESupported.Store(ice != nil) h.remoteICESupported.Store(ice != nil)
@@ -127,18 +151,18 @@ func (h *Handshaker) buildOfferAnswer() OfferAnswer {
answer := OfferAnswer{ answer := OfferAnswer{
WgListenPort: h.config.LocalWgPort, WgListenPort: h.config.LocalWgPort,
Version: version.NetbirdVersion(), Version: version.NetbirdVersion(),
RosenpassPubKey: h.config.RosenpassConfig.PubKey, RosenpassPubKey: h.config.RosenpassPubKey,
RosenpassAddr: h.config.RosenpassConfig.Addr, RosenpassAddr: h.config.RosenpassAddr,
} }
if h.ice != nil && h.RemoteICESupported() { if h.ice != nil && h.RemoteICESupported() {
uFrag, pwd := h.ice.GetLocalUserCredentials() creds := h.ice.Credentials()
sid := h.ice.SessionID() answer.IceCredentials = IceCredentials{creds.UFrag, creds.Pwd}
answer.IceCredentials = IceCredentials{uFrag, pwd} sid := creds.SessionID
answer.SessionID = &sid answer.SessionID = &sid
} }
if addr, ip, err := h.relay.RelayInstanceAddress(); err == nil { if addr, ip, err := h.relayManager.RelayInstanceAddress(); err == nil {
answer.RelaySrvAddress = addr answer.RelaySrvAddress = addr
answer.RelaySrvIP = ip answer.RelaySrvIP = ip
} }
@@ -146,11 +170,11 @@ func (h *Handshaker) buildOfferAnswer() OfferAnswer {
return answer return answer
} }
// updateRemoteICEState refreshes the remote ICE support flag from a received // UpdateRemoteICEState refreshes the remote ICE support flag from a received
// offer or answer and closes the ICE worker when the remote peer stopped // offer or answer and closes the ICE worker when the remote peer stopped
// sending ICE credentials. Runs on the Conn event loop. // sending ICE credentials. Runs on the Conn event loop.
func (h *Handshaker) updateRemoteICEState(offer *OfferAnswer) { func (h *Handshaker) UpdateRemoteICEState(offer *OfferAnswer) {
hasICE := offer.hasICECredentials() hasICE := offer.HasICECredentials()
prev := h.remoteICESupported.Swap(hasICE) prev := h.remoteICESupported.Swap(hasICE)
if prev != hasICE { if prev != hasICE {
if hasICE { if hasICE {
@@ -1,4 +1,4 @@
package peer package signaling
import ( import (
"github.com/pion/ice/v4" "github.com/pion/ice/v4"
+21 -14
View File
@@ -14,6 +14,7 @@ import (
"github.com/netbirdio/netbird/client/iface" "github.com/netbirdio/netbird/client/iface"
"github.com/netbirdio/netbird/client/iface/udpmux" "github.com/netbirdio/netbird/client/iface/udpmux"
icemaker "github.com/netbirdio/netbird/client/internal/peer/ice" icemaker "github.com/netbirdio/netbird/client/internal/peer/ice"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
"github.com/netbirdio/netbird/client/internal/peer/status" "github.com/netbirdio/netbird/client/internal/peer/status"
"github.com/netbirdio/netbird/client/internal/portforward" "github.com/netbirdio/netbird/client/internal/portforward"
"github.com/netbirdio/netbird/client/internal/stdnet" "github.com/netbirdio/netbird/client/internal/stdnet"
@@ -42,7 +43,7 @@ type WorkerICE struct {
log *log.Entry log *log.Entry
config ConnConfig config ConnConfig
conn iceCallbacks conn iceCallbacks
signaler *Signaler signaler *signaling.Signaler
iFaceDiscover stdnet.ExternalIFaceDiscover iFaceDiscover stdnet.ExternalIFaceDiscover
statusRecorder *status.Recorder statusRecorder *status.Recorder
portForwardManager *portforward.Manager portForwardManager *portforward.Manager
@@ -55,12 +56,12 @@ type WorkerICE struct {
// connectedAgent is the agent whose connection was last reported ready; guarded by muxAgent // connectedAgent is the agent whose connection was last reported ready; guarded by muxAgent
connectedAgent *icemaker.ThreadSafeAgent connectedAgent *icemaker.ThreadSafeAgent
// remoteSessionID represents the peer's session identifier from the latest remote offer. // remoteSessionID represents the peer's session identifier from the latest remote offer.
remoteSessionID ICESessionID remoteSessionID icemaker.SessionID
// sessionID is used to track the current session ID of the ICE agent // sessionID is used to track the current session ID of the ICE agent
// increase by one when disconnecting the agent // increase by one when disconnecting the agent
// with it the remote peer can discard the already deprecated offer/answer // with it the remote peer can discard the already deprecated offer/answer
// Without it the remote peer may recreate a workable ICE connection // Without it the remote peer may recreate a workable ICE connection
sessionID ICESessionID sessionID icemaker.SessionID
remoteSessionChanged bool remoteSessionChanged bool
muxAgent sync.Mutex muxAgent sync.Mutex
@@ -71,8 +72,8 @@ type WorkerICE struct {
portForwardAttempted bool portForwardAttempted bool
} }
func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn iceCallbacks, signaler *Signaler, ifaceDiscover stdnet.ExternalIFaceDiscover, statusRecorder *status.Recorder, portForwardManager *portforward.Manager, hasRelayOnLocally bool) (*WorkerICE, error) { func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn iceCallbacks, signaler *signaling.Signaler, ifaceDiscover stdnet.ExternalIFaceDiscover, statusRecorder *status.Recorder, portForwardManager *portforward.Manager, hasRelayOnLocally bool) (*WorkerICE, error) {
sessionID, err := NewICESessionID() sessionID, err := icemaker.NewSessionID()
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -99,7 +100,7 @@ func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn i
return w, nil return w, nil
} }
func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) { func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
w.log.Debugf("OnNewOffer for ICE, serial: %s", remoteOfferAnswer.SessionIDString()) w.log.Debugf("OnNewOffer for ICE, serial: %s", remoteOfferAnswer.SessionIDString())
w.muxAgent.Lock() w.muxAgent.Lock()
defer w.muxAgent.Unlock() defer w.muxAgent.Unlock()
@@ -123,7 +124,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) {
} }
} }
sessionID, err := NewICESessionID() sessionID, err := icemaker.NewSessionID()
if err != nil { if err != nil {
w.log.Errorf("failed to create new session ID: %s", err) w.log.Errorf("failed to create new session ID: %s", err)
} }
@@ -190,8 +191,14 @@ func (w *WorkerICE) OnRemoteCandidate(candidate ice.Candidate, haRoutes route.HA
} }
} }
func (w *WorkerICE) GetLocalUserCredentials() (frag string, pwd string) { func (w *WorkerICE) Credentials() signaling.Credentials {
return w.localUfrag, w.localPwd w.muxAgent.Lock()
defer w.muxAgent.Unlock()
return signaling.Credentials{
UFrag: w.localUfrag,
Pwd: w.localPwd,
SessionID: w.sessionID,
}
} }
func (w *WorkerICE) InProgress() bool { func (w *WorkerICE) InProgress() bool {
@@ -242,7 +249,7 @@ func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates []
return agent, nil return agent, nil
} }
func (w *WorkerICE) SessionID() ICESessionID { func (w *WorkerICE) getSessionID() icemaker.SessionID {
w.muxAgent.Lock() w.muxAgent.Lock()
defer w.muxAgent.Unlock() defer w.muxAgent.Unlock()
@@ -252,7 +259,7 @@ func (w *WorkerICE) SessionID() ICESessionID {
// will block until connection succeeded // will block until connection succeeded
// but it won't release if ICE Agent went into Disconnected or Failed state, // but it won't release if ICE Agent went into Disconnected or Failed state,
// so we have to cancel it with the provided context once agent detected a broken connection // so we have to cancel it with the provided context once agent detected a broken connection
func (w *WorkerICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) { func (w *WorkerICE) connect(ctx context.Context, dialerCancel context.CancelFunc, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) {
w.log.Debugf("gather candidates") w.log.Debugf("gather candidates")
if err := agent.GatherCandidates(); err != nil { if err := agent.GatherCandidates(); err != nil {
w.log.Warnf("failed to gather candidates: %s", err) w.log.Warnf("failed to gather candidates: %s", err)
@@ -333,7 +340,7 @@ func (w *WorkerICE) closeAgent(agent *icemaker.ThreadSafeAgent, cancel context.C
if w.agent == agent { if w.agent == agent {
// consider to remove from here and move to the OnNewOffer // consider to remove from here and move to the OnNewOffer
sessionID, err := NewICESessionID() sessionID, err := icemaker.NewSessionID()
if err != nil { if err != nil {
w.log.Errorf("failed to create new session ID: %s", err) w.log.Errorf("failed to create new session ID: %s", err)
} }
@@ -488,7 +495,7 @@ func (w *WorkerICE) onICESelectedCandidatePair(agent *icemaker.ThreadSafeAgent,
} }
func (w *WorkerICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) { func (w *WorkerICE) logSuccessfulPaths(agent *icemaker.ThreadSafeAgent) {
sessionID := w.SessionID() sessionID := w.getSessionID()
stats := agent.GetCandidatePairsStats() stats := agent.GetCandidatePairsStats()
localCandidates, _ := agent.GetLocalCandidates() localCandidates, _ := agent.GetLocalCandidates()
remoteCandidates, _ := agent.GetRemoteCandidates() remoteCandidates, _ := agent.GetRemoteCandidates()
@@ -554,7 +561,7 @@ func (w *WorkerICE) onConnectionStateChange(agent *icemaker.ThreadSafeAgent, dia
} }
} }
func (w *WorkerICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *OfferAnswer) (*ice.Conn, error) { func (w *WorkerICE) turnAgentDial(ctx context.Context, agent *icemaker.ThreadSafeAgent, remoteOfferAnswer *signaling.OfferAnswer) (*ice.Conn, error) {
if isController(w.config) { if isController(w.config) {
return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd) return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd)
} else { } else {
+3 -2
View File
@@ -10,6 +10,7 @@ import (
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
"github.com/netbirdio/netbird/client/internal/peer/signaling"
relayClient "github.com/netbirdio/netbird/shared/relay/client" relayClient "github.com/netbirdio/netbird/shared/relay/client"
) )
@@ -50,7 +51,7 @@ func NewWorkerRelay(ctx context.Context, log *log.Entry, ctrl bool, config ConnC
return r return r
} }
func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) { func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) {
if !w.isRelaySupported(remoteOfferAnswer) { if !w.isRelaySupported(remoteOfferAnswer) {
w.log.Infof("Relay is not supported by remote peer") w.log.Infof("Relay is not supported by remote peer")
w.relaySupportedOnRemotePeer.Store(false) w.relaySupportedOnRemotePeer.Store(false)
@@ -124,7 +125,7 @@ func (w *WorkerRelay) CloseConn() {
} }
} }
func (w *WorkerRelay) isRelaySupported(answer *OfferAnswer) bool { func (w *WorkerRelay) isRelaySupported(answer *signaling.OfferAnswer) bool {
if !w.relayManager.HasRelayAddress() { if !w.relayManager.HasRelayAddress() {
return false return false
} }