diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 9cba516cf..6ce46b2a6 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -228,6 +228,7 @@ type ParticipantParams struct { EnableParticipantDataBlob bool EnableStartAtDesiredQuality bool MigrationWaitDuration time.Duration + ExcludeIPv6LocalCandidates bool } type ParticipantImpl struct { @@ -2057,6 +2058,7 @@ func (p *ParticipantImpl) setupTransportManager() error { UseOneShotSignallingMode: p.params.UseOneShotSignallingMode, FireOnTrackBySdp: p.params.FireOnTrackBySdp, EnableDataTracks: p.params.EnableDataTracks, + ExcludeIPv6LocalCandidates: p.params.ExcludeIPv6LocalCandidates, } if p.params.SyncStreams && p.params.PlayoutDelay.GetEnabled() && p.params.ClientInfo.isFirefox() { // we will disable playout delay for Firefox if the user is expecting diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index d94ef169b..27ff8a3a3 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -319,6 +319,7 @@ type TransportParams struct { IsSendSide bool AllowPlayoutDelay bool UseOneShotSignallingMode bool + ExcludeIPv6LocalCandidates bool FireOnTrackBySdp bool DataChannelMaxBufferedAmount uint64 DatachannelSlowThreshold int @@ -1712,31 +1713,26 @@ func (t *PCTransport) GetAnswer() (webrtc.SessionDescription, uint32, error) { cld := t.pc.CurrentLocalDescription() - // add local candidates to ICE connection details - parsed, err := cld.Unmarshal() - if err == nil { - addLocalICECandidates := func(attrs []sdp.Attribute) { - for _, a := range attrs { - if a.IsICECandidate() { - c, err := ice.UnmarshalCandidate(a.Value) - if err != nil { - continue - } - t.connectionDetails.AddLocalICECandidate(c, false, false) - } - } - } + preferTCP := t.preferTCP.Load() + if t.isCandidateFilterActive(preferTCP) { + t.params.Logger.Debugw("local answer (unfiltered)", "sdp", cld.SDP) + } - addLocalICECandidates(parsed.Attributes) - for _, m := range parsed.MediaDescriptions { - addLocalICECandidates(m.Attributes) - } + // + // Filter after setting local description as pion expects the answer + // to match between CreateAnswer and SetLocalDescription. + // Filtered answer is sent to remote so that remote does not + // see filtered candidates. + // + filteredAnswer := t.filterCandidates(*cld, preferTCP, true) + if t.isCandidateFilterActive(preferTCP) { + t.params.Logger.Debugw("local answer (filtered)", "sdp", filteredAnswer.SDP) } answerId := t.remoteOfferId.Load() t.localAnswerId.Store(answerId) - return *cld, answerId, nil + return filteredAnswer, answerId, nil } func (t *PCTransport) GetICESessionUfrag() (string, error) { @@ -1899,13 +1895,13 @@ func (t *PCTransport) HandleICERestartSDPFragment(sdpFragment string) (string, e t.connectionDetails.AddRemoteICECandidate(c, false, false, false) } - ans, err := t.pc.CreateAnswer(nil) + answer, err := t.pc.CreateAnswer(nil) if err != nil { t.params.Logger.Warnw("could not create answer", err) return "", err } - if err = t.pc.SetLocalDescription(ans); err != nil { + if err = t.pc.SetLocalDescription(answer); err != nil { t.params.Logger.Warnw("could not set local description", err) return "", err } @@ -1915,31 +1911,30 @@ func (t *PCTransport) HandleICERestartSDPFragment(sdpFragment string) (string, e cld := t.pc.CurrentLocalDescription() + preferTCP := t.preferTCP.Load() + if t.isCandidateFilterActive(preferTCP) { + t.params.Logger.Debugw("local answer (unfiltered)", "sdp", cld.SDP) + } + + // + // Filter after setting local description as pion expects the answer + // to match between CreateAnswer and SetLocalDescription. + // Filtered answer is sent to remote so that remote does not + // see filtered candidates. + // + filteredAnswer := t.filterCandidates(*cld, preferTCP, true) + if t.isCandidateFilterActive(preferTCP) { + t.params.Logger.Debugw("local answer (filtered)", "sdp", filteredAnswer.SDP) + } + // add local candidates to ICE connection details - parsedAnswer, err := cld.Unmarshal() + parsedFilteredAnswer, err := filteredAnswer.Unmarshal() if err != nil { t.params.Logger.Warnw("could not parse local description", err) return "", err } - addLocalICECandidates := func(attrs []sdp.Attribute) { - for _, a := range attrs { - if a.IsICECandidate() { - c, err := ice.UnmarshalCandidate(a.Value) - if err != nil { - continue - } - t.connectionDetails.AddLocalICECandidate(c, false, false) - } - } - } - - addLocalICECandidates(parsedAnswer.Attributes) - for _, m := range parsedAnswer.MediaDescriptions { - addLocalICECandidates(m.Attributes) - } - - parsedFragmentAnswer, err := lksdp.ExtractSDPFragment(parsedAnswer) + parsedFragmentAnswer, err := lksdp.ExtractSDPFragment(parsedFilteredAnswer) if err != nil { t.params.Logger.Warnw("could not extract SDP fragment", err) return "", err @@ -2402,9 +2397,15 @@ func (t *PCTransport) handleLocalICECandidate(e event) error { filtered := false if c != nil { if t.preferTCP.Load() && c.Protocol != webrtc.ICEProtocolTCP { - t.params.Logger.Debugw("filtering out local candidate", "candidate", c.String()) + t.params.Logger.Debugw("filtering out local candidate, TCP prefered", "candidate", c.String()) filtered = true } + if !filtered && t.params.ExcludeIPv6LocalCandidates { + if IsIPv6(c.Address) { + t.params.Logger.Debugw("filtering out local candidate, IPv6 excluded", "candidate", c.String()) + filtered = true + } + } t.connectionDetails.AddLocalCandidate(c, filtered, true) } @@ -2469,6 +2470,10 @@ func (t *PCTransport) setNegotiationState(state transport.NegotiationState) { } } +func (t *PCTransport) isCandidateFilterActive(preferTCP bool) bool { + return preferTCP || t.params.ExcludeIPv6LocalCandidates +} + func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, isLocal bool) webrtc.SessionDescription { parsed, err := sd.Unmarshal() if err != nil { @@ -2486,12 +2491,10 @@ func (t *PCTransport) filterCandidates(sd webrtc.SessionDescription, preferTCP, filteredAttrs = append(filteredAttrs, a) continue } - excluded := preferTCP && !c.NetworkType().IsTCP() - if !excluded { - if !t.params.Config.UseMDNS && types.IsICECandidateMDNS(c) { - excluded = true - } - } + excluded := + (preferTCP && !c.NetworkType().IsTCP()) || + (t.params.ExcludeIPv6LocalCandidates && isLocal && c.NetworkType().IsIPv6()) || + (!t.params.Config.UseMDNS && types.IsICECandidateMDNS(c)) if !excluded { filteredAttrs = append(filteredAttrs, a) } @@ -2634,7 +2637,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { } preferTCP := t.preferTCP.Load() - if preferTCP { + if t.isCandidateFilterActive(preferTCP) { t.params.Logger.Debugw("local offer (unfiltered)", "sdp", offer.SDP) } @@ -2662,7 +2665,7 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error { // see filtered candidates. // offer = t.filterCandidates(offer, preferTCP, true) - if preferTCP { + if t.isCandidateFilterActive(preferTCP) { t.params.Logger.Debugw("local offer (filtered)", "sdp", offer.SDP) } @@ -2728,11 +2731,11 @@ func (t *PCTransport) isRemoteOfferRestartICE(parsed *sdp.SessionDescription) (s func (t *PCTransport) setRemoteDescription(sd webrtc.SessionDescription) error { // filter before setting remote description so that pion does not see filtered remote candidates preferTCP := t.preferTCP.Load() - if preferTCP { + if t.isCandidateFilterActive(preferTCP) { t.params.Logger.Debugw("remote description (unfiltered)", "type", sd.Type, "sdp", sd.SDP) } sd = t.filterCandidates(sd, preferTCP, false) - if preferTCP { + if t.isCandidateFilterActive(preferTCP) { t.params.Logger.Debugw("remote description (filtered)", "type", sd.Type, "sdp", sd.SDP) } @@ -2792,7 +2795,7 @@ func (t *PCTransport) createAndSendAnswer() error { } preferTCP := t.preferTCP.Load() - if preferTCP { + if t.isCandidateFilterActive(preferTCP) { t.params.Logger.Debugw("local answer (unfiltered)", "sdp", answer.SDP) } @@ -2808,7 +2811,7 @@ func (t *PCTransport) createAndSendAnswer() error { // see filtered candidates. // answer = t.filterCandidates(answer, preferTCP, true) - if preferTCP { + if t.isCandidateFilterActive(preferTCP) { t.params.Logger.Debugw("local answer (filtered)", "sdp", answer.SDP) } diff --git a/pkg/rtc/transportmanager.go b/pkg/rtc/transportmanager.go index 886217d7f..6259d1a7c 100644 --- a/pkg/rtc/transportmanager.go +++ b/pkg/rtc/transportmanager.go @@ -98,6 +98,7 @@ type TransportManagerParams struct { UseOneShotSignallingMode bool FireOnTrackBySdp bool EnableDataTracks bool + ExcludeIPv6LocalCandidates bool } type TransportManager struct { @@ -168,6 +169,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro DatachannelLossyTargetLatency: params.DatachannelLossyTargetLatency, FireOnTrackBySdp: params.FireOnTrackBySdp, EnableDataTracks: params.EnableDataTracks, + ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, }) if err != nil { return nil, err @@ -194,6 +196,7 @@ func NewTransportManager(params TransportManagerParams) (*TransportManager, erro Handler: TransportManagerTransportHandler{params.SubscriberHandler, t, lgr}, FireOnTrackBySdp: params.FireOnTrackBySdp, EnableDataTracks: params.EnableDataTracks, + ExcludeIPv6LocalCandidates: params.ExcludeIPv6LocalCandidates, }) if err != nil { return nil, err diff --git a/pkg/rtc/utils.go b/pkg/rtc/utils.go index 83d9b786d..fab143262 100644 --- a/pkg/rtc/utils.go +++ b/pkg/rtc/utils.go @@ -173,6 +173,11 @@ func MaybeTruncateIP(addr string) string { return addr[:len(addr)-3] + "..." } +func IsIPv6(addr string) bool { + ipAddr := net.ParseIP(addr) + return ipAddr != nil && ipAddr.To4() == nil +} + func ChunkProtoBatch[T proto.Message](batch []T, target int) [][]T { var chunks [][]T var start, size int diff --git a/pkg/service/wire_gen.go b/pkg/service/wire_gen.go index 0da441132..b64fbb063 100644 --- a/pkg/service/wire_gen.go +++ b/pkg/service/wire_gen.go @@ -86,23 +86,23 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } rtcEgressLauncher := NewEgressLauncher(egressClient, ioInfoService, objectStore) topicFormatter := rpc.NewTopicFormatter() - v, err := rpc.NewTypedRoomClient(clientParams) + roomClient, err := rpc.NewTypedRoomClient(clientParams) if err != nil { return nil, err } - v2, err := rpc.NewTypedParticipantClient(clientParams) + participantClient, err := rpc.NewTypedParticipantClient(clientParams) if err != nil { return nil, err } - roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, v, v2) + roomService, err := NewRoomService(limitConfig, apiConfig, router, roomAllocator, objectStore, rtcEgressLauncher, topicFormatter, roomClient, participantClient) if err != nil { return nil, err } - v3, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) + agentDispatchInternalClient, err := rpc.NewTypedAgentDispatchInternalClient(clientParams) if err != nil { return nil, err } - agentDispatchService := NewAgentDispatchService(limitConfig, v3, topicFormatter, roomAllocator, router) + agentDispatchService := NewAgentDispatchService(limitConfig, agentDispatchInternalClient, topicFormatter, roomAllocator, router) egressService := NewEgressService(egressClient, rtcEgressLauncher, ioInfoService, roomService) ingressConfig := getIngressConfig(conf) ingressClient, err := rpc.NewIngressClient(clientParams) @@ -117,11 +117,11 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live } sipService := NewSIPService(sipConfig, nodeID, messageBus, sipClient, sipStore, roomService, telemetryService) rtcService := NewRTCService(conf, roomAllocator, router, telemetryService) - v4, err := rpc.NewTypedWHIPParticipantClient(clientParams) + whipParticipantClient, err := rpc.NewTypedWHIPParticipantClient(clientParams) if err != nil { return nil, err } - serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, v4) + serviceWHIPService, err := NewWHIPService(conf, router, roomAllocator, clientParams, topicFormatter, whipParticipantClient) if err != nil { return nil, err } @@ -146,8 +146,8 @@ func InitializeServer(conf *config.Config, currentNode routing.LocalNode) (*Live if err != nil { return nil, err } - v5 := getTURNAuthHandlerFunc(turnAuthHandler) - server, err := newInProcessTurnServer(conf, v5) + authHandler := getTURNAuthHandlerFunc(turnAuthHandler) + server, err := newInProcessTurnServer(conf, authHandler) if err != nil { return nil, err }