feat: Add TURN proxying to enable offline deployments (#1000)

* Add turnconn

* Add option for passing ICE servers

* Log TURN remote address

* Add TURN server to coder start

coderd/coderd.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/go-chi/chi/v5"
+	"github.com/pion/webrtc/v3"
 	"google.golang.org/api/idtoken"
 
 	chitrace "gopkg.in/DataDog/dd-trace-go.v1/contrib/go-chi/chi.v5"
@@ -20,23 +21,25 @@ import (
 	"github.com/coder/coder/coderd/gitsshkey"
 	"github.com/coder/coder/coderd/httpapi"
 	"github.com/coder/coder/coderd/httpmw"
+	"github.com/coder/coder/coderd/turnconn"
 	"github.com/coder/coder/codersdk"
 	"github.com/coder/coder/site"
 )
 
 // Options are requires parameters for Coder to start.
 type Options struct {
+	AccessURL *url.URL
+	Logger    slog.Logger
+	Database  database.Store
+	Pubsub    database.Pubsub
+
 	AgentConnectionUpdateFrequency time.Duration
-	AccessURL                      *url.URL
-	Logger                         slog.Logger
-	Database                       database.Store
-	Pubsub                         database.Pubsub
-
-	AWSCertificates      awsidentity.Certificates
-	GoogleTokenValidator *idtoken.Validator
-
-	SecureAuthCookie   bool
-	SSHKeygenAlgorithm gitsshkey.Algorithm
+	AWSCertificates                awsidentity.Certificates
+	GoogleTokenValidator           *idtoken.Validator
+	ICEServers                     []webrtc.ICEServer
+	SecureAuthCookie               bool
+	SSHKeygenAlgorithm             gitsshkey.Algorithm
+	TURNServer                     *turnconn.Server
 }
 
 // New constructs the Coder API into an HTTP handler.
@@ -175,6 +178,8 @@ func New(options *Options) (http.Handler, func()) {
 				r.Use(httpmw.ExtractWorkspaceAgent(options.Database))
 				r.Get("/", api.workspaceAgentListen)
 				r.Get("/gitsshkey", api.agentGitSSHKey)
+				r.Get("/turn", api.workspaceAgentTurn)
+				r.Get("/iceservers", api.workspaceAgentICEServers)
 			})
 			r.Route("/{workspaceagent}", func(r chi.Router) {
 				r.Use(
@@ -183,6 +188,8 @@ func New(options *Options) (http.Handler, func()) {
 				)
 				r.Get("/", api.workspaceAgent)
 				r.Get("/dial", api.workspaceAgentDial)
+				r.Get("/turn", api.workspaceAgentTurn)
+				r.Get("/iceservers", api.workspaceAgentICEServers)
 			})
 		})
 		r.Route("/workspaceresources/{workspaceresource}", func(r chi.Router) {

coderd/coderdtest/coderdtest.go

@@ -39,6 +39,7 @@ import (
 	"github.com/coder/coder/coderd/database/databasefake"
 	"github.com/coder/coder/coderd/database/postgres"
 	"github.com/coder/coder/coderd/gitsshkey"
+	"github.com/coder/coder/coderd/turnconn"
 	"github.com/coder/coder/codersdk"
 	"github.com/coder/coder/cryptorand"
 	"github.com/coder/coder/provisioner/echo"
@@ -91,9 +92,8 @@ func New(t *testing.T, options *Options) *codersdk.Client {
 	}
 
 	srv := httptest.NewUnstartedServer(nil)
+	ctx, cancelFunc := context.WithCancel(context.Background())
 	srv.Config.BaseContext = func(_ net.Listener) context.Context {
-		ctx, cancelFunc := context.WithCancel(context.Background())
-		t.Cleanup(cancelFunc)
 		return ctx
 	}
 	srv.Start()
@@ -106,6 +106,9 @@ func New(t *testing.T, options *Options) *codersdk.Client {
 		options.SSHKeygenAlgorithm = gitsshkey.AlgorithmEd25519
 	}
 
+	turnServer, err := turnconn.New(nil)
+	require.NoError(t, err)
+
 	// We set the handler after server creation for the access URL.
 	srv.Config.Handler, closeWait = coderd.New(&coderd.Options{
 		AgentConnectionUpdateFrequency: 150 * time.Millisecond,
@@ -117,8 +120,11 @@ func New(t *testing.T, options *Options) *codersdk.Client {
 		AWSCertificates:      options.AWSInstanceIdentity,
 		GoogleTokenValidator: options.GoogleInstanceIdentity,
 		SSHKeygenAlgorithm:   options.SSHKeygenAlgorithm,
+		TURNServer:           turnServer,
 	})
 	t.Cleanup(func() {
+		cancelFunc()
+		_ = turnServer.Close()
 		srv.Close()
 		closeWait()
 	})

coderd/httpmw/workspaceagent.go

@@ -38,7 +38,7 @@ func ExtractWorkspaceAgent(db database.Store) func(http.Handler) http.Handler {
 			token, err := uuid.Parse(cookie.Value)
 			if err != nil {
 				httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
-					Message: fmt.Sprintf("parse token: %s", err),
+					Message: fmt.Sprintf("parse token %q: %s", cookie.Value, err),
 				})
 				return
 			}

coderd/turnconn/turnconn.go

@@ -0,0 +1,203 @@
+package turnconn
+
+import (
+	"io"
+	"net"
+	"sync"
+
+	"github.com/pion/logging"
+	"github.com/pion/turn/v2"
+	"github.com/pion/webrtc/v3"
+	"golang.org/x/net/proxy"
+	"golang.org/x/xerrors"
+)
+
+var (
+	// reservedAddress is a magic address that's used exclusively
+	// for proxying via Coder. We don't proxy all TURN connections,
+	// because that'd exclude the possibility of a customer using
+	// their own TURN server.
+	reservedAddress = "127.0.0.1:12345"
+	credential      = "coder"
+	localhost       = &net.TCPAddr{
+		IP: net.IPv4(127, 0, 0, 1),
+	}
+
+	// Proxy is a an ICE Server that uses a special hostname
+	// to indicate traffic should be proxied.
+	Proxy = webrtc.ICEServer{
+		URLs:       []string{"turns:" + reservedAddress},
+		Username:   "coder",
+		Credential: credential,
+	}
+)
+
+// New constructs a new TURN server binding to the relay address provided.
+// The relay address is used to broadcast the location of an accepted connection.
+func New(relayAddress *turn.RelayAddressGeneratorStatic) (*Server, error) {
+	if relayAddress == nil {
+		relayAddress = &turn.RelayAddressGeneratorStatic{
+			RelayAddress: localhost.IP,
+			Address:      "127.0.0.1",
+		}
+	}
+	logger := logging.NewDefaultLoggerFactory()
+	logger.DefaultLogLevel = logging.LogLevelDebug
+	server := &Server{
+		conns:  make(chan net.Conn, 1),
+		closed: make(chan struct{}),
+	}
+	server.listener = &listener{
+		srv: server,
+	}
+	var err error
+	server.turn, err = turn.NewServer(turn.ServerConfig{
+		AuthHandler: func(username, realm string, srcAddr net.Addr) (key []byte, ok bool) {
+			// TURN connections require credentials. It's not important
+			// for our use-case, because our listener is entirely in-memory.
+			return turn.GenerateAuthKey(Proxy.Username, "", credential), true
+		},
+		ListenerConfigs: []turn.ListenerConfig{{
+			Listener:              server.listener,
+			RelayAddressGenerator: relayAddress,
+		}},
+		LoggerFactory: logger,
+	})
+	if err != nil {
+		return nil, xerrors.Errorf("create server: %w", err)
+	}
+
+	return server, nil
+}
+
+// Server accepts and connects TURN allocations.
+//
+// This is a thin wrapper around pion/turn that pipes
+// connections directly to the in-memory handler.
+type Server struct {
+	listener *listener
+	turn     *turn.Server
+
+	closeMutex sync.Mutex
+	closed     chan (struct{})
+	conns      chan (net.Conn)
+}
+
+// Accept consumes a new connection into the TURN server.
+// A unique remote address must exist per-connection.
+// pion/turn indexes allocations based on the address.
+func (s *Server) Accept(nc net.Conn, remoteAddress, localAddress *net.TCPAddr) *Conn {
+	if localAddress == nil {
+		localAddress = localhost
+	}
+	conn := &Conn{
+		Conn:          nc,
+		remoteAddress: remoteAddress,
+		localAddress:  localAddress,
+		closed:        make(chan struct{}),
+	}
+	s.conns <- conn
+	return conn
+}
+
+// Close ends the TURN server.
+func (s *Server) Close() error {
+	s.closeMutex.Lock()
+	defer s.closeMutex.Unlock()
+	if s.isClosed() {
+		return nil
+	}
+	err := s.turn.Close()
+	close(s.conns)
+	close(s.closed)
+	return err
+}
+
+func (s *Server) isClosed() bool {
+	select {
+	case <-s.closed:
+		return true
+	default:
+		return false
+	}
+}
+
+// listener implements net.Listener for the TURN
+// server to consume.
+type listener struct {
+	srv *Server
+}
+
+func (l *listener) Accept() (net.Conn, error) {
+	conn, ok := <-l.srv.conns
+	if !ok {
+		return nil, io.EOF
+	}
+	return conn, nil
+}
+
+func (*listener) Close() error {
+	return nil
+}
+
+func (*listener) Addr() net.Addr {
+	return nil
+}
+
+type Conn struct {
+	net.Conn
+	closed        chan struct{}
+	localAddress  *net.TCPAddr
+	remoteAddress *net.TCPAddr
+}
+
+func (c *Conn) LocalAddr() net.Addr {
+	return c.localAddress
+}
+
+func (c *Conn) RemoteAddr() net.Addr {
+	return c.remoteAddress
+}
+
+// Closed returns a channel which is closed when
+// the connection is.
+func (c *Conn) Closed() <-chan struct{} {
+	return c.closed
+}
+
+func (c *Conn) Close() error {
+	err := c.Conn.Close()
+	select {
+	case <-c.closed:
+	default:
+		close(c.closed)
+	}
+	return err
+}
+
+type dialer func(network, addr string) (c net.Conn, err error)
+
+func (d dialer) Dial(network, addr string) (c net.Conn, err error) {
+	return d(network, addr)
+}
+
+// ProxyDialer accepts a proxy function that's called when the connection
+// address matches the reserved host in the "Proxy" ICE server.
+//
+// This should be passed to WebRTC connections as an ICE dialer.
+func ProxyDialer(proxyFunc func() (c net.Conn, err error)) proxy.Dialer {
+	return dialer(func(network, addr string) (net.Conn, error) {
+		if addr != reservedAddress {
+			return proxy.Direct.Dial(network, addr)
+		}
+		netConn, err := proxyFunc()
+		if err != nil {
+			return nil, err
+		}
+		return &Conn{
+			localAddress: localhost,
+			closed:       make(chan struct{}),
+			Conn:         netConn,
+		}, nil
+	})
+}

coderd/turnconn/turnconn_test.go

@@ -0,0 +1,106 @@
+package turnconn_test
+
+import (
+	"net"
+	"sync"
+	"testing"
+
+	"github.com/pion/webrtc/v3"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+
+	"cdr.dev/slog"
+	"cdr.dev/slog/sloggers/slogtest"
+	"github.com/coder/coder/coderd/turnconn"
+	"github.com/coder/coder/peer"
+)
+
+func TestMain(m *testing.M) {
+	goleak.VerifyTestMain(m)
+}
+
+func TestTURNConn(t *testing.T) {
+	t.Parallel()
+	turnServer, err := turnconn.New(nil)
+	require.NoError(t, err)
+	defer turnServer.Close()
+
+	logger := slogtest.Make(t, nil).Leveled(slog.LevelDebug)
+
+	clientDialer, clientTURN := net.Pipe()
+	turnServer.Accept(clientTURN, &net.TCPAddr{
+		IP:   net.IPv4(127, 0, 0, 1),
+		Port: 16000,
+	}, nil)
+	require.NoError(t, err)
+	clientSettings := webrtc.SettingEngine{}
+	clientSettings.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeTCP4, webrtc.NetworkTypeTCP6})
+	clientSettings.SetRelayAcceptanceMinWait(0)
+	clientSettings.SetICEProxyDialer(turnconn.ProxyDialer(func() (net.Conn, error) {
+		return clientDialer, nil
+	}))
+	client, err := peer.Client([]webrtc.ICEServer{turnconn.Proxy}, &peer.ConnOptions{
+		SettingEngine: clientSettings,
+		Logger:        logger.Named("client"),
+	})
+	require.NoError(t, err)
+
+	serverDialer, serverTURN := net.Pipe()
+	turnServer.Accept(serverTURN, &net.TCPAddr{
+		IP:   net.IPv4(127, 0, 0, 1),
+		Port: 16001,
+	}, nil)
+	require.NoError(t, err)
+	serverSettings := webrtc.SettingEngine{}
+	serverSettings.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeTCP4, webrtc.NetworkTypeTCP6})
+	serverSettings.SetRelayAcceptanceMinWait(0)
+	serverSettings.SetICEProxyDialer(turnconn.ProxyDialer(func() (net.Conn, error) {
+		return serverDialer, nil
+	}))
+	server, err := peer.Server([]webrtc.ICEServer{turnconn.Proxy}, &peer.ConnOptions{
+		SettingEngine: serverSettings,
+		Logger:        logger.Named("server"),
+	})
+	require.NoError(t, err)
+	exchange(t, client, server)
+
+	_, err = client.Ping()
+	require.NoError(t, err)
+}
+
+func exchange(t *testing.T, client, server *peer.Conn) {
+	var wg sync.WaitGroup
+	wg.Add(2)
+	t.Cleanup(func() {
+		_ = client.Close()
+		_ = server.Close()
+
+		wg.Wait()
+	})
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case c := <-server.LocalCandidate():
+				client.AddRemoteCandidate(c)
+			case c := <-server.LocalSessionDescription():
+				client.SetRemoteSessionDescription(c)
+			case <-server.Closed():
+				return
+			}
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case c := <-client.LocalCandidate():
+				server.AddRemoteCandidate(c)
+			case c := <-client.LocalSessionDescription():
+				server.SetRemoteSessionDescription(c)
+			case <-client.Closed():
+				return
+			}
+		}
+	}()
+}

coderd/workspaceagents.go

@@ -5,7 +5,9 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
+	"strconv"
 	"time"
 
 	"github.com/hashicorp/yamux"
@@ -219,6 +221,59 @@ func (api *api) workspaceAgentListen(rw http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (api *api) workspaceAgentICEServers(rw http.ResponseWriter, _ *http.Request) {
+	httpapi.Write(rw, http.StatusOK, api.ICEServers)
+}
+
+// workspaceAgentTurn proxies a WebSocket connection to the TURN server.
+func (api *api) workspaceAgentTurn(rw http.ResponseWriter, r *http.Request) {
+	api.websocketWaitMutex.Lock()
+	api.websocketWaitGroup.Add(1)
+	api.websocketWaitMutex.Unlock()
+	defer api.websocketWaitGroup.Done()
+
+	localAddress, _ := r.Context().Value(http.LocalAddrContextKey).(*net.TCPAddr)
+	remoteAddress := &net.TCPAddr{
+		IP: net.ParseIP(r.RemoteAddr),
+	}
+	// By default requests have the remote address and port.
+	host, port, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
+			Message: fmt.Sprintf("get remote address: %s", err),
+		})
+		return
+	}
+	remoteAddress.IP = net.ParseIP(host)
+	remoteAddress.Port, err = strconv.Atoi(port)
+	if err != nil {
+		httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
+			Message: fmt.Sprintf("remote address %q has no parsable port: %s", r.RemoteAddr, err),
+		})
+		return
+	}
+
+	wsConn, err := websocket.Accept(rw, r, &websocket.AcceptOptions{
+		CompressionMode: websocket.CompressionDisabled,
+	})
+	if err != nil {
+		httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
+			Message: fmt.Sprintf("accept websocket: %s", err),
+		})
+		return
+	}
+	defer func() {
+		_ = wsConn.Close(websocket.StatusNormalClosure, "")
+	}()
+	netConn := websocket.NetConn(r.Context(), wsConn, websocket.MessageBinary)
+	api.Logger.Debug(r.Context(), "accepting turn connection", slog.F("remote-address", r.RemoteAddr), slog.F("local-address", localAddress))
+	select {
+	case <-api.TURNServer.Accept(netConn, remoteAddress, localAddress).Closed():
+	case <-r.Context().Done():
+	}
+	api.Logger.Debug(r.Context(), "completed turn connection", slog.F("remote-address", r.RemoteAddr), slog.F("local-address", localAddress))
+}
+
 func convertWorkspaceAgent(dbAgent database.WorkspaceAgent, agentUpdateFrequency time.Duration) (codersdk.WorkspaceAgent, error) {
 	var envs map[string]string
 	if dbAgent.EnvironmentVariables.Valid {

coderd/workspaceagents_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
+	"github.com/pion/webrtc/v3"
 	"github.com/stretchr/testify/require"
 
 	"cdr.dev/slog"
@@ -89,16 +90,65 @@ func TestWorkspaceAgentListen(t *testing.T) {
 
 	agentClient := codersdk.New(client.URL)
 	agentClient.SessionToken = authToken
-	agentCloser := agent.New(agentClient.ListenWorkspaceAgent, &peer.ConnOptions{
-		Logger: slogtest.Make(t, nil),
-	})
+	agentCloser := agent.New(agentClient.ListenWorkspaceAgent, slogtest.Make(t, nil).Named("agent").Leveled(slog.LevelDebug))
 	t.Cleanup(func() {
 		_ = agentCloser.Close()
 	})
 	resources := coderdtest.AwaitWorkspaceAgents(t, client, workspace.LatestBuild.ID)
-	conn, err := client.DialWorkspaceAgent(context.Background(), resources[0].Agents[0].ID, nil, &peer.ConnOptions{
-		Logger: slogtest.Make(t, nil).Named("client").Leveled(slog.LevelDebug),
-	})
+	conn, err := client.DialWorkspaceAgent(context.Background(), resources[0].Agents[0].ID, nil)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = conn.Close()
+	})
+	_, err = conn.Ping()
+	require.NoError(t, err)
+}
+
+func TestWorkspaceAgentTURN(t *testing.T) {
+	t.Parallel()
+	client := coderdtest.New(t, nil)
+	user := coderdtest.CreateFirstUser(t, client)
+	daemonCloser := coderdtest.NewProvisionerDaemon(t, client)
+	authToken := uuid.NewString()
+	version := coderdtest.CreateTemplateVersion(t, client, user.OrganizationID, &echo.Responses{
+		Parse:           echo.ParseComplete,
+		ProvisionDryRun: echo.ProvisionComplete,
+		Provision: []*proto.Provision_Response{{
+			Type: &proto.Provision_Response_Complete{
+				Complete: &proto.Provision_Complete{
+					Resources: []*proto.Resource{{
+						Name: "example",
+						Type: "aws_instance",
+						Agents: []*proto.Agent{{
+							Id: uuid.NewString(),
+							Auth: &proto.Agent_Token{
+								Token: authToken,
+							},
+						}},
+					}},
+				},
+			},
+		}},
+	})
+	template := coderdtest.CreateTemplate(t, client, user.OrganizationID, version.ID)
+	coderdtest.AwaitTemplateVersionJob(t, client, version.ID)
+	workspace := coderdtest.CreateWorkspace(t, client, codersdk.Me, template.ID)
+	coderdtest.AwaitWorkspaceBuildJob(t, client, workspace.LatestBuild.ID)
+	daemonCloser.Close()
+
+	agentClient := codersdk.New(client.URL)
+	agentClient.SessionToken = authToken
+	agentCloser := agent.New(agentClient.ListenWorkspaceAgent, slogtest.Make(t, nil))
+	t.Cleanup(func() {
+		_ = agentCloser.Close()
+	})
+	resources := coderdtest.AwaitWorkspaceAgents(t, client, workspace.LatestBuild.ID)
+	opts := &peer.ConnOptions{
+		Logger: slogtest.Make(t, nil).Named("client"),
+	}
+	// Force a TURN connection!
+	opts.SettingEngine.SetNetworkTypes([]webrtc.NetworkType{webrtc.NetworkTypeTCP4})
+	conn, err := client.DialWorkspaceAgent(context.Background(), resources[0].Agents[0].ID, opts)
 	require.NoError(t, err)
 	t.Cleanup(func() {
 		_ = conn.Close()