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

Initial TURN implementation

Browse Source 
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
This commit is contained in:
Sean DuBois
2019-05-27 23:30:50 -07:00
parent 78c7f4e989
commit 2a96a308d5
5 changed files with 229 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent.go
+20
View File
@@ -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
candidate.go
+1
View File
@@ -31,6 +31,7 @@ type Candidate interface {
Equal(other Candidate) bool
addr() net.Addr
agent() *Agent
close() error
seen(outbound bool)
candidate_base.go
+44 -35
View File
@@ -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
}
candidate_relay.go
+53
View File
@@ -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)
}
gather.go
+111 -48
View File
@@ -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.
//