.gitlab-ci.yml

@@ -54,7 +54,6 @@ variables:
         golang-golang-x-sys-dev
         golang-golang-x-text-dev
         golang-golang-x-xerrors-dev
-        lbzip2
 
 # use Go installed as part of the official, Debian-based Docker images
 .golang-docker-debian-template: &golang-docker-debian-template
@@ -63,7 +62,6 @@ variables:
     - apt-get -qy install --no-install-recommends
         ca-certificates
         git
-        lbzip2
 
 .go-test: &go-test
   - gofmt -d .
@@ -93,7 +91,7 @@ variables:
 # -- jobs ------------------------------------------------------------
 
 android:
-  image: golang:1.21-$DEBIAN_STABLE
+  image: golang:1.23-$DEBIAN_STABLE
   variables:
     ANDROID_HOME: /usr/lib/android-sdk
     LANG: C.UTF-8
@@ -115,7 +113,6 @@ android:
         unzip
         wget
         ca-certificates
-        lbzip2
 
     - ndk=android-ndk-r21e-linux-x86_64.zip
     - wget --continue --no-verbose https://dl.google.com/android/repository/$ndk
@@ -299,9 +296,8 @@ build-container:
     name: gcr.io/kaniko-project/executor:debug
     entrypoint: [""]
   script:
-    - if [ $CI_COMMIT_REF_NAME == "main" ]; then export TAG='latest'; fi
+    - if [ $CI_COMMIT_REF_NAME == "main" ]; then export TAG='nightly'; fi
     - >-
-      echo "Building Docker image with tag: $TAG"
       /kaniko/executor
       --context "${CI_PROJECT_DIR}"
       --dockerfile "${CI_PROJECT_DIR}/Dockerfile"
@@ -321,7 +317,7 @@ merge-manifests:
       name: mplatform/manifest-tool:alpine
       entrypoint: [""]
   script:
-    - if [ $CI_COMMIT_REF_NAME == "main" ]; then export TAG='latest'; fi
+    - if [ $CI_COMMIT_REF_NAME == "main" ]; then export TAG='nightly'; fi
     - >-
       manifest-tool
       --username="${CI_REGISTRY_USER}"
@@ -339,11 +335,14 @@ merge-manifests:
 # If this is a tag, then we want to additionally tag the image as `stable`
 tag-container-release:
   stage: container-build
+  needs:
+    - job: merge-manifests
+      artifacts: false
   image: quay.io/podman/stable
   allow_failure: false
   variables:
     IMAGE_TAG: $CI_REGISTRY_IMAGE:$CI_COMMIT_TAG
-    RELEASE_TAG: $CI_REGISTRY_IMAGE:stable
+    RELEASE_TAG: $CI_REGISTRY_IMAGE:latest
   script:
     - echo "Tagging docker image with stable tag"
     - echo -n "$CI_JOB_TOKEN" | podman login -u gitlab-ci-token --password-stdin $CI_REGISTRY
@@ -359,7 +358,7 @@ clean-image-tags:
   needs:
     - job: merge-manifests
       artifacts: false
-  image: containers.torproject.org/tpo/tpa/base-images:bookworm
+  image: containers.torproject.org/tpo/tpa/base-images/debian:bookworm
   before_script:
     - *apt-template
     - apt-get install -y jq curl

ChangeLog

+Changes in version v2.10.0 - 2024-11-07
+- Issue 40402: Add proxy event for when client has connected
+- Issue 40405: Prevent panic for duplicate SnowflakeConn.Close() calls
+- Enable local time for proxy logging
+- Have proxy summary statistics log average transfer rate
+- Issue 40210: Remove duplicate poll interval loop in proxy
+- Issue 40371: Prevent broker and proxy from rejecting clients without ICE candidates
+- Issue 40392: Allow the proxy and probetest to set multiple STUN URLs
+- Issue 40387: Fix error in probetest NAT check
+- Fix proxy panic on invalid relayURL
+- Set empty pattern if broker bridge-list is empty
+- Improve documentation of Ephemeral[Min,Max]Port
+- Fix resource leak and NAT check in probetest
+- Fix memory leak from failed NAT check
+- Improve NAT check logging
+- Issue 40230: Send answer even if ICE gathering is not complete
+- Improve broker error message on unknown bridge fingerprint
+- Don't proxy private IP addresses
+- Only accept ws:// and wss:// relay addresses
+- Issue 40373: Add cli flag and SnowflakeProxy field to modify proxy poll interval
+- Use %w not $v in fmt.Errorf
+- Updates to documentation
+- Adjust copy buffer size to improve proxy performance
+- Improve descriptions of cli flags
+- Cosmetic changes for code readability
+- Issue 40367: Deduplicate prometheus metrics names
+- Report the version of snowflake to the tor process
+- Issue 40365: Indicate whether the repo was modified in the version string
+- Simplify NAT checking logic
+- Issue 40354: Use ptutil library for safelog and prometheus metrics
+- Add cli flag to set a listen address for proxy prometheus metrics
+- Issue 40345: Integrate docker image with release process
+- Bump versions of dependencies
+
 Changes in version v2.9.2 - 2024-03-18
 - Issue 40288: Add integration testing with Shadow
 - Issue 40345: Automatically build and push containers to our registry

Dockerfile

-FROM docker.io/library/golang:1.22 AS build
+FROM docker.io/library/golang:1.23 AS build
 
 # Set some labels
 # io.containers.autoupdate label will instruct podman to reach out to the corres

broker/bridge-list.go

@@ -32,7 +32,7 @@ import (
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/bridgefingerprint"
 )
 
-var ErrBridgeNotFound = errors.New("bridge not found")
+var ErrBridgeNotFound = errors.New("bridge with requested fingerprint is unknown to the broker")
 
 func NewBridgeListHolder() BridgeListHolderFileBased {
 	return &bridgeListHolder{}

broker/broker.go

@@ -27,8 +27,8 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
+	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/ptutil/safelog"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/namematcher"
-	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/safelog"
 	"golang.org/x/crypto/acme/autocert"
 )
 
@@ -53,7 +53,11 @@ func (ctx *BrokerContext) GetBridgeInfo(fingerprint bridgefingerprint.Fingerprin
 	return ctx.bridgeList.GetBridgeInfo(fingerprint)
 }
 
-func NewBrokerContext(metricsLogger *log.Logger) *BrokerContext {
+func NewBrokerContext(
+	metricsLogger *log.Logger,
+	allowedRelayPattern,
+	presumedPatternForLegacyClient string,
+) *BrokerContext {
 	snowflakes := new(SnowflakeHeap)
 	heap.Init(snowflakes)
 	rSnowflakes := new(SnowflakeHeap)
@@ -75,12 +79,14 @@ func NewBrokerContext(metricsLogger *log.Logger) *BrokerContext {
 	bridgeListHolder.LoadBridgeInfo(bytes.NewReader([]byte(DefaultBridges)))
 
 	return &BrokerContext{
-		snowflakes:           snowflakes,
-		restrictedSnowflakes: rSnowflakes,
-		idToSnowflake:        make(map[string]*Snowflake),
-		proxyPolls:           make(chan *ProxyPoll),
-		metrics:              metrics,
-		bridgeList:           bridgeListHolder,
+		snowflakes:                     snowflakes,
+		restrictedSnowflakes:           rSnowflakes,
+		idToSnowflake:                  make(map[string]*Snowflake),
+		proxyPolls:                     make(chan *ProxyPoll),
+		metrics:                        metrics,
+		bridgeList:                     bridgeListHolder,
+		allowedRelayPattern:            allowedRelayPattern,
+		presumedPatternForLegacyClient: presumedPatternForLegacyClient,
 	}
 }
 
@@ -161,12 +167,10 @@ func (ctx *BrokerContext) AddSnowflake(id string, proxyType string, natType stri
 	return snowflake
 }
 
-func (ctx *BrokerContext) InstallBridgeListProfile(reader io.Reader, relayPattern, presumedPatternForLegacyClient string) error {
+func (ctx *BrokerContext) InstallBridgeListProfile(reader io.Reader) error {
 	if err := ctx.bridgeList.LoadBridgeInfo(reader); err != nil {
 		return err
 	}
-	ctx.allowedRelayPattern = relayPattern
-	ctx.presumedPatternForLegacyClient = presumedPatternForLegacyClient
 	return nil
 }
 
@@ -210,7 +214,7 @@ func main() {
 	flag.StringVar(&geoipDatabase, "geoipdb", "/usr/share/tor/geoip", "path to correctly formatted geoip database mapping IPv4 address ranges to country codes")
 	flag.StringVar(&geoip6Database, "geoip6db", "/usr/share/tor/geoip6", "path to correctly formatted geoip database mapping IPv6 address ranges to country codes")
 	flag.StringVar(&bridgeListFilePath, "bridge-list-path", "", "file path for bridgeListFile")
-	flag.StringVar(&allowedRelayPattern, "allowed-relay-pattern", "", "allowed pattern for relay host name")
+	flag.StringVar(&allowedRelayPattern, "allowed-relay-pattern", "", "allowed pattern for relay host name. The broker will reject proxies whose AcceptedRelayPattern is more restrictive than this")
 	flag.StringVar(&presumedPatternForLegacyClient, "default-relay-pattern", "", "presumed pattern for legacy client")
 	flag.StringVar(&brokerSQSQueueName, "broker-sqs-name", "", "name of broker SQS queue to listen for incoming messages on")
 	flag.StringVar(&brokerSQSQueueRegion, "broker-sqs-region", "", "name of AWS region of broker SQS queue")
@@ -244,14 +248,14 @@ func main() {
 
 	metricsLogger := log.New(metricsFile, "", 0)
 
-	ctx := NewBrokerContext(metricsLogger)
+	ctx := NewBrokerContext(metricsLogger, allowedRelayPattern, presumedPatternForLegacyClient)
 
 	if bridgeListFilePath != "" {
 		bridgeListFile, err := os.Open(bridgeListFilePath)
 		if err != nil {
 			log.Fatal(err.Error())
 		}
-		err = ctx.InstallBridgeListProfile(bridgeListFile, allowedRelayPattern, presumedPatternForLegacyClient)
+		err = ctx.InstallBridgeListProfile(bridgeListFile)
 		if err != nil {
 			log.Fatal(err.Error())
 		}

broker/http.go

@@ -139,13 +139,6 @@ func clientOffers(i *IPC, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	err = validateSDP(body)
-	if err != nil {
-		log.Println("Error client SDP: ", err.Error())
-		w.WriteHeader(http.StatusBadRequest)
-		return
-	}
-
 	// Handle the legacy version
 	//
 	// We support two client message formats. The legacy format is for backwards

broker/ipc.go

@@ -185,7 +185,10 @@ func (i *IPC) ClientOffers(arg messages.Arg, response *[]byte) error {
 	}
 
 	if _, err := i.ctx.GetBridgeInfo(BridgeFingerprint); err != nil {
-		return err
+		return sendClientResponse(
+			&messages.ClientPollResponse{Error: err.Error()},
+			response,
+		)
 	}
 
 	offer.fingerprint = BridgeFingerprint.ToBytes()

broker/metrics.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 	"gitlab.torproject.org/tpo/anti-censorship/geoip"
+	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/ptutil/safeprom"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/messages"
 )
 
@@ -330,14 +331,14 @@ func sumMapValues(m *map[messages.RendezvousMethod]uint) uint {
 type PromMetrics struct {
 	registry         *prometheus.Registry
 	ProxyTotal       *prometheus.CounterVec
-	ProxyPollTotal   *RoundedCounterVec
-	ClientPollTotal  *RoundedCounterVec
+	ProxyPollTotal   *safeprom.CounterVec
+	ClientPollTotal  *safeprom.CounterVec
 	AvailableProxies *prometheus.GaugeVec
 
-	ProxyPollWithRelayURLExtensionTotal    *RoundedCounterVec
-	ProxyPollWithoutRelayURLExtensionTotal *RoundedCounterVec
+	ProxyPollWithRelayURLExtensionTotal    *safeprom.CounterVec
+	ProxyPollWithoutRelayURLExtensionTotal *safeprom.CounterVec
 
-	ProxyPollRejectedForRelayURLExtensionTotal *RoundedCounterVec
+	ProxyPollRejectedForRelayURLExtensionTotal *safeprom.CounterVec
 }
 
 // Initialize metrics for prometheus exporter
@@ -364,7 +365,7 @@ func initPrometheus() *PromMetrics {
 		[]string{"type", "nat"},
 	)
 
-	promMetrics.ProxyPollTotal = NewRoundedCounterVec(
+	promMetrics.ProxyPollTotal = safeprom.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: prometheusNamespace,
 			Name:      "rounded_proxy_poll_total",
@@ -373,7 +374,7 @@ func initPrometheus() *PromMetrics {
 		[]string{"nat", "status"},
 	)
 
-	promMetrics.ProxyPollWithRelayURLExtensionTotal = NewRoundedCounterVec(
+	promMetrics.ProxyPollWithRelayURLExtensionTotal = safeprom.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: prometheusNamespace,
 			Name:      "rounded_proxy_poll_with_relay_url_extension_total",
@@ -382,7 +383,7 @@ func initPrometheus() *PromMetrics {
 		[]string{"nat", "type"},
 	)
 
-	promMetrics.ProxyPollWithoutRelayURLExtensionTotal = NewRoundedCounterVec(
+	promMetrics.ProxyPollWithoutRelayURLExtensionTotal = safeprom.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: prometheusNamespace,
 			Name:      "rounded_proxy_poll_without_relay_url_extension_total",
@@ -391,7 +392,7 @@ func initPrometheus() *PromMetrics {
 		[]string{"nat", "type"},
 	)
 
-	promMetrics.ProxyPollRejectedForRelayURLExtensionTotal = NewRoundedCounterVec(
+	promMetrics.ProxyPollRejectedForRelayURLExtensionTotal = safeprom.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: prometheusNamespace,
 			Name:      "rounded_proxy_poll_rejected_relay_url_extension_total",
@@ -400,7 +401,7 @@ func initPrometheus() *PromMetrics {
 		[]string{"nat", "type"},
 	)
 
-	promMetrics.ClientPollTotal = NewRoundedCounterVec(
+	promMetrics.ClientPollTotal = safeprom.NewCounterVec(
 		prometheus.CounterOpts{
 			Namespace: prometheusNamespace,
 			Name:      "rounded_client_poll_total",

broker/prometheus.go

-/*
-Implements some additional prometheus metrics that we need for privacy preserving
-counts of users and proxies
-*/
-
-package main
-
-import (
-	"sync/atomic"
-
-	"github.com/prometheus/client_golang/prometheus"
-	dto "github.com/prometheus/client_model/go"
-	"google.golang.org/protobuf/proto"
-)
-
-// New Prometheus counter type that produces rounded counts of metrics
-// for privacy preserving reasons
-type RoundedCounter interface {
-	prometheus.Metric
-
-	Inc()
-}
-
-type roundedCounter struct {
-	total uint64 //reflects the true count
-	value uint64 //reflects the rounded count
-
-	desc       *prometheus.Desc
-	labelPairs []*dto.LabelPair
-}
-
-// Implements the RoundedCounter interface
-func (c *roundedCounter) Inc() {
-	atomic.AddUint64(&c.total, 1)
-	if c.total > c.value {
-		atomic.AddUint64(&c.value, 8)
-	}
-}
-
-// Implements the prometheus.Metric interface
-func (c *roundedCounter) Desc() *prometheus.Desc {
-	return c.desc
-}
-
-// Implements the prometheus.Metric interface
-func (c *roundedCounter) Write(m *dto.Metric) error {
-	m.Label = c.labelPairs
-
-	m.Counter = &dto.Counter{Value: proto.Float64(float64(c.value))}
-	return nil
-}
-
-// New prometheus vector type that will track RoundedCounter metrics
-// accross multiple labels
-type RoundedCounterVec struct {
-	*prometheus.MetricVec
-}
-
-func NewRoundedCounterVec(opts prometheus.CounterOpts, labelNames []string) *RoundedCounterVec {
-	desc := prometheus.NewDesc(
-		prometheus.BuildFQName(opts.Namespace, opts.Subsystem, opts.Name),
-		opts.Help,
-		labelNames,
-		opts.ConstLabels,
-	)
-	return &RoundedCounterVec{
-		MetricVec: prometheus.NewMetricVec(desc, func(lvs ...string) prometheus.Metric {
-			if len(lvs) != len(labelNames) {
-				panic("inconsistent cardinality")
-			}
-			return &roundedCounter{desc: desc, labelPairs: prometheus.MakeLabelPairs(desc, lvs)}
-		}),
-	}
-}
-
-// Helper function to return the underlying RoundedCounter metric from MetricVec
-func (v *RoundedCounterVec) With(labels prometheus.Labels) RoundedCounter {
-	metric, err := v.GetMetricWith(labels)
-	if err != nil {
-		panic(err)
-	}
-	return metric.(RoundedCounter)
-}

broker/snowflake-broker_test.go

@@ -89,7 +89,7 @@ func TestBroker(t *testing.T) {
 
 	Convey("Context", t, func() {
 		buf := new(bytes.Buffer)
-		ctx := NewBrokerContext(log.New(buf, "", 0))
+		ctx := NewBrokerContext(log.New(buf, "", 0), "", "")
 		i := &IPC{ctx}
 
 		Convey("Adds Snowflake", func() {
@@ -137,14 +137,6 @@ func TestBroker(t *testing.T) {
 			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
 			So(err, ShouldBeNil)
 
-			Convey("with HTTP Bad Request when client offer contains invalid SDP", func() {
-				data, err = createClientOffer("fake", NATUnknown, "")
-				invalidRequest, err := http.NewRequest("POST", "snowflake.broker/client", data)
-				So(err, ShouldBeNil)
-				clientOffers(i, w, invalidRequest)
-				So(w.Code, ShouldEqual, http.StatusBadRequest)
-			})
-
 			Convey("with error when no snowflakes are available.", func() {
 				clientOffers(i, w, r)
 				So(w.Code, ShouldEqual, http.StatusOK)
@@ -246,14 +238,6 @@ client-sqs-ips
 			So(err, ShouldBeNil)
 			r.Header.Set("Snowflake-NAT-TYPE", "restricted")
 
-			Convey("with HTTP Bad Request when client offer contains invalid SDP", func() {
-				invalidOffer := bytes.NewReader([]byte("{test}"))
-				invalidRequest, err := http.NewRequest("POST", "snowflake.broker/client", invalidOffer)
-				So(err, ShouldBeNil)
-				clientOffers(i, w, invalidRequest)
-				So(w.Code, ShouldEqual, http.StatusBadRequest)
-			})
-
 			Convey("with 503 when no snowflakes are available.", func() {
 				clientOffers(i, w, r)
 				So(w.Code, ShouldEqual, http.StatusServiceUnavailable)
@@ -503,7 +487,7 @@ client-sqs-ips
 	})
 
 	Convey("End-To-End", t, func() {
-		ctx := NewBrokerContext(NullLogger())
+		ctx := NewBrokerContext(NullLogger(), "", "")
 		i := &IPC{ctx}
 
 		Convey("Check for client/proxy data race", func() {
@@ -656,7 +640,7 @@ func TestSnowflakeHeap(t *testing.T) {
 func TestInvalidGeoipFile(t *testing.T) {
 	Convey("Geoip", t, func() {
 		// Make sure things behave properly if geoip file fails to load
-		ctx := NewBrokerContext(NullLogger())
+		ctx := NewBrokerContext(NullLogger(), "", "")
 		if err := ctx.metrics.LoadGeoipDatabases("invalid_filename", "invalid_filename6"); err != nil {
 			log.Printf("loading geo ip databases returned error: %v", err)
 		}
@@ -670,7 +654,7 @@ func TestMetrics(t *testing.T) {
 	Convey("Test metrics...", t, func() {
 		done := make(chan bool)
 		buf := new(bytes.Buffer)
-		ctx := NewBrokerContext(log.New(buf, "", 0))
+		ctx := NewBrokerContext(log.New(buf, "", 0), "", "")
 		i := &IPC{ctx}
 
 		err := ctx.metrics.LoadGeoipDatabases("test_geoip", "test_geoip6")

broker/sqs_test.go

@@ -22,7 +22,7 @@ func TestSQS(t *testing.T) {
 
 	Convey("Context", t, func() {
 		buf := new(bytes.Buffer)
-		ipcCtx := NewBrokerContext(log.New(buf, "", 0))
+		ipcCtx := NewBrokerContext(log.New(buf, "", 0), "", "")
 		i := &IPC{ipcCtx}
 
 		var logBuffer bytes.Buffer

broker/test_bridgeList.txt

+{"displayName":"flakey", "webSocketAddress":"wss://snowflake.torproject.net", "fingerprint":"2B280B23E1107BB62ABFC40DDCC8824814F80A72"}
+{"displayName":"second", "webSocketAddress":"wss://02.snowflake.torproject.net", "fingerprint":"8838024498816A039FCBBAB14E6F40A0843051FA"}

client/lib/peers.go

@@ -29,6 +29,7 @@ type Peers struct {
 	melt chan struct{}
 
 	collectLock sync.Mutex
+	closeOnce   sync.Once
 }
 
 // NewPeers constructs a fresh container of remote peers.
@@ -122,17 +123,19 @@ func (p *Peers) purgeClosedPeers() {
 // End closes all active connections to Peers contained here, and stops the
 // collection of future Peers.
 func (p *Peers) End() {
-	close(p.melt)
-	p.collectLock.Lock()
-	defer p.collectLock.Unlock()
-	close(p.snowflakeChan)
-	cnt := p.Count()
-	for e := p.activePeers.Front(); e != nil; {
-		next := e.Next()
-		conn := e.Value.(*WebRTCPeer)
-		conn.Close()
-		p.activePeers.Remove(e)
-		e = next
-	}
-	log.Printf("WebRTC: melted all %d snowflakes.", cnt)
+	p.closeOnce.Do(func() {
+		close(p.melt)
+		p.collectLock.Lock()
+		defer p.collectLock.Unlock()
+		close(p.snowflakeChan)
+		cnt := p.Count()
+		for e := p.activePeers.Front(); e != nil; {
+			next := e.Next()
+			conn := e.Value.(*WebRTCPeer)
+			conn.Close()
+			p.activePeers.Remove(e)
+			e = next
+		}
+		log.Printf("WebRTC: melted all %d snowflakes.", cnt)
+	})
 }

client/lib/rendezvous.go

@@ -7,15 +7,15 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
-	"net/url"
-
 	"log"
 	"net/http"
+	"net/url"
 	"sync"
 	"time"
 
 	"github.com/pion/webrtc/v3"
 	utls "github.com/refraction-networking/utls"
+
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/certs"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/event"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/messages"
@@ -79,7 +79,7 @@ func newBrokerChannelFromConfig(config ClientConfig) (*BrokerChannel, error) {
 	if config.UTLSClientID != "" {
 		utlsClientHelloID, err := utlsutil.NameToUTLSID(config.UTLSClientID)
 		if err != nil {
-			return nil, fmt.Errorf("unable to create broker channel: %v", err)
+			return nil, fmt.Errorf("unable to create broker channel: %w", err)
 		}
 		utlsConfig := &utls.Config{
 			RootCAs: certs.GetRootCAs(),
@@ -125,7 +125,8 @@ func newBrokerChannelFromConfig(config ClientConfig) (*BrokerChannel, error) {
 // Negotiate uses a RendezvousMethod to send the client's WebRTC SDP offer
 // and receive a snowflake proxy WebRTC SDP answer in return.
 func (bc *BrokerChannel) Negotiate(offer *webrtc.SessionDescription) (
-	*webrtc.SessionDescription, error) {
+	*webrtc.SessionDescription, error,
+) {
 	// Ideally, we could specify an `RTCIceTransportPolicy` that would handle
 	// this for us.  However, "public" was removed from the draft spec.
 	// See https://developer.mozilla.org/en-US/docs/Web/API/RTCConfiguration#RTCIceTransportPolicy_enum
@@ -201,7 +202,8 @@ func NewWebRTCDialerWithEvents(broker *BrokerChannel, iceServers []webrtc.ICESer
 
 // NewWebRTCDialerWithEventsAndProxy constructs a new WebRTCDialer.
 func NewWebRTCDialerWithEventsAndProxy(broker *BrokerChannel, iceServers []webrtc.ICEServer, max int,
-	eventLogger event.SnowflakeEventReceiver, proxy *url.URL) *WebRTCDialer {
+	eventLogger event.SnowflakeEventReceiver, proxy *url.URL,
+) *WebRTCDialer {
 	config := webrtc.Configuration{
 		ICEServers: iceServers,
 	}

client/lib/rendezvous_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"testing"
 
@@ -13,11 +14,13 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"github.com/golang/mock/gomock"
+	"github.com/pion/webrtc/v3"
 	. "github.com/smartystreets/goconvey/convey"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/amp"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/messages"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/nat"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/sqsclient"
+	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/util"
 )
 
 // mockTransport's RoundTrip method returns a response with a fake status and
@@ -386,3 +389,51 @@ func TestSQSRendezvous(t *testing.T) {
 		})
 	})
 }
+
+func TestBrokerChannel(t *testing.T) {
+	Convey("Requests a proxy and handles response", t, func() {
+		answerSdp := &webrtc.SessionDescription{
+			Type: webrtc.SDPTypeAnswer,
+			SDP:  "test",
+		}
+		answerSdpStr, _ := util.SerializeSessionDescription(answerSdp)
+		serverResponse, _ := (&messages.ClientPollResponse{
+			Answer: answerSdpStr,
+		}).EncodePollResponse()
+
+		offerSdp := &webrtc.SessionDescription{
+			Type: webrtc.SDPTypeOffer,
+			SDP:  "test",
+		}
+
+		requestBodyChan := make(chan []byte)
+		mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			body, _ := io.ReadAll(r.Body)
+			go func() {
+				requestBodyChan <- body
+			}()
+			w.Write(serverResponse)
+		}))
+		defer mockServer.Close()
+
+		brokerChannel, err := newBrokerChannelFromConfig(ClientConfig{
+			BrokerURL:         mockServer.URL,
+			BridgeFingerprint: "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA",
+		})
+		So(err, ShouldBeNil)
+		brokerChannel.SetNATType(nat.NATRestricted)
+
+		answerSdpReturned, err := brokerChannel.Negotiate(offerSdp)
+		So(err, ShouldBeNil)
+		So(answerSdpReturned, ShouldEqual, answerSdp)
+
+		body := <-requestBodyChan
+		pollReq, err := messages.DecodeClientPollRequest(body)
+		So(err, ShouldBeNil)
+		So(pollReq.Fingerprint, ShouldEqual, "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
+		So(pollReq.NAT, ShouldEqual, nat.NATRestricted)
+		requestSdp, err := util.DeserializeSessionDescription(pollReq.Offer)
+		So(err, ShouldBeNil)
+		So(requestSdp, ShouldEqual, offerSdp)
+	})
+}

client/lib/snowflake.go

@@ -243,14 +243,19 @@ type SnowflakeConn struct {
 //
 // The collection of snowflake proxies for this connection is stopped.
 func (conn *SnowflakeConn) Close() error {
+	var err error
 	log.Printf("---- SnowflakeConn: closed stream %v ---", conn.ID())
-	conn.Stream.Close()
+	err = conn.Stream.Close()
 	log.Printf("---- SnowflakeConn: end collecting snowflakes ---")
 	conn.snowflakes.End()
-	conn.pconn.Close()
+	if inerr := conn.pconn.Close(); err == nil {
+		err = inerr
+	}
 	log.Printf("---- SnowflakeConn: discarding finished session ---")
-	conn.sess.Close()
-	return nil // TODO: return errors if any of the above do
+	if inerr := conn.sess.Close(); err == nil {
+		err = inerr
+	}
+	return err
 }
 
 // loop through all provided STUN servers until we exhaust the list or find

client/lib/webrtc.go

@@ -4,11 +4,9 @@ import (
 	"crypto/rand"
 	"encoding/hex"
 	"errors"
-	"fmt"
 	"io"
 	"log"
 	"net/url"
-	"strings"
 	"sync"
 	"time"
 
@@ -301,9 +299,6 @@ func (c *WebRTCPeer) preparePeerConnection(config *webrtc.Configuration) error {
 
 	<-done // Wait for ICE candidate gathering to complete.
 
-	if !strings.Contains(c.pc.LocalDescription().SDP, "\na=candidate:") {
-		return fmt.Errorf("SDP offer contains no candidate")
-	}
 	return nil
 }
 

client/snowflake.go

@@ -16,11 +16,11 @@ import (
 	"syscall"
 
 	pt "gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/goptlib"
+	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/ptutil/safelog"
 
 	sf "gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/client/lib"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/event"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/proxy"
-	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/safelog"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/version"
 )
 
@@ -169,8 +169,8 @@ func main() {
 	sqsCredsStr := flag.String("sqscreds", "", "credentials to access SQS Queue")
 	logFilename := flag.String("log", "", "name of log file")
 	logToStateDir := flag.Bool("log-to-state-dir", false, "resolve the log file relative to tor's pt state dir")
-	keepLocalAddresses := flag.Bool("keep-local-addresses", false, "keep local LAN address ICE candidates")
-	unsafeLogging := flag.Bool("unsafe-logging", false, "prevent logs from being scrubbed")
+	keepLocalAddresses := flag.Bool("keep-local-addresses", false, "keep local LAN address ICE candidates.\nThis is usually pointless because Snowflake proxies don't usually reside on the same local network as the client.")
+	unsafeLogging := flag.Bool("unsafe-logging", false, "keep IP addresses and other sensitive info in the logs")
 	max := flag.Int("max", DefaultSnowflakeCapacity,
 		"capacity for number of multiplexed WebRTC peers")
 	versionFlag := flag.Bool("version", false, "display version info to stderr and quit")
@@ -263,6 +263,7 @@ func main() {
 			pt.ProxyDone()
 		}
 	}
+	pt.ReportVersion("snowflake-client", version.GetVersion())
 	listeners := make([]net.Listener, 0)
 	shutdown := make(chan struct{})
 	var wg sync.WaitGroup

client/torrc

@@ -3,7 +3,7 @@ DataDirectory datadir
 
 ClientTransportPlugin snowflake exec ./client -log snowflake.log
 
-Bridge snowflake 192.0.2.3:80 2B280B23E1107BB62ABFC40DDCC8824814F80A72 fingerprint=2B280B23E1107BB62ABFC40DDCC8824814F80A72 url=https://snowflake-broker.torproject.net.global.prod.fastly.net/ fronts=foursquare.com,github.githubassets.com ice=stun:stun.l.google.com:19302,stun:stun.antisip.com:3478,stun:stun.bluesip.net:3478,stun:stun.dus.net:3478,stun:stun.epygi.com:3478,stun:stun.sonetel.com:3478,stun:stun.uls.co.za:3478,stun:stun.voipgate.com:3478,stun:stun.voys.nl:3478 utls-imitate=hellorandomizedalpn
-Bridge snowflake 192.0.2.4:80 8838024498816A039FCBBAB14E6F40A0843051FA fingerprint=8838024498816A039FCBBAB14E6F40A0843051FA url=https://snowflake-broker.torproject.net.global.prod.fastly.net/ fronts=foursquare.com,github.githubassets.com ice=stun:stun.l.google.com:19302,stun:stun.antisip.com:3478,stun:stun.bluesip.net:3478,stun:stun.dus.net:3478,stun:stun.epygi.com:3478,stun:stun.sonetel.net:3478,stun:stun.uls.co.za:3478,stun:stun.voipgate.com:3478,stun:stun.voys.nl:3478 utls-imitate=hellorandomizedalpn
+Bridge snowflake 192.0.2.3:80 2B280B23E1107BB62ABFC40DDCC8824814F80A72 fingerprint=2B280B23E1107BB62ABFC40DDCC8824814F80A72 url=https://1098762253.rsc.cdn77.org/ fronts=www.cdn77.com,www.phpmyadmin.net ice=stun:stun.l.google.com:19302,stun:stun.antisip.com:3478,stun:stun.bluesip.net:3478,stun:stun.dus.net:3478,stun:stun.epygi.com:3478,stun:stun.sonetel.com:3478,stun:stun.uls.co.za:3478,stun:stun.voipgate.com:3478,stun:stun.voys.nl:3478 utls-imitate=hellorandomizedalpn
+Bridge snowflake 192.0.2.4:80 8838024498816A039FCBBAB14E6F40A0843051FA fingerprint=8838024498816A039FCBBAB14E6F40A0843051FA url=https://1098762253.rsc.cdn77.org/ fronts=www.cdn77.com,www.phpmyadmin.net ice=stun:stun.l.google.com:19302,stun:stun.antisip.com:3478,stun:stun.bluesip.net:3478,stun:stun.dus.net:3478,stun:stun.epygi.com:3478,stun:stun.sonetel.net:3478,stun:stun.uls.co.za:3478,stun:stun.voipgate.com:3478,stun:stun.voys.nl:3478 utls-imitate=hellorandomizedalpn
 
 SocksPort auto

common/event/interface.go

@@ -5,7 +5,7 @@ import (
 	"time"
 
 	"github.com/pion/webrtc/v3"
-	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/safelog"
+	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/ptutil/safelog"
 )
 
 type SnowflakeEvent interface {
@@ -67,6 +67,14 @@ func (e EventOnProxyStarting) String() string {
 	return "Proxy starting"
 }
 
+type EventOnProxyClientConnected struct {
+	SnowflakeEvent
+}
+
+func (e EventOnProxyClientConnected) String() string {
+	return fmt.Sprintf("client connected")
+}
+
 type EventOnProxyConnectionOver struct {
 	SnowflakeEvent
 	InboundTraffic  int64
@@ -86,8 +94,10 @@ type EventOnProxyStats struct {
 }
 
 func (e EventOnProxyStats) String() string {
-	statString := fmt.Sprintf("In the last %v, there were %v completed connections. Traffic Relayed ↓ %v %v, ↑ %v %v.",
-		e.SummaryInterval.String(), e.ConnectionCount, e.InboundBytes, e.InboundUnit, e.OutboundBytes, e.OutboundUnit)
+	statString := fmt.Sprintf("In the last %v, there were %v completed connections. Traffic Relayed ↓ %v %v (%.2f %v%s), ↑ %v %v (%.2f %v%s).",
+		e.SummaryInterval.String(), e.ConnectionCount,
+		e.InboundBytes, e.InboundUnit, float64(e.InboundBytes)/e.SummaryInterval.Seconds(), e.InboundUnit, "/s",
+		e.OutboundBytes, e.OutboundUnit, float64(e.OutboundBytes)/e.SummaryInterval.Seconds(), e.OutboundUnit, "/s")
 	return statString
 }