Fix regression with UDPMuxDefault and NAT1To1IPs

PR #496 added a mechanism to close connections of duplicate candidates
and avoid memory leaks. When UDPMuxDefault and NAT1To1IPs are used
together, candidates may share the same connection, and candidates may
be equal, since their IPs are overridden by the one specified in
NAT1To1IPs. When a duplicate candidate is found, its connection is
closed, making impossible to use any other candidate that shares the
same connection.

This PR fixes the issue as it prevents duplicate candidates from
creating connections and getting their connection closed.

gather.go

@@ -255,6 +255,7 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin
  }
 
  localAddresses := a.udpMux.GetListenAddresses()
+ existingConfigs := make(map[CandidateHostConfig]struct{})
 
  for _, addr := range localAddresses {
  udpAddr, ok := addr.(*net.UDPAddr)
@@ -272,17 +273,26 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin
  candidateIP = mappedIP
  }
 
- conn, err := a.udpMux.GetConn(a.localUfrag, udpAddr)
- if err != nil {
- return err
- }
  hostConfig := CandidateHostConfig{
  Network: udp,
  Address: candidateIP.String(),
  Port: udpAddr.Port,
  Component: ComponentRTP,
  }
 
+ // Detect a duplicate candidate before calling addCandidate().
+ // otherwise, addCandidate() detects the duplicate candidate
+ // and close its connection, invalidating all candidates
+ // that share the same connection.
+ if _, ok := existingConfigs[hostConfig]; ok {
+ continue
+ }
+
+ conn, err := a.udpMux.GetConn(a.localUfrag, udpAddr)
+ if err != nil {
+ return err
+ }
+
  c, err := NewCandidateHost(&hostConfig)
  if err != nil {
  closeConnAndLog(conn, a.log, fmt.Sprintf("Failed to create host mux candidate: %s %d: %v", candidateIP, udpAddr.Port, err))
@@ -297,6 +307,8 @@ func (a *Agent) gatherCandidatesLocalUDPMux(ctx context.Context) error { //nolin
  closeConnAndLog(conn, a.log, fmt.Sprintf("Failed to add candidate: %s %d: %v", candidateIP, udpAddr.Port, err))
  continue
  }
+
+ existingConfigs[hostConfig] = struct{}{}
  }
 
  return nil

gather_test.go

@@ -599,6 +599,51 @@ func TestTURNProxyDialer(t *testing.T) {
  assert.NoError(t, a.Close())
 }
 
+// TestUDPMuxDefaultWithNAT1To1IPsUsage asserts that candidates
+// are given and connections are valid when using UDPMuxDefault and NAT1To1IPs.
+func TestUDPMuxDefaultWithNAT1To1IPsUsage(t *testing.T) {
+ report := test.CheckRoutines(t)
+ defer report()
+
+ lim := test.TimeOut(time.Second * 30)
+ defer lim.Stop()
+
+ conn, err := net.ListenPacket("udp4", ":0")
+ assert.NoError(t, err)
+ defer func() {
+ _ = conn.Close()
+ }()
+
+ mux := NewUDPMuxDefault(UDPMuxParams{
+ UDPConn: conn,
+ })
+ defer func() {
+ _ = mux.Close()
+ }()
+
+ a, err := NewAgent(&AgentConfig{
+ NAT1To1IPs: []string{"1.2.3.4"},
+ NAT1To1IPCandidateType: CandidateTypeHost,
+ UDPMux: mux,
+ })
+ assert.NoError(t, err)
+
+ gatherCandidateDone := make(chan struct{})
+ assert.NoError(t, a.OnCandidate(func(c Candidate) {
+ if c == nil {
+ close(gatherCandidateDone)
+ } else {
+ assert.Equal(t, "1.2.3.4", c.Address())
+ }
+ }))
+ assert.NoError(t, a.GatherCandidates())
+ <-gatherCandidateDone
+
+ assert.NotEqual(t, 0, len(mux.connsIPv4))
+
+ assert.NoError(t, a.Close())
+}
+
 // Assert that candidates are given for each mux in a MultiUDPMux
 func TestMultiUDPMuxUsage(t *testing.T) {
  report := test.CheckRoutines(t)