From 2a96a308d5cbd7abe51e0ab1f60725c9583734ef Mon Sep 17 00:00:00 2001 From: Sean DuBois Date: Sun, 26 May 2019 00:56:10 -0700 Subject: [PATCH] Initial TURN implementation This implementation is not fully tested, and we don't handle all error case yet. We will continue to work on it though. Tests for send/recv and shutdown are in the works. Relates to #47 --- agent.go | 20 ++++++ candidate.go | 1 + candidate_base.go | 79 ++++++++++++---------- candidate_relay.go | 53 +++++++++++++++ gather.go | 159 +++++++++++++++++++++++++++++++-------------- 5 files changed, 229 insertions(+), 83 deletions(-) diff --git a/agent.go b/agent.go index e2d56c1..a60ace3 100644 --- a/agent.go +++ b/agent.go @@ -33,6 +33,10 @@ const ( maxPendingBindingRequests = 50 ) +var ( + defaultCandidateTypes = []CandidateType{CandidateTypeHost, CandidateTypeServerReflexive, CandidateTypeRelay} +) + type candidatePairs []*candidatePair func (cp candidatePairs) Len() int { return len(cp) } @@ -80,6 +84,8 @@ type Agent struct { portmin uint16 portmax uint16 + candidateTypes []CandidateType + // How long should a pair stay quiet before we declare it dead? // 0 means never timeout connectionTimeout time.Duration @@ -233,6 +239,12 @@ func NewAgent(config *AgentConfig) (*Agent, error) { a.taskLoopInterval = config.taskLoopInterval } + if config.CandidateTypes == nil || len(config.CandidateTypes) == 0 { + a.candidateTypes = defaultCandidateTypes + } else { + a.candidateTypes = config.CandidateTypes + } + go a.taskLoop() // Initialize local candidates @@ -478,6 +490,14 @@ func (a *Agent) addRemoteCandidate(c Candidate) { set = append(set, c) a.remoteCandidates[c.NetworkType()] = set + + for _, l := range a.localCandidates[NetworkTypeUDP4] { + if localRelay, ok := l.(*CandidateRelay); ok { + if err := localRelay.addPermission(c); err != nil { + a.log.Errorf("Failed to create TURN permission %v", err) + } + } + } } // GetLocalCandidates returns the local candidates diff --git a/candidate.go b/candidate.go index 21cb749..e073c21 100644 --- a/candidate.go +++ b/candidate.go @@ -31,6 +31,7 @@ type Candidate interface { Equal(other Candidate) bool addr() net.Addr + agent() *Agent close() error seen(outbound bool) diff --git a/candidate_base.go b/candidate_base.go index c394c01..89dad86 100644 --- a/candidate_base.go +++ b/candidate_base.go @@ -6,6 +6,7 @@ import ( "sync" "time" + "github.com/pion/logging" "github.com/pion/stun" ) @@ -22,10 +23,10 @@ type candidateBase struct { lastSent time.Time lastReceived time.Time - agent *Agent - conn net.PacketConn - closeCh chan struct{} - closedCh chan struct{} + currAgent *Agent + conn net.PacketConn + closeCh chan struct{} + closedCh chan struct{} } // IP returns Candidate IP @@ -65,7 +66,7 @@ func (c *candidateBase) RelatedAddress() *CandidateRelatedAddress { // start runs the candidate using the provided connection func (c *candidateBase) start(a *Agent, conn net.PacketConn) { - c.agent = a + c.currAgent = a c.conn = conn c.closeCh = make(chan struct{}) c.closedCh = make(chan struct{}) @@ -78,7 +79,7 @@ func (c *candidateBase) recvLoop() { close(c.closedCh) }() - log := c.agent.log + log := c.agent().log buffer := make([]byte, receiveMTU) for { n, srcAddr, err := c.conn.ReadFrom(buffer) @@ -86,43 +87,47 @@ func (c *candidateBase) recvLoop() { return } - if stun.IsMessage(buffer[:n]) { - m := &stun.Message{ - Raw: make([]byte, n), - } - // Explicitly copy raw buffer so Message can own the memory. - copy(m.Raw, buffer[:n]) - if err = m.Decode(); err != nil { - log.Warnf("Failed to handle decode ICE from %s to %s: %v", c.addr(), srcAddr, err) - continue - } - err = c.agent.run(func(agent *Agent) { - agent.handleInbound(m, c, srcAddr) - }) - if err != nil { - log.Warnf("Failed to handle message: %v", err) - } + handleInboundCandidateMsg(c, buffer[:n], srcAddr, log) + } +} - continue +func handleInboundCandidateMsg(c Candidate, buffer []byte, srcAddr net.Addr, log logging.LeveledLogger) { + if stun.IsMessage(buffer) { + m := &stun.Message{ + Raw: make([]byte, len(buffer)), } - - isValidRemoteCandidate := make(chan bool, 1) - err = c.agent.run(func(agent *Agent) { - isValidRemoteCandidate <- agent.noSTUNSeen(c, srcAddr) + // Explicitly copy raw buffer so Message can own the memory. + copy(m.Raw, buffer) + if err := m.Decode(); err != nil { + log.Warnf("Failed to handle decode ICE from %s to %s: %v", c.addr(), srcAddr, err) + return + } + err := c.agent().run(func(agent *Agent) { + agent.handleInbound(m, c, srcAddr) }) - if err != nil { log.Warnf("Failed to handle message: %v", err) - } else if !<-isValidRemoteCandidate { - log.Warnf("Discarded message from %s, not a valid remote candidate", c.addr()) } - // NOTE This will return packetio.ErrFull if the buffer ever manages to fill up. - _, err = c.agent.buffer.Write(buffer[:n]) - if err != nil { - log.Warnf("failed to write packet") - } + return } + + isValidRemoteCandidate := make(chan bool, 1) + err := c.agent().run(func(agent *Agent) { + isValidRemoteCandidate <- agent.noSTUNSeen(c, srcAddr) + }) + + if err != nil { + log.Warnf("Failed to handle message: %v", err) + } else if !<-isValidRemoteCandidate { + log.Warnf("Discarded message from %s, not a valid remote candidate", c.addr()) + } + + // NOTE This will return packetio.ErrFull if the buffer ever manages to fill up. + if _, err := c.agent().buffer.Write(buffer); err != nil { + log.Warnf("failed to write packet") + } + } // close stops the recvLoop @@ -223,3 +228,7 @@ func (c *candidateBase) addr() net.Addr { Port: c.Port(), } } + +func (c *candidateBase) agent() *Agent { + return c.currAgent +} diff --git a/candidate_relay.go b/candidate_relay.go index 2264361..09b92bf 100644 --- a/candidate_relay.go +++ b/candidate_relay.go @@ -1,12 +1,18 @@ package ice import ( + "errors" "net" + + "github.com/pion/turnc" ) // CandidateRelay ... type CandidateRelay struct { candidateBase + + allocation *turnc.Allocation + permissions map[string]*turnc.Permission } // NewCandidateRelay creates a new relay candidate @@ -28,5 +34,52 @@ func NewCandidateRelay(network string, ip net.IP, port int, component uint16, re Port: relPort, }, }, + permissions: map[string]*turnc.Permission{}, }, nil } + +func (c *CandidateRelay) setAllocation(a *turnc.Allocation) { + c.allocation = a +} + +func (c *CandidateRelay) start(a *Agent, conn net.PacketConn) { + c.currAgent = a +} + +func (c *CandidateRelay) close() error { + return nil +} + +func (c *CandidateRelay) addPermission(dst Candidate) error { + permission, err := c.allocation.Create(dst.addr()) + if err != nil { + return err + } + + c.lock.Lock() + c.permissions[dst.String()] = permission + c.lock.Unlock() + + go func(remoteAddr net.Addr) { + log := c.agent().log + buffer := make([]byte, receiveMTU) + for { + n, err := permission.Read(buffer) + if err != nil { + return + } + + handleInboundCandidateMsg(c, buffer[:n], remoteAddr, log) + } + }(dst.addr()) + return nil +} + +func (c *CandidateRelay) writeTo(raw []byte, dst Candidate) (int, error) { + permission, ok := c.permissions[dst.String()] + if !ok { + return 0, errors.New("no permission created for remote candidate") + } + + return permission.Write(raw) +} diff --git a/gather.go b/gather.go index 3efec15..c03f689 100644 --- a/gather.go +++ b/gather.go @@ -6,6 +6,7 @@ import ( "time" "github.com/pion/stun" + "github.com/pion/turnc" ) func localInterfaces(networkTypes []NetworkType) (ips []net.IP) { @@ -124,8 +125,19 @@ func (a *Agent) gatherCandidates() { } <-gatherStateUpdated - a.gatherCandidatesLocal(a.networkTypes) - a.gatherCandidatesSrflx(a.urls, a.networkTypes) + for _, t := range a.candidateTypes { + switch t { + case CandidateTypeHost: + a.gatherCandidatesLocal(a.networkTypes) + case CandidateTypeServerReflexive: + a.gatherCandidatesSrflx(a.urls, a.networkTypes) + case CandidateTypeRelay: + if err := a.gatherCandidatesRelay(a.urls); err != nil { + a.log.Errorf("Failed to gather relay candidates: %v\n", err) + } + } + } + if err := a.run(func(agent *Agent) { if a.onCandidateHdlr != nil { go a.onCandidateHdlr(nil) @@ -183,60 +195,58 @@ func (a *Agent) gatherCandidatesSrflx(urls []*URL, networkTypes []NetworkType) { for _, networkType := range networkTypes { network := networkType.String() for _, url := range urls { + if url.Scheme != SchemeTypeSTUN { + continue + } + hostPort := fmt.Sprintf("%s:%d", url.Host, url.Port) serverAddr, err := net.ResolveUDPAddr(network, hostPort) if err != nil { a.log.Warnf("failed to resolve stun host: %s: %v", hostPort, err) continue } + for _, ip := range localIPs { - switch url.Scheme { - case SchemeTypeSTUN: - 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 + 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 + } + + xoraddr, err := getXORMappedAddr(conn, serverAddr, time.Second*5) + if err != nil { + a.log.Warnf("could not get server reflexive address %s %s: %v\n", network, url, err) + continue + } + + laddr := conn.LocalAddr().(*net.UDPAddr) + 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 + } + + if err := a.run(func(agent *Agent) { + set := a.localCandidates[c.NetworkType()] + set = append(set, c) + a.localCandidates[c.NetworkType()] = set + }); err != nil { + a.log.Warnf("Failed to append to localCandidates: %v\n", err) + continue + } + + c.start(a, conn) + + if err := a.run(func(agent *Agent) { + if a.onCandidateHdlr != nil { + go a.onCandidateHdlr(c) } - - xoraddr, err := getXORMappedAddr(conn, serverAddr, time.Second*5) - if err != nil { - a.log.Warnf("could not get server reflexive address %s %s: %v\n", network, url, err) - continue - } - - laddr := conn.LocalAddr().(*net.UDPAddr) - 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 - } - - if err := a.run(func(agent *Agent) { - set := a.localCandidates[c.NetworkType()] - set = append(set, c) - a.localCandidates[c.NetworkType()] = set - }); err != nil { - a.log.Warnf("Failed to append to localCandidates: %v\n", err) - continue - } - - c.start(a, conn) - - if err := a.run(func(agent *Agent) { - if a.onCandidateHdlr != nil { - go a.onCandidateHdlr(c) - } - }); err != nil { - a.log.Warnf("Failed to run onCandidateHdlr task: %v\n", err) - continue - } - - default: - a.log.Warnf("scheme %s is not implemented\n", url.Scheme) + }); err != nil { + a.log.Warnf("Failed to run onCandidateHdlr task: %v\n", err) continue } } @@ -244,6 +254,59 @@ func (a *Agent) gatherCandidatesSrflx(urls []*URL, networkTypes []NetworkType) { } } +func (a *Agent) gatherCandidatesRelay(urls []*URL) error { + network := NetworkTypeUDP4.String() // TODO IPv6 + for _, url := range urls { + switch { + case url.Scheme != SchemeTypeTURN: + continue + case url.Username == "": + return ErrUsernameEmpty + case url.Password == "": + return ErrPasswordEmpty + } + + raddr, err := net.ResolveUDPAddr(network, fmt.Sprintf("%s:%d", url.Host, url.Port)) + if err != nil { + return err + } + c, err := net.DialUDP(network, nil, raddr) + if err != nil { + return err + } + client, clientErr := turnc.New(turnc.Options{ + Conn: c, + Username: url.Username, + Password: url.Password, + }) + if clientErr != nil { + return clientErr + } + allocation, allocErr := client.Allocate() + if allocErr != nil { + return allocErr + } + + laddr := c.LocalAddr().(*net.UDPAddr) + ip := allocation.Relayed().IP + port := allocation.Relayed().Port + + candidate, err := NewCandidateRelay(network, ip, port, ComponentRTP, laddr.IP.String(), laddr.Port) + if err != nil { + a.log.Warnf("Failed to create server reflexive candidate: %s %s %d: %v\n", network, ip, port, err) + continue + } + candidate.setAllocation(allocation) + + set := a.localCandidates[candidate.NetworkType()] + set = append(set, candidate) + a.localCandidates[candidate.NetworkType()] = set + candidate.start(a, nil) + } + + return nil +} + // getXORMappedAddr initiates a stun requests to serverAddr using conn, reads the response and returns // the XORMappedAddress returned by the stun server. //