diff --git a/address.go b/address.go new file mode 100644 index 0000000..5b97443 --- /dev/null +++ b/address.go @@ -0,0 +1,177 @@ +package ice + +import ( + "strconv" + "strings" + + "github.com/pkg/errors" +) + +// TODO: Migrate address parsing to STUN/TURN packages? + +var ( + // ErrServerType indicates the server type could not be parsed + ErrServerType = errors.New("unknown server type") + + // ErrSTUNQuery indicates query arguments are provided in a STUN URL + ErrSTUNQuery = errors.New("queries not supported in stun address") + + // ErrInvalidQuery indicates an unsupported query is provided + ErrInvalidQuery = errors.New("invalid query") + + // ErrTransportType indicates an unsupported transport type was provided + ErrTransportType = errors.New("invalid transport type") + + // ErrHost indicates the server hostname could not be parsed + ErrHost = errors.New("invalid hostname") + + // ErrPort indicates the server port could not be parsed + ErrPort = errors.New("invalid port") +) + +// ServerType indicates the type of server used +type ServerType int + +const ( + // ServerTypeSTUN indicates the URL represents a STUN server + ServerTypeSTUN ServerType = iota + 1 + + // ServerTypeTURN indicates the URL represents a TURN server + ServerTypeTURN +) + +func (t ServerType) String() string { + switch t { + case ServerTypeSTUN: + return "stun" + case ServerTypeTURN: + return "turn" + default: + return ErrUnknownType.Error() + } +} + +// TransportType indicates the transport that is used +type TransportType int + +const ( + // TransportUDP indicates the URL uses a UDP transport + TransportUDP TransportType = iota + 1 + + // TransportTCP indicates the URL uses a TCP transport + TransportTCP +) + +func (t TransportType) String() string { + switch t { + case TransportUDP: + return "udp" + case TransportTCP: + return "tcp" + default: + return ErrUnknownType.Error() + } +} + +// URL represents a STUN (rfc7064) or TURN (rfc7065) URL +type URL struct { + Type ServerType + Secure bool + Host string + Port int + TransportType TransportType +} + +// NewURL creates a new URL by parsing a STUN (rfc7064) or TURN (rfc7065) uri string +func NewURL(address string) (URL, error) { + var result URL + + var scheme string + scheme, address = split(address, ":") + + switch strings.ToLower(scheme) { + case "stun": + result.Type = ServerTypeSTUN + result.Secure = false + + case "stuns": + result.Type = ServerTypeSTUN + result.Secure = true + + case "turn": + result.Type = ServerTypeTURN + result.Secure = false + + case "turns": + result.Type = ServerTypeTURN + result.Secure = true + + default: + return result, ErrServerType + } + + var query string + address, query = split(address, "?") + + if query != "" { + if result.Type == ServerTypeSTUN { + return result, ErrSTUNQuery + } + key, value := split(query, "=") + if strings.ToLower(key) != "transport" { + return result, ErrInvalidQuery + } + switch strings.ToLower(value) { + case "udp": + result.TransportType = TransportUDP + case "tcp": + result.TransportType = TransportTCP + default: + return result, ErrTransportType + } + } else { + if result.Secure { + result.TransportType = TransportTCP + } else { + result.TransportType = TransportUDP + } + } + + var host string + var port string + colon := strings.IndexByte(address, ':') + if colon == -1 { + host = address + if result.Secure { + port = "5349" + } else { + port = "3478" + } + } else if i := strings.IndexByte(address, ']'); i != -1 { + host = strings.TrimPrefix(address[:i], "[") + port = address[i+1+len(":"):] + } else { + host = address[:colon] + port = address[colon+len(":"):] + } + if host == "" { + return result, ErrHost + } + result.Host = strings.ToLower(host) + + var err error + result.Port, err = strconv.Atoi(port) + if err != nil { + return result, ErrPort + } + + return result, nil +} + +func split(s string, c string) (string, string) { + i := strings.Index(s, c) + if i < 0 { + return s, "" + } + return s[:i], s[i+len(c):] +} diff --git a/address_test.go b/address_test.go new file mode 100644 index 0000000..5a66914 --- /dev/null +++ b/address_test.go @@ -0,0 +1,68 @@ +package ice + +import "testing" + +func TestNewURL(t *testing.T) { + t.Run("Success", func(t *testing.T) { + testCases := []struct { + rawURL string + expectedType ServerType + expectedSecure bool + expectedHost string + expectedPort int + expectedTransportType TransportType + }{ + {"stun:google.de", ServerTypeSTUN, false, "google.de", 3478, TransportUDP}, + {"stun:google.de:1234", ServerTypeSTUN, false, "google.de", 1234, TransportUDP}, + {"stuns:google.de", ServerTypeSTUN, true, "google.de", 5349, TransportTCP}, + {"stun:[::1]:123", ServerTypeSTUN, false, "::1", 123, TransportUDP}, + {"turn:google.de", ServerTypeTURN, false, "google.de", 3478, TransportUDP}, + {"turns:google.de", ServerTypeTURN, true, "google.de", 5349, TransportTCP}, + {"turn:google.de?transport=udp", ServerTypeTURN, false, "google.de", 3478, TransportUDP}, + {"turn:google.de?transport=tcp", ServerTypeTURN, false, "google.de", 3478, TransportTCP}, + } + + for i, testCase := range testCases { + url, err := NewURL(testCase.rawURL) + if err != nil { + t.Errorf("Case %d: got error: %v", i, err) + } + + if url.Type != testCase.expectedType || + url.Secure != testCase.expectedSecure || + url.Host != testCase.expectedHost || + url.Port != testCase.expectedPort || + url.TransportType != testCase.expectedTransportType { + t.Errorf("Case %d: got %s %t %s %d %s", + i, + url.Type, + url.Secure, + url.Host, + url.Port, + url.TransportType, + ) + } + } + }) + t.Run("Failure", func(t *testing.T) { + testCases := []struct { + rawURL string + expectedErr error + }{ + {"", ErrServerType}, + {":::", ErrServerType}, + {"google.de", ErrServerType}, + {"stun:", ErrHost}, + {"stun:google.de:abc", ErrPort}, + {"stun:google.de?transport=udp", ErrSTUNQuery}, + {"turn:google.de?trans=udp", ErrInvalidQuery}, + {"turn:google.de?transport=ip", ErrTransportType}, + } + + for i, testCase := range testCases { + if _, err := NewURL(testCase.rawURL); err != testCase.expectedErr { + t.Errorf("Case %d: got error '%v' expected '%v'", i, err, testCase.expectedErr) + } + } + }) +} diff --git a/agent.go b/agent.go new file mode 100644 index 0000000..58ec295 --- /dev/null +++ b/agent.go @@ -0,0 +1,307 @@ +package ice + +import ( + "fmt" + "math/rand" + "net" + "sync" + "time" + + "github.com/pions/pkg/stun" + "github.com/pions/webrtc/internal/util" + "github.com/pkg/errors" +) + +// OutboundCallback is the user defined Callback that is called when ICE traffic needs to sent +type OutboundCallback func(raw []byte, local *stun.TransportAddr, remote *net.UDPAddr) + +// Agent represents the ICE agent +type Agent struct { + sync.RWMutex + + outboundCallback OutboundCallback + iceNotifier func(ConnectionState) + + tieBreaker uint64 + connectionState ConnectionState + gatheringState GatheringState + + haveStarted bool + isControlling bool + taskLoopChan chan bool + + LocalUfrag string + LocalPwd string + LocalCandidates []Candidate + + remoteUfrag string + remotePwd string + remoteCandidates []Candidate + + selectedPair struct { + // lastUpdateTime ? + remote Candidate + local Candidate + } +} + +const ( + agentTickerBaseInterval = 3 * time.Second + stunTimeout = 10 * time.Second +) + +// NewAgent creates a new Agent +func NewAgent(outboundCallback OutboundCallback, iceNotifier func(ConnectionState)) *Agent { + return &Agent{ + outboundCallback: outboundCallback, + iceNotifier: iceNotifier, + + tieBreaker: rand.Uint64(), + gatheringState: GatheringStateComplete, // TODO trickle-ice + connectionState: ConnectionStateNew, + + LocalUfrag: util.RandSeq(16), + LocalPwd: util.RandSeq(32), + } +} + +// Start starts the agent +func (a *Agent) Start(isControlling bool, remoteUfrag, remotePwd string) error { + a.Lock() + defer a.Unlock() + + if a.haveStarted { + return errors.Errorf("Attempted to start agent twice") + } else if remoteUfrag == "" { + return errors.Errorf("remoteUfrag is empty") + } else if remotePwd == "" { + return errors.Errorf("remotePwd is empty") + } + + a.isControlling = isControlling + a.remoteUfrag = remoteUfrag + a.remotePwd = remotePwd + + go a.agentTaskLoop() + return nil +} + +func (a *Agent) pingCandidate(local, remote Candidate) { + msg, err := stun.Build(stun.ClassRequest, stun.MethodBinding, stun.GenerateTransactionId(), + &stun.Username{Username: a.remoteUfrag + ":" + a.LocalUfrag}, + &stun.UseCandidate{}, + &stun.IceControlling{TieBreaker: a.tieBreaker}, + &stun.Priority{Priority: uint32(local.GetBase().Priority(HostCandidatePreference, 1))}, + &stun.MessageIntegrity{ + Key: []byte(a.remotePwd), + }, + &stun.Fingerprint{}, + ) + if err != nil { + fmt.Println(err) + return + } + + a.outboundCallback(msg.Pack(), &stun.TransportAddr{ + IP: net.ParseIP(local.GetBase().Address), + Port: local.GetBase().Port, + }, &net.UDPAddr{ + IP: net.ParseIP(remote.GetBase().Address), + Port: remote.GetBase().Port, + }) +} + +func (a *Agent) updateConnectionState(newState ConnectionState) { + a.connectionState = newState + a.iceNotifier(a.connectionState) +} + +func (a *Agent) setSelectedPair(local, remote Candidate) { + a.selectedPair.remote = remote + a.selectedPair.local = local + a.updateConnectionState(ConnectionStateConnected) +} + +func (a *Agent) agentTaskLoop() { + // TODO this should be dynamic, and grow when the connection is stable + t := time.NewTicker(agentTickerBaseInterval) + a.updateConnectionState(ConnectionStateChecking) + + assertSelectedPairValid := func() bool { + if a.selectedPair.remote == nil || a.selectedPair.local == nil { + return false + } else if time.Since(a.selectedPair.remote.GetBase().LastSeen) > stunTimeout { + a.selectedPair.remote = nil + a.selectedPair.local = nil + a.updateConnectionState(ConnectionStateDisconnected) + return false + } + + return true + } + + for { + select { + case <-t.C: + a.Lock() + if a.isControlling { + if assertSelectedPairValid() { + a.Unlock() + continue + } + + for _, localCandidate := range a.LocalCandidates { + for _, remoteCandidate := range a.remoteCandidates { + a.pingCandidate(localCandidate, remoteCandidate) + } + } + } else { + assertSelectedPairValid() + } + a.Unlock() + case <-a.taskLoopChan: + t.Stop() + return + } + } +} + +// AddRemoteCandidate adds a new remote candidate +func (a *Agent) AddRemoteCandidate(c Candidate) { + a.Lock() + defer a.Unlock() + a.remoteCandidates = append(a.remoteCandidates, c) +} + +// AddLocalCandidate adds a new local candidate +func (a *Agent) AddLocalCandidate(c Candidate) { + a.Lock() + defer a.Unlock() + a.LocalCandidates = append(a.LocalCandidates, c) +} + +// Close cleans up the Agent +func (a *Agent) Close() { + close(a.taskLoopChan) +} + +func getTransportAddrCandidate(candidates []Candidate, addr *stun.TransportAddr) Candidate { + for _, c := range candidates { + if c.GetBase().Address == addr.IP.String() && c.GetBase().Port == addr.Port { + return c + } + } + return nil +} + +func getUDPAddrCandidate(candidates []Candidate, addr *net.UDPAddr) Candidate { + for _, c := range candidates { + if c.GetBase().Address == addr.IP.String() && c.GetBase().Port == addr.Port { + return c + } + } + return nil +} + +func (a *Agent) sendBindingSuccess(m *stun.Message, local *stun.TransportAddr, remote *net.UDPAddr) { + if out, err := stun.Build(stun.ClassSuccessResponse, stun.MethodBinding, m.TransactionID, + &stun.XorMappedAddress{ + XorAddress: stun.XorAddress{ + IP: remote.IP, + Port: remote.Port, + }, + }, + &stun.MessageIntegrity{ + Key: []byte(a.LocalPwd), + }, + &stun.Fingerprint{}, + ); err != nil { + fmt.Printf("Failed to handle inbound ICE from: %s to: %s error: %s", local.String(), remote.String(), err.Error()) + } else { + a.outboundCallback(out.Pack(), local, remote) + } + +} + +func (a *Agent) handleInboundControlled(m *stun.Message, local *stun.TransportAddr, remote *net.UDPAddr, localCandidate, remoteCandidate Candidate) { + if _, isControlled := m.GetOneAttribute(stun.AttrIceControlled); isControlled && !a.isControlling { + fmt.Println("inbound isControlled && a.isControlling == false") + return + } + + if _, useCandidateFound := m.GetOneAttribute(stun.AttrUseCandidate); useCandidateFound { + a.setSelectedPair(localCandidate, remoteCandidate) + } + a.sendBindingSuccess(m, local, remote) +} + +func (a *Agent) handleInboundControlling(m *stun.Message, local *stun.TransportAddr, remote *net.UDPAddr, localCandidate, remoteCandidate Candidate) { + if _, isControlling := m.GetOneAttribute(stun.AttrIceControlling); isControlling && a.isControlling { + fmt.Println("inbound isControlling && a.isControlling == true") + return + } else if _, useCandidate := m.GetOneAttribute(stun.AttrUseCandidate); useCandidate && a.isControlling { + fmt.Println("useCandidate && a.isControlling == true") + return + } + + if m.Class == stun.ClassSuccessResponse && m.Method == stun.MethodBinding { + //Binding success! + if a.selectedPair.remote == nil && a.selectedPair.local == nil { + a.setSelectedPair(localCandidate, remoteCandidate) + } + } else { + a.sendBindingSuccess(m, local, remote) + } +} + +// HandleInbound processes traffic from a remote candidate +func (a *Agent) HandleInbound(buf []byte, local *stun.TransportAddr, remote *net.UDPAddr) { + a.Lock() + defer a.Unlock() + + localCandidate := getTransportAddrCandidate(a.LocalCandidates, local) + if localCandidate == nil { + // TODO debug + // fmt.Printf("Could not find local candidate for %s:%d ", local.IP.String(), local.Value) + return + } + + remoteCandidate := getUDPAddrCandidate(a.remoteCandidates, remote) + if remoteCandidate == nil { + // TODO debug + // fmt.Printf("Could not find remote candidate for %s:%d ", remote.IP.String(), remote.Value) + return + } + remoteCandidate.GetBase().LastSeen = time.Now() + + m, err := stun.NewMessage(buf) + if err != nil { + fmt.Println(fmt.Sprintf("Failed to handle decode ICE from: %s to: %s error: %s", local.String(), remote.String(), err.Error())) + return + } + + if a.isControlling { + a.handleInboundControlling(m, local, remote, localCandidate, remoteCandidate) + } else { + a.handleInboundControlled(m, local, remote, localCandidate, remoteCandidate) + } + +} + +// SelectedPair gets the current selected pair's Addresses (or returns nil) +func (a *Agent) SelectedPair() (local *stun.TransportAddr, remote *net.UDPAddr) { + a.RLock() + defer a.RUnlock() + + if a.selectedPair.remote == nil || a.selectedPair.local == nil { + return nil, nil + } + + return &stun.TransportAddr{ + IP: net.ParseIP(a.selectedPair.local.GetBase().Address), + Port: a.selectedPair.local.GetBase().Port, + }, &net.UDPAddr{ + IP: net.ParseIP(a.selectedPair.remote.GetBase().Address), + Port: a.selectedPair.remote.GetBase().Port, + } +} diff --git a/candidate.go b/candidate.go new file mode 100644 index 0000000..8c45f24 --- /dev/null +++ b/candidate.go @@ -0,0 +1,66 @@ +package ice + +import ( + "math/rand" + "time" +) + +// Preference enums when generate Priority +const ( + HostCandidatePreference uint16 = 126 + SrflxCandidatePreference uint16 = 100 +) + +// Candidate represents an ICE candidate +type Candidate interface { + GetBase() *CandidateBase +} + +// CandidateBase represents an ICE candidate, a base with enough attributes +// for host candidates, see CandidateSrflx and CandidateRelay for more +type CandidateBase struct { + Protocol TransportType + Address string + Port int + LastSeen time.Time +} + +// Priority computes the priority for this ICE Candidate +func (c *CandidateBase) Priority(typePreference uint16, component uint16) uint16 { + localPreference := uint16(rand.Uint32() / 2) + return (2^24)*typePreference + + (2^8)*localPreference + + (2^0)*(256-component) +} + +// CandidateHost is a Candidate of typ Host +type CandidateHost struct { + CandidateBase +} + +// GetBase returns the CandidateBase, attributes shared between all Candidates +func (c *CandidateHost) GetBase() *CandidateBase { + return &c.CandidateBase +} + +// IP for CandidateHost +func (c *CandidateHost) Address() string { + return c.CandidateBase.Address +} + +// Value for CandidateHost +func (c *CandidateHost) Port() int { + return c.CandidateBase.Port +} + +// CandidateSrflx is a Candidate of typ Server-Reflexive +type CandidateSrflx struct { + CandidateBase + RemoteAddress string + RemotePort int +} + +// GetBase returns the CandidateBase, attributes shared between all Candidates +func (c *CandidateSrflx) GetBase() *CandidateBase { + return &c.CandidateBase +} diff --git a/ice.go b/ice.go new file mode 100644 index 0000000..6ea9fda --- /dev/null +++ b/ice.go @@ -0,0 +1,81 @@ +package ice + +import "github.com/pkg/errors" + +// ErrUnknownType indicates a Unknown info +var ErrUnknownType = errors.New("Unknown") + +// ConnectionState is an enum showing the state of a ICE Connection +type ConnectionState int + +// List of supported States +const ( + // ConnectionStateNew ICE agent is gathering addresses + ConnectionStateNew = iota + 1 + + // ConnectionStateChecking ICE agent has been given local and remote candidates, and is attempting to find a match + ConnectionStateChecking + + // ConnectionStateConnected ICE agent has a pairing, but is still checking other pairs + ConnectionStateConnected + + // ConnectionStateCompleted ICE agent has finished + ConnectionStateCompleted + + // ConnectionStateFailed ICE agent never could successfully connect + ConnectionStateFailed + + // ConnectionStateDisconnected ICE agent connected successfully, but has entered a failed state + ConnectionStateDisconnected + + // ConnectionStateClosed ICE agent has finished and is no longer handling requests + ConnectionStateClosed +) + +func (c ConnectionState) String() string { + switch c { + case ConnectionStateNew: + return "New" + case ConnectionStateChecking: + return "Checking" + case ConnectionStateConnected: + return "Connected" + case ConnectionStateCompleted: + return "Completed" + case ConnectionStateFailed: + return "Failed" + case ConnectionStateDisconnected: + return "Disconnected" + case ConnectionStateClosed: + return "Closed" + default: + return "Invalid" + } +} + +// GatheringState describes the state of the candidate gathering process +type GatheringState int + +const ( + // GatheringStateNew indicates candidate gatering is not yet started + GatheringStateNew GatheringState = iota + 1 + + // GatheringStateGathering indicates candidate gatering is ongoing + GatheringStateGathering + + // GatheringStateComplete indicates candidate gatering has been completed + GatheringStateComplete +) + +func (t GatheringState) String() string { + switch t { + case GatheringStateNew: + return "new" + case GatheringStateGathering: + return "gathering" + case GatheringStateComplete: + return "complete" + default: + return ErrUnknownType.Error() + } +}