Skip to content

Commit 9586aea

Browse files
bclswl0827bclswl0827
committed
Handle duplicate local candidates safely
Return ErrDuplicateCandidate when AddLocalCandidate receives a candidate that is already registered. Preserve ownership of the caller-provided candidate and packet connection instead of closing them. Keep the existing cleanup behavior for candidates gathered internally. Move the local candidate tests into agent_test.go and cover both duplicate-handling paths.
1 parent 5e87d3b commit 9586aea

5 files changed

Lines changed: 189 additions & 148 deletions

File tree

‎agent.go‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1037,7 +1037,7 @@ func (a *Agent) AddLocalCandidate(cand Candidate, candidateConn net.PacketConn)
10371037
return ErrCandidatePacketConnNil
10381038
}
10391039

1040-
return a.addCandidate(a.loop, cand, candidateConn)
1040+
return a.addCandidate(a.loop, cand, candidateConn, true)
10411041
}
10421042

10431043
func (a *Agent) resolveAndAddMulticastCandidate(cand *CandidateHost) {
@@ -1364,15 +1364,21 @@ func (a *Agent) shouldAcceptRemoteCandidate(cand Candidate) bool {
13641364
return true
13651365
}
13661366

1367-
func (a *Agent) addCandidate(ctx context.Context, cand Candidate, candidateConn net.PacketConn) error {
1367+
func (a *Agent) addCandidate(ctx context.Context, cand Candidate, candidateConn net.PacketConn, errorOnDuplicate bool) error {
13681368
if err := ctx.Err(); err != nil {
13691369
return err
13701370
}
13711371

1372-
return a.loop.Run(ctx, func(context.Context) {
1372+
duplicate := false
1373+
err := a.loop.Run(ctx, func(context.Context) {
13731374
set := a.localCandidates[cand.NetworkType()]
13741375
for _, candidate := range set {
13751376
if candidate.Equal(cand) {
1377+
if errorOnDuplicate {
1378+
duplicate = true
1379+
return
1380+
}
1381+
13761382
a.log.Debugf("Ignore duplicate candidate: %s", cand)
13771383
if err := cand.close(); err != nil {
13781384
a.log.Warnf("Failed to close duplicate candidate: %v", err)
@@ -1403,6 +1409,14 @@ func (a *Agent) addCandidate(ctx context.Context, cand Candidate, candidateConn
14031409
a.candidateNotifier.EnqueueCandidate(cand)
14041410
}
14051411
})
1412+
if err != nil {
1413+
return err
1414+
}
1415+
if duplicate {
1416+
return ErrDuplicateCandidate
1417+
}
1418+
1419+
return nil
14061420
}
14071421

14081422
func (a *Agent) setCandidateExtensions(cand Candidate) {

‎agent_test.go‎

Lines changed: 163 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,6 +51,30 @@ type blockingWritePacketConn struct {
5151
closeOnce sync.Once
5252
}
5353

54+
type localCandidatePacketConn struct {
55+
addr net.Addr
56+
closeCount atomic.Int32
57+
}
58+
59+
func (c *localCandidatePacketConn) ReadFrom([]byte) (int, net.Addr, error) {
60+
return 0, c.addr, io.EOF
61+
}
62+
63+
func (c *localCandidatePacketConn) WriteTo(payload []byte, _ net.Addr) (int, error) {
64+
return len(payload), nil
65+
}
66+
67+
func (c *localCandidatePacketConn) Close() error {
68+
c.closeCount.Add(1)
69+
70+
return nil
71+
}
72+
73+
func (c *localCandidatePacketConn) LocalAddr() net.Addr { return c.addr }
74+
func (c *localCandidatePacketConn) SetDeadline(time.Time) error { return nil }
75+
func (c *localCandidatePacketConn) SetReadDeadline(time.Time) error { return nil }
76+
func (c *localCandidatePacketConn) SetWriteDeadline(time.Time) error { return nil }
77+
5478
func newBlockingWritePacketConn() *blockingWritePacketConn {
5579
return &blockingWritePacketConn{
5680
writeStarted: make(chan struct{}),
@@ -2711,6 +2735,143 @@ func TestAddRemoteCandidateHonorsRemoteIPFilter(t *testing.T) {
27112735
}, time.Second, 10*time.Millisecond)
27122736
}
27132737

2738+
func TestAddLocalCandidateRegistersExternalRelay(t *testing.T) {
2739+
agent, err := NewAgentWithOptions(
2740+
WithCandidateTypes([]CandidateType{}),
2741+
WithMulticastDNSMode(MulticastDNSModeDisabled),
2742+
)
2743+
require.NoError(t, err)
2744+
defer func() { require.NoError(t, agent.Close()) }()
2745+
2746+
gatheringComplete := make(chan struct{})
2747+
candidates := make(chan Candidate, 1)
2748+
require.NoError(t, agent.OnCandidate(func(candidate Candidate) {
2749+
if candidate == nil {
2750+
close(gatheringComplete)
2751+
2752+
return
2753+
}
2754+
candidates <- candidate
2755+
}))
2756+
2757+
candidate, err := NewCandidateRelay(&CandidateRelayConfig{
2758+
Network: NetworkTypeUDP4.String(),
2759+
Address: "192.0.2.10",
2760+
Port: 5000,
2761+
Component: ComponentRTP,
2762+
RelayProtocol: "custom",
2763+
})
2764+
require.NoError(t, err)
2765+
packetConn := &localCandidatePacketConn{
2766+
addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 10), Port: 5000},
2767+
}
2768+
2769+
require.NoError(t, agent.AddLocalCandidate(candidate, packetConn))
2770+
localCandidates, err := agent.GetLocalCandidates()
2771+
require.NoError(t, err)
2772+
require.Len(t, localCandidates, 1)
2773+
relayCandidate, ok := localCandidates[0].(*CandidateRelay)
2774+
require.True(t, ok)
2775+
require.Equal(t, "custom", relayCandidate.RelayProtocol())
2776+
2777+
select {
2778+
case got := <-candidates:
2779+
require.Equal(t, candidate, got)
2780+
case <-time.After(time.Second):
2781+
require.FailNow(t, "timed out waiting for local candidate callback")
2782+
}
2783+
2784+
require.NoError(t, agent.GatherCandidates())
2785+
select {
2786+
case <-gatheringComplete:
2787+
case <-time.After(time.Second):
2788+
require.FailNow(t, "timed out waiting for gathering completion")
2789+
}
2790+
}
2791+
2792+
func TestAddLocalCandidateRejectsNilPacketConn(t *testing.T) {
2793+
agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled))
2794+
require.NoError(t, err)
2795+
defer func() { require.NoError(t, agent.Close()) }()
2796+
2797+
candidate, err := NewCandidateRelay(&CandidateRelayConfig{
2798+
Network: NetworkTypeUDP4.String(),
2799+
Address: "192.0.2.11",
2800+
Port: 5001,
2801+
Component: ComponentRTP,
2802+
})
2803+
require.NoError(t, err)
2804+
require.ErrorIs(t, agent.AddLocalCandidate(candidate, nil), ErrCandidatePacketConnNil)
2805+
}
2806+
2807+
func TestAddLocalCandidateRejectsDuplicateWithoutClosing(t *testing.T) {
2808+
agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled))
2809+
require.NoError(t, err)
2810+
defer func() { require.NoError(t, agent.Close()) }()
2811+
2812+
require.NoError(t, agent.OnCandidate(func(Candidate) {}))
2813+
var candidateCloseCount atomic.Int32
2814+
candidate, err := NewCandidateRelay(&CandidateRelayConfig{
2815+
Network: NetworkTypeUDP4.String(),
2816+
Address: "192.0.2.12",
2817+
Port: 5002,
2818+
Component: ComponentRTP,
2819+
OnClose: func() error {
2820+
candidateCloseCount.Add(1)
2821+
2822+
return nil
2823+
},
2824+
})
2825+
require.NoError(t, err)
2826+
packetConn := &localCandidatePacketConn{
2827+
addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 12), Port: 5002},
2828+
}
2829+
2830+
require.NoError(t, agent.AddLocalCandidate(candidate, packetConn))
2831+
require.ErrorIs(t, agent.AddLocalCandidate(candidate, packetConn), ErrDuplicateCandidate)
2832+
require.Zero(t, candidateCloseCount.Load())
2833+
require.Zero(t, packetConn.closeCount.Load())
2834+
2835+
localCandidates, err := agent.GetLocalCandidates()
2836+
require.NoError(t, err)
2837+
require.Equal(t, []Candidate{candidate}, localCandidates)
2838+
}
2839+
2840+
func TestAddCandidateClosesDuplicate(t *testing.T) {
2841+
agent, err := NewAgentWithOptions(WithMulticastDNSMode(MulticastDNSModeDisabled))
2842+
require.NoError(t, err)
2843+
defer func() { require.NoError(t, agent.Close()) }()
2844+
2845+
require.NoError(t, agent.OnCandidate(func(Candidate) {}))
2846+
config := CandidateRelayConfig{
2847+
Network: NetworkTypeUDP4.String(),
2848+
Address: "192.0.2.13",
2849+
Port: 5003,
2850+
Component: ComponentRTP,
2851+
}
2852+
first, err := NewCandidateRelay(&config)
2853+
require.NoError(t, err)
2854+
var duplicateCloseCount atomic.Int32
2855+
config.OnClose = func() error {
2856+
duplicateCloseCount.Add(1)
2857+
2858+
return nil
2859+
}
2860+
duplicate, err := NewCandidateRelay(&config)
2861+
require.NoError(t, err)
2862+
firstConn := &localCandidatePacketConn{
2863+
addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 13), Port: 5003},
2864+
}
2865+
duplicateConn := &localCandidatePacketConn{
2866+
addr: &net.UDPAddr{IP: net.IPv4(192, 0, 2, 13), Port: 5003},
2867+
}
2868+
2869+
require.NoError(t, agent.addCandidate(context.Background(), first, firstConn, false))
2870+
require.NoError(t, agent.addCandidate(context.Background(), duplicate, duplicateConn, false))
2871+
require.Equal(t, int32(1), duplicateCloseCount.Load())
2872+
require.Equal(t, int32(1), duplicateConn.closeCount.Load())
2873+
}
2874+
27142875
func TestGetLocalCandidates(t *testing.T) {
27152876
var config AgentConfig
27162877

@@ -2736,7 +2897,7 @@ func TestGetLocalCandidates(t *testing.T) {
27362897

27372898
expectedCandidates = append(expectedCandidates, cand)
27382899

2739-
err = agent.addCandidate(context.Background(), cand, dummyConn)
2900+
err = agent.addCandidate(context.Background(), cand, dummyConn, false)
27402901
require.NoError(t, err)
27412902
}
27422903

@@ -3376,7 +3537,7 @@ func TestSetCandidatesUfrag(t *testing.T) {
33763537
cand, errCand := NewCandidateHost(&cfg)
33773538
require.NoError(t, errCand)
33783539

3379-
err = agent.addCandidate(context.Background(), cand, dummyConn)
3540+
err = agent.addCandidate(context.Background(), cand, dummyConn, false)
33803541
require.NoError(t, err)
33813542
}
33823543

‎errors.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,9 @@ var (
6868
// a packet connection.
6969
ErrCandidatePacketConnNil = errors.New("candidate packet connection is nil")
7070

71+
// ErrDuplicateCandidate indicates a local candidate has already been added.
72+
ErrDuplicateCandidate = errors.New("candidate already added")
73+
7174
// ErrMultipleGatherAttempted indicates GatherCandidates has been called multiple times.
7275
ErrMultipleGatherAttempted = errors.New("attempting to gather candidates during gathering state")
7376

‎gather.go‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -484,7 +484,7 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
484484
continue
485485
}
486486

487-
if err := a.addCandidate(ctx, candidateHost, connAndPort.conn); err != nil {
487+
if err := a.addCandidate(ctx, candidateHost, connAndPort.conn, false); err != nil {
488488
if closeErr := candidateHost.close(); closeErr != nil {
489489
a.log.Warnf("Failed to close candidate: %v", closeErr)
490490
}
@@ -593,7 +593,7 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin
593593
continue
594594
}
595595

596-
if err := a.addCandidate(ctx, c, conn); err != nil {
596+
if err := a.addCandidate(ctx, c, conn, false); err != nil {
597597
if closeErr := c.close(); closeErr != nil {
598598
a.log.Warnf("Failed to close candidate: %v", closeErr)
599599
}
@@ -708,7 +708,7 @@ func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes []
708708
continue
709709
}
710710

711-
if err := a.addCandidate(ctx, c, currentConn); err != nil {
711+
if err := a.addCandidate(ctx, c, currentConn, false); err != nil {
712712
if closeErr := c.close(); closeErr != nil {
713713
a.log.Warnf("Failed to close candidate: %v", closeErr)
714714
}
@@ -797,7 +797,7 @@ func (a *Agent) gatherCandidatesSrflxUDPMux(ctx context.Context, urls []*stun.UR
797797
return
798798
}
799799

800-
if err := a.addCandidate(ctx, c, conn); err != nil {
800+
if err := a.addCandidate(ctx, c, conn, false); err != nil {
801801
if closeErr := c.close(); closeErr != nil {
802802
a.log.Warnf("Failed to close candidate: %v", closeErr)
803803
}
@@ -921,7 +921,7 @@ func (a *Agent) gatherCandidatesSrflx(ctx context.Context, urls []*stun.URI, net
921921
return
922922
}
923923

924-
if err := a.addCandidate(ctx, c, conn); err != nil {
924+
if err := a.addCandidate(ctx, c, conn, false); err != nil {
925925
if closeErr := c.close(); closeErr != nil {
926926
a.log.Warnf("Failed to close candidate: %v", closeErr)
927927
}
@@ -1362,7 +1362,7 @@ func (a *Agent) createRelayCandidate(ctx context.Context, ep relayEndpoint, ip n
13621362
return err
13631363
}
13641364

1365-
if err := a.addCandidate(ctx, candidate, ep.conn); err != nil {
1365+
if err := a.addCandidate(ctx, candidate, ep.conn, false); err != nil {
13661366
if closeErr := candidate.close(); closeErr != nil {
13671367
a.log.Warnf("Failed to close candidate: %v", closeErr)
13681368
}

0 commit comments

Comments
 (0)