feat(snowflake): upgrade to v2 (+ small tweaks) (#667)

This diff contains the following changes and enhancements:

1. upgrade snowflake to v2

2. observe that we were not changing defaults from outside of snowflake.go, so remove code allowing to do that;

3. bump the timeout to 600 seconds (it seems 300 was not always enough based on my testing);

4. add useful knob to disable `torsf` progress (it's really annoying on console, we should do something about this);

5. ptx.go: avoid printing an error when the connection has just been closed;

6. snowflake: test AMP cache, see that it's not working currently, so leave it disabled.

Related issues: https://github.com/ooni/probe/issues/1845, https://github.com/ooni/probe/issues/1894, and https://github.com/ooni/probe/issues/1917.

go.mod

@@ -5,7 +5,7 @@ go 1.16
 require (
 	filippo.io/age v1.0.0
 	git.torproject.org/pluggable-transports/goptlib.git v1.2.0
-	git.torproject.org/pluggable-transports/snowflake.git v1.1.0
+	git.torproject.org/pluggable-transports/snowflake.git/v2 v2.0.1
 	github.com/alecthomas/kingpin v2.2.6+incompatible
 	github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 // indirect
 	github.com/apex/log v1.9.0

go.sum

@@ -17,8 +17,8 @@ git.torproject.org/pluggable-transports/goptlib.git v1.0.0/go.mod h1:YT4XMSkuEXb
 git.torproject.org/pluggable-transports/goptlib.git v1.1.0/go.mod h1:YT4XMSkuEXbtqlydr9+OxqFAyspUv0Gr9qhM3B++o/Q=
 git.torproject.org/pluggable-transports/goptlib.git v1.2.0 h1:0qRF7Dw5qXd0FtZkjWUiAh5GTutRtDGL4GXUDJ4qMHs=
 git.torproject.org/pluggable-transports/goptlib.git v1.2.0/go.mod h1:4PBMl1dg7/3vMWSoWb46eGWlrxkUyn/CAJmxhDLAlDs=
-git.torproject.org/pluggable-transports/snowflake.git v1.1.0 h1:rl/LloEeBG1sqdZdVxdW1Gmb/c3ZjdvT5o3RV8iaDg4=
+git.torproject.org/pluggable-transports/snowflake.git/v2 v2.0.1 h1:0hivepCHxfUr+GBypo9G2hJvKuYnOxNnm7/yPHoKuo4=
-git.torproject.org/pluggable-transports/snowflake.git v1.1.0/go.mod h1:+a2yI6dfEjwEnqPgXTtKobeHDEdgJa3ANZN4bjSQk+M=
+git.torproject.org/pluggable-transports/snowflake.git/v2 v2.0.1/go.mod h1:Ja4gwzbs3jyXCx6PQMM6Vigj09gOl9ox7A3lHi1s+2I=
 github.com/AlecAivazis/survey/v2 v2.0.5/go.mod h1:WYBhg6f0y/fNYUuesWQc0PKbJcEliGcYHB9sNT3Bg74=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
@@ -679,6 +679,7 @@ gitlab.com/yawning/obfs4.git v0.0.0-20220102012252-cbf3f3cfa09c/go.mod h1:M/ySGu
 gitlab.com/yawning/utls.git v0.0.11-1/go.mod h1:eYdrOOCoedNc3xw50kJ/s8JquyxeS5kr3vkFZFPTI9w=
 gitlab.com/yawning/utls.git v0.0.12-1 h1:RL6O0MP2YI0KghuEU/uGN6+8b4183eqNWoYgx7CXD0U=
 gitlab.com/yawning/utls.git v0.0.12-1/go.mod h1:3ONKiSFR9Im/c3t5RKmMJTVdmZN496FNyk3mjrY1dyo=
+gitlab.torproject.org/tpo/anti-censorship/geoip v0.0.0-20210928150955-7ce4b3d98d01/go.mod h1:K3LOI4H8fa6j+7E10ViHeGEQV10304FG4j94ypmKLjY=
 go.etcd.io/bbolt v1.3.2/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/etcd v0.0.0-20191023171146-3cf2f69b5738/go.mod h1:dnLIgRNXwCJa5e+c6mIZCrds/GIG4ncV9HhK5PX7jPg=

internal/engine/experiment/torsf/torsf.go

@@ -16,10 +16,12 @@ import (
 )
 
 // testVersion is the tor experiment version.
-const testVersion = "0.1.0"
+const testVersion = "0.1.1"
 
 // Config contains the experiment config.
-type Config struct{}
+type Config struct {
+	DisableProgress bool `ooni:"Disable printing progress messages"`
+}
 
 // TestKeys contains the experiment's result.
 type TestKeys struct {
@@ -74,7 +76,7 @@ func (m *Measurer) Run(
 	testkeys := &TestKeys{}
 	measurement.TestKeys = testkeys
 	start := time.Now()
-	const maxRuntime = 300 * time.Second
+	const maxRuntime = 600 * time.Second
 	ctx, cancel := context.WithTimeout(ctx, maxRuntime)
 	defer cancel()
 	errch := make(chan error)
@@ -87,8 +89,10 @@ func (m *Measurer) Run(
 			callbacks.OnProgress(1.0, "torsf experiment is finished")
 			return err
 		case <-ticker.C:
-			progress := time.Since(start).Seconds() / maxRuntime.Seconds()
+			if !m.config.DisableProgress {
-			callbacks.OnProgress(progress, "torsf experiment is running")
+				progress := time.Since(start).Seconds() / maxRuntime.Seconds()
+				callbacks.OnProgress(progress, "torsf experiment is running")
+			}
 		}
 	}
 }

internal/engine/experiment/torsf/torsf_test.go

@@ -18,7 +18,7 @@ func TestExperimentNameAndVersion(t *testing.T) {
 	if m.ExperimentName() != "torsf" {
 		t.Fatal("invalid experiment name")
 	}
-	if m.ExperimentVersion() != "0.1.0" {
+	if m.ExperimentVersion() != "0.1.1" {
 		t.Fatal("invalid experiment version")
 	}
 }

internal/ptx/ptx.go

@@ -39,6 +39,7 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net"
 	"strings"
@@ -184,7 +185,9 @@ func (lst *Listener) acceptLoop(ctx context.Context, ln ptxSocksListener) {
 			if err, ok := err.(net.Error); ok && err.Temporary() {
 				continue
 			}
-			lst.logger().Warnf("ptx: socks accept error: %s", err)
+			if !errors.Is(err, net.ErrClosed) {
+				lst.logger().Warnf("ptx: socks accept error: %s", err)
+			}
 			return
 		}
 		go lst.handleSocksConn(ctx, conn)

internal/ptx/snowflake.go

@@ -4,35 +4,16 @@ import (
 	"context"
 	"net"
 
-	sflib "git.torproject.org/pluggable-transports/snowflake.git/client/lib"
+	sflib "git.torproject.org/pluggable-transports/snowflake.git/v2/client/lib"
 	"github.com/ooni/probe-cli/v3/internal/stuninput"
 )
 
 // SnowflakeDialer is a dialer for snowflake. When optional fields are
 // not specified, we use defaults from the snowflake repository.
 type SnowflakeDialer struct {
-	// BrokerURL is the optional broker URL. If not specified,
-	// we will be using a sensible default value.
-	BrokerURL string
-
-	// FrontDomain is the domain to use for fronting. If not
-	// specified, we will be using a sensible default.
-	FrontDomain string
-
-	// ICEAddresses contains the addresses to use for ICE. If not
-	// specified, we will be using a sensible default.
-	ICEAddresses []string
-
-	// MaxSnowflakes is the maximum number of snowflakes we
-	// should create per dialer. If negative or zero, we will
-	// be using a sensible default.
-	MaxSnowflakes int
-
 	// newClientTransport is an optional hook for creating
 	// an alternative snowflakeTransport in testing.
-	newClientTransport func(brokerURL string, frontDomain string,
+	newClientTransport func(config sflib.ClientConfig) (snowflakeTransport, error)
-		iceAddresses []string, keepLocalAddresses bool,
-		maxSnowflakes int) (snowflakeTransport, error)
 }
 
 // snowflakeTransport is anything that allows us to dial a snowflake
@@ -50,10 +31,14 @@ func (d *SnowflakeDialer) DialContext(ctx context.Context) (net.Conn, error) {
 func (d *SnowflakeDialer) dialContext(
 	ctx context.Context) (net.Conn, chan interface{}, error) {
 	done := make(chan interface{})
-	txp, err := d.newSnowflakeClient(
+	txp, err := d.newSnowflakeClient(sflib.ClientConfig{
-		d.brokerURL(), d.frontDomain(), d.iceAddresses(),
+		BrokerURL:          d.brokerURL(),
-		false, d.maxSnowflakes(),
+		AmpCacheURL:        d.ampCacheURL(),
-	)
+		FrontDomain:        d.frontDomain(),
+		ICEAddresses:       d.iceAddresses(),
+		KeepLocalAddresses: false,
+		Max:                d.maxSnowflakes(),
+	})
 	if err != nil {
 		return nil, nil, err
 	}
@@ -83,47 +68,42 @@ func (d *SnowflakeDialer) dialContext(
 
 // newSnowflakeClient allows us to call a mock rather than
 // the real sflib.NewSnowflakeClient.
-func (d *SnowflakeDialer) newSnowflakeClient(brokerURL string, frontDomain string,
+func (d *SnowflakeDialer) newSnowflakeClient(config sflib.ClientConfig) (snowflakeTransport, error) {
-	iceAddresses []string, keepLocalAddresses bool,
-	maxSnowflakes int) (snowflakeTransport, error) {
 	if d.newClientTransport != nil {
-		return d.newClientTransport(brokerURL, frontDomain, iceAddresses,
+		return d.newClientTransport(config)
-			keepLocalAddresses, maxSnowflakes)
 	}
-	return sflib.NewSnowflakeClient(
+	return sflib.NewSnowflakeClient(config)
-		brokerURL, frontDomain, iceAddresses,
+}
-		keepLocalAddresses, maxSnowflakes)
+
+// ampCacheURL returns a suitable AMP cache URL.
+func (d *SnowflakeDialer) ampCacheURL() string {
+	// I tried using the following AMP cache and always got:
+	//
+	// 2022/01/19 16:51:28 AMP cache rendezvous response: 500 Internal Server Error
+	//
+	// So I disabled the AMP cache until we figure it out.
+	//
+	//return "https://cdn.ampproject.org/"
+	return ""
 }
 
 // brokerURL returns a suitable broker URL.
 func (d *SnowflakeDialer) brokerURL() string {
-	if d.BrokerURL != "" {
-		return d.BrokerURL
-	}
 	return "https://snowflake-broker.torproject.net.global.prod.fastly.net/"
 }
 
 // frontDomain returns a suitable front domain.
 func (d *SnowflakeDialer) frontDomain() string {
-	if d.FrontDomain != "" {
-		return d.FrontDomain
-	}
 	return "cdn.sstatic.net"
 }
 
 // iceAddresses returns suitable ICE addresses.
 func (d *SnowflakeDialer) iceAddresses() []string {
-	if len(d.ICEAddresses) > 0 {
-		return d.ICEAddresses
-	}
 	return stuninput.AsSnowflakeInput()
 }
 
 // maxSnowflakes returns the number of snowflakes to collect.
 func (d *SnowflakeDialer) maxSnowflakes() int {
-	if d.MaxSnowflakes > 0 {
-		return d.MaxSnowflakes
-	}
 	return 1
 }
 

internal/ptx/snowflake_test.go

@@ -6,6 +6,7 @@ import (
 	"net"
 	"testing"
 
+	sflib "git.torproject.org/pluggable-transports/snowflake.git/v2/client/lib"
 	"github.com/ooni/probe-cli/v3/internal/atomicx"
 	"github.com/ooni/probe-cli/v3/internal/model/mocks"
 )
@@ -47,7 +48,7 @@ var _ snowflakeTransport = &mockableSnowflakeTransport{}
 
 func TestSnowflakeDialerWorksWithMocks(t *testing.T) {
 	sfd := &SnowflakeDialer{
-		newClientTransport: func(brokerURL, frontDomain string, iceAddresses []string, keepLocalAddresses bool, maxSnowflakes int) (snowflakeTransport, error) {
+		newClientTransport: func(config sflib.ClientConfig) (snowflakeTransport, error) {
 			return &mockableSnowflakeTransport{
 				MockDial: func() (net.Conn, error) {
 					return &mocks.Conn{
@@ -79,7 +80,7 @@ func TestSnowflakeDialerWorksWithMocks(t *testing.T) {
 func TestSnowflakeDialerCannotCreateTransport(t *testing.T) {
 	expected := errors.New("mocked error")
 	sfd := &SnowflakeDialer{
-		newClientTransport: func(brokerURL, frontDomain string, iceAddresses []string, keepLocalAddresses bool, maxSnowflakes int) (snowflakeTransport, error) {
+		newClientTransport: func(config sflib.ClientConfig) (snowflakeTransport, error) {
 			return nil, expected
 		},
 	}
@@ -95,7 +96,7 @@ func TestSnowflakeDialerCannotCreateTransport(t *testing.T) {
 func TestSnowflakeDialerCannotCreateConnWithNoContextExpiration(t *testing.T) {
 	expected := errors.New("mocked error")
 	sfd := &SnowflakeDialer{
-		newClientTransport: func(brokerURL, frontDomain string, iceAddresses []string, keepLocalAddresses bool, maxSnowflakes int) (snowflakeTransport, error) {
+		newClientTransport: func(config sflib.ClientConfig) (snowflakeTransport, error) {
 			return &mockableSnowflakeTransport{
 				MockDial: func() (net.Conn, error) {
 					return nil, expected
@@ -117,7 +118,7 @@ func TestSnowflakeDialerCannotCreateConnWithContextExpiration(t *testing.T) {
 	defer cancel()
 	expected := errors.New("mocked error")
 	sfd := &SnowflakeDialer{
-		newClientTransport: func(brokerURL, frontDomain string, iceAddresses []string, keepLocalAddresses bool, maxSnowflakes int) (snowflakeTransport, error) {
+		newClientTransport: func(config sflib.ClientConfig) (snowflakeTransport, error) {
 			return &mockableSnowflakeTransport{
 				MockDial: func() (net.Conn, error) {
 					cancel() // before returning to the caller
@@ -140,7 +141,7 @@ func TestSnowflakeDialerWorksWithWithCancelledContext(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	sfd := &SnowflakeDialer{
-		newClientTransport: func(brokerURL, frontDomain string, iceAddresses []string, keepLocalAddresses bool, maxSnowflakes int) (snowflakeTransport, error) {
+		newClientTransport: func(config sflib.ClientConfig) (snowflakeTransport, error) {
 			return &mockableSnowflakeTransport{
 				MockDial: func() (net.Conn, error) {
 					cancel() // cause a cancel before we can really have a conn
@@ -167,24 +168,3 @@ func TestSnowflakeDialerWorksWithWithCancelledContext(t *testing.T) {
 		t.Fatal("the goroutine did not call close")
 	}
 }
-
-func TestSnowflakeWeCanSetCustomValues(t *testing.T) {
-	sfd := &SnowflakeDialer{
-		BrokerURL:     "antani",
-		FrontDomain:   "mascetti",
-		ICEAddresses:  []string{"melandri"},
-		MaxSnowflakes: 11,
-	}
-	if sfd.brokerURL() != "antani" {
-		t.Fatal("invalid broker URL")
-	}
-	if sfd.frontDomain() != "mascetti" {
-		t.Fatal("invalid front domain")
-	}
-	if v := sfd.iceAddresses(); len(v) != 1 || v[0] != "melandri" {
-		t.Fatal("invalid ICE addresses")
-	}
-	if sfd.maxSnowflakes() != 11 {
-		t.Fatal("invalid max number of snowflakes")
-	}
-}