Implement Active TCP-ICE

agent.go

@@ -130,6 +130,10 @@ type Agent struct {
 	insecureSkipVerify bool
 
 	proxyDialer proxy.Dialer
+
+	tcpConnections map[string]net.Conn
+
+	tcpActive bool
 }
 
 type task struct {
@@ -358,6 +362,9 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 		return nil, err
 	}
 
+	a.tcpActive = config.ActiveTCP
+	a.tcpConnections = map[string]net.Conn{}
+
 	return a, nil
 }
 
@@ -596,6 +603,7 @@ func (a *Agent) pingAllCandidates() {
 		a.log.Warn("pingAllCandidates called with no candidate pairs. Connection is not possible yet.")
 	}
 
+	// TODO: have to get the remote passive TCP candidate into the checklist
 	for _, p := range a.checklist {
 		if p.state == CandidatePairStateWaiting {
 			p.state = CandidatePairStateInProgress
@@ -803,6 +811,21 @@ func (a *Agent) addRemoteCandidate(c Candidate) {
 		}
 	}
 
+	// Open a TCP socket if the remote candidate is passive TCP
+	// TODO: only open the socket if the remote actually exists
+	if c.TCPType() == TCPTypePassive {
+		conn, err := net.Dial("tcp", c.Address())
+		if err != nil {
+			// TODO: what is the typical Pion error handling strategy here?
+			a.log.Warnf("Failed to open a TCP connection to %s: %v", c.Address(), err)
+			return
+		}
+
+		a.tcpConnections[c.Address()] = conn
+		// TODO:
+		// a.addPair(
+	}
+
 	a.requestConnectivityCheck()
 }
 

agent_config.go

@@ -143,6 +143,7 @@ type AgentConfig struct {
 	// TCPMux will be used for multiplexing incoming TCP connections for ICE TCP.
 	// Currently only passive candidates are supported. This functionality is
 	// experimental and the API might change in the future.
+	// TODO: update these docs when active candidate support begins to work
 	TCPMux TCPMux
 
 	// UDPMux is used for multiplexing multiple incoming UDP connections on a single port
@@ -153,6 +154,11 @@ type AgentConfig struct {
 	// Proxy Dialer is a dialer that should be implemented by the user based on golang.org/x/net/proxy
 	// dial interface in order to support corporate proxies
 	ProxyDialer proxy.Dialer
+
+	// ActiveTCP will initialize an active TCP host candidate on each local
+	// network interface if set to true.
+	// TODO: should it default to true?
+	ActiveTCP bool
 }
 
 // initWithDefaults populates an agent and falls back to defaults if fields are unset

agent_tcpmux_test.go

@@ -0,0 +1,99 @@
+//go:build !js
+// +build !js
+
+package ice
+
+import (
+	"net"
+	"testing"
+	"time"
+
+	"github.com/pion/logging"
+	"github.com/pion/transport/test"
+	"github.com/stretchr/testify/require"
+)
+
+// TestTCPMuxAgent is an end to end test over TCP mux, ensuring two agents could
+// connect over TCP mux.
+func TestTCPMuxAgent(t *testing.T) {
+	report := test.CheckRoutines(t)
+	defer report()
+
+	lim := test.TimeOut(time.Second * 30)
+	defer lim.Stop()
+
+	const muxPort = 7686
+
+	listener, err := net.ListenTCP("tcp", &net.TCPAddr{
+		Port: muxPort,
+	})
+	require.NoError(t, err, "error starting listener")
+	defer func() {
+		_ = listener.Close()
+	}()
+
+	loggerFactory := logging.NewDefaultLoggerFactory()
+	tcpMux := NewTCPMuxDefault(TCPMuxParams{
+		Listener:       listener,
+		Logger:         loggerFactory.NewLogger("ice"),
+		ReadBufferSize: 20,
+	})
+
+	defer func() {
+		_ = tcpMux.Close()
+	}()
+
+	require.NotNil(t, tcpMux.LocalAddr(), "tcpMux.LocalAddr() is nil")
+
+	muxedPassiveAgent, err := NewAgent(&AgentConfig{
+		TCPMux:         tcpMux,
+		CandidateTypes: []CandidateType{CandidateTypeHost},
+		NetworkTypes: []NetworkType{
+			NetworkTypeTCP4,
+		},
+		LoggerFactory: loggerFactory,
+	})
+	require.NoError(t, err)
+
+	loggerFactory.DefaultLogLevel.Set(logging.LogLevelDebug)
+	activeAgent, err := NewAgent(&AgentConfig{
+		CandidateTypes: []CandidateType{CandidateTypeHost},
+		NetworkTypes:   []NetworkType{NetworkTypeTCP4},
+		LoggerFactory:  loggerFactory,
+		ActiveTCP:      true,
+	})
+	require.NoError(t, err)
+
+	conn, muxedConn := connect(muxedPassiveAgent, activeAgent)
+
+	pair := muxedPassiveAgent.getSelectedPair()
+	require.NotNil(t, pair)
+	require.Equal(t, muxPort, pair.Local.Port())
+
+	// send a packet to Mux
+	data := []byte("hello world")
+	_, err = conn.Write(data)
+	require.NoError(t, err)
+
+	buffer := make([]byte, 1024)
+	n, err := muxedConn.Read(buffer)
+	require.NoError(t, err)
+	require.Equal(t, data, buffer[:n])
+
+	// send a packet from Mux
+	_, err = muxedConn.Write(data)
+	require.NoError(t, err)
+
+	n, err = conn.Read(buffer)
+	require.NoError(t, err)
+	require.Equal(t, data, buffer[:n])
+
+	// close it down
+	require.NoError(t, conn.Close())
+	require.NoError(t, muxedConn.Close())
+	require.NoError(t, tcpMux.Close())
+}
+
+func ActiveTCPIsOnByDefault(t *testing.T) {
+	require.FailNow(t, "TODO: Should Active TCP ICE be on by default?")
+}

candidate_base.go

@@ -202,6 +202,7 @@ func (c *candidateBase) start(a *Agent, conn net.PacketConn, initializedCh <-cha
 	c.closeCh = make(chan struct{})
 	c.closedCh = make(chan struct{})
 
+	// TODO: should active TCP candidates even do this? I need to read/ask.
 	go c.recvLoop(initializedCh)
 }
 
@@ -219,6 +220,7 @@ func (c *candidateBase) recvLoop(initializedCh <-chan struct{}) {
 	log := c.agent().log
 	buffer := make([]byte, receiveMTU)
 	for {
+		// FIXME: exempt active TCP candidates from this entire function
 		n, srcAddr, err := c.conn.ReadFrom(buffer)
 		if err != nil {
 			return

gather.go

@@ -163,6 +163,8 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
 			address = a.mDNSName
 		}
 
+		haveSetupActiveTCP := false
+
 		for network := range networks {
 			var port int
 			var conn net.PacketConn
@@ -171,19 +173,37 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
 
 			switch network {
 			case tcp:
-				// Handle ICE TCP passive mode
-				a.log.Debugf("GetConn by ufrag: %s\n", a.localUfrag)
-				conn, err = a.tcpMux.GetConnByUfrag(a.localUfrag)
-				if err != nil {
-					if !errors.Is(err, ErrTCPMuxNotInitialized) {
-						a.log.Warnf("error getting tcp conn by ufrag: %s %s %s\n", network, ip, a.localUfrag)
+				// FIXME: the control flow here is kind of jank
+				if a.tcpActive && !haveSetupActiveTCP {
+					a.log.Debugf("Reserving one local active candidate")
+					// TODO: is the nil here ok? we'll be connecting them to remote
+					// candidates later. Maybe reify that in a struct or
+					// something.
+					a.tcpConnections[ip.String()] = nil
+					haveSetupActiveTCP = true
+					tcpType = TCPTypeActive
+					port = 0 // TODO: not sure what to do here yet
+				} else {
+					// Handle ICE TCP passive mode
+					a.log.Debugf("GetConn by ufrag: %s\n", a.localUfrag)
+					conn, err = a.tcpMux.GetConnByUfrag(a.localUfrag)
+					if err != nil {
+						if !errors.Is(err, ErrTCPMuxNotInitialized) {
+							a.log.Warnf("error getting tcp conn by ufrag: %s %s %s\n", network, ip, a.localUfrag)
+						}
+
+						if !a.tcpActive {
+							continue
+						}
+
+						tcpType = TCPTypeActive
+
 					}
-					continue
+					port = conn.LocalAddr().(*net.TCPAddr).Port
+					tcpType = TCPTypePassive
+					// is there a way to verify that the listen address is even
+					// accessible from the current interface.
 				}
-				port = conn.LocalAddr().(*net.TCPAddr).Port
-				tcpType = TCPTypePassive
-				// is there a way to verify that the listen address is even
-				// accessible from the current interface.
 			case udp:
 				conn, err = listenUDPInPortRange(a.net, a.log, int(a.portmax), int(a.portmin), network, &net.UDPAddr{IP: ip, Port: 0})
 				if err != nil {