Adding WATER reverse listener

go.mod

@@ -48,9 +48,8 @@ require (
 	github.com/hashicorp/golang-lru v0.5.4
 	github.com/mitchellh/panicwrap v1.0.0
 	github.com/op/go-logging v0.0.0-20160315200505-970db520ece7
-	github.com/oschwald/geoip2-golang v1.8.0
-	github.com/prometheus/client_golang v1.19.1
 	github.com/refraction-networking/utls v1.3.3
+	github.com/refraction-networking/water v0.6.4
 	github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726
 	github.com/spaolacci/murmur3 v1.1.0
 	github.com/stretchr/testify v1.8.4
@@ -131,6 +130,7 @@ require (
 	github.com/mmcloughlin/avo v0.0.0-20200803215136-443f81d77104 // indirect
 	github.com/nwaples/rardecode v1.1.2 // indirect
 	github.com/onsi/ginkgo/v2 v2.12.0 // indirect
+	github.com/oschwald/geoip2-golang v1.8.0 // indirect
 	github.com/oschwald/maxminddb-golang v1.10.0 // indirect
 	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect
 	github.com/pierrec/lz4/v4 v4.1.12 // indirect
@@ -152,6 +152,7 @@ require (
 	github.com/pion/webrtc/v3 v3.2.6 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/prometheus/client_golang v1.19.1 // indirect
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.48.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
@@ -162,6 +163,7 @@ require (
 	github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 // indirect
 	github.com/templexxx/cpu v0.0.8 // indirect
 	github.com/templexxx/xorsimd v0.4.1 // indirect
+	github.com/tetratelabs/wazero v1.7.0 // indirect
 	github.com/ti-mo/conntrack v0.3.0 // indirect
 	github.com/ti-mo/netfilter v0.3.1 // indirect
 	github.com/tjfoc/gmsm v1.3.2 // indirect
@@ -180,13 +182,13 @@ require (
 	go.uber.org/multierr v1.8.0 // indirect
 	go.uber.org/zap v1.21.0 // indirect
 	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/exp v0.0.0-20240110193028-0dcbfd608b1e // indirect
+	golang.org/x/exp v0.0.0-20240119083558-1b970713d09a // indirect
 	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/oauth2 v0.16.0 // indirect
 	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.17.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
-	golang.org/x/tools v0.16.0 // indirect
+	golang.org/x/tools v0.17.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20231002182017-d307bd883b97 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20231002182017-d307bd883b97 // indirect
@@ -199,3 +201,5 @@ require (
 
 // Waiting on https://github.com/mitchellh/panicwrap/pull/27 to be merged upstream
 replace github.com/mitchellh/panicwrap v1.0.0 => github.com/getlantern/panicwrap v0.0.0-20200707191944-9ba45baf8e51
+
+replace github.com/tetratelabs/wazero => github.com/refraction-networking/wazero v1.7.1-w

go.sum

@@ -423,6 +423,10 @@ github.com/quic-go/quic-go v0.40.0 h1:GYd1iznlKm7dpHD7pOVpUvItgMPo/jrMgDWZhMCecq
 github.com/quic-go/quic-go v0.40.0/go.mod h1:PeN7kuVJ4xZbxSv/4OX6S1USOX8MJvydwpTx31vx60c=
 github.com/refraction-networking/utls v1.3.3 h1:f/TBLX7KBciRyFH3bwupp+CE4fzoYKCirhdRcC490sw=
 github.com/refraction-networking/utls v1.3.3/go.mod h1:DlecWW1LMlMJu+9qpzzQqdHDT/C2LAe03EdpLUz/RL8=
+github.com/refraction-networking/water v0.6.4 h1:7E0nFdSyVXMZ5bz7GPHz1ALBQDUiyHbiBfPZ4eSBmIM=
+github.com/refraction-networking/water v0.6.4/go.mod h1:UDjvBH4nYp0UkrJm9GJ1G96cpELH31/BIjoTiWkYoME=
+github.com/refraction-networking/wazero v1.7.1-w h1:z7Ty5PsMkJEDBCsn3ELUjceQGBT0FMVGldOSpDK3giQ=
+github.com/refraction-networking/wazero v1.7.1-w/go.mod h1:ytl6Zuh20R/eROuyDaGPkp82O9C/DJfXAwJfQ3X6/7Y=
 github.com/riobard/go-bloom v0.0.0-20200614022211-cdc8013cb5b3 h1:f/FNXud6gA3MNr8meMVVGxhp+QBTqY91tM8HjEuMjGg=
 github.com/riobard/go-bloom v0.0.0-20200614022211-cdc8013cb5b3/go.mod h1:HgjTstvQsPGkxUsCd2KWxErBblirPizecHcpD3ffK+s=
 github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
@@ -550,8 +554,8 @@ golang.org/x/crypto v0.9.0/go.mod h1:yrmDGqONDYtNj3tH8X9dzUun2m2lzPa9ngI6/RUPGR0
 golang.org/x/crypto v0.18.0 h1:PGVlW0xEltQnzFZ55hkuX5+KLyrMYhHld1YHO4AKcdc=
 golang.org/x/crypto v0.18.0/go.mod h1:R0j02AL6hcrfOiy9T4ZYp/rcWeMxM3L6QYxlOuEG1mg=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
-golang.org/x/exp v0.0.0-20240110193028-0dcbfd608b1e h1:723BNChdd0c2Wk6WOE320qGBiPtYx0F0Bbm1kriShfE=
-golang.org/x/exp v0.0.0-20240110193028-0dcbfd608b1e/go.mod h1:iRJReGqOEeBhDZGkGbynYwcHlctCvnjTYIamk7uXpHI=
+golang.org/x/exp v0.0.0-20240119083558-1b970713d09a h1:Q8/wZp0KX97QFTc2ywcOE0YRjZPVIx+MXInMzdvQqcA=
+golang.org/x/exp v0.0.0-20240119083558-1b970713d09a/go.mod h1:idGWGoKP1toJGkd5/ig9ZLuPcZBC3ewk7SzmH0uou08=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
@@ -677,8 +681,8 @@ golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4f
 golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
 golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
-golang.org/x/tools v0.16.0 h1:GO788SKMRunPIBCXiQyo2AaexLstOrVhuAL5YwsckQM=
-golang.org/x/tools v0.16.0/go.mod h1:kYVVN6I1mBNoB1OX+noeBjbRk4IUEPa7JJ+TJMEooJ0=
+golang.org/x/tools v0.17.0 h1:FvmRgNOcs3kOa+T20R1uhfP9F6HgG2mfxDv1vrx1Htc=
+golang.org/x/tools v0.17.0/go.mod h1:xsh6VxdV005rRVaS6SSAf9oiAqljS7UZUacMZ8Bnsps=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

http-proxy/main.go

@@ -186,6 +186,9 @@ var (
 
 	algenevaAddr = flag.String("algeneva-addr", "", "Address at which to listen for algenAddr connections.")
 
+	waterAddr = flag.String("water-addr", "", "Address at which to listen for WATER connections.")
+	waterWASM = flag.String("water-wasm", "", "Base64 encoded WASM for WATER")
+
 	track = flag.String("track", "", "The track this proxy is running on")
 )
 
@@ -469,6 +472,8 @@ func main() {
 		BroflakeCert:                       os.Getenv("BROFLAKE_CERT"),
 		BroflakeKey:                        os.Getenv("BROFLAKE_KEY"),
 		AlgenevaAddr:                       *algenevaAddr,
+		WaterAddr:                          *waterAddr,
+		WaterWASM:                          *waterWASM,
 	}
 	if *maxmindLicenseKey != "" {
 		log.Debug("Will use Maxmind for geolocating clients")

http_proxy.go

@@ -65,6 +65,7 @@ import (
 	"github.com/getlantern/http-proxy-lantern/v2/tlslistener"
 	"github.com/getlantern/http-proxy-lantern/v2/tlsmasq"
 	"github.com/getlantern/http-proxy-lantern/v2/tokenfilter"
+	"github.com/getlantern/http-proxy-lantern/v2/water"
 	"github.com/getlantern/http-proxy-lantern/v2/wss"
 
 	algeneva "github.com/getlantern/lantern-algeneva"
@@ -184,6 +185,9 @@ type Proxy struct {
 
 	AlgenevaAddr string
 
+	WaterAddr string
+	WaterWASM string
+
 	throttleConfig throttle.Config
 	instrument     instrument.Instrument
 }
@@ -649,6 +653,8 @@ func (p *Proxy) buildOTELOpts(endpoint string, includeProxyName bool) *otel.Opts
 		opts.Addr = p.BroflakeAddr
 	} else if p.AlgenevaAddr != "" {
 		opts.Addr = p.AlgenevaAddr
+	} else if p.WaterAddr != "" {
+		opts.Addr = p.WaterAddr
 	}
 	if includeProxyName {
 		opts.ProxyName = proxyName
@@ -960,6 +966,19 @@ func (p *Proxy) listenAlgeneva(baseListen func(string) (net.Listener, error)) li
 	}
 }
 
+func (p *Proxy) listenWATER() listenerBuilderFN {
+	return func(addr string) (net.Listener, error) {
+		ctx := context.Background()
+		waterListener, err := water.NewWATERListener(ctx, addr, p.WaterWASM)
+		if err != nil {
+			return nil, err
+		}
+
+		log.Debugf("Listening for water at %v", waterListener.Addr())
+		return waterListener, nil
+	}
+}
+
 func (p *Proxy) setupPacketForward() error {
 	if runtime.GOOS != "linux" {
 		log.Debugf("Ignoring packet forward on %v", runtime.GOOS)

protoListeners.go

@@ -29,5 +29,6 @@ func getProtoListenersArgs(p *Proxy) []protoListenerArgs {
 			p.ShadowsocksMultiplexAddr,
 			p.wrapMultiplexing(p.listenShadowsocks),
 		},
+		{"water_reverse", p.WaterAddr, p.wrapMultiplexing(p.listenWATER())},
 	}
 }

water/listener.go

@@ -0,0 +1,33 @@
+package water
+
+import (
+	"context"
+	"encoding/base64"
+	"net"
+
+	"github.com/getlantern/golog"
+	"github.com/refraction-networking/water"
+	_ "github.com/refraction-networking/water/transport/v0"
+)
+
+var log = golog.LoggerFor("water")
+
+func NewWATERListener(ctx context.Context, address string, wasm string) (net.Listener, error) {
+	decodedWASM, err := base64.StdEncoding.DecodeString(wasm)
+	if err != nil {
+		log.Errorf("failed to decode WASM base64: %v", err)
+		return nil, err
+	}
+
+	cfg := &water.Config{
+		TransportModuleBin: decodedWASM,
+	}
+
+	waterListener, err := cfg.ListenContext(ctx, "tcp", address)
+	if err != nil {
+		log.Errorf("error creating water listener: %v", err)
+		return nil, err
+	}
+
+	return waterListener, nil
+}

water/listener_test.go

@@ -0,0 +1,86 @@
+package water
+
+import (
+	"bytes"
+	"context"
+	"embed"
+	"encoding/base64"
+	"io"
+	"net"
+	"testing"
+
+	"github.com/refraction-networking/water"
+	_ "github.com/refraction-networking/water/transport/v0"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+//go:embed testdata/*
+var testData embed.FS
+
+func TestWATERListener(t *testing.T) {
+	addr := "127.0.0.1:8888"
+
+	f, err := testData.Open("testdata/reverse.go.wasm")
+	require.Nil(t, err)
+
+	wasm, err := io.ReadAll(f)
+	require.Nil(t, err)
+
+	b64WASM := base64.StdEncoding.EncodeToString(wasm)
+
+	ctx := context.Background()
+
+	cfg := &water.Config{
+		TransportModuleBin: wasm,
+	}
+
+	ll, err := NewWATERListener(ctx, addr, b64WASM)
+	require.Nil(t, err)
+
+	messageRequest := "hello"
+	expectedResponse := "world"
+	// running listener
+	go func() {
+		for {
+			var conn net.Conn
+			conn, err = ll.Accept()
+			if err != nil {
+				t.Error(err)
+			}
+
+			go func() {
+				buf := make([]byte, 2*len(messageRequest))
+				n, err := conn.Read(buf)
+				if err != nil {
+					log.Errorf("error reading: %v", err)
+					return
+				}
+
+				buf = buf[:n]
+				if !bytes.Equal(buf, []byte(messageRequest)) {
+					log.Errorf("unexpected request %v %v", buf, messageRequest)
+					return
+				}
+				conn.Write([]byte(expectedResponse))
+			}()
+		}
+	}()
+
+	dialer, err := water.NewDialerWithContext(ctx, cfg)
+	require.Nil(t, err)
+
+	conn, err := dialer.DialContext(ctx, "tcp", ll.Addr().String())
+	require.Nil(t, err)
+	defer conn.Close()
+
+	n, err := conn.Write([]byte(messageRequest))
+	assert.Nil(t, err)
+	assert.Equal(t, len(messageRequest), n)
+
+	buf := make([]byte, 1024)
+	n, err = conn.Read(buf)
+	assert.Nil(t, err)
+	assert.Equal(t, len(expectedResponse), n)
+	assert.Equal(t, expectedResponse, string(buf[:n]))
+}