diff --git a/client/internal/engine.go b/client/internal/engine.go index 7b2fc7b26..c86d9ce39 100644 --- a/client/internal/engine.go +++ b/client/internal/engine.go @@ -48,6 +48,7 @@ import ( "github.com/netbirdio/netbird/client/internal/peer" "github.com/netbirdio/netbird/client/internal/peer/guard" 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/portforward" "github.com/netbirdio/netbird/client/internal/profilemanager" @@ -182,7 +183,7 @@ type EngineServices struct { type Engine struct { // signal is a Signal Service client signal signal.Client - signaler *peer.Signaler + signaler *signaling.Signaler // mgmClient is a Management Service client mgmClient mgm.Client // peerConns is a map that holds all the peers that are known to this peer @@ -318,7 +319,7 @@ func NewEngine( ctx: ctx, cancel: cancel, signal: services.SignalClient, - signaler: peer.NewSignaler(services.SignalClient, config.WgPrivateKey), + signaler: signaling.NewSignaler(services.SignalClient, config.WgPrivateKey), mgmClient: services.MgmClient, relayManager: services.RelayManager, peerStore: peerstore.NewConnStore(), @@ -2747,7 +2748,7 @@ func createFile(path string) error { return file.Close() } -func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) { +func convertToOfferAnswer(msg *sProto.Message) (*signaling.OfferAnswer, error) { remoteCred, err := signal.UnMarshalCredential(msg) if err != nil { return nil, err @@ -2763,9 +2764,9 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) { } // Handle optional SessionID - var sessionID *peer.ICESessionID + var sessionID *icemaker.SessionID 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) sessionID = nil // Set to nil if conversion fails } else { @@ -2775,8 +2776,8 @@ func convertToOfferAnswer(msg *sProto.Message) (*peer.OfferAnswer, error) { relayIP := decodeRelayIP(msg.GetBody().GetRelayServerIP()) - offerAnswer := peer.OfferAnswer{ - IceCredentials: peer.IceCredentials{ + offerAnswer := signaling.OfferAnswer{ + IceCredentials: signaling.IceCredentials{ UFrag: remoteCred.UFrag, Pwd: remoteCred.Pwd, }, diff --git a/client/internal/peer/conn.go b/client/internal/peer/conn.go index 649010b3b..315567263 100644 --- a/client/internal/peer/conn.go +++ b/client/internal/peer/conn.go @@ -21,6 +21,7 @@ import ( "github.com/netbirdio/netbird/client/internal/peer/guard" icemaker "github.com/netbirdio/netbird/client/internal/peer/ice" "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/status" "github.com/netbirdio/netbird/client/internal/peer/wg_watcher" @@ -50,7 +51,7 @@ type MetricsRecorder interface { type ServiceDependencies struct { StatusRecorder *status.Recorder - Signaler *Signaler + Signaler *signaling.Signaler IFaceDiscover stdnet.ExternalIFaceDiscover RelayManager *relayClient.Manager SrWatcher *guard.SRWatcher @@ -109,7 +110,7 @@ type Conn struct { ctxCancel context.CancelFunc config ConnConfig statusRecorder *status.Recorder - signaler *Signaler + signaler *signaling.Signaler iFaceDiscover stdnet.ExternalIFaceDiscover relayManager *relayClient.Manager srWatcher *guard.SRWatcher @@ -136,7 +137,7 @@ type Conn struct { // dials spawned by the event loop, keeping only the newest offer while a // dial is running. relayDialInFlight bool - pendingRelayOffer *OfferAnswer + pendingRelayOffer *signaling.OfferAnswer wgWatcher *wg_watcher.WGWatcher wgWatcherWg sync.WaitGroup @@ -150,7 +151,7 @@ type Conn struct { wgProxyICE wgproxy.Proxy wgProxyRelay wgproxy.Proxy - handshaker *Handshaker + handshaker *signaling.Handshaker guard *guard.Guard wg sync.WaitGroup @@ -236,7 +237,16 @@ func (conn *Conn) open(engineCtx context.Context, firstPacket []byte) error { 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) @@ -296,7 +306,7 @@ func (conn *Conn) Close(signalToRemote bool) { // 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 // 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.Log.Infof("OnRemoteOffer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay) 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 // 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.Log.Infof("OnRemoteAnswer, on status ICE: %s, status Relay: %s", conn.statusICE, conn.statusRelay) 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 // remote ICE support state, dispatches the offer to the relay and ICE workers // and answers it. -func (conn *Conn) handleRemoteOffer(offer *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()) +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.metricsStages.RecordSignalingReceived() - conn.handshaker.updateRemoteICEState(offer) + conn.handshaker.UpdateRemoteICEState(offer) conn.dispatchOfferToRelay(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 // as an offer, without answering it. -func (conn *Conn) handleRemoteAnswer(answer *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()) +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.metricsStages.RecordSignalingReceived() - conn.handshaker.updateRemoteICEState(answer) + conn.handshaker.UpdateRemoteICEState(answer) conn.dispatchOfferToRelay(answer) conn.dispatchOfferToICE(answer) } @@ -544,7 +554,7 @@ func (conn *Conn) handleRemoteCandidate(e evRemoteCandidate) { 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() { return } @@ -554,7 +564,7 @@ func (conn *Conn) dispatchOfferToICE(offer *OfferAnswer) { // 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 // 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 { conn.pendingRelayOffer = offer return diff --git a/client/internal/peer/conn_test.go b/client/internal/peer/conn_test.go index 09a382b28..bcb6be5ba 100644 --- a/client/internal/peer/conn_test.go +++ b/client/internal/peer/conn_test.go @@ -15,6 +15,7 @@ import ( "github.com/netbirdio/netbird/client/iface" "github.com/netbirdio/netbird/client/internal/peer/guard" "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/stdnet" "github.com/netbirdio/netbird/util" @@ -76,8 +77,8 @@ func TestConn_DiscardMessagesWhenNotOpened(t *testing.T) { conn, err := NewConn(connConf, sd) require.NoError(t, err) - offerAnswer := OfferAnswer{ - IceCredentials: IceCredentials{ + offerAnswer := signaling.OfferAnswer{ + IceCredentials: signaling.IceCredentials{ UFrag: "test", Pwd: "test", }, diff --git a/client/internal/peer/event.go b/client/internal/peer/event.go index f05e6f80a..2e893fc17 100644 --- a/client/internal/peer/event.go +++ b/client/internal/peer/event.go @@ -5,6 +5,7 @@ import ( "github.com/pion/ice/v4" + "github.com/netbirdio/netbird/client/internal/peer/signaling" "github.com/netbirdio/netbird/route" ) @@ -21,11 +22,11 @@ type evClose struct { } type evRemoteOffer struct { - offer OfferAnswer + offer signaling.OfferAnswer } type evRemoteAnswer struct { - answer OfferAnswer + answer signaling.OfferAnswer } type evRemoteCandidate struct { diff --git a/client/internal/peer/session_id.go b/client/internal/peer/ice/session_id.go similarity index 66% rename from client/internal/peer/session_id.go rename to client/internal/peer/ice/session_id.go index 4f630adc0..98af9e131 100644 --- a/client/internal/peer/session_id.go +++ b/client/internal/peer/ice/session_id.go @@ -1,4 +1,4 @@ -package peer +package ice import ( "crypto/rand" @@ -9,26 +9,26 @@ import ( const sessionIDSize = 5 -type ICESessionID string +type SessionID string -// NewICESessionID generates a new session ID for distinguishing sessions -func NewICESessionID() (ICESessionID, error) { +// NewSessionID generates a new session ID for distinguishing sessions +func NewSessionID() (SessionID, error) { b := make([]byte, sessionIDSize) if _, err := io.ReadFull(rand.Reader, b); err != nil { 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 { 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 -func (id ICESessionID) Bytes() ([]byte, error) { +func (id SessionID) Bytes() ([]byte, error) { if len(id) == 0 { return nil, fmt.Errorf("ICE session ID is empty") } @@ -42,6 +42,6 @@ func (id ICESessionID) Bytes() ([]byte, error) { return b, nil } -func (id ICESessionID) String() string { +func (id SessionID) String() string { return string(id) } diff --git a/client/internal/peer/mailbox_test.go b/client/internal/peer/mailbox_test.go index 287647bc4..20e509886 100644 --- a/client/internal/peer/mailbox_test.go +++ b/client/internal/peer/mailbox_test.go @@ -3,6 +3,7 @@ package peer import ( "testing" + "github.com/netbirdio/netbird/client/internal/peer/signaling" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -10,9 +11,9 @@ import ( func TestMailbox_OfferCoalescing(t *testing.T) { mb := newMailbox() - require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{WgListenPort: 1}})) - require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{WgListenPort: 2}})) - require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{WgListenPort: 3}})) + require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 1}})) + require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 2}})) + require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{WgListenPort: 3}})) evs := mb.drain() 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(evRemoteOffer{offer: OfferAnswer{}})) + require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}})) evs := mb.drain() 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) { 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})) @@ -66,8 +67,8 @@ func TestMailbox_DrainOrder(t *testing.T) { mb := newMailbox() require.True(t, mb.post(evGuardTick{})) - require.True(t, mb.post(evRemoteAnswer{answer: OfferAnswer{}})) - require.True(t, mb.post(evRemoteOffer{offer: OfferAnswer{}})) + require.True(t, mb.post(evRemoteAnswer{answer: signaling.OfferAnswer{}})) + require.True(t, mb.post(evRemoteOffer{offer: signaling.OfferAnswer{}})) require.True(t, mb.post(evRelayDown{})) require.True(t, mb.post(evICEDown{sessionChanged: true})) require.True(t, mb.post(evClose{})) diff --git a/client/internal/peer/handshaker.go b/client/internal/peer/signaling/handshaker.go similarity index 70% rename from client/internal/peer/handshaker.go rename to client/internal/peer/signaling/handshaker.go index 73b8445ca..6e774f2c8 100644 --- a/client/internal/peer/handshaker.go +++ b/client/internal/peer/signaling/handshaker.go @@ -1,4 +1,4 @@ -package peer +package signaling import ( "errors" @@ -8,6 +8,8 @@ import ( 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" ) @@ -45,10 +47,10 @@ type OfferAnswer struct { // fails. Zero value if the peer did not advertise an IP. RelaySrvIP netip.Addr // 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 != "" } @@ -59,29 +61,51 @@ func (o *OfferAnswer) SessionIDString() 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 // and answers and tracking whether the remote peer supports ICE. Incoming // message processing is driven by the Conn event loop. type Handshaker struct { - mu sync.Mutex - log *log.Entry - config ConnConfig - signaler *Signaler - ice *WorkerICE - relay *WorkerRelay + mu sync.Mutex + log *log.Entry + config Config + signaler *Signaler + ice ICEWorker + relayManager *relayClient.Manager // 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. 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{ - log: log, - config: config, - signaler: signaler, - ice: ice, - relay: relay, + log: log, + config: config, + signaler: signaler, + ice: ice, + relayManager: relayManager, } // assume remote supports ICE until we learn otherwise from received offers h.remoteICESupported.Store(ice != nil) @@ -127,18 +151,18 @@ func (h *Handshaker) buildOfferAnswer() OfferAnswer { answer := OfferAnswer{ WgListenPort: h.config.LocalWgPort, Version: version.NetbirdVersion(), - RosenpassPubKey: h.config.RosenpassConfig.PubKey, - RosenpassAddr: h.config.RosenpassConfig.Addr, + RosenpassPubKey: h.config.RosenpassPubKey, + RosenpassAddr: h.config.RosenpassAddr, } if h.ice != nil && h.RemoteICESupported() { - uFrag, pwd := h.ice.GetLocalUserCredentials() - sid := h.ice.SessionID() - answer.IceCredentials = IceCredentials{uFrag, pwd} + creds := h.ice.Credentials() + answer.IceCredentials = IceCredentials{creds.UFrag, creds.Pwd} + sid := creds.SessionID 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.RelaySrvIP = ip } @@ -146,11 +170,11 @@ func (h *Handshaker) buildOfferAnswer() OfferAnswer { 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 // sending ICE credentials. Runs on the Conn event loop. -func (h *Handshaker) updateRemoteICEState(offer *OfferAnswer) { - hasICE := offer.hasICECredentials() +func (h *Handshaker) UpdateRemoteICEState(offer *OfferAnswer) { + hasICE := offer.HasICECredentials() prev := h.remoteICESupported.Swap(hasICE) if prev != hasICE { if hasICE { diff --git a/client/internal/peer/signaler.go b/client/internal/peer/signaling/signaler.go similarity index 99% rename from client/internal/peer/signaler.go rename to client/internal/peer/signaling/signaler.go index 5e437d96b..75fff2a1d 100644 --- a/client/internal/peer/signaler.go +++ b/client/internal/peer/signaling/signaler.go @@ -1,4 +1,4 @@ -package peer +package signaling import ( "github.com/pion/ice/v4" diff --git a/client/internal/peer/worker_ice.go b/client/internal/peer/worker_ice.go index acc4c7972..47796609b 100644 --- a/client/internal/peer/worker_ice.go +++ b/client/internal/peer/worker_ice.go @@ -14,6 +14,7 @@ import ( "github.com/netbirdio/netbird/client/iface" "github.com/netbirdio/netbird/client/iface/udpmux" 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/portforward" "github.com/netbirdio/netbird/client/internal/stdnet" @@ -42,7 +43,7 @@ type WorkerICE struct { log *log.Entry config ConnConfig conn iceCallbacks - signaler *Signaler + signaler *signaling.Signaler iFaceDiscover stdnet.ExternalIFaceDiscover statusRecorder *status.Recorder portForwardManager *portforward.Manager @@ -55,12 +56,12 @@ type WorkerICE struct { // connectedAgent is the agent whose connection was last reported ready; guarded by muxAgent connectedAgent *icemaker.ThreadSafeAgent // 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 // increase by one when disconnecting the agent // with it the remote peer can discard the already deprecated offer/answer // Without it the remote peer may recreate a workable ICE connection - sessionID ICESessionID + sessionID icemaker.SessionID remoteSessionChanged bool muxAgent sync.Mutex @@ -71,8 +72,8 @@ type WorkerICE struct { 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) { - sessionID, err := NewICESessionID() +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 := icemaker.NewSessionID() if err != nil { return nil, err } @@ -99,7 +100,7 @@ func NewWorkerICE(ctx context.Context, log *log.Entry, config ConnConfig, conn i 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.muxAgent.Lock() defer w.muxAgent.Unlock() @@ -123,7 +124,7 @@ func (w *WorkerICE) OnNewOffer(remoteOfferAnswer *OfferAnswer) { } } - sessionID, err := NewICESessionID() + sessionID, err := icemaker.NewSessionID() if err != nil { 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) { - return w.localUfrag, w.localPwd +func (w *WorkerICE) Credentials() signaling.Credentials { + w.muxAgent.Lock() + defer w.muxAgent.Unlock() + return signaling.Credentials{ + UFrag: w.localUfrag, + Pwd: w.localPwd, + SessionID: w.sessionID, + } } func (w *WorkerICE) InProgress() bool { @@ -242,7 +249,7 @@ func (w *WorkerICE) reCreateAgent(dialerCancel context.CancelFunc, candidates [] return agent, nil } -func (w *WorkerICE) SessionID() ICESessionID { +func (w *WorkerICE) getSessionID() icemaker.SessionID { w.muxAgent.Lock() defer w.muxAgent.Unlock() @@ -252,7 +259,7 @@ func (w *WorkerICE) SessionID() ICESessionID { // will block until connection succeeded // 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 -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") if err := agent.GatherCandidates(); err != nil { 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 { // consider to remove from here and move to the OnNewOffer - sessionID, err := NewICESessionID() + sessionID, err := icemaker.NewSessionID() if err != nil { 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) { - sessionID := w.SessionID() + sessionID := w.getSessionID() stats := agent.GetCandidatePairsStats() localCandidates, _ := agent.GetLocalCandidates() 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) { return agent.Dial(ctx, remoteOfferAnswer.IceCredentials.UFrag, remoteOfferAnswer.IceCredentials.Pwd) } else { diff --git a/client/internal/peer/worker_relay.go b/client/internal/peer/worker_relay.go index 72322db27..255d12d5e 100644 --- a/client/internal/peer/worker_relay.go +++ b/client/internal/peer/worker_relay.go @@ -10,6 +10,7 @@ import ( log "github.com/sirupsen/logrus" + "github.com/netbirdio/netbird/client/internal/peer/signaling" 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 } -func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *OfferAnswer) { +func (w *WorkerRelay) OnNewOffer(remoteOfferAnswer *signaling.OfferAnswer) { if !w.isRelaySupported(remoteOfferAnswer) { w.log.Infof("Relay is not supported by remote peer") 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() { return false }