Implement active TCP candidate type

Author: ashellunts

active_tcp_test.go

@@ -0,0 +1,94 @@
+// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+//go:build !js
+// +build !js
+
+package ice
+
+import (
+	"net"
+	"testing"
+
+	"github.com/pion/logging"
+	"github.com/stretchr/testify/require"
+)
+
+func TestActiveTCP(t *testing.T) {
+	r := require.New(t)
+
+	const port = 7686
+
+	listener, err := net.ListenTCP("tcp", &net.TCPAddr{
+		IP:   net.IPv4(127, 0, 0, 1),
+		Port: port,
+	})
+	r.NoError(err)
+	defer func() {
+		_ = listener.Close()
+	}()
+
+	loggerFactory := logging.NewDefaultLoggerFactory()
+	loggerFactory.DefaultLogLevel.Set(logging.LogLevelTrace)
+
+	tcpMux := NewTCPMuxDefault(TCPMuxParams{
+		Listener:       listener,
+		Logger:         loggerFactory.NewLogger("passive-ice"),
+		ReadBufferSize: 20,
+	})
+
+	defer func() {
+		_ = tcpMux.Close()
+	}()
+
+	r.NotNil(tcpMux.LocalAddr(), "tcpMux.LocalAddr() is nil")
+
+	passiveAgent, err := NewAgent(&AgentConfig{
+		TCPMux:          tcpMux,
+		CandidateTypes:  []CandidateType{CandidateTypeHost},
+		NetworkTypes:    []NetworkType{NetworkTypeTCP4},
+		LoggerFactory:   loggerFactory,
+		ActiveTCP:       false,
+		IncludeLoopback: true,
+	})
+	r.NoError(err)
+	r.NotNil(passiveAgent)
+
+	activeAgent, err := NewAgent(&AgentConfig{
+		CandidateTypes: []CandidateType{CandidateTypeHost},
+		NetworkTypes:   []NetworkType{NetworkTypeTCP4},
+		LoggerFactory:  loggerFactory,
+		ActiveTCP:      true,
+	})
+	r.NoError(err)
+	r.NotNil(activeAgent)
+
+	passiveAgentConn, activeAgenConn := connect(passiveAgent, activeAgent)
+	r.NotNil(passiveAgentConn)
+	r.NotNil(activeAgenConn)
+
+	pair := passiveAgent.getSelectedPair()
+	r.NotNil(pair)
+	r.Equal(port, pair.Local.Port())
+
+	data := []byte("hello world")
+	_, err = passiveAgentConn.Write(data)
+	r.NoError(err)
+
+	buffer := make([]byte, 1024)
+	n, err := activeAgenConn.Read(buffer)
+	r.NoError(err)
+	r.Equal(data, buffer[:n])
+
+	data2 := []byte("hello world 2")
+	_, err = activeAgenConn.Write(data2)
+	r.NoError(err)
+
+	n, err = passiveAgentConn.Read(buffer)
+	r.NoError(err)
+	r.Equal(data2, buffer[:n])
+
+	r.NoError(activeAgenConn.Close())
+	r.NoError(passiveAgentConn.Close())
+	r.NoError(tcpMux.Close())
+}

agent.go

@@ -138,6 +138,7 @@ type Agent struct {
 
 	interfaceFilter func(string) bool
 	ipFilter        func(net.IP) bool
+	ActiveTCP       bool
 	includeLoopback bool
 
 	insecureSkipVerify bool
@@ -312,6 +313,8 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 
 		ipFilter: config.IPFilter,
 
+		ActiveTCP: config.ActiveTCP,
+
 		insecureSkipVerify: config.InsecureSkipVerify,
 
 		includeLoopback: config.IncludeLoopback,
@@ -578,6 +581,45 @@ func (a *Agent) getBestValidCandidatePair() *CandidatePair {
 }
 
 func (a *Agent) addPair(local, remote Candidate) *CandidatePair {
+	if local.TCPType() == TCPTypeActive && remote.TCPType() == TCPTypePassive {
+		a.log.Debugf("artur, addPair: local %s, remote %s", local, remote)
+		addressToConnect := remote.Address() + ":" + fmt.Sprint(remote.Port())
+		a.log.Debugf("artur, addressToConnect %s", addressToConnect)
+
+		// connect
+		conn, err := net.Dial("tcp", addressToConnect)
+		if err != nil {
+			a.log.Errorf("Failed to dial TCP address %s: %v", addressToConnect, err)
+			return nil
+		}
+		a.log.Debugf("artur, socket connected, local %s, remote %s", conn.LocalAddr(), conn.RemoteAddr())
+
+		// create PacketCon from tcp connection
+		packetConn := newTCPPacketConn(tcpPacketParams{
+			ReadBuffer: 8,
+			LocalAddr:  conn.LocalAddr(),
+			Logger:     a.log,
+		})
+
+		if err = packetConn.AddConn(conn, nil); err != nil {
+			a.log.Errorf("Failed to add TCP connection: %v", err)
+			return nil
+		}
+
+		localAddress, ok := conn.LocalAddr().(*net.TCPAddr)
+		if !ok {
+			a.log.Errorf("Failed to cast local address to TCP address")
+			return nil
+		}
+
+		localCandidateHost, ok := local.(*CandidateHost)
+		if !ok {
+			a.log.Errorf("Failed to cast local candidate to CandidateHost")
+			return nil
+		}
+		localCandidateHost.port = localAddress.Port // this causes a data race with candidateBase.Port()
+		local.start(a, packetConn, a.startedCh)
+	}
 	p := newCandidatePair(local, remote, a.isControlling)
 	a.checklist = append(a.checklist, p)
 	return p
@@ -754,7 +796,9 @@ func (a *Agent) addCandidate(ctx context.Context, c Candidate, candidateConn net
 			}
 		}
 
-		c.start(a, candidateConn, a.startedCh)
+		if c.TCPType() != TCPTypeActive {
+			c.start(a, candidateConn, a.startedCh)
+		}
 
 		set = append(set, c)
 		a.localCandidates[c.NetworkType()] = set

agent_config.go

@@ -144,6 +144,8 @@ type AgentConfig struct {
 	// the ips which are used to gather ICE candidates.
 	IPFilter func(net.IP) bool
 
+	ActiveTCP bool
+
 	// InsecureSkipVerify controls if self-signed certificates are accepted when connecting
 	// to TURN servers via TLS or DTLS
 	InsecureSkipVerify bool

agent_test.go

@@ -9,7 +9,6 @@ package ice
 import (
 	"context"
 	"errors"
-	"fmt"
 	"net"
 	"strconv"
 	"sync"
@@ -1505,7 +1504,6 @@ func TestLiteLifecycle(t *testing.T) {
 	bFailed := make(chan interface{})
 
 	require.NoError(t, bAgent.OnConnectionStateChange(func(c ConnectionState) {
-		fmt.Println(c)
 		switch c {
 		case ConnectionStateConnected:
 			close(bConnected)

candidate_base.go

@@ -255,6 +255,8 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
 	a := c.agent()
 
 	if stun.IsMessage(buf) {
+		a.log.Tracef("handleInboundPacket ICE to %s from %s", c.addr(), srcAddr)
+
 		m := &stun.Message{
 			Raw: make([]byte, len(buf)),
 		}
@@ -263,7 +265,7 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
 		copy(m.Raw, buf)
 
 		if err := m.Decode(); err != nil {
-			a.log.Warnf("Failed to handle decode ICE from %s to %s: %v", c.addr(), srcAddr, err)
+			a.log.Warnf("Failed to handle decode ICE to %s from %s: %v", c.addr(), srcAddr, err)
 			return
 		}
 
@@ -279,7 +281,7 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
 	if !c.validateSTUNTrafficCache(srcAddr) {
 		remoteCandidate, valid := a.validateNonSTUNTraffic(c, srcAddr) //nolint:contextcheck
 		if !valid {
-			a.log.Warnf("Discarded message from %s, not a valid remote candidate", c.addr())
+			a.log.Warnf("Discarded message to %s, not a valid remote candidate", c.addr())
 			return
 		}
 		c.addRemoteCandidateCache(remoteCandidate, srcAddr)

gather.go

@@ -165,44 +165,49 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
 
 			switch network {
 			case tcp:
-				if a.tcpMux == nil {
-					continue
-				}
-
-				// Handle ICE TCP passive mode
-				var muxConns []net.PacketConn
-				if multi, ok := a.tcpMux.(AllConnsGetter); ok {
-					a.log.Debugf("GetAllConns by ufrag: %s", a.localUfrag)
-					muxConns, err = multi.GetAllConns(a.localUfrag, mappedIP.To4() == nil, ip)
-					if err != nil {
-						a.log.Warnf("Failed to get all TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
-						continue
-					}
+				if a.ActiveTCP {
+					conns = append(conns, connAndPort{nil, 0})
+					tcpType = TCPTypeActive
 				} else {
-					a.log.Debugf("GetConn by ufrag: %s", a.localUfrag)
-					conn, err := a.tcpMux.GetConnByUfrag(a.localUfrag, mappedIP.To4() == nil, ip)
-					if err != nil {
-						a.log.Warnf("Failed to get TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
+					// Handle ICE TCP passive mode
+					if a.tcpMux == nil {
 						continue
 					}
-					muxConns = []net.PacketConn{conn}
-				}
 
-				// Extract the port for each PacketConn we got.
-				for _, conn := range muxConns {
-					if tcpConn, ok := conn.LocalAddr().(*net.TCPAddr); ok {
-						conns = append(conns, connAndPort{conn, tcpConn.Port})
+					var muxConns []net.PacketConn
+					if multi, ok := a.tcpMux.(AllConnsGetter); ok {
+						a.log.Debugf("GetAllConns by ufrag: %s", a.localUfrag)
+						muxConns, err = multi.GetAllConns(a.localUfrag, mappedIP.To4() == nil, ip)
+						if err != nil {
+							a.log.Warnf("Failed to get all TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
+							continue
+						}
 					} else {
-						a.log.Warnf("Failed to get port of connection from TCPMux: %s %s %s", network, ip, a.localUfrag)
+						a.log.Debugf("GetConn by ufrag: %s", a.localUfrag)
+						conn, err := a.tcpMux.GetConnByUfrag(a.localUfrag, mappedIP.To4() == nil, ip)
+						if err != nil {
+							a.log.Warnf("Failed to get TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
+							continue
+						}
+						muxConns = []net.PacketConn{conn}
 					}
+
+					// Extract the port for each PacketConn we got.
+					for _, conn := range muxConns {
+						if tcpConn, ok := conn.LocalAddr().(*net.TCPAddr); ok {
+							conns = append(conns, connAndPort{conn, tcpConn.Port})
+						} else {
+							a.log.Warnf("Failed to get port of connection from TCPMux: %s %s %s", network, ip, a.localUfrag)
+						}
+					}
+					if len(conns) == 0 {
+						// Didn't succeed with any, try the next network.
+						continue
+					}
+					tcpType = TCPTypePassive
+					// Is there a way to verify that the listen address is even
+					// accessible from the current interface.
 				}
-				if len(conns) == 0 {
-					// Didn't succeed with any, try the next network.
-					continue
-				}
-				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 {