feat(miniooni): implement torsf tunnel (#921)

This diff adds to miniooni support for using the torsf tunnel. Such a
tunnel consists of a snowflake pluggable transport in front of a custom
instance of tor and requires tor to be installed.

The usage is like:

```
./miniooni --tunnel=torsf [...]
```

The default snowflake rendezvous method is "domain_fronting". You can
select the AMP cache instead using "amp":

```
./miniooni --snowflake-rendezvous=amp --tunnel=torsf [...]
```

Part of https://github.com/ooni/probe/issues/1955

internal/cmd/miniooni/main.go

@@ -39,6 +39,7 @@ type Options struct {
 	Random              bool
 	RepeatEvery         int64
 	ReportFile          string
+	SnowflakeRendezvous string
 	TorArgs             []string
 	TorBinary           string
 	Tunnel              string
@@ -125,6 +126,13 @@ func main() {
 		"set the output report file path (default: \"report.jsonl\")",
 	)
 
+	flags.StringVar(
+		&globalOptions.SnowflakeRendezvous,
+		"snowflake-rendezvous",
+		"domain_fronting",
+		"rendezvous method for --tunnel=torsf (one of: \"domain_fronting\" and \"amp\")",
+	)
+
 	flags.StringSliceVar(
 		&globalOptions.TorArgs,
 		"tor-args",
@@ -143,7 +151,7 @@ func main() {
 		&globalOptions.Tunnel,
 		"tunnel",
 		"",
-		"tunnel to use to communicate with the OONI backend (one of: tor, psiphon)",
+		"tunnel to use to communicate with the OONI backend (one of: psiphon, tor, torsf)",
 	)
 
 	flags.BoolVarP(

internal/cmd/miniooni/session.go

@@ -39,6 +39,7 @@ func newSessionOrPanic(ctx context.Context, currentOptions *Options,
 		KVStore:             kvstore,
 		Logger:              logger,
 		ProxyURL:            proxyURL,
+		SnowflakeRendezvous: currentOptions.SnowflakeRendezvous,
 		SoftwareName:        softwareName,
 		SoftwareVersion:     softwareVersion,
 		TorArgs:             currentOptions.TorArgs,

internal/engine/session.go

@@ -35,6 +35,10 @@ type SessionConfig struct {
 	TorArgs                []string
 	TorBinary              string
 
+	// SnowflakeRendezvous is the rendezvous method
+	// to be used by the torsf tunnel
+	SnowflakeRendezvous string
+
 	// TunnelDir is the directory where we should store
 	// the state of persistent tunnels. This field is
 	// optional _unless_ you want to use tunnels. In such
@@ -171,12 +175,13 @@ func NewSession(ctx context.Context, config SessionConfig) (*Session, error) {
 	proxyURL := config.ProxyURL
 	if proxyURL != nil {
 		switch proxyURL.Scheme {
-		case "psiphon", "tor", "fake":
+		case "psiphon", "tor", "torsf", "fake":
 			config.Logger.Infof(
 				"starting '%s' tunnel; please be patient...", proxyURL.Scheme)
 			tunnel, _, err := tunnel.Start(ctx, &tunnel.Config{
 				Logger:              config.Logger,
 				Name:                proxyURL.Scheme,
+				SnowflakeRendezvous: config.SnowflakeRendezvous,
 				Session:             &sessionTunnelEarlySession{},
 				TorArgs:             config.TorArgs,
 				TorBinary:           config.TorBinary,

internal/ptx/ptx.go

@@ -74,6 +74,10 @@ type Listener struct {
 	// counts the bytes consumed by the experiment.
 	ExperimentByteCounter *bytecounter.Counter
 
+	// ListenSocks is OPTIONAL and allows you to override the
+	// function called by default to listen for SOCKS5.
+	ListenSocks func(network string, laddr string) (SocksListener, error)
+
 	// Logger is the OPTIONAL logger. When not set, this library
 	// will not emit logs. (But the underlying pluggable transport
 	// may still emit its own log messages.)
@@ -98,10 +102,7 @@ type Listener struct {
 	laddr net.Addr
 
 	// listener allows us to stop the listener.
-	listener ptxSocksListener
+	listener SocksListener
-
-	// overrideListenSocks allows us to override pt.ListenSocks.
-	overrideListenSocks func(network string, laddr string) (ptxSocksListener, error)
 }
 
 // logger returns the Logger, if set, or the defaultLogger.
@@ -148,7 +149,7 @@ func (lst *Listener) forwardWithContext(ctx context.Context, left, right net.Con
 // handleSocksConn handles a new SocksConn connection by establishing
 // the corresponding PT connection and forwarding traffic. This
 // function TAKES OWNERSHIP of the socksConn argument.
-func (lst *Listener) handleSocksConn(ctx context.Context, socksConn ptxSocksConn) error {
+func (lst *Listener) handleSocksConn(ctx context.Context, socksConn SocksConn) error {
 	err := socksConn.Grant(&net.TCPAddr{IP: net.IPv4zero, Port: 0})
 	if err != nil {
 		lst.logger().Warnf("ptx: socksConn.Grant error: %s", err)
@@ -169,10 +170,10 @@ func (lst *Listener) handleSocksConn(ctx context.Context, socksConn ptxSocksConn
 	return nil                                     // used for testing
 }
 
-// ptxSocksListener is a pt.SocksListener-like structure.
+// SocksListener is the listener for socks connections.
-type ptxSocksListener interface {
+type SocksListener interface {
 	// AcceptSocks accepts a socks conn
-	AcceptSocks() (ptxSocksConn, error)
+	AcceptSocks() (SocksConn, error)
 
 	// Addr returns the listening address.
 	Addr() net.Addr
@@ -181,8 +182,8 @@ type ptxSocksListener interface {
 	Close() error
 }
 
-// ptxSocksConn is a pt.SocksConn-like structure.
+// SocksConn is a SOCKS connection.
-type ptxSocksConn interface {
+type SocksConn interface {
 	// net.Conn is the embedded interface.
 	net.Conn
 
@@ -192,7 +193,7 @@ type ptxSocksConn interface {
 
 // acceptLoop accepts and handles local socks connection. This function
 // DOES NOT take ownership of the socks listener.
-func (lst *Listener) acceptLoop(ctx context.Context, ln ptxSocksListener) {
+func (lst *Listener) acceptLoop(ctx context.Context, ln SocksListener) {
 	for {
 		conn, err := ln.AcceptSocks()
 		if err != nil {
@@ -243,15 +244,15 @@ func (lst *Listener) Start() error {
 }
 
 // listenSocks calles either pt.ListenSocks or lst.overrideListenSocks.
-func (lst *Listener) listenSocks(network string, laddr string) (ptxSocksListener, error) {
+func (lst *Listener) listenSocks(network string, laddr string) (SocksListener, error) {
-	if lst.overrideListenSocks != nil {
+	if lst.ListenSocks != nil {
-		return lst.overrideListenSocks(network, laddr)
+		return lst.ListenSocks(network, laddr)
 	}
 	return lst.castListener(pt.ListenSocks(network, laddr))
 }
 
 // castListener casts a pt.SocksListener to ptxSocksListener.
-func (lst *Listener) castListener(in *pt.SocksListener, err error) (ptxSocksListener, error) {
+func (lst *Listener) castListener(in *pt.SocksListener, err error) (SocksListener, error) {
 	if err != nil {
 		return nil, err
 	}
@@ -264,7 +265,7 @@ type ptxSocksListenerAdapter struct {
 }
 
 // AcceptSocks adapts pt.SocksListener.AcceptSocks to ptxSockListener.AcceptSocks.
-func (la *ptxSocksListenerAdapter) AcceptSocks() (ptxSocksConn, error) {
+func (la *ptxSocksListenerAdapter) AcceptSocks() (SocksConn, error) {
 	return la.SocksListener.AcceptSocks()
 }
 

internal/ptx/ptx_test.go

@@ -73,7 +73,7 @@ func TestListenerWorksWithFakeDialer(t *testing.T) {
 func TestListenerCannotListen(t *testing.T) {
 	expected := errors.New("mocked error")
 	lst := &Listener{
-		overrideListenSocks: func(network, laddr string) (ptxSocksListener, error) {
+		ListenSocks: func(network, laddr string) (SocksListener, error) {
 			return nil, expected
 		},
 	}
@@ -193,7 +193,7 @@ func TestListenerForwardWithNaturalTermination(t *testing.T) {
 // mockableSocksListener is a mockable ptxSocksListener.
 type mockableSocksListener struct {
 	// MockAcceptSocks allows to mock AcceptSocks.
-	MockAcceptSocks func() (ptxSocksConn, error)
+	MockAcceptSocks func() (SocksConn, error)
 
 	// MockAddr allows to mock Addr.
 	MockAddr func() net.Addr
@@ -203,7 +203,7 @@ type mockableSocksListener struct {
 }
 
 // AcceptSocks implemements ptxSocksListener.AcceptSocks.
-func (m *mockableSocksListener) AcceptSocks() (ptxSocksConn, error) {
+func (m *mockableSocksListener) AcceptSocks() (SocksConn, error) {
 	return m.MockAcceptSocks()
 }
 
@@ -220,7 +220,7 @@ func (m *mockableSocksListener) Close() error {
 func TestListenerLoopWithTemporaryError(t *testing.T) {
 	isclosed := &atomicx.Int64{}
 	sl := &mockableSocksListener{
-		MockAcceptSocks: func() (ptxSocksConn, error) {
+		MockAcceptSocks: func() (SocksConn, error) {
 			if isclosed.Load() > 0 {
 				return nil, io.EOF
 			}

internal/tunnel/config.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cretz/bine/control"
 	"github.com/cretz/bine/tor"
 	"github.com/ooni/probe-cli/v3/internal/model"
+	"github.com/ooni/probe-cli/v3/internal/ptx"
 	"golang.org/x/sys/execabs"
 )
 
@@ -28,6 +29,10 @@ type Config struct {
 	// of obtaining a valid psiphon configuration.
 	Session Session
 
+	// SnowflakeRendezvous is the OPTIONAL rendezvous
+	// method for snowflake
+	SnowflakeRendezvous string
+
 	// TunnelDir is the MANDATORY directory in which the tunnel SHOULD
 	// store its state, if any. If this field is empty, the
 	// Start function fails with ErrEmptyTunnelDir.
@@ -56,6 +61,18 @@ type Config struct {
 	// testNetListen allows us to mock net.Listen in testing code.
 	testNetListen func(network string, address string) (net.Listener, error)
 
+	// testSfListenSocks is OPTIONAL and allows to override the
+	// ListenSocks field of a ptx.Listener.
+	testSfListenSocks func(network string, laddr string) (ptx.SocksListener, error)
+
+	// testSfNewPTXListener is OPTIONAL and allows us to wrap the
+	// constructed ptx.Listener for testing purposes.
+	testSfWrapPTXListener func(torsfPTXListener) torsfPTXListener
+
+	// testSfTorStart is OPTIONAL and allows us to override the
+	// call to torStart inside the torsf tunnel.
+	testSfTorStart func(ctx context.Context, config *Config) (Tunnel, DebugInfo, error)
+
 	// testSocks5New allows us to mock socks5.New in testing code.
 	testSocks5New func(conf *socks5.Config) (*socks5.Server, error)
 
@@ -74,6 +91,14 @@ type Config struct {
 	testTorGetInfo func(ctrl *control.Conn, keys ...string) ([]*control.KeyVal, error)
 }
 
+// snowflakeRendezvousMethod returns the rendezvous method that snowflake should use
+func (c *Config) snowflakeRendezvousMethod() string {
+	if c.SnowflakeRendezvous != "" {
+		return c.SnowflakeRendezvous
+	}
+	return "domain_fronting"
+}
+
 // logger returns the logger to use.
 func (c *Config) logger() model.Logger {
 	if c.Logger != nil {

internal/tunnel/torsf.go

@@ -0,0 +1,122 @@
+package tunnel
+
+//
+// torsf: Tor+snowflake tunnel
+//
+
+import (
+	"context"
+	"net/url"
+	"time"
+
+	"github.com/ooni/probe-cli/v3/internal/bytecounter"
+	"github.com/ooni/probe-cli/v3/internal/ptx"
+)
+
+// torsfStart starts the torsf (tor+snowflake) tunnel
+func torsfStart(ctx context.Context, config *Config) (Tunnel, DebugInfo, error) {
+	config.logger().Infof("tunnel: starting snowflake with %s rendezvous method", config.snowflakeRendezvousMethod())
+	if err := ctx.Err(); err != nil {
+		return nil, DebugInfo{}, err
+	}
+
+	// 1. start a listener using snowflake
+	sfdialer, err := newSnowflakeDialer(config)
+	if err != nil {
+		return nil, DebugInfo{}, err
+	}
+	ptl := config.sfNewPTXListener(ctx, sfdialer)
+	if err := ptl.Start(); err != nil {
+		return nil, DebugInfo{}, err
+	}
+
+	// 2. append arguments to the configuration
+	extraArguments := []string{
+		"UseBridges", "1",
+		"ClientTransportPlugin", ptl.AsClientTransportPluginArgument(),
+		"Bridge", sfdialer.AsBridgeArgument(),
+	}
+	config.TorArgs = append(config.TorArgs, extraArguments...)
+
+	// 3. start tor as we would normally do
+	torTunnel, debugInfo, err := config.sfTorStart(ctx, config)
+	debugInfo.Name = "torsf"
+	if err != nil {
+		ptl.Stop()
+		return nil, debugInfo, err
+	}
+
+	// 4. wrap the tunnel and the listener
+	tsft := &torsfTunnel{
+		torTunnel:  torTunnel,
+		sfListener: ptl,
+	}
+	return tsft, debugInfo, nil
+}
+
+func (c *Config) sfNewPTXListener(ctx context.Context, sfdialer *ptx.SnowflakeDialer) (out torsfPTXListener) {
+	out = &ptx.Listener{
+		ExperimentByteCounter: nil,
+		ListenSocks:           c.testSfListenSocks,
+		Logger:                c.logger(),
+		PTDialer:              sfdialer,
+		SessionByteCounter:    bytecounter.ContextSessionByteCounter(ctx),
+	}
+	if c.testSfWrapPTXListener != nil {
+		out = c.testSfWrapPTXListener(out)
+	}
+	return
+}
+
+// torsfPTXListener is an abstract ptx.Listener.
+type torsfPTXListener interface {
+	Start() error
+	Stop()
+	AsClientTransportPluginArgument() string
+}
+
+func (c *Config) sfTorStart(ctx context.Context, config *Config) (Tunnel, DebugInfo, error) {
+	if c.testSfTorStart != nil {
+		return c.testSfTorStart(ctx, config)
+	}
+	return torStart(ctx, config)
+}
+
+// newSnowflakeDialer returns the correct snowflake dialer.
+func newSnowflakeDialer(config *Config) (*ptx.SnowflakeDialer, error) {
+	rm, err := ptx.NewSnowflakeRendezvousMethod(config.snowflakeRendezvousMethod())
+	if err != nil {
+		return nil, err
+	}
+	sfDialer := ptx.NewSnowflakeDialerWithRendezvousMethod(rm)
+	return sfDialer, nil
+}
+
+// torsfTunnel implements Tunnel
+type torsfTunnel struct {
+	torTunnel  Tunnel
+	sfListener torsfListener
+}
+
+// torsfListener is torsfTunnel's view of a ptx listener for snowflake
+type torsfListener interface {
+	Stop()
+}
+
+var _ Tunnel = &torsfTunnel{}
+
+// BootstrapTime implements Tunnel
+func (tt *torsfTunnel) BootstrapTime() time.Duration {
+	return tt.torTunnel.BootstrapTime()
+}
+
+// SOCKS5ProxyURL implements Tunnel
+func (tt *torsfTunnel) SOCKS5ProxyURL() *url.URL {
+	return tt.torTunnel.SOCKS5ProxyURL()
+}
+
+// Stop implements Tunnel
+func (tt *torsfTunnel) Stop() {
+	tt.torTunnel.Stop()
+	tt.sfListener.Stop()
+}

internal/tunnel/torsf_test.go

@@ -0,0 +1,201 @@
+package tunnel
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"sync"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/ooni/probe-cli/v3/internal/atomicx"
+	"github.com/ooni/probe-cli/v3/internal/model"
+	"github.com/ooni/probe-cli/v3/internal/model/mocks"
+	"github.com/ooni/probe-cli/v3/internal/ptx"
+)
+
+type torsfPTXListenerWrapper struct {
+	torsfPTXListener
+	counter *atomicx.Int64
+}
+
+func (tw *torsfPTXListenerWrapper) Stop() {
+	tw.counter.Add(1)
+	tw.torsfPTXListener.Stop()
+}
+
+func Test_torsfStart(t *testing.T) {
+	t.Run("newSnowflakeDialer fails", func(t *testing.T) {
+		ctx := context.Background()
+		config := &Config{
+			Name:                 "torsf",
+			Session:              &MockableSession{},
+			SnowflakeRendezvous:  "antani", // should cause failure
+			TunnelDir:            filepath.Join(os.TempDir(), "torsf-xx"),
+			Logger:               model.DiscardLogger,
+			TorArgs:              []string{},
+			TorBinary:            "",
+			testExecabsLookPath:  nil,
+			testMkdirAll:         nil,
+			testNetListen:        nil,
+			testSocks5New:        nil,
+			testTorStart:         nil,
+			testTorProtocolInfo:  nil,
+			testTorEnableNetwork: nil,
+			testTorGetInfo:       nil,
+		}
+		expectDebugInfo := DebugInfo{}
+		tun, debugInfo, err := torsfStart(ctx, config)
+		if !errors.Is(err, ptx.ErrSnowflakeNoSuchRendezvousMethod) {
+			t.Fatal("unexpected err", err)
+		}
+		if tun != nil {
+			t.Fatal("expected nil tun")
+		}
+		if diff := cmp.Diff(expectDebugInfo, debugInfo); diff != "" {
+			t.Fatal(diff)
+		}
+	})
+
+	t.Run("ptl.Start fails", func(t *testing.T) {
+		ctx := context.Background()
+		expected := errors.New("mocked error")
+		config := &Config{
+			Name:                "torsf",
+			Session:             &MockableSession{},
+			SnowflakeRendezvous: "", // is the default
+			TunnelDir:           filepath.Join(os.TempDir(), "torsf-xx"),
+			Logger:              model.DiscardLogger,
+			TorArgs:             []string{},
+			TorBinary:           "",
+			testExecabsLookPath: nil,
+			testMkdirAll:        nil,
+			testNetListen:       nil,
+			testSfListenSocks: func(network, laddr string) (ptx.SocksListener, error) {
+				return nil, expected
+			},
+			testSocks5New:        nil,
+			testTorStart:         nil,
+			testTorProtocolInfo:  nil,
+			testTorEnableNetwork: nil,
+			testTorGetInfo:       nil,
+		}
+		expectDebugInfo := DebugInfo{}
+		tun, debugInfo, err := torsfStart(ctx, config)
+		if !errors.Is(err, expected) {
+			t.Fatal("unexpected err", err)
+		}
+		if tun != nil {
+			t.Fatal("expected nil tun")
+		}
+		if diff := cmp.Diff(expectDebugInfo, debugInfo); diff != "" {
+			t.Fatal(diff)
+		}
+	})
+
+	t.Run("torStart fails", func(t *testing.T) {
+		ctx := context.Background()
+		stopCounter := &atomicx.Int64{}
+		expected := errors.New("expected err")
+		config := &Config{
+			Name:                "torsf",
+			Session:             &MockableSession{},
+			SnowflakeRendezvous: "", // is the default
+			TunnelDir:           filepath.Join(os.TempDir(), "torsf-xx"),
+			Logger:              model.DiscardLogger,
+			TorArgs:             []string{},
+			TorBinary:           "",
+			testExecabsLookPath: nil,
+			testMkdirAll:        nil,
+			testNetListen:       nil,
+			testSfListenSocks:   nil,
+			testSfWrapPTXListener: func(tp torsfPTXListener) torsfPTXListener {
+				return &torsfPTXListenerWrapper{
+					torsfPTXListener: tp,
+					counter:          stopCounter,
+				}
+			},
+			testSfTorStart: func(ctx context.Context, config *Config) (Tunnel, DebugInfo, error) {
+				return nil, DebugInfo{}, expected
+			},
+			testSocks5New:        nil,
+			testTorStart:         nil,
+			testTorProtocolInfo:  nil,
+			testTorEnableNetwork: nil,
+			testTorGetInfo:       nil,
+		}
+		expectDebugInfo := DebugInfo{
+			Name: "torsf",
+		}
+		tun, debugInfo, err := torsfStart(ctx, config)
+		if !errors.Is(err, expected) {
+			t.Fatal("unexpected err", err)
+		}
+		if tun != nil {
+			t.Fatal("expected nil tun")
+		}
+		if diff := cmp.Diff(expectDebugInfo, debugInfo); diff != "" {
+			t.Fatal(diff)
+		}
+		if stopCounter.Load() != 1 {
+			t.Fatal("did not call stop")
+		}
+	})
+
+	t.Run("on success", func(t *testing.T) {
+		ctx := context.Background()
+		expectDebugInfo := DebugInfo{
+			Name: "torsf",
+		}
+		config := &Config{
+			Name:                "torsf",
+			Session:             &MockableSession{},
+			SnowflakeRendezvous: "", // is the default
+			TunnelDir:           filepath.Join(os.TempDir(), "torsf-xx"),
+			Logger:              model.DiscardLogger,
+			TorArgs:             []string{},
+			TorBinary:           "",
+			testExecabsLookPath: nil,
+			testMkdirAll:        nil,
+			testNetListen:       nil,
+			testSfListenSocks:   nil,
+			testSfTorStart: func(ctx context.Context, config *Config) (Tunnel, DebugInfo, error) {
+				tun := &fakeTunnel{
+					addr: &mocks.Addr{
+						MockString: func() string {
+							return "127.0.0.1:5555"
+						},
+					},
+					bootstrapTime: 123,
+					listener: &mocks.Listener{
+						MockClose: func() error {
+							return nil
+						},
+					},
+					once: sync.Once{},
+				}
+				return tun, expectDebugInfo, nil
+			},
+			testSocks5New:        nil,
+			testTorStart:         nil,
+			testTorProtocolInfo:  nil,
+			testTorEnableNetwork: nil,
+			testTorGetInfo:       nil,
+		}
+		tun, debugInfo, err := torsfStart(ctx, config)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if diff := cmp.Diff(expectDebugInfo, debugInfo); diff != "" {
+			t.Fatal(diff)
+		}
+		if tun.BootstrapTime() != 123 {
+			t.Fatal("invalid bootstrap time")
+		}
+		if tun.SOCKS5ProxyURL().String() != "socks5://127.0.0.1:5555" {
+			t.Fatal("invalid socks5 proxy URL")
+		}
+		tun.Stop()
+	})
+}

internal/tunnel/tunnel.go

@@ -124,6 +124,8 @@ func Start(ctx context.Context, config *Config) (Tunnel, DebugInfo, error) {
 		return fakeStart(ctx, config)
 	case "psiphon":
 		return psiphonStart(ctx, config)
+	case "torsf":
+		return torsfStart(ctx, config)
 	case "tor":
 		return torStart(ctx, config)
 	default:

internal/tunnel/tunnel_test.go

@@ -59,6 +59,22 @@ func TestStartTorWithCancelledContext(t *testing.T) {
 	}
 }
 
+func TestStartTorsfWithCancelledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // fail immediately
+	tun, _, err := tunnel.Start(ctx, &tunnel.Config{
+		Name:      "torsf",
+		Session:   &tunnel.MockableSession{},
+		TunnelDir: "testdata",
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatal("not the error we expected")
+	}
+	if tun != nil {
+		t.Fatal("expected nil tunnel here")
+	}
+}
+
 func TestStartInvalidTunnel(t *testing.T) {
 	ctx := context.Background()
 	tun, _, err := tunnel.Start(ctx, &tunnel.Config{