diff --git a/agent.go b/agent.go index 8f53659..8ce8fd1 100644 --- a/agent.go +++ b/agent.go @@ -3,8 +3,6 @@ package ice import ( - "bytes" - "encoding/binary" "fmt" "math/rand" "net" @@ -12,8 +10,6 @@ import ( "sync" "time" - "errors" - "github.com/pion/logging" "github.com/pion/stun" "github.com/pion/transport/packetio" @@ -35,6 +31,18 @@ const ( stunAttrHeaderLength = 4 ) +type candidatePairs []*candidatePair + +func (cp candidatePairs) Len() int { return len(cp) } +func (cp candidatePairs) Swap(i, j int) { cp[i], cp[j] = cp[j], cp[i] } + +type byPairPriority struct{ candidatePairs } + +// NB: Reverse sort so our candidates start at highest priority +func (bp byPairPriority) Less(i, j int) bool { + return bp.candidatePairs[i].Priority() > bp.candidatePairs[j].Priority() +} + // Agent represents the ICE agent type Agent struct { onConnectionStateChangeHdlr func(ConnectionState) @@ -196,8 +204,8 @@ func NewAgent(config *AgentConfig) (*Agent, error) { } // Initialize local candidates - a.gatherCandidatesLocal(config.NetworkTypes) - a.gatherCandidatesReflective(config.Urls, config.NetworkTypes) + gatherCandidatesLocal(a, config.NetworkTypes) + gatherCandidatesReflective(a, config.Urls, config.NetworkTypes) go a.taskLoop() return a, nil @@ -226,130 +234,6 @@ func (a *Agent) onSelectedCandidatePairChange(p *candidatePair) { } } -func (a *Agent) listenUDP(network string, laddr *net.UDPAddr) (*net.UDPConn, error) { - if (laddr.Port != 0) || ((a.portmin == 0) && (a.portmax == 0)) { - return net.ListenUDP(network, laddr) - } - var i, j int - i = int(a.portmin) - if i == 0 { - i = 1 - } - j = int(a.portmax) - if j == 0 { - j = 0xFFFF - } - for i <= j { - c, e := net.ListenUDP(network, &net.UDPAddr{IP: laddr.IP, Port: i}) - if e == nil { - return c, e - } - i++ - } - return nil, ErrPort -} - -func (a *Agent) gatherCandidatesLocal(networkTypes []NetworkType) { - localIPs := localInterfaces(networkTypes) - for _, ip := range localIPs { - for _, network := range supportedNetworks { - conn, err := a.listenUDP(network, &net.UDPAddr{IP: ip, Port: 0}) - if err != nil { - a.log.Warnf("could not listen %s %s\n", network, ip) - continue - } - - port := conn.LocalAddr().(*net.UDPAddr).Port - c, err := NewCandidateHost(network, ip, port, ComponentRTP) - if err != nil { - a.log.Warnf("Failed to create host candidate: %s %s %d: %v\n", network, ip, port, err) - continue - } - - networkType := c.NetworkType - set := a.localCandidates[networkType] - set = append(set, c) - a.localCandidates[networkType] = set - - c.start(a, conn) - } - } -} - -func (a *Agent) gatherCandidatesReflective(urls []*URL, networkTypes []NetworkType) { - for _, networkType := range networkTypes { - network := networkType.String() - for _, url := range urls { - switch url.Scheme { - case SchemeTypeSTUN: - laddr, xoraddr, err := allocateUDP(network, url) - if err != nil { - a.log.Warnf("could not allocate %s %s: %v\n", network, url, err) - continue - } - conn, err := net.ListenUDP(network, laddr) - if err != nil { - a.log.Warnf("could not listen %s %s: %v\n", network, laddr, err) - } - - ip := xoraddr.IP - port := xoraddr.Port - relIP := laddr.IP.String() - relPort := laddr.Port - c, err := NewCandidateServerReflexive(network, ip, port, ComponentRTP, relIP, relPort) - if err != nil { - a.log.Warnf("Failed to create server reflexive candidate: %s %s %d: %v\n", network, ip, port, err) - continue - } - - networkType := c.NetworkType - set := a.localCandidates[networkType] - set = append(set, c) - a.localCandidates[networkType] = set - - c.start(a, conn) - - default: - a.log.Warnf("scheme %s is not implemented\n", url.Scheme) - continue - } - } - } -} - -func allocateUDP(network string, url *URL) (*net.UDPAddr, *stun.XorAddress, error) { - // TODO Do we want the timeout to be configurable? - client, err := stun.NewClient(network, fmt.Sprintf("%s:%d", url.Host, url.Port), time.Second*5) - if err != nil { - return nil, nil, flattenErrs([]error{errors.New("failed to create STUN client"), err}) - } - localAddr, ok := client.LocalAddr().(*net.UDPAddr) - if !ok { - return nil, nil, fmt.Errorf("failed to cast STUN client to UDPAddr") - } - - resp, err := client.Request() - if err != nil { - return nil, nil, flattenErrs([]error{errors.New("failed to make STUN request"), err}) - } - - if err = client.Close(); err != nil { - return nil, nil, flattenErrs([]error{errors.New("failed to close STUN client"), err}) - } - - attr, ok := resp.GetOneAttribute(stun.AttrXORMappedAddress) - if !ok { - return nil, nil, fmt.Errorf("got response from STUN server that did not contain XORAddress") - } - - var addr stun.XorAddress - if err = addr.Unpack(resp, attr); err != nil { - return nil, nil, flattenErrs([]error{errors.New("failed to unpack STUN XorAddress response"), err}) - } - - return localAddr, &addr, nil -} - func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remotePwd string) error { switch { case a.haveStarted: @@ -429,18 +313,6 @@ func (a *Agent) updateConnectionState(newState ConnectionState) { } } -type candidatePairs []*candidatePair - -func (cp candidatePairs) Len() int { return len(cp) } -func (cp candidatePairs) Swap(i, j int) { cp[i], cp[j] = cp[j], cp[i] } - -type byPairPriority struct{ candidatePairs } - -// NB: Reverse sort so our candidates start at highest priority -func (bp byPairPriority) Less(i, j int) bool { - return bp.candidatePairs[i].Priority() > bp.candidatePairs[j].Priority() -} - func (a *Agent) setValidPair(local, remote *Candidate, selected, controlling bool) { // TODO: avoid duplicates p := newCandidatePair(local, remote, controlling) @@ -737,98 +609,6 @@ func (a *Agent) handleInboundControlling(m *stun.Message, localCandidate, remote } } -// handleNewPeerReflexiveCandidate adds an unseen remote transport address -// to the remote candidate list as a peer-reflexive candidate. -func (a *Agent) handleNewPeerReflexiveCandidate(local *Candidate, remote net.Addr) error { - var ip net.IP - var port int - - switch addr := remote.(type) { - case *net.UDPAddr: - ip = addr.IP - port = addr.Port - case *net.TCPAddr: - ip = addr.IP - port = addr.Port - default: - return fmt.Errorf("unsupported address type %T", addr) - } - - pflxCandidate, err := NewCandidatePeerReflexive( - local.NetworkType.String(), // assume, same as that of local - ip, - port, - local.Component, - "", // unknown at this moment. TODO: need a review - 0, // unknown at this moment. TODO: need a review - ) - - if err != nil { - return flattenErrs([]error{fmt.Errorf("failed to create peer-reflexive candidate: %v", remote), err}) - } - - // Add pflxCandidate to the remote candidate list - a.addRemoteCandidate(pflxCandidate) - return nil -} - -func (a *Agent) assertInboundUsername(m *stun.Message) error { - usernameAttr := &stun.Username{} - usernameRawAttr, usernameFound := m.GetOneAttribute(stun.AttrUsername) - - if !usernameFound { - return fmt.Errorf("inbound packet missing Username") - } else if err := usernameAttr.Unpack(m, usernameRawAttr); err != nil { - return err - } - - expectedUsername := a.localUfrag + ":" + a.remoteUfrag - if usernameAttr.Username != expectedUsername { - return fmt.Errorf("username mismatch expected(%x) actual(%x)", expectedUsername, usernameAttr.Username) - } - - return nil -} - -func (a *Agent) assertInboundMessageIntegrity(m *stun.Message, key []byte) error { - messageIntegrityAttr := &stun.MessageIntegrity{} - messageIntegrityRawAttr, messageIntegrityAttrFound := m.GetOneAttribute(stun.AttrMessageIntegrity) - - if !messageIntegrityAttrFound { - return fmt.Errorf("inbound packet missing MessageIntegrity") - } else if err := messageIntegrityAttr.Unpack(m, messageIntegrityRawAttr); err != nil { - return err - } - - tailLength := messageIntegrityRawAttr.Length + stunAttrHeaderLength - rawCopy := make([]byte, len(m.Raw)) - copy(rawCopy, m.Raw) - - // If we have a fingerprint we need to exclude it from the MessageIntegrity computation - if rawFingerprint, hasFingerprint := m.GetOneAttribute(stun.AttrFingerprint); hasFingerprint { - fingerprintLength := rawFingerprint.Length + stunAttrHeaderLength - tailLength += fingerprintLength - - // Rewrite the packet header to be new length (excluding values we don't care about) - currLength := binary.BigEndian.Uint16(rawCopy[2:4]) - binary.BigEndian.PutUint16(rawCopy[2:], currLength-fingerprintLength) - } - - lengthToHash := len(rawCopy) - int(tailLength) - if lengthToHash < 1 { - return fmt.Errorf("unable to assert MessageIntegrity, length calculation failed (%d)", lengthToHash) - } - - computedMessageIntegrity, err := stun.MessageIntegrityCalculateHMAC(key, rawCopy[:lengthToHash]) - if err != nil { - return err - } else if !bytes.Equal(computedMessageIntegrity, messageIntegrityRawAttr.Value) { - return fmt.Errorf("messageIntegrity mismatch expected(%x) actual(%x)", computedMessageIntegrity, messageIntegrityRawAttr.Value) - } - - return nil -} - // handleInbound processes STUN traffic from a remote candidate func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr) { if m == nil || local == nil { @@ -838,15 +618,15 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr switch { case m.Method == stun.MethodBinding && m.Class == stun.ClassSuccessResponse: - if err := a.assertInboundMessageIntegrity(m, []byte(a.remotePwd)); err != nil { + if err := assertInboundMessageIntegrity(m, []byte(a.remotePwd)); err != nil { a.log.Warnf("discard message from (%s), %v", remote, err) return } case m.Method == stun.MethodBinding && m.Class == stun.ClassRequest: - if err := a.assertInboundUsername(m); err != nil { + if err := assertInboundUsername(m, a.localUfrag+":"+a.remoteUfrag); err != nil { a.log.Warnf("discard message from (%s), %v", remote, err) return - } else if err := a.assertInboundMessageIntegrity(m, []byte(a.localPwd)); err != nil { + } else if err := assertInboundMessageIntegrity(m, []byte(a.localPwd)); err != nil { a.log.Warnf("discard message from (%s), %v", remote, err) return } @@ -857,11 +637,12 @@ func (a *Agent) handleInbound(m *stun.Message, local *Candidate, remote net.Addr remoteCandidate := a.findRemoteCandidate(local.NetworkType, remote) if remoteCandidate == nil { a.log.Debugf("detected a new peer-reflexive candiate: %s ", remote) - err := a.handleNewPeerReflexiveCandidate(local, remote) + pflxCandidate, err := handleNewPeerReflexiveCandidate(local, remote) if err != nil { - // Log warning, then move on.. a.log.Warn(err.Error()) } + + a.addRemoteCandidate(pflxCandidate) return } remoteCandidate.seen(false) diff --git a/agent_test.go b/agent_test.go index a3597eb..210823b 100644 --- a/agent_test.go +++ b/agent_test.go @@ -291,10 +291,11 @@ func TestHandlePeerReflexive(t *testing.T) { remote := &net.TCPAddr{IP: net.ParseIP("172.17.0.3"), Port: 999} - err = a.handleNewPeerReflexiveCandidate(local, remote) + candidate, err := handleNewPeerReflexiveCandidate(local, remote) if err != nil { t.Fatalf("handleNewPeerReflexiveCandidate() should not fail: %v", err) } + a.addRemoteCandidate(candidate) // length of remote candidate list must be one now if len(a.remoteCandidates) != 1 { diff --git a/candidate.go b/candidate.go index 2eaa49f..3edab15 100644 --- a/candidate.go +++ b/candidate.go @@ -117,6 +117,39 @@ func NewCandidateRelay(network string, ip net.IP, port int, component uint16, re }, nil } +// handleNewPeerReflexiveCandidate creates a ReflexiveCandidate from a remote transport address +func handleNewPeerReflexiveCandidate(local *Candidate, remote net.Addr) (*Candidate, error) { + var ip net.IP + var port int + + switch addr := remote.(type) { + case *net.UDPAddr: + ip = addr.IP + port = addr.Port + case *net.TCPAddr: + ip = addr.IP + port = addr.Port + default: + return nil, fmt.Errorf("unsupported address type %T", addr) + } + + pflxCandidate, err := NewCandidatePeerReflexive( + local.NetworkType.String(), // assume, same as that of local + ip, + port, + local.Component, + "", // unknown at this moment. TODO: need a review + 0, // unknown at this moment. TODO: need a review + ) + + if err != nil { + return nil, flattenErrs([]error{fmt.Errorf("failed to create peer-reflexive candidate: %v", remote), err}) + } + + // Add pflxCandidate to the remote candidate list + return pflxCandidate, nil +} + // start runs the candidate using the provided connection func (c *Candidate) start(a *Agent, conn net.PacketConn) { c.agent = a diff --git a/gather.go b/gather.go new file mode 100644 index 0000000..fed2056 --- /dev/null +++ b/gather.go @@ -0,0 +1,153 @@ +package ice + +import "net" + +func localInterfaces(networkTypes []NetworkType) (ips []net.IP) { + ifaces, err := net.Interfaces() + if err != nil { + return ips + } + + var IPv4Requested, IPv6Requested bool + for _, typ := range networkTypes { + if typ.IsIPv4() { + IPv4Requested = true + } + + if typ.IsIPv6() { + IPv6Requested = true + } + } + + for _, iface := range ifaces { + if iface.Flags&net.FlagUp == 0 { + continue // interface down + } + if iface.Flags&net.FlagLoopback != 0 { + continue // loopback interface + } + + addrs, err := iface.Addrs() + if err != nil { + return ips + } + + for _, addr := range addrs { + var ip net.IP + switch addr := addr.(type) { + case *net.IPNet: + ip = addr.IP + case *net.IPAddr: + ip = addr.IP + + } + if ip == nil || ip.IsLoopback() { + continue + } + + if ipv4 := ip.To4(); ipv4 == nil { + if !IPv6Requested { + continue + } else if !isSupportedIPv6(ip) { + continue + } + } else if !IPv4Requested { + continue + } + + ips = append(ips, ip) + } + } + return ips +} + +func listenUDP(portMax, portMin int, network string, laddr *net.UDPAddr) (*net.UDPConn, error) { + if (laddr.Port != 0) || ((portMin == 0) && (portMax == 0)) { + return net.ListenUDP(network, laddr) + } + var i, j int + i = portMin + if i == 0 { + i = 1 + } + j = portMax + if j == 0 { + j = 0xFFFF + } + for i <= j { + c, e := net.ListenUDP(network, &net.UDPAddr{IP: laddr.IP, Port: i}) + if e == nil { + return c, e + } + i++ + } + return nil, ErrPort +} + +func gatherCandidatesLocal(a *Agent, networkTypes []NetworkType) { + localIPs := localInterfaces(networkTypes) + for _, ip := range localIPs { + for _, network := range supportedNetworks { + conn, err := listenUDP(int(a.portmax), int(a.portmin), network, &net.UDPAddr{IP: ip, Port: 0}) + if err != nil { + a.log.Warnf("could not listen %s %s\n", network, ip) + continue + } + + port := conn.LocalAddr().(*net.UDPAddr).Port + c, err := NewCandidateHost(network, ip, port, ComponentRTP) + if err != nil { + a.log.Warnf("Failed to create host candidate: %s %s %d: %v\n", network, ip, port, err) + continue + } + + networkType := c.NetworkType + set := a.localCandidates[networkType] + set = append(set, c) + a.localCandidates[networkType] = set + + c.start(a, conn) + } + } +} + +func gatherCandidatesReflective(a *Agent, urls []*URL, networkTypes []NetworkType) { + for _, networkType := range networkTypes { + network := networkType.String() + for _, url := range urls { + switch url.Scheme { + case SchemeTypeSTUN: + laddr, xoraddr, err := allocateUDP(network, url) + if err != nil { + a.log.Warnf("could not allocate %s %s: %v\n", network, url, err) + continue + } + conn, err := net.ListenUDP(network, laddr) + if err != nil { + a.log.Warnf("could not listen %s %s: %v\n", network, laddr, err) + } + + ip := xoraddr.IP + port := xoraddr.Port + relIP := laddr.IP.String() + relPort := laddr.Port + c, err := NewCandidateServerReflexive(network, ip, port, ComponentRTP, relIP, relPort) + if err != nil { + a.log.Warnf("Failed to create server reflexive candidate: %s %s %d: %v\n", network, ip, port, err) + continue + } + + networkType := c.NetworkType + set := a.localCandidates[networkType] + set = append(set, c) + a.localCandidates[networkType] = set + + c.start(a, conn) + + default: + a.log.Warnf("scheme %s is not implemented\n", url.Scheme) + continue + } + } + } +} diff --git a/stun.go b/stun.go new file mode 100644 index 0000000..b7627cf --- /dev/null +++ b/stun.go @@ -0,0 +1,101 @@ +package ice + +import ( + "bytes" + "encoding/binary" + "errors" + "fmt" + "net" + "time" + + "github.com/pion/stun" +) + +func assertInboundUsername(m *stun.Message, expectedUsername string) error { + usernameAttr := &stun.Username{} + usernameRawAttr, usernameFound := m.GetOneAttribute(stun.AttrUsername) + + if !usernameFound { + return fmt.Errorf("inbound packet missing Username") + } else if err := usernameAttr.Unpack(m, usernameRawAttr); err != nil { + return err + } + + if usernameAttr.Username != expectedUsername { + return fmt.Errorf("username mismatch expected(%x) actual(%x)", expectedUsername, usernameAttr.Username) + } + + return nil +} + +func assertInboundMessageIntegrity(m *stun.Message, key []byte) error { + messageIntegrityAttr := &stun.MessageIntegrity{} + messageIntegrityRawAttr, messageIntegrityAttrFound := m.GetOneAttribute(stun.AttrMessageIntegrity) + + if !messageIntegrityAttrFound { + return fmt.Errorf("inbound packet missing MessageIntegrity") + } else if err := messageIntegrityAttr.Unpack(m, messageIntegrityRawAttr); err != nil { + return err + } + + tailLength := messageIntegrityRawAttr.Length + stunAttrHeaderLength + rawCopy := make([]byte, len(m.Raw)) + copy(rawCopy, m.Raw) + + // If we have a fingerprint we need to exclude it from the MessageIntegrity computation + if rawFingerprint, hasFingerprint := m.GetOneAttribute(stun.AttrFingerprint); hasFingerprint { + fingerprintLength := rawFingerprint.Length + stunAttrHeaderLength + tailLength += fingerprintLength + + // Rewrite the packet header to be new length (excluding values we don't care about) + currLength := binary.BigEndian.Uint16(rawCopy[2:4]) + binary.BigEndian.PutUint16(rawCopy[2:], currLength-fingerprintLength) + } + + lengthToHash := len(rawCopy) - int(tailLength) + if lengthToHash < 1 { + return fmt.Errorf("unable to assert MessageIntegrity, length calculation failed (%d)", lengthToHash) + } + + computedMessageIntegrity, err := stun.MessageIntegrityCalculateHMAC(key, rawCopy[:lengthToHash]) + if err != nil { + return err + } else if !bytes.Equal(computedMessageIntegrity, messageIntegrityRawAttr.Value) { + return fmt.Errorf("messageIntegrity mismatch expected(%x) actual(%x)", computedMessageIntegrity, messageIntegrityRawAttr.Value) + } + + return nil +} + +func allocateUDP(network string, url *URL) (*net.UDPAddr, *stun.XorAddress, error) { + // TODO Do we want the timeout to be configurable? + client, err := stun.NewClient(network, fmt.Sprintf("%s:%d", url.Host, url.Port), time.Second*5) + if err != nil { + return nil, nil, flattenErrs([]error{errors.New("failed to create STUN client"), err}) + } + localAddr, ok := client.LocalAddr().(*net.UDPAddr) + if !ok { + return nil, nil, fmt.Errorf("failed to cast STUN client to UDPAddr") + } + + resp, err := client.Request() + if err != nil { + return nil, nil, flattenErrs([]error{errors.New("failed to make STUN request"), err}) + } + + if err = client.Close(); err != nil { + return nil, nil, flattenErrs([]error{errors.New("failed to close STUN client"), err}) + } + + attr, ok := resp.GetOneAttribute(stun.AttrXORMappedAddress) + if !ok { + return nil, nil, fmt.Errorf("got response from STUN server that did not contain XORAddress") + } + + var addr stun.XorAddress + if err = addr.Unpack(resp, attr); err != nil { + return nil, nil, flattenErrs([]error{errors.New("failed to unpack STUN XorAddress response"), err}) + } + + return localAddr, &addr, nil +} diff --git a/util.go b/util.go index 0d565f2..59fbf99 100644 --- a/util.go +++ b/util.go @@ -9,65 +9,6 @@ import ( "time" ) -func localInterfaces(networkTypes []NetworkType) (ips []net.IP) { - ifaces, err := net.Interfaces() - if err != nil { - return ips - } - - var IPv4Requested, IPv6Requested bool - for _, typ := range networkTypes { - if typ.IsIPv4() { - IPv4Requested = true - } - - if typ.IsIPv6() { - IPv6Requested = true - } - } - - for _, iface := range ifaces { - if iface.Flags&net.FlagUp == 0 { - continue // interface down - } - if iface.Flags&net.FlagLoopback != 0 { - continue // loopback interface - } - - addrs, err := iface.Addrs() - if err != nil { - return ips - } - - for _, addr := range addrs { - var ip net.IP - switch addr := addr.(type) { - case *net.IPNet: - ip = addr.IP - case *net.IPAddr: - ip = addr.IP - - } - if ip == nil || ip.IsLoopback() { - continue - } - - if ipv4 := ip.To4(); ipv4 == nil { - if !IPv6Requested { - continue - } else if !isSupportedIPv6(ip) { - continue - } - } else if !IPv4Requested { - continue - } - - ips = append(ips, ip) - } - } - return ips -} - type atomicError struct{ v atomic.Value } func (a *atomicError) Store(err error) {