netbird/ice
0
0
Fork 0
mirror of https://github.com/netbirdio/ice.git synced 2026-05-22 17:10:58 -07:00
Code Issues Packages Projects Releases Wiki Activity

Implement regular nomination

Browse Source 
In this first iteration we nominate the first valid pair we found.
This commit is contained in:
Hugo Arregui
2019-05-02 14:32:27 -03:00
committed by Hugo Arregui
parent ea5cdd03c0
commit cafd8d7860
4 changed files with 460 additions and 264 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent.go
+104 -123
View File
@@ -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
})
agent_test.go
+135 -140
View File
@@ -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) {
selection.go
+220
View File
@@ -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)
}
}
transport.go
+1 -1
View File
@@ -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
}