16 "github.com/anacrolix/log"
17 "github.com/anacrolix/missinggo/v2/iter"
18 qt "github.com/frankban/quicktest"
19 "github.com/stretchr/testify/assert"
20 "github.com/stretchr/testify/require"
22 "github.com/anacrolix/torrent/internal/testutil"
25 // Check that after completing leeching, a leecher transitions to a seeding
26 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
27 func TestHolepunchConnect(t *testing.T) {
29 greetingTempDir, mi := testutil.GreetingTestTorrent()
30 defer os.RemoveAll(greetingTempDir)
32 cfg := TestingConfig(t)
34 cfg.MaxAllocPeerRequestDataPerConn = 4
35 cfg.DataDir = greetingTempDir
38 cfg.AcceptPeerConnections = false
39 // Listening, even without accepting, still means the leecher-leecher completes the dial to the seeder, and so it
40 // won't attempt to holepunch.
42 seeder, err := NewClient(cfg)
43 require.NoError(t, err)
45 defer testutil.ExportStatusWriter(seeder, "s", t)()
46 seederTorrent, ok, err := seeder.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
47 require.NoError(t, err)
49 seederTorrent.VerifyData()
51 cfg = TestingConfig(t)
53 cfg.DataDir = t.TempDir()
54 cfg.AlwaysWantConns = true
55 // This way the leecher leecher will still try to use this peer as a relay, but won't be told
56 // about the seeder via PEX.
57 //cfg.DisablePEX = true
59 leecher, err := NewClient(cfg)
60 require.NoError(t, err)
62 defer testutil.ExportStatusWriter(leecher, "l", t)()
64 cfg = TestingConfig(t)
66 cfg.DataDir = t.TempDir()
67 cfg.MaxAllocPeerRequestDataPerConn = 4
69 cfg.NominalDialTimeout = time.Second
70 //cfg.DisableUTP = true
71 leecherLeecher, _ := NewClient(cfg)
72 require.NoError(t, err)
73 defer leecherLeecher.Close()
74 defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
75 leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *TorrentSpec) {
76 ret = TorrentSpecFromMetaInfo(mi)
81 require.NoError(t, err)
83 llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *TorrentSpec) {
84 ret = TorrentSpecFromMetaInfo(mi)
88 require.NoError(t, err)
97 qt.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), qt.IsNil)
99 go seederTorrent.AddClientPeer(leecher)
100 waitForConns(seederTorrent)
101 go llg.AddClientPeer(leecher)
103 time.Sleep(time.Second)
105 targetAddr := seeder.ListenAddrs()[0]
106 log.Printf("trying to initiate to %v", targetAddr)
107 initiateConn(outgoingConnOpts{
112 requireRendezvous: true,
113 skipHolepunchRendezvous: false,
114 HeaderObfuscationPolicy: llg.cl.config.HeaderObfuscationPolicy,
119 c.Check(seeder.dialedSuccessfullyAfterHolepunchConnect, qt.Not(qt.HasLen), 0)
120 c.Check(leecherLeecher.probablyOnlyConnectedDueToHolepunch, qt.Not(qt.HasLen), 0)
122 llClientStats := leecherLeecher.Stats()
123 c.Check(llClientStats.NumPeersUndialableWithoutHolepunch, qt.Not(qt.Equals), 0)
124 c.Check(llClientStats.NumPeersUndialableWithoutHolepunchDialedAfterHolepunchConnect, qt.Not(qt.Equals), 0)
125 c.Check(llClientStats.NumPeersProbablyOnlyConnectedDueToHolepunch, qt.Not(qt.Equals), 0)
128 func waitForConns(t *Torrent) {
139 // Show that dialling TCP will complete before the other side accepts.
140 func TestDialTcpNotAccepting(t *testing.T) {
141 l, err := net.Listen("tcp", "localhost:0")
143 c.Check(err, qt.IsNil)
145 dialedConn, err := net.Dial("tcp", l.Addr().String())
146 c.Assert(err, qt.IsNil)
150 func TestTcpSimultaneousOpen(t *testing.T) {
151 const network = "tcp"
152 ctx := context.Background()
153 makeDialer := func(localPort int, remoteAddr string) func() (net.Conn, error) {
154 dialer := net.Dialer{
155 LocalAddr: &net.TCPAddr{
159 return func() (net.Conn, error) {
160 return dialer.DialContext(ctx, network, remoteAddr)
164 first, second := randPortPair()
165 t.Logf("ports are %v and %v", first, second)
166 err := testSimultaneousOpen(
168 makeDialer(first, fmt.Sprintf("localhost:%d", second)),
169 makeDialer(second, fmt.Sprintf("localhost:%d", first)),
171 c.Assert(err, qt.IsNil)
174 func randIntInRange(low, high int) int {
175 return rand.Intn(high-low+1) + low
178 func randDynamicPort() int {
179 return randIntInRange(49152, 65535)
182 func randPortPair() (first int, second int) {
183 first = randDynamicPort()
185 second = randDynamicPort()
192 func writeMsg(conn net.Conn) {
193 conn.Write([]byte(defaultMsg))
194 // Writing must be closed so the reader will get EOF and stop reading.
198 func readMsg(conn net.Conn) error {
199 msgBytes, err := io.ReadAll(conn)
203 msgStr := string(msgBytes)
204 if msgStr != defaultMsg {
205 return fmt.Errorf("read %q", msgStr)
210 var errMsgNotReceived = errors.New("msg not received in time")
212 // Runs two dialers simultaneously, then sends a message on one connection and check it reads from
213 // the other, thereby showing that both dials obtained endpoints to the same connection.
214 func testSimultaneousOpen(
215 cleanup func(func()),
216 firstDialer, secondDialer func() (net.Conn, error),
218 errs := make(chan error)
219 var dialsDone sync.WaitGroup
221 dialsDone.Add(numDials)
222 signal := make(chan struct{})
223 var dialersDone sync.WaitGroup
224 dialersDone.Add(numDials)
226 dialer func() (net.Conn, error),
227 onSignal func(net.Conn),
229 defer dialersDone.Done()
230 conn, err := dialer()
247 func(conn net.Conn) {
254 func(conn net.Conn) {
255 gotMsg := make(chan error, 1)
257 gotMsg <- readMsg(conn)
260 case err := <-gotMsg:
262 case <-time.After(time.Second):
263 errs <- errMsgNotReceived
268 for range iter.N(numDials) {
275 for range iter.N(numDials) {
285 const defaultMsg = "hello"
287 // Show that uTP doesn't implement simultaneous open. When two sockets dial each other, they both
288 // get separate connections. This means that holepunch connect may result in an accept (and dial)
289 // for one or both peers involved.
290 func TestUtpSimultaneousOpen(t *testing.T) {
292 const network = "udp"
293 ctx := context.Background()
294 newUtpSocket := func(addr string) utpSocket {
295 socket, err := NewUtpSocket(
298 func(net.Addr) bool {
303 c.Assert(err, qt.IsNil)
306 first := newUtpSocket("localhost:3000")
308 second := newUtpSocket("localhost:3001")
310 getDial := func(sock utpSocket, addr string) func() (net.Conn, error) {
311 return func() (net.Conn, error) {
312 return sock.DialContext(ctx, network, addr)
315 err := testSimultaneousOpen(
317 getDial(first, "localhost:3001"),
318 getDial(second, "localhost:3000"),
320 c.Assert(err, qt.ErrorIs, errMsgNotReceived)
323 func testDirectDialMsg(c *qt.C, r, w net.Conn) {
326 c.Assert(err, qt.IsNil)
329 // Show that dialling one socket and accepting from the other results in them having ends of the
331 func TestUtpDirectDialMsg(t *testing.T) {
333 const network = "udp"
334 ctx := context.Background()
335 newUtpSocket := func(addr string) utpSocket {
336 socket, err := NewUtpSocket(network, addr, func(net.Addr) bool {
339 c.Assert(err, qt.IsNil)
342 first := newUtpSocket("localhost:0")
344 second := newUtpSocket("localhost:0")
346 writer, err := first.DialContext(ctx, network, second.Addr().String())
347 c.Assert(err, qt.IsNil)
349 reader, err := second.Accept()
351 c.Assert(err, qt.IsNil)
352 testDirectDialMsg(c, reader, writer)