[connect-ip] add client tunnel command and skeleton of end-to-end test

cmd/tunnelproxy/main.go

@@ -87,7 +87,11 @@ func main() {
 	g.Go(func() error {
 		log.Infof("Starting Tunnel Proxy server")
 
-		return srv.Start(ctx, mgr)
+		if err := srv.SetupWithManager(mgr); err != nil {
+			log.Fatalf("Unable to setup Tunnel Proxy server: %v", err)
+		}
+
+		return srv.Start(ctx)
 	})
 
 	g.Go(func() error {

go.mod

@@ -15,7 +15,7 @@ require (
 	github.com/dgraph-io/badger/v4 v4.5.0
 	github.com/docker/docker v27.2.0+incompatible
 	github.com/dpeckett/contextio v0.5.1
-	github.com/dpeckett/network v0.3.1
+	github.com/dpeckett/network v0.3.3
 	github.com/dpeckett/triemap v0.3.1
 	github.com/envoyproxy/gateway v0.5.0-rc.1.0.20240618131507-bdff5d56b59d
 	github.com/envoyproxy/go-control-plane v0.13.0

go.sum

@@ -895,6 +895,8 @@ github.com/dpeckett/contextio v0.5.1 h1:w19s6EThbZuRpa2z/Lu06v6+o3rrZhbBzmkol6en
 github.com/dpeckett/contextio v0.5.1/go.mod h1:IY/CQ1ee6y4C5j/mU0X0M/D84s2FxNisggbNClTPndc=
 github.com/dpeckett/network v0.3.1 h1:rMDRLc85zc3v4mGcGfbOrNA9Kx69K2Xr8bD/Hc9MERY=
 github.com/dpeckett/network v0.3.1/go.mod h1:83quX+FE+BdOAKFEm5Om+QdI/1ZEQVNUBZSPl7V7erk=
+github.com/dpeckett/network v0.3.3 h1:ghUiAOddpgKSBokp7/CVhjkO+ZjsuczeerwB+i9dBdw=
+github.com/dpeckett/network v0.3.3/go.mod h1:AyOdc+YGAT7zWoDDd/r8Pv4kGNlKrboVcug0FKaoqbA=
 github.com/dpeckett/triemap v0.3.1 h1:jzxCyKs/ATw9uCdD2bd0xFTPLIP9uZwX0iZUOOOIDoc=
 github.com/dpeckett/triemap v0.3.1/go.mod h1:pBxNH+K6m5I4lVo+W7u6JEanxP13adD4t2XYVMxfmTo=
 github.com/dunglas/httpsfv v1.0.2 h1:iERDp/YAfnojSDJ7PW3dj1AReJz4MrwbECSSE59JWL0=

pkg/connip/client.go

@@ -151,7 +151,7 @@ func (t *ClientTransport) Connect(ctx context.Context, serverAddr string) error
 	t.NetstackNetwork = t.tun.Network(resolveConf)
 
 	// TODO (dpeckett): how to bubble up errors from this?
-	go splice(t.tun, t.conn)
+	go Splice(t.tun, t.conn)
 
 	return nil
 }
@@ -181,7 +181,12 @@ func (t *ClientTransport) Close() error {
 	return closeErr
 }
 
-// FowardToLoopback forwards all inbound traffic to the loopback interface.
-func (t *ClientTransport) FowardToLoopback(ctx context.Context) error {
-	return t.tun.ForwardTo(ctx, network.Loopback())
+// FowardTo forwards all inbound traffic to the upstream network.
+func (t *ClientTransport) FowardTo(ctx context.Context, upstream network.Network) error {
+	return t.tun.ForwardTo(ctx, upstream)
+}
+
+// LocalAddresses returns the local addresses assigned to the client.
+func (t *ClientTransport) LocalAddresses() ([]netip.Prefix, error) {
+	return t.tun.LocalAddresses()
 }

pkg/connip/muxed_connection.go

@@ -108,6 +108,8 @@ func (m *MuxedConnection) ReadPacket(pkt []byte) (int, error) {
 }
 
 func (m *MuxedConnection) WritePacket(pkt []byte) ([]byte, error) {
+	slog.Debug("Write packet to connection", slog.Int("bytes", len(pkt)))
+
 	var dstIP netip.Addr
 	switch pkt[0] >> 4 {
 	case 6:

pkg/connip/splice.go

@@ -12,22 +12,18 @@ import (
 	"github.com/apoxy-dev/apoxy-cli/pkg/netstack"
 )
 
-func splice(tun tun.Device, conn Connection) error {
+func Splice(tun tun.Device, conn Connection) error {
 	var g errgroup.Group
 
 	g.Go(func() error {
+		defer conn.Close()
+
 		var pkt [netstack.IPv6MinMTU]byte
 		sizes := make([]int, 1)
 
 		for {
 			_, err := tun.Read([][]byte{pkt[:]}, sizes, 0)
 			if err != nil {
-				if errors.Is(err, net.ErrClosed) {
-					// TUN device is closed, exit the loop.
-					// TODO: is this the correct error
-					return nil
-				}
-
 				return fmt.Errorf("failed to read from TUN: %w", err)
 			}
 
@@ -54,9 +50,6 @@ func splice(tun tun.Device, conn Connection) error {
 		for {
 			n, err := conn.ReadPacket(pkt[:])
 			if err != nil {
-				if errors.Is(err, net.ErrClosed) {
-					return nil
-				}
 				return fmt.Errorf("failed to read from connection: %w", err)
 			}
 
@@ -69,5 +62,9 @@ func splice(tun tun.Device, conn Connection) error {
 		}
 	})
 
-	return g.Wait()
+	if err := g.Wait(); err != nil && !errors.Is(err, net.ErrClosed) {
+		return fmt.Errorf("failed to splice: %w", err)
+	}
+
+	return nil
 }

pkg/tunnel/client.go

@@ -0,0 +1,184 @@
+package tunnel
+
+import (
+	"context"
+	"crypto/x509"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/netip"
+	"strconv"
+
+	"github.com/apoxy-dev/apoxy-cli/pkg/connip"
+	"github.com/apoxy-dev/apoxy-cli/pkg/socksproxy"
+	"github.com/dpeckett/network"
+)
+
+type TunnelClientOption func(*tunnelClientOptions)
+
+type tunnelClientOptions struct {
+	serverAddr      string
+	uuid            string
+	authToken       string
+	pcapPath        string
+	rootCAs         *x509.CertPool
+	socksListenAddr string
+}
+
+func defaultClientOptions() *tunnelClientOptions {
+	return &tunnelClientOptions{
+		serverAddr:      "localhost:8443",
+		socksListenAddr: "localhost:1080",
+	}
+}
+
+// WithServerAddr sets the server address that the tunnel client will connect to.
+// The address should be in the format "host:port".
+func WithServerAddr(addr string) TunnelClientOption {
+	return func(o *tunnelClientOptions) {
+		o.serverAddr = addr
+	}
+}
+
+// WithUUID sets the UUID for the tunnel client.
+func WithUUID(uuid string) TunnelClientOption {
+	return func(o *tunnelClientOptions) {
+		o.uuid = uuid
+	}
+}
+
+// WithAuthToken sets the authentication token for the tunnel client.
+func WithAuthToken(token string) TunnelClientOption {
+	return func(o *tunnelClientOptions) {
+		o.authToken = token
+	}
+}
+
+// WithPcapPath sets the optional path to a packet capture file for the tunnel client.
+func WithPcapPath(path string) TunnelClientOption {
+	return func(o *tunnelClientOptions) {
+		o.pcapPath = path
+	}
+}
+
+// WithRootCAs sets the optional root CA certificates for TLS verification.
+func WithRootCAs(caCerts *x509.CertPool) TunnelClientOption {
+	return func(o *tunnelClientOptions) {
+		o.rootCAs = caCerts
+	}
+}
+
+// WithSocksListenAddr sets the listen address for the local SOCKS5 proxy server.
+func WithSocksListenAddr(addr string) TunnelClientOption {
+	return func(o *tunnelClientOptions) {
+		o.socksListenAddr = addr
+	}
+}
+
+type TunnelClient struct {
+	options   *tunnelClientOptions
+	transport *connip.ClientTransport
+	proxy     *socksproxy.ProxyServer
+
+	tunnelCtx       context.Context
+	tunnelCtxCancel context.CancelFunc
+}
+
+// NewTunnelClient creates a new SOCKS5 proxy and loopback reverse proxy,
+// that forwards and receives traffic via QUIC tunnels.
+func NewTunnelClient(opts ...TunnelClientOption) (*TunnelClient, error) {
+	options := defaultClientOptions()
+	for _, opt := range opts {
+		opt(options)
+	}
+
+	// Validate the options.
+	if options.uuid == "" {
+		return nil, fmt.Errorf("uuid is required")
+	}
+
+	if options.authToken == "" {
+		return nil, fmt.Errorf("auth token is required")
+	}
+
+	// Create the transport layer for the tunnel client.
+	transport := connip.NewClientTransport(&connip.ClientConfig{
+		UUID:      options.uuid,
+		AuthToken: options.authToken,
+		PcapPath:  options.pcapPath,
+		RootCAs:   options.rootCAs,
+	})
+
+	proxy := socksproxy.NewServer(options.socksListenAddr, transport, network.Host())
+
+	return &TunnelClient{
+		options:   options,
+		transport: transport,
+		proxy:     proxy,
+	}, nil
+}
+
+// Start establishes a connection to the server and begins forwarding traffic.
+func (c *TunnelClient) Start(ctx context.Context) error {
+	c.tunnelCtx, c.tunnelCtxCancel = context.WithCancel(ctx)
+
+	// Connect to the server using the transport layer.
+	if err := c.transport.Connect(ctx, c.options.serverAddr); err != nil {
+		return fmt.Errorf("failed to connect to server: %w", err)
+	}
+
+	_, socksListenPortStr, err := net.SplitHostPort(c.options.socksListenAddr)
+	if err != nil {
+		return fmt.Errorf("failed to parse SOCKS listen address: %w", err)
+	}
+
+	socksListenPort, err := strconv.Atoi(socksListenPortStr)
+	if err != nil {
+		return fmt.Errorf("failed to parse SOCKS listen port: %w", err)
+	}
+
+	slog.Info("Forwarding all inbound traffic to loopback interface")
+
+	// Forward all inbound traffic to the loopback interface.
+	// This allows the tunnel client to act as a reverse proxy.
+	if err := c.transport.FowardTo(c.tunnelCtx, network.Filtered(&network.FilteredNetworkConfig{
+		// Otherwise we could be DoS'd by a network loop.
+		DeniedPorts: []uint16{uint16(socksListenPort)},
+		Upstream:    network.Loopback(),
+	})); err != nil {
+		return fmt.Errorf("failed to forward to loopback: %w", err)
+	}
+
+	slog.Info("Starting SOCKS5 proxy", slog.String("listenAddr", c.options.socksListenAddr))
+
+	// Start the SOCKS5 proxy for forwarding outbound traffic.
+	go func() {
+		if err := c.proxy.ListenAndServe(c.tunnelCtx); err != nil {
+			slog.Error("SOCKS proxy error", slog.String("error", err.Error()))
+		}
+	}()
+
+	return nil
+}
+
+// Stop closes the tunnel client and stops forwarding traffic.
+func (t *TunnelClient) Stop() error {
+	// Stop any background tasks (and the SOCKS5 proxy).
+	if t.tunnelCtxCancel != nil {
+		t.tunnelCtxCancel()
+	}
+
+	// Close the transport layer.
+	if t.transport != nil {
+		if err := t.transport.Close(); err != nil {
+			return fmt.Errorf("failed to close transport: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// Get the local addresses assigned to the tunnel client.
+func (c *TunnelClient) LocalAddresses() ([]netip.Prefix, error) {
+	return c.transport.LocalAddresses()
+}