feat(UnderlyingNetwork): add support for ListenTCP (#1278)

The lack of this support already created some difficulties inside the
testingx package and I am increasinglty sick of it.

While there, see to increase the testing coverage of the netxlite
package.

While there acknowledge and commit workaround for
ooni/probe#2537.

Added while looking into moving forward with making beacons fully
testable using netem, per ooni/probe#2531.

internal/mocks/underlyingnetwork.go

@@ -17,6 +17,8 @@ type UnderlyingNetwork struct {
 
 	MockDialContext func(ctx context.Context, network, address string) (net.Conn, error)
 
+	MockListenTCP func(network string, addr *net.TCPAddr) (net.Listener, error)
+
 	MockListenUDP func(network string, addr *net.UDPAddr) (model.UDPLikeConn, error)
 
 	MockGetaddrinfoLookupANY func(ctx context.Context, domain string) ([]string, string, error)
@@ -38,6 +40,10 @@ func (un *UnderlyingNetwork) DialContext(ctx context.Context, network, address s
 	return un.MockDialContext(ctx, network, address)
 }
 
+func (un *UnderlyingNetwork) ListenTCP(network string, addr *net.TCPAddr) (net.Listener, error) {
+	return un.MockListenTCP(network, addr)
+}
+
 func (un *UnderlyingNetwork) ListenUDP(network string, addr *net.UDPAddr) (model.UDPLikeConn, error) {
 	return un.MockListenUDP(network, addr)
 }

internal/mocks/underlyingnetwork_test.go

@@ -55,6 +55,22 @@ func TestUnderlyingNetwork(t *testing.T) {
 		}
 	})
 
+	t.Run("ListenTCP", func(t *testing.T) {
+		expect := errors.New("mocked error")
+		un := &UnderlyingNetwork{
+			MockListenTCP: func(network string, addr *net.TCPAddr) (net.Listener, error) {
+				return nil, expect
+			},
+		}
+		listener, err := un.ListenTCP("tcp", &net.TCPAddr{})
+		if !errors.Is(err, expect) {
+			t.Fatal("unexpected err", err)
+		}
+		if listener != nil {
+			t.Fatal("expected nil listener")
+		}
+	})
+
 	t.Run("ListenUDP", func(t *testing.T) {
 		expect := errors.New("mocked error")
 		un := &UnderlyingNetwork{

internal/model/measurement_test.go

@@ -6,8 +6,20 @@ import (
 	"errors"
 	"fmt"
 	"testing"
+	"time"
 )
 
+func TestMeasurementFormatTimeNowUTC(t *testing.T) {
+	t.Run("produces a string using the correct date format", func(t *testing.T) {
+		out := MeasurementFormatTimeNowUTC()
+		result, err := time.Parse(MeasurementDateFormat, out)
+		if err != nil {
+			t.Fatal(err)
+		}
+		_ = result
+	})
+}
+
 func TestMeasurementTargetMarshalJSON(t *testing.T) {
 	var mt MeasurementTarget
 	data, err := json.Marshal(mt)

internal/model/netx.go

@@ -562,6 +562,9 @@ type UnderlyingNetwork interface {
 	// GetaddrinfoResolverNetwork returns the resolver network.
 	GetaddrinfoResolverNetwork() string
 
+	// ListenTCP is equivalent to net.ListenTCP.
+	ListenTCP(network string, addr *net.TCPAddr) (net.Listener, error)
+
 	// ListenUDP is equivalent to net.ListenUDP.
 	ListenUDP(network string, addr *net.UDPAddr) (UDPLikeConn, error)
 }

internal/netxlite/dialer_test.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"io"
 	"net"
+	"runtime"
 	"strings"
 	"sync"
 	"testing"
@@ -104,7 +105,8 @@ func TestDialerSystem(t *testing.T) {
 			defaultTp := &DefaultTProxy{}
 			tp := &mocks.UnderlyingNetwork{
 				MockDialTimeout: func() time.Duration {
-					// Note: this test is notoriously flaky on Windows
+					// Note: this test is notoriously flaky on Windows as documented by
+					// TODO(https://github.com/ooni/probe/issues/2537)
 					return time.Nanosecond
 				},
 				MockDialContext: defaultTp.DialContext,
@@ -115,7 +117,10 @@ func TestDialerSystem(t *testing.T) {
 			conn, err := d.DialContext(ctx, "tcp", "dns.google:443")
 			stop := time.Now()
 			if err == nil || !strings.HasSuffix(err.Error(), "i/o timeout") {
-				t.Fatal(err)
+				if runtime.GOOS == "windows" {
+					t.Skip("https://github.com/ooni/probe/issues/2537")
+				}
+				t.Fatal("unexpected error", err)
 			}
 			if conn != nil {
 				t.Fatal("unexpected conn")

internal/netxlite/dnsovergetaddrinfo_test.go

@@ -13,6 +13,14 @@ import (
 	"github.com/ooni/probe-cli/v3/internal/mocks"
 )
 
+func TestNewDNSOverGetaddrinfoTransport(t *testing.T) {
+	txp := NewDNSOverGetaddrinfoTransport()
+	underlying := txp.(*dnsOverGetaddrinfoTransport)
+	if underlying.provider.underlying != nil {
+		t.Fatal("expected to see a nil underlying network")
+	}
+}
+
 func TestDNSOverGetaddrinfo(t *testing.T) {
 	t.Run("RequiresPadding", func(t *testing.T) {
 		txp := &dnsOverGetaddrinfoTransport{}

internal/netxlite/netem.go

@@ -43,6 +43,11 @@ func (a *NetemUnderlyingNetworkAdapter) GetaddrinfoResolverNetwork() string {
 	return a.UNet.GetaddrinfoResolverNetwork()
 }
 
+// ListenTCP implements model.UnderlyingNetwork
+func (a *NetemUnderlyingNetworkAdapter) ListenTCP(network string, addr *net.TCPAddr) (net.Listener, error) {
+	return a.UNet.ListenTCP(network, addr)
+}
+
 // ListenUDP implements model.UnderlyingNetwork
 func (a *NetemUnderlyingNetworkAdapter) ListenUDP(network string, addr *net.UDPAddr) (model.UDPLikeConn, error) {
 	return a.UNet.ListenUDP(network, addr)

internal/netxlite/netem_test.go

@@ -0,0 +1,63 @@
+package netxlite
+
+import (
+	"context"
+	"net"
+	"sync"
+	"testing"
+
+	"github.com/apex/log"
+	"github.com/ooni/netem"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
+)
+
+func TestNetemUnderlyingNetworkAdapter(t *testing.T) {
+
+	// This test case explicitly ensures we can use the adapter to listen for TCP
+	t.Run("ListenTCP", func(t *testing.T) {
+		// create a star network topology
+		topology := runtimex.Try1(netem.NewStarTopology(log.Log))
+		defer topology.Close()
+
+		// constants for the IP address we're using
+		const (
+			clientAddress = "130.192.91.211"
+			serverAddress = "93.184.216.34"
+		)
+
+		// create the stacks
+		serverStack := runtimex.Try1(topology.AddHost(serverAddress, "0.0.0.0", &netem.LinkConfig{}))
+		clientStack := runtimex.Try1(topology.AddHost(clientAddress, "0.0.0.0", &netem.LinkConfig{}))
+
+		// wrap the server stack and create listening socket
+		serverAdapter := &NetemUnderlyingNetworkAdapter{serverStack}
+		serverEndpoint := &net.TCPAddr{IP: net.ParseIP(serverAddress), Port: 54321}
+		listener := runtimex.Try1(serverAdapter.ListenTCP("tcp", serverEndpoint))
+		defer listener.Close()
+
+		// listen in a background goroutine
+		wg := &sync.WaitGroup{}
+		wg.Add(1)
+		go func() {
+			conn := runtimex.Try1(listener.Accept())
+			conn.Close()
+			wg.Done()
+		}()
+
+		// wrap the client stack
+		clientAdapter := &NetemUnderlyingNetworkAdapter{clientStack}
+
+		// connect in a background goroutine
+		wg.Add(1)
+		go func() {
+			ctx := context.Background()
+			conn := runtimex.Try1(clientAdapter.DialContext(ctx, "tcp", serverEndpoint.String()))
+			conn.Close()
+			wg.Done()
+		}()
+
+		// wait for all operations to complete
+		wg.Wait()
+	})
+
+}

internal/netxlite/netx_test.go

@@ -14,7 +14,8 @@ import (
 	"github.com/quic-go/quic-go/http3"
 )
 
-func TestNetx(t *testing.T) {
+// This test ensures that a Netx wrapping a netem.UNet is WAI
+func TestNetxWithNetem(t *testing.T) {
 	// create a star network topology
 	topology := runtimex.Try1(netem.NewStarTopology(log.Log))
 	defer topology.Close()

internal/netxlite/tproxy.go

@@ -86,6 +86,11 @@ func (tp *DefaultTProxy) DialContext(ctx context.Context, network, address strin
 	return d.DialContext(ctx, network, address)
 }
 
+// ListenTCP implements UnderlyingNetwork.
+func (tp *DefaultTProxy) ListenTCP(network string, addr *net.TCPAddr) (net.Listener, error) {
+	return net.ListenTCP(network, addr)
+}
+
 // ListenUDP implements UnderlyingNetwork.
 func (tp *DefaultTProxy) ListenUDP(network string, addr *net.UDPAddr) (model.UDPLikeConn, error) {
 	return net.ListenUDP(network, addr)

internal/netxlite/tproxy_test.go

@@ -6,11 +6,13 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/ooni/probe-cli/v3/internal/mocks"
 	"github.com/ooni/probe-cli/v3/internal/model"
+	"github.com/ooni/probe-cli/v3/internal/runtimex"
 )
 
 func TestTproxyNilSafeProvider(t *testing.T) {
@@ -99,3 +101,33 @@ func TestWithCustomTProxy(t *testing.T) {
 		})
 	})
 }
+
+// We generally do not listen here as part of other tests, since the listening
+// functionality is mainly only use for testingx. So, here's a specific test for that.
+func TestTproxyListenTCP(t *testing.T) {
+	tproxy := &DefaultTProxy{}
+
+	listener := runtimex.Try1(tproxy.ListenTCP("tcp", &net.TCPAddr{}))
+	serverEndpoint := listener.Addr().String()
+
+	// listen in a background goroutine
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+	go func() {
+		conn := runtimex.Try1(listener.Accept())
+		conn.Close()
+		wg.Done()
+	}()
+
+	// dial in a background goroutine
+	wg.Add(1)
+	go func() {
+		ctx := context.Background()
+		conn := runtimex.Try1(tproxy.DialContext(ctx, "tcp", serverEndpoint))
+		conn.Close()
+		wg.Done()
+	}()
+
+	// wait for the goroutines to finish
+	wg.Wait()
+}