Merge pull request #1 from getlantern/feat/migrating-listener

migrating water integration into one repository

dialer/dialer.go

@@ -0,0 +1,39 @@
+// Package dialer holds the dialer implementation for the water transport.
+package dialer
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/getlantern/golog"
+	"github.com/getlantern/lantern-water/logger"
+	"github.com/refraction-networking/water"
+	_ "github.com/refraction-networking/water/transport/v1"
+)
+
+// DialerParameters are used when creating a new dialer.
+type DialerParameters struct {
+	// An optional golog.Logger used for keeping compatibility with http-proxy
+	// and flashlight logger. If not defined the dialer will use the default
+	// water logger.
+	Logger    golog.Logger
+	Transport string // Specifies transport being used.
+	WASM      []byte // The WASM module to use.
+}
+
+// NewDialer creates a new water dialer with the given parameters.
+func NewDialer(ctx context.Context, params DialerParameters) (water.Dialer, error) {
+	cfg := &water.Config{
+		TransportModuleBin: params.WASM,
+	}
+
+	if params.Logger != nil {
+		cfg.OverrideLogger = slog.New(logger.NewLogHandler(params.Logger, params.Transport))
+	}
+
+	dialer, err := water.NewDialerWithContext(ctx, cfg)
+	if err != nil {
+		return nil, err
+	}
+	return dialer, nil
+}

dialer/dialer_test.go

@@ -0,0 +1,93 @@
+package dialer
+
+import (
+	"bytes"
+	"context"
+	"embed"
+	"io"
+	"net"
+	"testing"
+
+	"github.com/getlantern/golog"
+	"github.com/getlantern/lantern-water/listener"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	_ "github.com/refraction-networking/water/transport/v1"
+)
+
+//go:embed testdata/*
+var testData embed.FS
+
+func TestNewDialer(t *testing.T) {
+	ctx := context.Background()
+	f, err := testData.Open("testdata/reverse_v1.wasm")
+	require.Nil(t, err)
+
+	wasm, err := io.ReadAll(f)
+	require.Nil(t, err)
+
+	listenerParameters := listener.ListenerParams{
+		Logger:    golog.LoggerFor("water_listener"),
+		Transport: "reverse_v1",
+		Address:   "127.0.0.1:3000",
+		WASM:      wasm,
+	}
+
+	ll, err := listener.NewWATERListener(ctx, listenerParameters)
+	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)
+				return
+			}
+
+			go func() {
+				if conn == nil {
+					t.Error("nil connection")
+					return
+				}
+				buf := make([]byte, 2*len(messageRequest))
+				n, connErr := conn.Read(buf)
+				if connErr != nil {
+					t.Errorf("error reading: %v", err)
+					return
+				}
+
+				buf = buf[:n]
+				if !bytes.Equal(buf, []byte(messageRequest)) {
+					t.Errorf("unexpected request %v %v", buf, messageRequest)
+					return
+				}
+				conn.Write([]byte(expectedResponse))
+			}()
+		}
+	}()
+
+	dialer, err := NewDialer(ctx, DialerParameters{
+		Logger:    golog.LoggerFor("water_dialer"),
+		Transport: "reverse_v1",
+		WASM:      wasm,
+	})
+
+	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]))
+}

downloader/downloader.go

@@ -0,0 +1,86 @@
+// Package downloader provides a WASM downloader that can download the WASM
+// file from a given URL. The downloader supports both HTTPS URLs and magnet links.
+package downloader
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+)
+
+//go:generate mockgen -package=downloader -destination=mocks_test.go . WASMDownloader,torrentClient,torrentInfo
+//go:generate mockgen -package=downloader -destination=torrent_reader_mock_test.go github.com/anacrolix/torrent Reader
+
+// WASMDownloader is an interface that defines the methods that a WASM downloader
+type WASMDownloader interface {
+	DownloadWASM(context.Context, io.Writer) error
+	Close() error
+}
+
+type downloader struct {
+	urls             []string
+	httpClient       *http.Client
+	httpDownloader   WASMDownloader
+	magnetDownloader WASMDownloader
+}
+
+// NewWaterWASMDownloader creates a new WASMDownloader instance.
+func NewWASMDownloader(urls []string, httpClient *http.Client) (WASMDownloader, error) {
+	if len(urls) == 0 {
+		return nil, fmt.Errorf("WASM downloader requires URLs to download but received empty list")
+	}
+	return &downloader{
+		urls:       urls,
+		httpClient: httpClient,
+	}, nil
+}
+
+func (d *downloader) Close() error {
+	if d.magnetDownloader != nil {
+		return d.magnetDownloader.Close()
+	}
+	return nil
+}
+
+// DownloadWASM downloads the WASM file from the given URLs, verifies the hash
+// sum and writes the file to the given writer.
+func (d *downloader) DownloadWASM(ctx context.Context, w io.Writer) error {
+	joinedErrs := errors.New("failed to download WASM from all URLs")
+	for _, url := range d.urls {
+		err := d.downloadWASM(ctx, w, url)
+		if err != nil {
+			joinedErrs = errors.Join(joinedErrs, err)
+			continue
+		}
+
+		return nil
+	}
+	return joinedErrs
+}
+
+// downloadWASM checks what kind of URL was given and downloads the WASM file
+// from the URL. It can be a HTTPS URL or a magnet link.
+func (d *downloader) downloadWASM(ctx context.Context, w io.Writer, url string) error {
+	switch {
+	case strings.HasPrefix(url, "http://"), strings.HasPrefix(url, "https://"):
+		if d.httpDownloader == nil {
+			d.httpDownloader = newHTTPSDownloader(d.httpClient, url)
+		}
+		return d.httpDownloader.DownloadWASM(ctx, w)
+	case strings.HasPrefix(url, "magnet:?"):
+		if d.magnetDownloader == nil {
+			var err error
+			downloader, err := newMagnetDownloader(ctx, url)
+			if err != nil {
+				return err
+			}
+			d.magnetDownloader = downloader
+		}
+		return d.magnetDownloader.DownloadWASM(ctx, w)
+	default:
+		return fmt.Errorf("unsupported protocol: %s", url)
+	}
+}

downloader/downloader_test.go

@@ -0,0 +1,135 @@
+package downloader
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	gomock "go.uber.org/mock/gomock"
+)
+
+func TestNewWASMDownloader(t *testing.T) {
+	var tests = []struct {
+		name            string
+		givenURLs       []string
+		givenHTTPClient *http.Client
+		assert          func(*testing.T, WASMDownloader, error)
+	}{
+		{
+			name: "it should return an error when providing an empty list of URLs",
+			assert: func(t *testing.T, d WASMDownloader, err error) {
+				assert.Error(t, err)
+				assert.Nil(t, d)
+			},
+		},
+		{
+			name:            "it should successfully return a wasm downloader",
+			givenURLs:       []string{"http://example.com"},
+			givenHTTPClient: http.DefaultClient,
+			assert: func(t *testing.T, wDownloader WASMDownloader, err error) {
+				assert.NoError(t, err)
+				d := wDownloader.(*downloader)
+				assert.Equal(t, []string{"http://example.com"}, d.urls)
+				assert.Equal(t, http.DefaultClient, d.httpClient)
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			d, err := NewWASMDownloader(tt.givenURLs, tt.givenHTTPClient)
+			tt.assert(t, d, err)
+		})
+	}
+}
+
+func TestDownloadWASM(t *testing.T) {
+	ctx := context.Background()
+
+	contentMessage := "content"
+	var tests = []struct {
+		name                string
+		givenHTTPClient     *http.Client
+		givenURLs           []string
+		givenWriter         io.Writer
+		setupHTTPDownloader func(ctrl *gomock.Controller) WASMDownloader
+		assert              func(*testing.T, io.Reader, error)
+	}{
+		{
+			name: "udp urls are unsupported",
+			givenURLs: []string{
+				"udp://example.com",
+			},
+			assert: func(t *testing.T, r io.Reader, err error) {
+				b, berr := io.ReadAll(r)
+				require.NoError(t, berr)
+				assert.Empty(t, b)
+				assert.Error(t, err)
+				assert.ErrorContains(t, err, "unsupported protocol")
+			},
+		},
+		{
+			name: "http download error",
+			givenURLs: []string{
+				"http://example.com",
+			},
+			setupHTTPDownloader: func(ctrl *gomock.Controller) WASMDownloader {
+				httpDownloader := NewMockWASMDownloader(ctrl)
+				httpDownloader.EXPECT().DownloadWASM(ctx, gomock.Any()).Return(assert.AnError)
+				return httpDownloader
+			},
+			assert: func(t *testing.T, r io.Reader, err error) {
+				b, berr := io.ReadAll(r)
+				require.NoError(t, berr)
+				assert.Empty(t, b)
+				assert.Error(t, err)
+				assert.ErrorContains(t, err, assert.AnError.Error())
+				assert.ErrorContains(t, err, "failed to download WASM from all URLs")
+			},
+		},
+		{
+			name: "success",
+			givenURLs: []string{
+				"http://example.com",
+			},
+			setupHTTPDownloader: func(ctrl *gomock.Controller) WASMDownloader {
+				httpDownloader := NewMockWASMDownloader(ctrl)
+				httpDownloader.EXPECT().DownloadWASM(ctx, gomock.Any()).DoAndReturn(
+					func(ctx context.Context, w io.Writer) error {
+						_, err := w.Write([]byte(contentMessage))
+						return err
+					})
+				return httpDownloader
+			},
+			assert: func(t *testing.T, r io.Reader, err error) {
+				b, berr := io.ReadAll(r)
+				require.NoError(t, berr)
+				assert.NoError(t, err)
+				assert.Equal(t, contentMessage, string(b))
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var httpDownloader WASMDownloader
+			if tt.setupHTTPDownloader != nil {
+				ctrl := gomock.NewController(t)
+				defer ctrl.Finish()
+				httpDownloader = tt.setupHTTPDownloader(ctrl)
+			}
+
+			b := &bytes.Buffer{}
+			d, err := NewWASMDownloader(tt.givenURLs, tt.givenHTTPClient)
+			require.NoError(t, err)
+
+			if httpDownloader != nil {
+				d.(*downloader).httpDownloader = httpDownloader
+			}
+			err = d.DownloadWASM(ctx, b)
+			tt.assert(t, b, err)
+		})
+	}
+}