Skip to content

Commit 1edec30

Browse files
ashelluntsashellunts
committed
Implement active TCP candidate type
1 parent 3172e05 commit 1edec30

File tree

6 files changed

+181
-36
lines changed

6 files changed

+181
-36
lines changed

‎active_tcp_test.go‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
// SPDX-FileCopyrightText: 2023 The Pion community <https://pion.ly>
2+
// SPDX-License-Identifier: MIT
3+
4+
//go:build !js
5+
// +build !js
6+
7+
package ice
8+
9+
import (
10+
"net"
11+
"testing"
12+
13+
"github.com/pion/logging"
14+
"github.com/stretchr/testify/require"
15+
)
16+
17+
func TestActiveTCP(t *testing.T) {
18+
r := require.New(t)
19+
20+
const port = 7686
21+
22+
listener, err := net.ListenTCP("tcp", &net.TCPAddr{
23+
IP: net.IPv4(127, 0, 0, 1),
24+
Port: port,
25+
})
26+
r.NoError(err)
27+
defer func() {
28+
_ = listener.Close()
29+
}()
30+
31+
loggerFactory := logging.NewDefaultLoggerFactory()
32+
loggerFactory.DefaultLogLevel.Set(logging.LogLevelTrace)
33+
34+
tcpMux := NewTCPMuxDefault(TCPMuxParams{
35+
Listener: listener,
36+
Logger: loggerFactory.NewLogger("passive-ice"),
37+
ReadBufferSize: 20,
38+
})
39+
40+
defer func() {
41+
_ = tcpMux.Close()
42+
}()
43+
44+
r.NotNil(tcpMux.LocalAddr(), "tcpMux.LocalAddr() is nil")
45+
46+
passiveAgent, err := NewAgent(&AgentConfig{
47+
TCPMux: tcpMux,
48+
CandidateTypes: []CandidateType{CandidateTypeHost},
49+
NetworkTypes: []NetworkType{NetworkTypeTCP4},
50+
LoggerFactory: loggerFactory,
51+
ActiveTCP: false,
52+
IncludeLoopback: true,
53+
})
54+
r.NoError(err)
55+
r.NotNil(passiveAgent)
56+
57+
activeAgent, err := NewAgent(&AgentConfig{
58+
CandidateTypes: []CandidateType{CandidateTypeHost},
59+
NetworkTypes: []NetworkType{NetworkTypeTCP4},
60+
LoggerFactory: loggerFactory,
61+
ActiveTCP: true,
62+
})
63+
r.NoError(err)
64+
r.NotNil(activeAgent)
65+
66+
passiveAgentConn, activeAgenConn := connect(passiveAgent, activeAgent)
67+
r.NotNil(passiveAgentConn)
68+
r.NotNil(activeAgenConn)
69+
70+
pair := passiveAgent.getSelectedPair()
71+
r.NotNil(pair)
72+
r.Equal(port, pair.Local.Port())
73+
74+
data := []byte("hello world")
75+
_, err = passiveAgentConn.Write(data)
76+
r.NoError(err)
77+
78+
buffer := make([]byte, 1024)
79+
n, err := activeAgenConn.Read(buffer)
80+
r.NoError(err)
81+
r.Equal(data, buffer[:n])
82+
83+
data2 := []byte("hello world 2")
84+
_, err = activeAgenConn.Write(data2)
85+
r.NoError(err)
86+
87+
n, err = passiveAgentConn.Read(buffer)
88+
r.NoError(err)
89+
r.Equal(data2, buffer[:n])
90+
91+
r.NoError(activeAgenConn.Close())
92+
r.NoError(passiveAgentConn.Close())
93+
r.NoError(tcpMux.Close())
94+
}

‎agent.go‎

Lines changed: 45 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,7 @@ type Agent struct {
138138

139139
interfaceFilter func(string) bool
140140
ipFilter func(net.IP) bool
141+
ActiveTCP bool
141142
includeLoopback bool
142143

143144
insecureSkipVerify bool
@@ -312,6 +313,8 @@ func NewAgent(config *AgentConfig) (*Agent, error) { //nolint:gocognit
312313

313314
ipFilter: config.IPFilter,
314315

316+
ActiveTCP: config.ActiveTCP,
317+
315318
insecureSkipVerify: config.InsecureSkipVerify,
316319

317320
includeLoopback: config.IncludeLoopback,
@@ -578,6 +581,45 @@ func (a *Agent) getBestValidCandidatePair() *CandidatePair {
578581
}
579582

580583
func (a *Agent) addPair(local, remote Candidate) *CandidatePair {
584+
if local.TCPType() == TCPTypeActive && remote.TCPType() == TCPTypePassive {
585+
a.log.Debugf("artur, addPair: local %s, remote %s", local, remote)
586+
addressToConnect := remote.Address() + ":" + fmt.Sprint(remote.Port())
587+
a.log.Debugf("artur, addressToConnect %s", addressToConnect)
588+
589+
// connect
590+
conn, err := net.Dial("tcp", addressToConnect)
591+
if err != nil {
592+
a.log.Errorf("Failed to dial TCP address %s: %v", addressToConnect, err)
593+
return nil
594+
}
595+
a.log.Debugf("artur, socket connected, local %s, remote %s", conn.LocalAddr(), conn.RemoteAddr())
596+
597+
// create PacketCon from tcp connection
598+
packetConn := newTCPPacketConn(tcpPacketParams{
599+
ReadBuffer: 8,
600+
LocalAddr: conn.LocalAddr(),
601+
Logger: a.log,
602+
})
603+
604+
if err = packetConn.AddConn(conn, nil); err != nil {
605+
a.log.Errorf("Failed to add TCP connection: %v", err)
606+
return nil
607+
}
608+
609+
localAddress, ok := conn.LocalAddr().(*net.TCPAddr)
610+
if !ok {
611+
a.log.Errorf("Failed to cast local address to TCP address")
612+
return nil
613+
}
614+
615+
localCandidateHost, ok := local.(*CandidateHost)
616+
if !ok {
617+
a.log.Errorf("Failed to cast local candidate to CandidateHost")
618+
return nil
619+
}
620+
localCandidateHost.port = localAddress.Port // this causes a data race with candidateBase.Port()
621+
local.start(a, packetConn, a.startedCh)
622+
}
581623
p := newCandidatePair(local, remote, a.isControlling)
582624
a.checklist = append(a.checklist, p)
583625
return p
@@ -754,7 +796,9 @@ func (a *Agent) addCandidate(ctx context.Context, c Candidate, candidateConn net
754796
}
755797
}
756798

757-
c.start(a, candidateConn, a.startedCh)
799+
if c.TCPType() != TCPTypeActive {
800+
c.start(a, candidateConn, a.startedCh)
801+
}
758802

759803
set = append(set, c)
760804
a.localCandidates[c.NetworkType()] = set

‎agent_config.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,8 @@ type AgentConfig struct {
144144
// the ips which are used to gather ICE candidates.
145145
IPFilter func(net.IP) bool
146146

147+
ActiveTCP bool
148+
147149
// InsecureSkipVerify controls if self-signed certificates are accepted when connecting
148150
// to TURN servers via TLS or DTLS
149151
InsecureSkipVerify bool

‎agent_test.go‎

Lines changed: 0 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,6 @@ package ice
99
import (
1010
"context"
1111
"errors"
12-
"fmt"
1312
"net"
1413
"strconv"
1514
"sync"
@@ -1505,7 +1504,6 @@ func TestLiteLifecycle(t *testing.T) {
15051504
bFailed := make(chan interface{})
15061505

15071506
require.NoError(t, bAgent.OnConnectionStateChange(func(c ConnectionState) {
1508-
fmt.Println(c)
15091507
switch c {
15101508
case ConnectionStateConnected:
15111509
close(bConnected)

‎candidate_base.go‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -255,6 +255,8 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
255255
a := c.agent()
256256

257257
if stun.IsMessage(buf) {
258+
a.log.Tracef("handleInboundPacket ICE to %s from %s", c.addr(), srcAddr)
259+
258260
m := &stun.Message{
259261
Raw: make([]byte, len(buf)),
260262
}
@@ -263,7 +265,7 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
263265
copy(m.Raw, buf)
264266

265267
if err := m.Decode(); err != nil {
266-
a.log.Warnf("Failed to handle decode ICE from %s to %s: %v", c.addr(), srcAddr, err)
268+
a.log.Warnf("Failed to handle decode ICE to %s from %s: %v", c.addr(), srcAddr, err)
267269
return
268270
}
269271

@@ -279,7 +281,7 @@ func (c *candidateBase) handleInboundPacket(buf []byte, srcAddr net.Addr) {
279281
if !c.validateSTUNTrafficCache(srcAddr) {
280282
remoteCandidate, valid := a.validateNonSTUNTraffic(c, srcAddr) //nolint:contextcheck
281283
if !valid {
282-
a.log.Warnf("Discarded message from %s, not a valid remote candidate", c.addr())
284+
a.log.Warnf("Discarded message to %s, not a valid remote candidate", c.addr())
283285
return
284286
}
285287
c.addRemoteCandidateCache(remoteCandidate, srcAddr)

‎gather.go‎

Lines changed: 36 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -165,44 +165,49 @@ func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []Networ
165165

166166
switch network {
167167
case tcp:
168-
if a.tcpMux == nil {
169-
continue
170-
}
171-
172-
// Handle ICE TCP passive mode
173-
var muxConns []net.PacketConn
174-
if multi, ok := a.tcpMux.(AllConnsGetter); ok {
175-
a.log.Debugf("GetAllConns by ufrag: %s", a.localUfrag)
176-
muxConns, err = multi.GetAllConns(a.localUfrag, mappedIP.To4() == nil, ip)
177-
if err != nil {
178-
a.log.Warnf("Failed to get all TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
179-
continue
180-
}
168+
if a.ActiveTCP {
169+
conns = append(conns, connAndPort{nil, 0})
170+
tcpType = TCPTypeActive
181171
} else {
182-
a.log.Debugf("GetConn by ufrag: %s", a.localUfrag)
183-
conn, err := a.tcpMux.GetConnByUfrag(a.localUfrag, mappedIP.To4() == nil, ip)
184-
if err != nil {
185-
a.log.Warnf("Failed to get TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
172+
// Handle ICE TCP passive mode
173+
if a.tcpMux == nil {
186174
continue
187175
}
188-
muxConns = []net.PacketConn{conn}
189-
}
190176

191-
// Extract the port for each PacketConn we got.
192-
for _, conn := range muxConns {
193-
if tcpConn, ok := conn.LocalAddr().(*net.TCPAddr); ok {
194-
conns = append(conns, connAndPort{conn, tcpConn.Port})
177+
var muxConns []net.PacketConn
178+
if multi, ok := a.tcpMux.(AllConnsGetter); ok {
179+
a.log.Debugf("GetAllConns by ufrag: %s", a.localUfrag)
180+
muxConns, err = multi.GetAllConns(a.localUfrag, mappedIP.To4() == nil, ip)
181+
if err != nil {
182+
a.log.Warnf("Failed to get all TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
183+
continue
184+
}
195185
} else {
196-
a.log.Warnf("Failed to get port of connection from TCPMux: %s %s %s", network, ip, a.localUfrag)
186+
a.log.Debugf("GetConn by ufrag: %s", a.localUfrag)
187+
conn, err := a.tcpMux.GetConnByUfrag(a.localUfrag, mappedIP.To4() == nil, ip)
188+
if err != nil {
189+
a.log.Warnf("Failed to get TCP connections by ufrag: %s %s %s", network, ip, a.localUfrag)
190+
continue
191+
}
192+
muxConns = []net.PacketConn{conn}
197193
}
194+
195+
// Extract the port for each PacketConn we got.
196+
for _, conn := range muxConns {
197+
if tcpConn, ok := conn.LocalAddr().(*net.TCPAddr); ok {
198+
conns = append(conns, connAndPort{conn, tcpConn.Port})
199+
} else {
200+
a.log.Warnf("Failed to get port of connection from TCPMux: %s %s %s", network, ip, a.localUfrag)
201+
}
202+
}
203+
if len(conns) == 0 {
204+
// Didn't succeed with any, try the next network.
205+
continue
206+
}
207+
tcpType = TCPTypePassive
208+
// Is there a way to verify that the listen address is even
209+
// accessible from the current interface.
198210
}
199-
if len(conns) == 0 {
200-
// Didn't succeed with any, try the next network.
201-
continue
202-
}
203-
tcpType = TCPTypePassive
204-
// Is there a way to verify that the listen address is even
205-
// accessible from the current interface.
206211
case udp:
207212
conn, err := listenUDPInPortRange(a.net, a.log, int(a.portMax), int(a.portMin), network, &net.UDPAddr{IP: ip, Port: 0})
208213
if err != nil {

0 commit comments

Comments
 (0)