test(transports): dual stack 1 (#2698)

Author: rlve

tests/libp2p/transports/stream_tests.nim

@@ -4,7 +4,8 @@
 {.used.}
 
 import chronos, stew/byteutils, std/random
-import ../../../libp2p/[stream/connection, transports/transport, muxers/muxer]
+import
+  ../../../libp2p/[stream/connection, transports/transport, muxers/muxer, multiaddress]
 import ../../tools/[stream, sync]
 import ./utils
 
@@ -57,6 +58,49 @@ template streamTransportTest*(
 
     runTransportTest(transportProvider, streamProvider, addressIP6.get())
 
+  asyncTest "start binds both IPv4 and IPv6 addresses":
+    if addressIP6.isNone:
+      skip() # ipv6 not supported
+      return
+
+    let server = transportProvider()
+    await server.start(@[addressIP4, addressIP6.get()])
+    defer:
+      await server.stop()
+
+    check:
+      server.addrs.len == 2
+      extractPort(server.addrs.addrByFamily(IP4)) > 0
+      extractPort(server.addrs.addrByFamily(IP6)) > 0
+
+  asyncTest "dual-stack server exchanges data with IPv4 and IPv6 clients":
+    if addressIP6.isNone:
+      skip() # ipv6 not supported
+      return
+
+    proc serverStreamHandler(stream: MuxedStream) {.async: (raises: []).} =
+      noExceptionWithStreamClose(stream):
+        var buffer: array[clientMessage.len, byte]
+        await stream.readExactly(addr buffer, clientMessage.len)
+        check string.fromBytes(buffer) == clientMessage
+        await stream.write(serverMessage)
+
+    proc clientStreamHandler(stream: MuxedStream) {.async: (raises: []).} =
+      noExceptionWithStreamClose(stream):
+        await stream.write(clientMessage)
+
+        var buffer: array[serverMessage.len, byte]
+        await stream.readExactly(addr buffer, serverMessage.len)
+        check string.fromBytes(buffer) == serverMessage
+
+    await dualStackStreamScenario(
+      @[addressIP4, addressIP6.get()],
+      transportProvider,
+      streamProvider,
+      serverStreamHandler,
+      clientStreamHandler,
+    )
+
   asyncTest "read/write Lp":
     proc serverStreamHandler(stream: MuxedStream) {.async: (raises: []).} =
       noExceptionWithStreamClose(stream):
@@ -76,6 +120,35 @@ template streamTransportTest*(
       clientStreamHandler,
     )
 
+  asyncTest "Connection.reset aborts the initiator stream":
+    var serverResetDone = newFuture[void]()
+
+    proc serverStreamHandler(stream: MuxedStream) {.async: (raises: []).} =
+      noExceptionWithStreamClose(stream):
+        let msg = await stream.readLp(100)
+        check msg == fromHex("1234")
+        await stream.reset()
+        serverResetDone.complete()
+
+    proc clientStreamHandler(stream: MuxedStream) {.async: (raises: []).} =
+      noExceptionWithStreamClose(stream):
+        await stream.writeLp(fromHex("1234"))
+        await serverResetDone
+
+        var buffer: array[1, byte]
+        check (await stream.readOnce(addr buffer[0], 1)) == 0
+
+        expect LPStreamResetError:
+          await stream.writeLp(fromHex("1234"))
+
+    await runSingleStreamScenario(
+      @[addressIP4],
+      transportProvider,
+      streamProvider,
+      serverStreamHandler,
+      clientStreamHandler,
+    )
+
   asyncTest "EOF handling - first readOnce at EOF + repeated reads":
     var serverHandlerDone = newFuture[void]()
 

tests/libp2p/transports/test_quic.nim

@@ -5,8 +5,13 @@
 
 import chronos, random, stew/byteutils
 import
-  ../../../libp2p/
-    [transports/transport, transports/quictransport, upgrademngrs/upgrade, muxers/muxer]
+  ../../../libp2p/[
+    transports/transport,
+    transports/quictransport,
+    upgrademngrs/upgrade,
+    muxers/muxer,
+    multiaddress,
+  ]
 import ../../tools/[unittest, crypto as cryptoTools, multiaddress]
 import ./basic_tests
 import ./stream_tests
@@ -50,35 +55,6 @@ suite "Quic transport":
     streamProvider,
   )
 
-  asyncTest "Connection.reset aborts the initiator stream":
-    var serverResetDone = newFuture[void]()
-
-    proc serverStreamHandler(stream: MuxedStream) {.async: (raises: []).} =
-      noExceptionWithStreamClose(stream):
-        let msg = await stream.readLp(100)
-        check msg == fromHex("1234")
-        await stream.reset()
-        serverResetDone.complete()
-
-    proc clientStreamHandler(stream: MuxedStream) {.async: (raises: []).} =
-      noExceptionWithStreamClose(stream):
-        await stream.writeLp(fromHex("1234"))
-        await serverResetDone
-
-        var buffer: array[1, byte]
-        check (await stream.readOnce(addr buffer[0], 1)) == 0
-
-        expect LPStreamResetError:
-          await stream.writeLp(fromHex("1234"))
-
-    await runSingleStreamScenario(
-      @[MultiAddress.init(addressIP4).get()],
-      quicTransProvider,
-      streamProvider,
-      serverStreamHandler,
-      clientStreamHandler,
-    )
-
   asyncTest "transport e2e - invalid cert - server":
     let server = await createQuicTransport(isServer = true, withInvalidCert = true)
     asyncSpawn createServerAcceptConn(server)()
@@ -131,9 +107,9 @@ suite "Quic transport":
       discard await client.dial("", server.addrs[0], Opt.some(wrongPeerId))
 
   asyncTest "should allow multiple local addresses":
-    let key = PrivateKey.random(ECDSA, rng()).tryGet()
-    let server = QuicTransport.new(Upgrade(), key)
-    await server.start(@[QuicAutoAddress, QuicAutoAddress])
+    let server = await createQuicTransport(
+      isServer = true, addresses = @[QuicAutoAddress, QuicAutoAddress]
+    )
     defer:
       await server.stop()
 
@@ -183,9 +159,9 @@ suite "Quic transport":
       extractPort(serverConn.observedAddr.get()) == clientListenPort
 
   asyncTest "dial uses dial-only endpoint with multiple listener matches":
-    let key = PrivateKey.random(ECDSA, rng()).tryGet()
-    let client = QuicTransport.new(Upgrade(), key)
-    await client.start(@[QuicAutoAddress, QuicAutoAddress])
+    let client = await createQuicTransport(
+      isServer = true, addresses = @[QuicAutoAddress, QuicAutoAddress]
+    )
     let server = await createQuicTransport(isServer = true)
     defer:
       await allFutures(client.stop(), server.stop())
@@ -227,6 +203,54 @@ suite "Quic transport":
       extractPort(serverConn.observedAddr.get()) ==
         extractPort(clientConn.localAddr.get())
 
+  asyncTest "dual-stack dialer reuses the matching-family listener":
+    # Dialing from the listener endpoint makes the remote observe the listen port.
+    # Port reuse and DCUtR hole punching depend on that.
+    let dialer = await createQuicTransport(
+      isServer = true, addresses = @[QuicAutoAddressIP4, QuicAutoAddressIP6]
+    )
+    let server =
+      await createQuicTransport(isServer = true, addresses = @[QuicAutoAddressIP6])
+    defer:
+      await allFutures(dialer.stop(), server.stop())
+
+    let dialerIPv6Port = extractPort(dialer.addrs.addrByFamily(IP6))
+
+    let acceptFut = server.accept()
+    let dialerConn = await dialer.dial("", server.addrs[0])
+    let serverConn = await acceptFut
+    defer:
+      await allFutures(dialerConn.close(), serverConn.close())
+
+    check:
+      serverConn.observedAddr.isSome()
+      # same port as the IPv6 listener means that listener was reused
+      extractPort(serverConn.observedAddr.get()) == dialerIPv6Port
+
+  asyncTest "dial uses an IPv6 dial-only endpoint when only an IPv4 listener exists":
+    # An IPv4 socket cannot carry an IPv6 dial, so the IPv4 listener cannot be
+    # reused and a separate IPv6 dial-only endpoint has to be opened.
+    let dialer =
+      await createQuicTransport(isServer = true, addresses = @[QuicAutoAddressIP4])
+    let server =
+      await createQuicTransport(isServer = true, addresses = @[QuicAutoAddressIP6])
+    defer:
+      await allFutures(dialer.stop(), server.stop())
+
+    let dialerIPv4Port = extractPort(dialer.addrs[0])
+
+    let acceptFut = server.accept()
+    let dialerConn = await dialer.dial("", server.addrs[0])
+    let serverConn = await acceptFut
+    defer:
+      await allFutures(dialerConn.close(), serverConn.close())
+
+    check:
+      dialer.addrs.len == 1
+      serverConn.observedAddr.isSome()
+      # a different port than the IPv4 listener means a separate endpoint was used
+      extractPort(serverConn.observedAddr.get()) != dialerIPv4Port
+
   asyncTest "server not accepting":
     let server = await createQuicTransport(isServer = true)
     # intentionally not calling createServerAcceptConn as server should not accept

tests/libp2p/transports/utils.nim

@@ -80,7 +80,7 @@ proc createQuicTransport*(
     isServer: bool = false,
     withInvalidCert: bool = false,
     privateKey: Opt[PrivateKey] = Opt.none(PrivateKey),
-    address: MultiAddress = QuicAutoAddress,
+    addresses: seq[MultiAddress] = @[QuicAutoAddress],
 ): Future[QuicTransport] {.async.} =
   let key =
     if privateKey.isNone:
@@ -95,8 +95,7 @@ proc createQuicTransport*(
       QuicTransport.new(Upgrade(), key)
 
   if isServer: # servers are started because they need to listen
-    let ma = @[address]
-    await trans.start(ma)
+    await trans.start(addresses)
 
   return trans
 
@@ -109,6 +108,13 @@ type StreamProvider* =
 
 type StreamHandler* = proc(stream: MuxedStream) {.async: (raises: []).}
 
+proc addrByFamily*(addrs: seq[MultiAddress], family: MaPattern): MultiAddress =
+  ## First address whose leading ip component matches `family` (IP4 or IP6).
+  for a in addrs:
+    if family.matchPartial(a):
+      return a
+  raiseAssert "no address for the requested family"
+
 proc extractPort*(ma: MultiAddress): int =
   var codec =
     if isTcpTransport(ma) or isWsTransport(ma):
@@ -148,15 +154,15 @@ proc serverHandlerSingleStream*(
   except CatchableError as exc:
     raiseAssert "should not fail: " & exc.msg
 
-proc clientRunSingleStream*(
-    server: Transport,
+proc clientRunSingleStreamTo*(
+    address: MultiAddress,
     transportProvider: TransportProvider,
     streamProvider: StreamProvider,
     handler: StreamHandler,
 ) {.async: (raises: []).} =
   try:
     let client = transportProvider()
-    let conn = await client.dial("", server.addrs[0])
+    let conn = await client.dial("", address)
     let muxer = streamProvider(conn)
 
     let stream = await muxer.newStream()
@@ -168,6 +174,16 @@ proc clientRunSingleStream*(
   except CatchableError as exc:
     raiseAssert "should not fail: " & exc.msg
 
+proc clientRunSingleStream*(
+    server: Transport,
+    transportProvider: TransportProvider,
+    streamProvider: StreamProvider,
+    handler: StreamHandler,
+) {.async: (raises: []).} =
+  await clientRunSingleStreamTo(
+    server.addrs[0], transportProvider, streamProvider, handler
+  )
+
 proc runSingleStreamScenario*(
     multiAddress: seq[MultiAddress],
     transportProvider: TransportProvider,
@@ -186,6 +202,39 @@ proc runSingleStreamScenario*(
   await allFutures(clientTask, serverTask)
   await server.stop()
 
+proc dualStackStreamScenario*(
+    listenAddrs: seq[MultiAddress],
+    transportProvider: TransportProvider,
+    streamProvider: StreamProvider,
+    serverStreamHandler: StreamHandler,
+    clientStreamHandler: StreamHandler,
+) {.async: (raises: [CancelledError, LPError]).} =
+  ## Start one server on `listenAddrs`, then dial it once per address family from
+  ## a fresh client, serving each accepted connection with `serverStreamHandler`.
+  let server = transportProvider()
+  await server.start(listenAddrs)
+  defer:
+    await server.stop()
+
+  # dial the server's resolved address of each family, not the wildcard input
+  let targets = @[server.addrs.addrByFamily(IP4), server.addrs.addrByFamily(IP6)]
+
+  # accept() is single-consumer, so serve the clients one after another
+  proc serveAll() {.async: (raises: []).} =
+    for _ in targets:
+      await serverHandlerSingleStream(server, streamProvider, serverStreamHandler)
+
+  let serverTask = serveAll()
+  var clientTasks: seq[Future[void]]
+  for target in targets:
+    clientTasks.add(
+      clientRunSingleStreamTo(
+        target, transportProvider, streamProvider, clientStreamHandler
+      )
+    )
+  await allFutures(clientTasks)
+  await serverTask
+
 proc countTransitions*(readOrder: seq[byte]): int =
   var transitions = 0
   for i in 1 ..< readOrder.len: