Implement better client IP per rendezvous method tracking for clients

Implement better client IP per rendezvous method tracking for clients Add tests for added code, fix existing tests chore(deps): update module github.com/miekg/dns to v1.1.58 Implement better client IP tracking for http and ampcache Add tests for added code, fix existing tests Implement GetCandidateAddrs from SDP Add getting client IP for SQS Bug fixes Bug fix for tests
2025-10-13 11:11:30 -04:00 · 2024-02-21 01:30:37 -05:00 · 2024-02-21 01:30:37 -05:00 · b512e242e8
commit b512e242e8
parent fe56eaddf4
11 changed files with 252 additions and 24 deletions
--- a/broker/amp.go
+++ b/broker/amp.go
@ -7,6 +7,7 @@ import (

 	"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/util"
 )

 // ampClientOffers is the AMP-speaking endpoint for client poll messages,
@ -35,7 +36,7 @@ func ampClientOffers(i *IPC, w http.ResponseWriter, r *http.Request) {
 	if err == nil {
 		arg := messages.Arg{
 			Body:             encPollReq,
-			RemoteAddr:       "",
+			RemoteAddr:       util.GetClientIp(r),
 			RendezvousMethod: messages.RendezvousAmpCache,
 		}
 		err = i.ClientOffers(arg, &response)
--- a/broker/http.go
+++ b/broker/http.go
@ -10,6 +10,7 @@ import (
 	"os"

 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/messages"
+	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/util"
 )

 const (
@ -102,7 +103,7 @@ func proxyPolls(i *IPC, w http.ResponseWriter, r *http.Request) {

 	arg := messages.Arg{
 		Body:       body,
-		RemoteAddr: r.RemoteAddr,
+		RemoteAddr: util.GetClientIp(r),
 	}

 	var response []byte
@ -167,7 +168,7 @@ func clientOffers(i *IPC, w http.ResponseWriter, r *http.Request) {

 	arg := messages.Arg{
 		Body:             body,
-		RemoteAddr:       "",
+		RemoteAddr:       util.GetClientIp(r),
 		RendezvousMethod: messages.RendezvousHttp,
 	}

@ -227,7 +228,7 @@ func proxyAnswers(i *IPC, w http.ResponseWriter, r *http.Request) {

 	arg := messages.Arg{
 		Body:       body,
-		RemoteAddr: "",
+		RemoteAddr: util.GetClientIp(r),
 	}

 	var response []byte
--- a/broker/ipc.go
+++ b/broker/ipc.go
@ -5,7 +5,6 @@ import (
 	"encoding/hex"
 	"fmt"
 	"log"
-	"net"
 	"time"

 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/bridgefingerprint"
@ -101,7 +100,7 @@ func (i *IPC) ProxyPolls(arg messages.Arg, response *[]byte) error {
 	}

 	// Log geoip stats
-	remoteIP, _, err := net.SplitHostPort(arg.RemoteAddr)
+	remoteIP := arg.RemoteAddr
 	if err != nil {
 		log.Println("Warning: cannot process proxy IP: ", err.Error())
 	} else {
@ -196,13 +195,7 @@ func (i *IPC) ClientOffers(arg messages.Arg, response *[]byte) error {
 		snowflake.offerChannel <- offer
 	} else {
 		i.ctx.metrics.lock.Lock()
-		i.ctx.metrics.clientDeniedCount[arg.RendezvousMethod]++
-		i.ctx.metrics.promMetrics.ClientPollTotal.With(prometheus.Labels{"nat": offer.natType, "status": "denied", "rendezvous_method": string(arg.RendezvousMethod)}).Inc()
-		if offer.natType == NATUnrestricted {
-			i.ctx.metrics.clientUnrestrictedDeniedCount[arg.RendezvousMethod]++
-		} else {
-			i.ctx.metrics.clientRestrictedDeniedCount[arg.RendezvousMethod]++
-		}
+		i.ctx.metrics.UpdateRendezvousStats(arg.RemoteAddr, arg.RendezvousMethod, offer.natType, false)
 		i.ctx.metrics.lock.Unlock()
 		resp := &messages.ClientPollResponse{Error: messages.StrNoProxies}
 		return sendClientResponse(resp, response)
@ -212,8 +205,7 @@ func (i *IPC) ClientOffers(arg messages.Arg, response *[]byte) error {
 	select {
 	case answer := <-snowflake.answerChannel:
 		i.ctx.metrics.lock.Lock()
-		i.ctx.metrics.clientProxyMatchCount[arg.RendezvousMethod]++
-		i.ctx.metrics.promMetrics.ClientPollTotal.With(prometheus.Labels{"nat": offer.natType, "status": "matched", "rendezvous_method": string(arg.RendezvousMethod)}).Inc()
+		i.ctx.metrics.UpdateRendezvousStats(arg.RemoteAddr, arg.RendezvousMethod, offer.natType, true)
 		i.ctx.metrics.lock.Unlock()
 		resp := &messages.ClientPollResponse{Answer: answer}
 		err = sendClientResponse(resp, response)
--- a/broker/metrics.go
+++ b/broker/metrics.go
@ -24,6 +24,12 @@ const (
 	metricsResolution   = 60 * 60 * 24 * time.Second //86400 seconds
 )

+var rendezvoudMethodList = [...]messages.RendezvousMethod{
+	messages.RendezvousHttp,
+	messages.RendezvousAmpCache,
+	messages.RendezvousSqs,
+}
+
 type CountryStats struct {
 	// map[proxyType][address]bool
 	proxies map[string]map[string]bool
@ -49,6 +55,8 @@ type Metrics struct {
 	clientUnrestrictedDeniedCount map[messages.RendezvousMethod]uint
 	clientProxyMatchCount         map[messages.RendezvousMethod]uint

+	rendezvousCountryStats map[messages.RendezvousMethod]map[string]int
+
 	proxyPollWithRelayURLExtension         uint
 	proxyPollWithoutRelayURLExtension      uint
 	proxyPollRejectedWithRelayURLExtension uint
@ -96,7 +104,6 @@ func (s CountryStats) Display() string {
 }

 func (m *Metrics) UpdateCountryStats(addr string, proxyType string, natType string) {
-
 	var country string
 	var ok bool

@ -137,7 +144,59 @@ func (m *Metrics) UpdateCountryStats(addr string, proxyType string, natType stri
 	default:
 		m.countryStats.natUnknown[addr] = true
 	}
+}

+func (m *Metrics) UpdateRendezvousStats(addr string, rendezvousMethod messages.RendezvousMethod, natType string, matched bool) {
+	ip := net.ParseIP(addr)
+	country := "??"
+	if m.geoipdb != nil {
+		country_by_addr, ok := m.geoipdb.GetCountryByAddr(ip)
+		if ok {
+			country = country_by_addr
+		}
+	}
+
+	var status string
+	if !matched {
+		m.clientDeniedCount[rendezvousMethod]++
+		if natType == NATUnrestricted {
+			m.clientUnrestrictedDeniedCount[rendezvousMethod]++
+		} else {
+			m.clientRestrictedDeniedCount[rendezvousMethod]++
+		}
+		status = "denied"
+	} else {
+		status = "matched"
+		m.clientProxyMatchCount[rendezvousMethod]++
+	}
+	m.rendezvousCountryStats[rendezvousMethod][country]++
+	m.promMetrics.ClientPollTotal.With(prometheus.Labels{
+		"nat":               natType,
+		"status":            status,
+		"rendezvous_method": string(rendezvousMethod),
+		"cc":                country,
+	}).Inc()
+}
+
+func (m *Metrics) DisplayRendezvousStatsByCountry(rendezvoudMethod messages.RendezvousMethod) string {
+	output := ""
+
+	// Use the records struct to sort our counts map by value.
+	rs := records{}
+	for cc, count := range m.rendezvousCountryStats[rendezvoudMethod] {
+		rs = append(rs, record{cc: cc, count: count})
+	}
+	sort.Sort(sort.Reverse(rs))
+	for _, r := range rs {
+		output += fmt.Sprintf("%s=%d,", r.cc, binCount(uint(r.count)))
+	}
+
+	// cut off trailing ","
+	if len(output) > 0 {
+		return output[:len(output)-1]
+	}
+
+	return output
 }

 func (m *Metrics) LoadGeoipDatabases(geoipDB string, geoip6DB string) error {
@ -157,6 +216,11 @@ func NewMetrics(metricsLogger *log.Logger) (*Metrics, error) {
 	m.clientUnrestrictedDeniedCount = make(map[messages.RendezvousMethod]uint)
 	m.clientProxyMatchCount = make(map[messages.RendezvousMethod]uint)

+	m.rendezvousCountryStats = make(map[messages.RendezvousMethod]map[string]int)
+	for _, rendezvousMethod := range rendezvoudMethodList {
+		m.rendezvousCountryStats[rendezvousMethod] = make(map[string]int)
+	}
+
 	m.countryStats = CountryStats{
 		counts:          make(map[string]int),
 		proxies:         make(map[string]map[string]bool),
@ -211,14 +275,11 @@ func (m *Metrics) printMetrics() {
 	m.logger.Println("client-unrestricted-denied-count", binCount(sumMapValues(&m.clientUnrestrictedDeniedCount)))
 	m.logger.Println("client-snowflake-match-count", binCount(sumMapValues(&m.clientProxyMatchCount)))

-	for _, rendezvousMethod := range [3]messages.RendezvousMethod{
-		messages.RendezvousHttp,
-		messages.RendezvousAmpCache,
-		messages.RendezvousSqs,
-	} {
+	for _, rendezvousMethod := range rendezvoudMethodList {
 		m.logger.Printf("client-%s-count %d\n", rendezvousMethod, binCount(
 			m.clientDeniedCount[rendezvousMethod]+m.clientProxyMatchCount[rendezvousMethod],
 		))
+		m.logger.Printf("client-%s-ips %s\n", rendezvousMethod, m.DisplayRendezvousStatsByCountry(rendezvousMethod))
 	}

 	m.logger.Println("snowflake-ips-nat-restricted", len(m.countryStats.natRestricted))
@ -237,6 +298,12 @@ func (m *Metrics) zeroMetrics() {
 	m.proxyPollWithRelayURLExtension = 0
 	m.proxyPollWithoutRelayURLExtension = 0
 	m.clientProxyMatchCount = make(map[messages.RendezvousMethod]uint)
+
+	m.rendezvousCountryStats = make(map[messages.RendezvousMethod]map[string]int)
+	for _, rendezvousMethod := range rendezvoudMethodList {
+		m.rendezvousCountryStats[rendezvousMethod] = make(map[string]int)
+	}
+
 	m.countryStats.counts = make(map[string]int)
 	for pType := range m.countryStats.proxies {
 		m.countryStats.proxies[pType] = make(map[string]bool)
@ -339,7 +406,7 @@ func initPrometheus() *PromMetrics {
 			Name:      "rounded_client_poll_total",
 			Help:      "The number of snowflake client polls, rounded up to a multiple of 8",
 		},
-		[]string{"nat", "status", "rendezvous_method"},
+		[]string{"nat", "status", "cc", "rendezvous_method"},
 	)

 	// We need to register our metrics so they can be exported.
--- a/broker/snowflake-broker_test.go
+++ b/broker/snowflake-broker_test.go
@ -157,8 +157,11 @@ client-restricted-denied-count 8
 client-unrestricted-denied-count 0
 client-snowflake-match-count 0
 client-http-count 8
+client-http-ips ??=8
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 0
+client-sqs-ips 
 `)
 			})

@ -184,8 +187,11 @@ client-restricted-denied-count 0
 client-unrestricted-denied-count 0
 client-snowflake-match-count 8
 client-http-count 8
+client-http-ips ??=8
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 0
+client-sqs-ips 
 `)
 			})

@ -260,8 +266,11 @@ client-restricted-denied-count 8
 client-unrestricted-denied-count 0
 client-snowflake-match-count 0
 client-http-count 8
+client-http-ips ??=8
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 0
+client-sqs-ips 
 `)
 			})

@ -287,8 +296,11 @@ client-restricted-denied-count 0
 client-unrestricted-denied-count 0
 client-snowflake-match-count 8
 client-http-count 8
+client-http-ips ??=8
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 0
+client-sqs-ips 
 `)
 			})

@ -340,8 +352,11 @@ client-restricted-denied-count 8
 client-unrestricted-denied-count 0
 client-snowflake-match-count 0
 client-http-count 0
+client-http-ips 
 client-ampcache-count 8
+client-ampcache-ips ??=8
 client-sqs-count 0
+client-sqs-ips 
 `)
 			})

@ -369,8 +384,11 @@ client-restricted-denied-count 0
 client-unrestricted-denied-count 0
 client-snowflake-match-count 8
 client-http-count 0
+client-http-ips 
 client-ampcache-count 8
+client-ampcache-ips ??=8
 client-sqs-count 0
+client-sqs-ips 
 `)
 			})

@ -728,8 +746,11 @@ client-restricted-denied-count 0
 client-unrestricted-denied-count 0
 client-snowflake-match-count 0
 client-http-count 0
+client-http-ips 
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 0
+client-sqs-ips 
 snowflake-ips-nat-restricted 0
 snowflake-ips-nat-unrestricted 0
 snowflake-ips-nat-unknown 1
@ -742,6 +763,7 @@ snowflake-ips-nat-unknown 1
 			data, err := createClientOffer(sdp, NATUnknown, "")
 			So(err, ShouldBeNil)
 			r, err := http.NewRequest("POST", "snowflake.broker/client", data)
+			r.RemoteAddr = "129.97.208.23:8888" //CA geoip
 			So(err, ShouldBeNil)

 			clientOffers(i, w, r)
@ -752,9 +774,11 @@ client-restricted-denied-count 8
 client-unrestricted-denied-count 0
 client-snowflake-match-count 0
 client-http-count 8
+client-http-ips CA=8
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 0
-`)
+client-sqs-ips `)

 			// Test reset
 			buf.Reset()
@ -774,8 +798,11 @@ client-restricted-denied-count 0
 client-unrestricted-denied-count 0
 client-snowflake-match-count 0
 client-http-count 0
+client-http-ips 
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 0
+client-sqs-ips 
 snowflake-ips-nat-restricted 0
 snowflake-ips-nat-unrestricted 0
 snowflake-ips-nat-unknown 0
--- a/broker/sqs.go
+++ b/broker/sqs.go
@ -12,6 +12,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/sqs/types"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/messages"
 	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/sqsclient"
+	"gitlab.torproject.org/tpo/anti-censorship/pluggable-transports/snowflake/v2/common/util"
 )

 const (
@ -144,9 +145,27 @@ func (r *sqsHandler) handleMessage(context context.Context, message *types.Messa
 	answerSQSURL := res.QueueUrl

 	encPollReq = []byte(*message.Body)
+
+	// Get best guess Client IP for geolocating
+	remoteAddr := ""
+	req, err := messages.DecodeClientPollRequest(encPollReq)
+	if err != nil {
+		log.Printf("SQSHandler: error encounted when decoding client poll request %s: %v\n", *clientID, err)
+	} else {
+		sdp, err := util.DeserializeSessionDescription(req.Offer)
+		if err != nil {
+			log.Printf("SQSHandler: error encounted when deserializing session desc %s: %v\n", *clientID, err)
+		} else {
+			candidateAddrs := util.GetCandidateAddrs(sdp.SDP)
+			if len(candidateAddrs) > 0 {
+				remoteAddr = candidateAddrs[0].String()
+			}
+		}
+	}
+
 	arg := messages.Arg{
 		Body:             encPollReq,
-		RemoteAddr:       "",
+		RemoteAddr:       remoteAddr,
 		RendezvousMethod: messages.RendezvousSqs,
 	}
 	err = r.IPC.ClientOffers(arg, &response)
--- a/broker/sqs_test.go
+++ b/broker/sqs_test.go
@ -195,8 +195,11 @@ client-restricted-denied-count 0
 client-unrestricted-denied-count 0
 client-snowflake-match-count 8
 client-http-count 0
+client-http-ips 
 client-ampcache-count 0
+client-ampcache-ips 
 client-sqs-count 8
+client-sqs-ips ??=8
 `)
 								wg.Done()
 							}