Implement active TCP candidate type

Author: ashellunts

agent.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -138,6 +139,7 @@ type Agent struct {
 
 	interfaceFilter func(string) bool
 	ipFilter        func(net.IP) bool
+	ActiveTCP       bool
 	includeLoopback bool
 
 	insecureSkipVerify bool
@@ -312,6 +314,8 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
 
 		ipFilter: config.IPFilter,
 
+		ActiveTCP: config.ActiveTCP,
+
 		insecureSkipVerify: config.InsecureSkipVerify,
 
 		includeLoopback: config.IncludeLoopback,
@@ -579,6 +583,40 @@ func (a *Agent) getBestValidCandidatePair() *CandidatePair {
 }
 
 func (a *Agent) addPair(local, remote Candidate) *CandidatePair {
+	if local.TCPType() == TCPTypeActive && remote.TCPType() == TCPTypePassive {
+		addressToConnect := net.JoinHostPort(remote.Address(), strconv.Itoa(remote.Port()))
+
+		conn, err := net.Dial("tcp", addressToConnect)
+		if err != nil {
+			a.log.Errorf("Failed to dial TCP address %s: %v", addressToConnect, err)
+			return nil
+		}
+
+		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
@@ -755,7 +793,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_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 TestAgentActiveTCP(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_config.go

@@ -145,6 +145,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

gather.go

@@ -166,44 +166,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 {