From cafd8d78606a3887716266a0d9bdf3c6f8959ccb Mon Sep 17 00:00:00 2001 From: Hugo Arregui Date: Mon, 29 Apr 2019 12:20:04 -0300 Subject: [PATCH] Implement regular nomination In this first iteration we nominate the first valid pair we found. --- agent.go | 227 +++++++++++++++++++---------------------- agent_test.go | 275 +++++++++++++++++++++++++------------------------- selection.go | 220 ++++++++++++++++++++++++++++++++++++++++ transport.go | 2 +- 4 files changed, 460 insertions(+), 264 deletions(-) create mode 100644 selection.go diff --git a/agent.go b/agent.go index 569c366..8b594e6 100644 --- a/agent.go +++ b/agent.go @@ -48,8 +48,9 @@ func (bp byPairPriority) Less(i, j int) bool { } type bindingRequest struct { - transactionID []byte - destination net.Addr + transactionID []byte + destination net.Addr + isUseCandidate bool } // Agent represents the ICE agent @@ -98,6 +99,7 @@ type Agent struct { remotePwd string remoteCandidates map[NetworkType][]*Candidate + selector pairCandidateSelector selectedPair *candidatePair validPairs []*candidatePair @@ -222,6 +224,7 @@ func NewAgent(config *AgentConfig) (*Agent, error) { gatherCandidatesReflective(a, config.Urls, config.NetworkTypes) go a.taskLoop() + return a, nil } @@ -257,10 +260,18 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP case remotePwd == "": return ErrRemotePwdEmpty } + a.haveStarted.Store(true) a.log.Debugf("Started agent: isControlling? %t, remoteUfrag: %q, remotePwd: %q", isControlling, remoteUfrag, remotePwd) return a.run(func(agent *Agent) { + if isControlling { + a.selector = &controllingSelector{agent: a, log: a.log} + } else { + a.selector = &controlledSelector{agent: a, log: a.log} + } + a.selector.Start() + agent.isControlling = isControlling agent.remoteUfrag = remoteUfrag agent.remotePwd = remotePwd @@ -274,57 +285,6 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP }) } -func (a *Agent) pingCandidate(local, remote *Candidate) { - var msg *stun.Message - var err error - - // The controlling agent MUST include the USE-CANDIDATE attribute in - // order to nominate a candidate pair (Section 8.1.1). The controlled - // agent MUST NOT include the USE-CANDIDATE attribute in a Binding - // request. - - transactionID := stun.GenerateTransactionID() - if a.isControlling { - msg, err = stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID, - &stun.Username{Username: a.remoteUfrag + ":" + a.localUfrag}, - &stun.UseCandidate{}, - &stun.IceControlling{TieBreaker: a.tieBreaker}, - &stun.Priority{Priority: local.Priority()}, - &stun.MessageIntegrity{ - Key: []byte(a.remotePwd), - }, - &stun.Fingerprint{}, - ) - } else { - msg, err = stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID, - &stun.Username{Username: a.remoteUfrag + ":" + a.localUfrag}, - &stun.IceControlled{TieBreaker: a.tieBreaker}, - &stun.Priority{Priority: local.Priority()}, - &stun.MessageIntegrity{ - Key: []byte(a.remotePwd), - }, - &stun.Fingerprint{}, - ) - } - - if err != nil { - a.log.Debug(err.Error()) - return - } - - a.log.Tracef("ping STUN from %s to %s\n", local.String(), remote.String()) - if overflow := len(a.pendingBindingRequests) - maxPendingBindingRequests; overflow > 1 { - a.log.Debugf("Discarded %d pending binding requests, pendingBindingRequests is full", overflow) - a.pendingBindingRequests = a.pendingBindingRequests[overflow:] - } - - a.pendingBindingRequests = append(a.pendingBindingRequests, bindingRequest{ - transactionID: transactionID, - destination: remote.addr(), - }) - a.sendSTUN(msg, local, remote) -} - func (a *Agent) updateConnectionState(newState ConnectionState) { if a.connectionState != newState { a.log.Infof("Setting new connection state: %s", newState) @@ -338,33 +298,52 @@ func (a *Agent) updateConnectionState(newState ConnectionState) { } } -func (a *Agent) setValidPair(local, remote *Candidate, selected, controlling bool) { - // TODO: avoid duplicates - p := newCandidatePair(local, remote, controlling) - a.log.Tracef("Found valid candidate pair: %s (selected? %t)", p, selected) - - if selected { - // Notify when the selected pair changes - if !a.selectedPair.Equal(p) { - a.onSelectedCandidatePairChange(p) +func (a *Agent) findValidPair(local, remote *Candidate) *candidatePair { + for _, p := range a.validPairs { + if p.local == local && p.remote == remote { + return p } - a.selectedPair = p - a.validPairs = nil - // TODO: only set state to connected on selecting final pair? - a.updateConnectionState(ConnectionStateConnected) - } else { - // keep track of pairs with succesfull bindings since any of them - // can be used for communication until the final pair is selected: - // https://tools.ietf.org/html/draft-ietf-ice-rfc5245bis-20#section-12 - a.validPairs = append(a.validPairs, p) - // Sort the candidate pairs by priority of the remotes - sort.Sort(byPairPriority{a.validPairs}) } + return nil +} + +func (a *Agent) addValidPair(local, remote *Candidate) *candidatePair { + p := a.findValidPair(local, remote) + if p != nil { + a.log.Tracef("Candidate pair is already valid: %s", p) + return p + } + + p = newCandidatePair(local, remote, a.isControlling) + a.log.Tracef("Found valid candidate pair: %s", p) + + // keep track of pairs with succesfull bindings since any of them + // can be used for communication until the final pair is selected: + // https://tools.ietf.org/html/draft-ietf-ice-rfc5245bis-20#section-12 + a.validPairs = append(a.validPairs, p) + return p +} + +func (a *Agent) setSelectedPair(p *candidatePair) { + a.log.Tracef("Set selected candidate pair: %s", p) + // Notify when the selected pair changes + a.onSelectedCandidatePairChange(p) + + a.selectedPair = p + a.updateConnectionState(ConnectionStateConnected) // Signal connected a.onConnectedOnce.Do(func() { close(a.onConnected) }) } +func (a *Agent) getBestValidPair() *candidatePair { + if len(a.validPairs) == 0 { + return nil + } + sort.Sort(byPairPriority{a.validPairs}) + return a.validPairs[0] +} + // A task is a type task func(*Agent) @@ -383,28 +362,29 @@ func (a *Agent) run(t task) error { } func (a *Agent) taskLoop() { - contactCandidates := func() { - if a.validateSelectedPair() { - a.log.Trace("checking keepalive") - a.checkKeepalive() - } else { - a.log.Trace("pinging all candidates") - a.pingAllCandidates() - } - } - for { - select { - case <-a.forceCandidateContact: - contactCandidates() - case <-a.connectivityChan: - contactCandidates() - case t := <-a.taskChan: - // Run the task - t(a) + if a.selector != nil { + select { + case <-a.forceCandidateContact: + a.selector.ContactCandidates() + case <-a.connectivityChan: + a.selector.ContactCandidates() + case t := <-a.taskChan: + // Run the task + t(a) - case <-a.done: - return + case <-a.done: + return + } + } else { + select { + case t := <-a.taskChan: + // Run the task + t(a) + + case <-a.done: + return + } } } } @@ -449,7 +429,7 @@ func (a *Agent) pingAllCandidates() { for _, localCandidate := range localCandidates { for _, remoteCandidate := range remoteCandidates { - a.pingCandidate(localCandidate, remoteCandidate) + a.selector.PingCandidate(localCandidate, remoteCandidate) } } @@ -573,6 +553,25 @@ func (a *Agent) findRemoteCandidate(networkType NetworkType, addr net.Addr) *Can return nil } +func (a *Agent) sendBindingRequest(m *stun.Message, local, remote *Candidate) { + a.log.Tracef("ping STUN from %s to %s\n", local.String(), remote.String()) + + if overflow := len(a.pendingBindingRequests) - (maxPendingBindingRequests - 1); overflow > 0 { + a.log.Debugf("Discarded %d pending binding requests, pendingBindingRequests is full", overflow) + a.pendingBindingRequests = a.pendingBindingRequests[overflow:] + } + + _, useCandidate := m.GetOneAttribute(stun.AttrUseCandidate) + + a.pendingBindingRequests = append(a.pendingBindingRequests, bindingRequest{ + transactionID: m.TransactionID, + destination: remote.addr(), + isUseCandidate: useCandidate, + }) + + a.sendSTUN(m, local, remote) +} + func (a *Agent) sendBindingSuccess(m *stun.Message, local, remote *Candidate) { base := remote if out, err := stun.Build(stun.ClassSuccessResponse, stun.MethodBinding, m.TransactionID, @@ -595,13 +594,12 @@ func (a *Agent) sendBindingSuccess(m *stun.Message, local, remote *Candidate) { // Assert that the passed TransactionID is in our pendingBindingRequests and returns the destination // If the bindingRequest was valid remove it from our pending cache -func (a *Agent) handleInboundBindingSuccess(id []byte) (bool, net.Addr) { +func (a *Agent) handleInboundBindingSuccess(id []byte) (bool, *bindingRequest) { for i := range a.pendingBindingRequests { if bytes.Equal(a.pendingBindingRequests[i].transactionID, id) { - defer func() { - a.pendingBindingRequests = append(a.pendingBindingRequests[:i], a.pendingBindingRequests[i+1:]...) - }() - return true, a.pendingBindingRequests[i].destination + validBindingRequest := a.pendingBindingRequests[i] + a.pendingBindingRequests = append(a.pendingBindingRequests[:i], a.pendingBindingRequests[i+1:]...) + return true, &validBindingRequest } } return false, nil @@ -612,7 +610,9 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr var err error if m == nil || local == nil { return - } else if m.Method != stun.MethodBinding || !(m.Class == stun.ClassSuccessResponse || m.Class == stun.ClassRequest) { + } + + if m.Method != stun.MethodBinding || !(m.Class == stun.ClassSuccessResponse || m.Class == stun.ClassRequest) { a.log.Tracef("unhandled STUN from %s to %s class(%s) method(%s)", remote.String(), local.String(), m.Method.String(), m.Class.String()) return } @@ -639,26 +639,12 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr return } - ok, transactionAddr := a.handleInboundBindingSuccess(m.TransactionID) - if !ok { - a.log.Errorf("discard message from (%s), invalid TransactionID %s", remote, m.TransactionID) - return - } - - // Assert that NAT is not symmetric - // https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1 - if !addrEqual(transactionAddr, remote) { - a.log.Debugf("discard message: transaction source and destination does not match expected(%s), actual(%s)", transactionAddr, remote) - return - } else if remoteCandidate == nil { // Should fail previous check, better to be safe though + if remoteCandidate == nil { a.log.Warnf("discard success message from (%s), no such remote", remote) return } - a.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String()) - - // Remember the working pair and select it when receiving a success response - a.setValidPair(local, remoteCandidate, true, true) + a.selector.HandleSucessResponse(m, local, remoteCandidate, remote) } else { if err = assertInboundUsername(m, a.localUfrag+":"+a.remoteUfrag); err != nil { a.log.Warnf("discard message from (%s), %v", remote, err) @@ -688,8 +674,7 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr a.log.Tracef("inbound STUN (Request) from %s to %s", remote.String(), local.String()) - // Send success response - a.sendBindingSuccess(m, local, remoteCandidate) + a.selector.HandleBindingRequest(m, local, remoteCandidate) } remoteCandidate.seen(false) @@ -707,7 +692,7 @@ func (a *Agent) noSTUNSeen(local *Candidate, remote net.Addr) bool { return true } -func (a *Agent) getBestPair() (*candidatePair, error) { +func (a *Agent) getSelectedPair() (*candidatePair, error) { res := make(chan *candidatePair) err := a.run(func(agent *Agent) { @@ -715,10 +700,6 @@ func (a *Agent) getBestPair() (*candidatePair, error) { res <- agent.selectedPair return } - for _, p := range agent.validPairs { - res <- p - return - } res <- nil }) diff --git a/agent_test.go b/agent_test.go index 0dd945c..654a353 100644 --- a/agent_test.go +++ b/agent_test.go @@ -37,16 +37,12 @@ func TestPairSearch(t *testing.T) { t.Fatalf("TestPairSearch is only a valid test if a.validPairs is empty on construction") } - cp, err := a.getBestPair() + cp := a.getBestValidPair() if cp != nil { t.Fatalf("No Candidate pairs should exist") } - if err == nil { - t.Fatalf("An error should have been reported (with no available candidate pairs)") - } - err = a.Close() if err != nil { @@ -112,11 +108,8 @@ func TestPairPriority(t *testing.T) { } for _, remote := range []*Candidate{relayRemote, srflxRemote, prflxRemote, hostRemote} { - a.setValidPair(hostLocal, remote, false, false) - bestPair, err := a.getBestPair() - if err != nil { - t.Fatalf("Failed to get best candidate pair: %s", err) - } + a.addValidPair(hostLocal, remote) + bestPair := a.getBestValidPair() if bestPair.String() != (&candidatePair{remote: remote, local: hostLocal}).String() { t.Fatalf("Unexpected bestPair %s (expected remote: %s)", bestPair, remote) } @@ -163,23 +156,14 @@ func TestOnSelectedCandidatePairChange(t *testing.T) { // select the pair if err = a.run(func(agent *Agent) { - agent.setValidPair(hostLocal, relayRemote, true, false) + p := newCandidatePair(hostLocal, relayRemote, false) + agent.setSelectedPair(p) }); err != nil { t.Fatalf("Failed to setValidPair(): %s", err) } + // ensure that the callback fired on setting the pair <-callbackCalled - // set the same pair; this should not invoke the callback - // if the callback is invoked now it will panic due - // to second close of the channel - if err = a.run(func(agent *Agent) { - agent.setValidPair(hostLocal, relayRemote, true, false) - }); err != nil { - t.Fatalf("Failed to setValidPair(): %s", err) - } - if err := a.Close(); err != nil { - t.Fatalf("Error on agent.Close(): %s", err) - } } type BadAddr struct{} @@ -191,6 +175,18 @@ func (ba *BadAddr) String() string { return "yyy" } +func runAgentTest(t *testing.T, config *AgentConfig, task func(a *Agent)) { + a, err := NewAgent(config) + + if err != nil { + t.Fatalf("Error constructing ice.Agent") + } + + if err := a.run(task); err != nil { + t.Fatalf("Agent run failure: %v", err) + } +} + func TestHandlePeerReflexive(t *testing.T) { // Limit runtime in case of deadlocks lim := test.TimeOut(time.Second * 2) @@ -198,127 +194,120 @@ func TestHandlePeerReflexive(t *testing.T) { t.Run("UDP pflx candidate from handleInbound()", func(t *testing.T) { var config AgentConfig - a, err := NewAgent(&config) + runAgentTest(t, &config, func(a *Agent) { + a.selector = &controllingSelector{agent: a, log: a.log} + ip := net.ParseIP("192.168.0.2") + local, err := NewCandidateHost("udp", ip, 777, 1) + local.conn = &mockPacketConn{} + if err != nil { + t.Fatalf("failed to create a new candidate: %v", err) + } - if err != nil { - t.Fatalf("Error constructing ice.Agent") - } + remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999} - ip := net.ParseIP("192.168.0.2") - local, err := NewCandidateHost("udp", ip, 777, 1) - local.conn = &mockPacketConn{} - if err != nil { - t.Fatalf("failed to create a new candidate: %v", err) - } + msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(), + &stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag}, + &stun.UseCandidate{}, + &stun.IceControlling{TieBreaker: a.tieBreaker}, + &stun.Priority{Priority: local.Priority()}, + &stun.MessageIntegrity{ + Key: []byte(a.localPwd), + }, + &stun.Fingerprint{}, + ) + if err != nil { + t.Fatal(err) + } - remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999} + a.handleInbound(msg, local, remote) - msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(), - &stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag}, - &stun.UseCandidate{}, - &stun.IceControlling{TieBreaker: a.tieBreaker}, - &stun.Priority{Priority: local.Priority()}, - &stun.MessageIntegrity{ - Key: []byte(a.localPwd), - }, - &stun.Fingerprint{}, - ) - if err != nil { - t.Fatal(err) - } + // length of remote candidate list must be one now + if len(a.remoteCandidates) != 1 { + t.Fatal("failed to add a network type to the remote candidate list") + } - a.handleInbound(msg, local, remote) + // length of remote candidate list for a network type must be 1 + set := a.remoteCandidates[local.NetworkType] + if len(set) != 1 { + t.Fatal("failed to add prflx candidate to remote candidate list") + } - // length of remote candidate list must be one now - if len(a.remoteCandidates) != 1 { - t.Fatal("failed to add a network type to the remote candidate list") - } + c := set[0] - // length of remote candidate list for a network type must be 1 - set := a.remoteCandidates[local.NetworkType] - if len(set) != 1 { - t.Fatal("failed to add prflx candidate to remote candidate list") - } + if c.Type != CandidateTypePeerReflexive { + t.Fatal("candidate type must be prflx") + } - c := set[0] + if !c.IP.Equal(net.ParseIP("172.17.0.3")) { + t.Fatal("IP address mismatch") + } - if c.Type != CandidateTypePeerReflexive { - t.Fatal("candidate type must be prflx") - } + if c.Port != 999 { + t.Fatal("Port number mismatch") + } - if !c.IP.Equal(net.ParseIP("172.17.0.3")) { - t.Fatal("IP address mismatch") - } - - if c.Port != 999 { - t.Fatal("Port number mismatch") - } - - err = a.Close() - if err != nil { - t.Fatalf("Close agent emits error %v", err) - } + err = a.Close() + if err != nil { + t.Fatalf("Close agent emits error %v", err) + } + }) }) t.Run("Bad network type with handleInbound()", func(t *testing.T) { var config AgentConfig - a, err := NewAgent(&config) + runAgentTest(t, &config, func(a *Agent) { + a.selector = &controllingSelector{agent: a, log: a.log} + ip := net.ParseIP("192.168.0.2") + local, err := NewCandidateHost("tcp", ip, 777, 1) + if err != nil { + t.Fatalf("failed to create a new candidate: %v", err) + } - if err != nil { - t.Fatal("Error constructing ice.Agent") - } + remote := &BadAddr{} - ip := net.ParseIP("192.168.0.2") - local, err := NewCandidateHost("tcp", ip, 777, 1) - if err != nil { - t.Fatalf("failed to create a new candidate: %v", err) - } + a.handleInbound(nil, local, remote) - remote := &BadAddr{} + if len(a.remoteCandidates) != 0 { + t.Fatal("bad address should not be added to the remote candidate list") + } - a.handleInbound(nil, local, remote) - - if len(a.remoteCandidates) != 0 { - t.Fatal("bad address should not be added to the remote candidate list") - } - - err = a.Close() - if err != nil { - t.Fatalf("Close agent emits error %v", err) - } + err = a.Close() + if err != nil { + t.Fatalf("Close agent emits error %v", err) + } + }) }) t.Run("Success from unknown remote, prflx candidate MUST only be created via Binding Request", func(t *testing.T) { - a, err := NewAgent(&AgentConfig{}) - if err != nil { - t.Fatalf("Error constructing ice.Agent") - } + var config AgentConfig + runAgentTest(t, &config, func(a *Agent) { + a.selector = &controllingSelector{agent: a, log: a.log} + a.pendingBindingRequests = []bindingRequest{ + {[]byte("ABC"), &net.UDPAddr{}, false}, + } - a.pendingBindingRequests = []bindingRequest{ - {[]byte("ABC"), &net.UDPAddr{}}, - } + local, err := NewCandidateHost("udp", net.ParseIP("192.168.0.2"), 777, 1) + local.conn = &mockPacketConn{} + if err != nil { + t.Fatalf("failed to create a new candidate: %v", err) + } - local, err := NewCandidateHost("udp", net.ParseIP("192.168.0.2"), 777, 1) - local.conn = &mockPacketConn{} - if err != nil { - t.Fatalf("failed to create a new candidate: %v", err) - } + remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999} + msg, err := stun.Build(stun.ClassSuccessResponse, stun.MethodBinding, []byte("ABC"), + &stun.MessageIntegrity{ + Key: []byte(a.remotePwd), + }, + &stun.Fingerprint{}, + ) + if err != nil { + t.Fatal(err) + } - remote := &net.UDPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999} - msg, err := stun.Build(stun.ClassSuccessResponse, stun.MethodBinding, []byte("ABC"), - &stun.MessageIntegrity{ - Key: []byte(a.remotePwd), - }, - &stun.Fingerprint{}, - ) - if err != nil { - t.Fatal(err) - } - - a.handleInbound(msg, local, remote) - if len(a.remoteCandidates) != 0 { - t.Fatal("unknown remote was able to create a candidate") - } + a.handleInbound(msg, local, remote) + if len(a.remoteCandidates) != 0 { + t.Fatal("unknown remote was able to create a candidate") + } + }) }) } @@ -431,32 +420,38 @@ func TestInboundValidity(t *testing.T) { t.Fatalf("Error constructing ice.Agent") } - a.handleInbound(buildMsg(stun.ClassRequest, a.localUfrag+":"+a.remoteUfrag, a.localPwd), local, remote) - if len(a.remoteCandidates) != 1 { - t.Fatal("Binding with valid values was unable to create prflx candidate") + err = a.run(func(a *Agent) { + a.selector = &controllingSelector{agent: a, log: a.log} + a.handleInbound(buildMsg(stun.ClassRequest, a.localUfrag+":"+a.remoteUfrag, a.localPwd), local, remote) + if len(a.remoteCandidates) != 1 { + t.Fatal("Binding with valid values was unable to create prflx candidate") + } + }) + + if err != nil { + t.Fatalf("Agent run failure: %v", err) } }) t.Run("Valid bind without fingerprint", func(t *testing.T) { - a, err := NewAgent(&AgentConfig{}) - if err != nil { - t.Fatalf("Error constructing ice.Agent") - } + var config AgentConfig + runAgentTest(t, &config, func(a *Agent) { + a.selector = &controllingSelector{agent: a, log: a.log} + msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(), + &stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag}, + &stun.MessageIntegrity{ + Key: []byte(a.localPwd), + }, + ) + if err != nil { + t.Fatal(err) + } - msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionID(), - &stun.Username{Username: a.localUfrag + ":" + a.remoteUfrag}, - &stun.MessageIntegrity{ - Key: []byte(a.localPwd), - }, - ) - if err != nil { - t.Fatal(err) - } - - a.handleInbound(msg, local, remote) - if len(a.remoteCandidates) != 1 { - t.Fatal("Binding with valid values (but no fingerprint) was unable to create prflx candidate") - } + a.handleInbound(msg, local, remote) + if len(a.remoteCandidates) != 1 { + t.Fatal("Binding with valid values (but no fingerprint) was unable to create prflx candidate") + } + }) }) t.Run("Success with invalid TransactionID", func(t *testing.T) { diff --git a/selection.go b/selection.go new file mode 100644 index 0000000..71fd8ba --- /dev/null +++ b/selection.go @@ -0,0 +1,220 @@ +package ice + +import ( + "net" + + "github.com/pion/logging" + "github.com/pion/stun" +) + +type pairCandidateSelector interface { + Start() + ContactCandidates() + PingCandidate(local, remote *Candidate) + HandleSucessResponse(m *stun.Message, local, remote *Candidate, remoteAddr net.Addr) + HandleBindingRequest(m *stun.Message, local, remote *Candidate) +} + +type controllingSelector struct { + agent *Agent + nominatedPair *candidatePair + log logging.LeveledLogger +} + +func (s *controllingSelector) Start() { +} + +func (s *controllingSelector) ContactCandidates() { + switch { + case s.agent.selectedPair != nil: + if s.agent.validateSelectedPair() { + s.log.Trace("checking keepalive") + s.agent.checkKeepalive() + } + case s.nominatedPair != nil: + s.nominatePair(s.nominatedPair) + default: + s.log.Trace("pinging all candidates") + s.agent.pingAllCandidates() + } +} + +func (s *controllingSelector) nominatePair(pair *candidatePair) { + transactionID := stun.GenerateTransactionID() + + // The controlling agent MUST include the USE-CANDIDATE attribute in + // order to nominate a candidate pair (Section 8.1.1). The controlled + // agent MUST NOT include the USE-CANDIDATE attribute in a Binding + // request. + msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID, + &stun.Username{Username: s.agent.remoteUfrag + ":" + s.agent.localUfrag}, + &stun.UseCandidate{}, + &stun.IceControlling{TieBreaker: s.agent.tieBreaker}, + &stun.Priority{Priority: pair.local.Priority()}, + &stun.MessageIntegrity{ + Key: []byte(s.agent.remotePwd), + }, + &stun.Fingerprint{}, + ) + + if err != nil { + s.log.Error(err.Error()) + return + } + + s.log.Tracef("ping STUN (nominate candidate pair) from %s to %s\n", pair.local.String(), pair.remote.String()) + s.agent.sendBindingRequest(msg, pair.local, pair.remote) +} + +func (s *controllingSelector) HandleBindingRequest(m *stun.Message, local, remote *Candidate) { + s.agent.sendBindingSuccess(m, local, remote) + + p := s.agent.findValidPair(local, remote) + if p != nil && s.nominatedPair == nil && s.agent.selectedPair == nil { + s.nominatedPair = p + s.nominatePair(p) + } +} + +func (s *controllingSelector) HandleSucessResponse(m *stun.Message, local, remote *Candidate, remoteAddr net.Addr) { + ok, pendingRequest := s.agent.handleInboundBindingSuccess(m.TransactionID) + if !ok { + s.log.Errorf("discard message from (%s), invalid TransactionID %s", remote, m.TransactionID) + return + } + + transactionAddr := pendingRequest.destination + + // Assert that NAT is not symmetric + // https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1 + if !addrEqual(transactionAddr, remoteAddr) { + s.log.Debugf("discard message: transaction source and destination does not match expected(%s), actual(%s)", transactionAddr, remote) + return + } + + s.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String()) + p := s.agent.addValidPair(local, remote) + + if pendingRequest.isUseCandidate { + s.agent.setSelectedPair(p) + } +} + +func (s *controllingSelector) PingCandidate(local, remote *Candidate) { + transactionID := stun.GenerateTransactionID() + + msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID, + &stun.Username{Username: s.agent.remoteUfrag + ":" + s.agent.localUfrag}, + &stun.IceControlling{TieBreaker: s.agent.tieBreaker}, + &stun.Priority{Priority: local.Priority()}, + &stun.MessageIntegrity{ + Key: []byte(s.agent.remotePwd), + }, + &stun.Fingerprint{}, + ) + + if err != nil { + s.log.Error(err.Error()) + return + } + + s.agent.sendBindingRequest(msg, local, remote) +} + +type controlledSelector struct { + agent *Agent + log logging.LeveledLogger +} + +func (s *controlledSelector) Start() {} + +func (s *controlledSelector) ContactCandidates() { + if s.agent.selectedPair != nil { + if s.agent.validateSelectedPair() { + s.log.Trace("checking keepalive") + s.agent.checkKeepalive() + } + } else { + s.log.Trace("pinging all candidates") + s.agent.pingAllCandidates() + } +} + +func (s *controlledSelector) PingCandidate(local, remote *Candidate) { + transactionID := stun.GenerateTransactionID() + + msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, transactionID, + &stun.Username{Username: s.agent.remoteUfrag + ":" + s.agent.localUfrag}, + &stun.IceControlled{TieBreaker: s.agent.tieBreaker}, + &stun.Priority{Priority: local.Priority()}, + &stun.MessageIntegrity{ + Key: []byte(s.agent.remotePwd), + }, + &stun.Fingerprint{}, + ) + + if err != nil { + s.log.Error(err.Error()) + return + } + + s.agent.sendBindingRequest(msg, local, remote) +} + +func (s *controlledSelector) HandleSucessResponse(m *stun.Message, local, remote *Candidate, remoteAddr net.Addr) { + // TODO according to the standard we should specifically answer a failed nomination: + // https://tools.ietf.org/html/rfc8445#section-7.3.1.5 + // If the controlled agent does not accept the request from the + // controlling agent, the controlled agent MUST reject the nomination + // request with an appropriate error code response (e.g., 400) + // [RFC5389]. + + ok, pendingRequest := s.agent.handleInboundBindingSuccess(m.TransactionID) + if !ok { + s.log.Errorf("discard message from (%s), invalid TransactionID %s", remote, m.TransactionID) + return + } + + transactionAddr := pendingRequest.destination + + // Assert that NAT is not symmetric + // https://tools.ietf.org/html/rfc8445#section-7.2.5.2.1 + if !addrEqual(transactionAddr, remoteAddr) { + s.log.Debugf("discard message: transaction source and destination does not match expected(%s), actual(%s)", transactionAddr, remote) + return + } + + s.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String()) + s.agent.addValidPair(local, remote) +} + +func (s *controlledSelector) HandleBindingRequest(m *stun.Message, local, remote *Candidate) { + _, useCandidate := m.GetOneAttribute(stun.AttrUseCandidate) + + if useCandidate { + // https://tools.ietf.org/html/rfc8445#section-7.3.1.5 + p := s.agent.findValidPair(local, remote) + + if p != nil { + // If the state of this pair is Succeeded, it means that the check + // previously sent by this pair produced a successful response and + // generated a valid pair (Section 7.2.5.3.2). The agent sets the + // nominated flag value of the valid pair to true. + s.agent.setSelectedPair(p) + s.agent.sendBindingSuccess(m, local, remote) + } else { + // If the received Binding request triggered a new check to be + // enqueued in the triggered-check queue (Section 7.3.1.4), once the + // check is sent and if it generates a successful response, and + // generates a valid pair, the agent sets the nominated flag of the + // pair to true. If the request fails (Section 7.2.5.2), the agent + // MUST remove the candidate pair from the valid list, set the + // candidate pair state to Failed, and set the checklist state to + // Failed. + s.PingCandidate(local, remote) + } + } else { + s.agent.sendBindingSuccess(m, local, remote) + s.PingCandidate(local, remote) + } +} diff --git a/transport.go b/transport.go index 4253a43..a74cefb 100644 --- a/transport.go +++ b/transport.go @@ -75,7 +75,7 @@ func (c *Conn) Write(p []byte) (int, error) { return 0, errors.New("the ICE conn can't write STUN messages") } - pair, err := c.agent.getBestPair() + pair, err := c.agent.getSelectedPair() if err != nil { return 0, err }