Release 2021.2.5

It also fixes the marshelling of CIDR into JSON since otherwise
it would show garbled characters as the mask.

CHANGES.md

@@ -1,22 +1,29 @@
-# Experimental: This is a new format for release notes. The format and availability is subject to change.
+**Experimental**: This is a new format for release notes. The format and availability is subject to change.
+
+## 2021.2.5
+
+### New Features
+- We introduce [Cloudflare One Routing](https://developers.cloudflare.com/cloudflare-one/tutorials/warp-to-tunnel) in beta mode, where users can create private networks with RFC 1918 IP addresses on Cloudflare's network. Clients running Cloudflare WARP client in the same organization can then connect to services made available by Argo Tunnel. Please share your feedback in the GitHub issue opened.
+
 
 ## 2021.2.4
 
 ### Bug Fixes
-- [configuration] fixes a case where `cloudflared` failed to read URLs in configuration files
-- [logging] cloudflared now logs reason for failed connections if the error is recoverable
+- Reverts the Improvement released in 2021.2.3 for CLI arguments as it introduced a regression where cloudflared failed to read URLs in configuration files.
+- Cloudflared now logs the reason for failed connections if the error is recoverable.
+
 
 ## 2021.2.3
  
 ### Backward Incompatible Changes
-- [db-connect] Removes db-connect from `cloudflared`. The Cloudflare Workers product will continue to support db-connect implementations with versions of `cloudflared` that predate this release and include support for db-connect.
+- Removes db-connect. The Cloudflare Workers product will continue to support db-connect implementations with versions of cloudflared that predate this release and include support for db-connect.
  
 ### New Features
-- [TCP connects] Introduces support for proxy configurations with websockets in arbitrary TCP connections (#318)
+- Introduces support for proxy configurations with websockets in arbitrary TCP connections (#318).
  
 ### Improvements
-- [command line interface] Nested commands (such as cloudflared tunnel run) now consider CLI arguments even if they appear earlier in the command. E.g. `cloudflared --config config.yaml tunnel run` will now behave similarly to `cloudflared tunnel --config config.yaml run`
+- Nested command line arguments (such as `cloudflared tunnel run`) now consider CLI arguments even if they appear earlier in the command. E.g. `cloudflared --config config.yaml tunnel run` will now behave similarly to `cloudflared tunnel --config config.yaml run`
  
 ### Bug Fixes
-- [dns-proxy] The maximum number of upstream connections is now limited by default which should fix reported issues of cloudflared exhausting CPU usage when faced with connectivity issues.
+- The maximum number of upstream connections is now limited by default which should fix reported issues of cloudflared exhausting CPU usage when faced with connectivity issues.
 

RELEASE_NOTES

@@ -1,3 +1,21 @@
+2021.2.5
+- 2021-02-23 Publish change notes for 2021.2.5
+- 2021-02-11 TUN-3838: ResponseWriter no longer reads and origin error tests
+- 2021-02-10 TUN-3895: Tests for socks stream handler
+- 2021-02-19 TUN-3939: Add logging that shows that Warp-routing is enabled
+- 2021-02-02 TUN-3817: Adds tests for websocket based streaming regression
+- 2021-02-04 TUN-3799: extended the Stream interface to take a logger and added debug logs for io.Copy errors
+- 2021-02-03 TUN-3855: Add ability to override target of 'access ssh' command to a different host for testing
+- 2021-02-04 TUN-3853: Respond with ws headers from the origin service rather than generating our own
+- 2021-02-08 TUN-3889: Move host header override logic to httpService
+- 2021-02-05 TUN-3868: Refactor singleTCPService and bridgeService to tcpOverWSService and rawTCPService
+- 2021-01-21 TUN-3753: Select http2 protocol when warp routing is enabled
+- 2021-01-26 TUN-3809: Allow routes ip show to output as JSON or YAML
+- 2021-01-11 TUN-3615: added support to  proxy tcp streams
+- 2021-01-17 TUN-3725: Warp-routing is independent of ingress
+- 2021-01-15 TUN-3764: Actively flush data for TCP streams
+- 2020-12-09 TUN-3617: Separate service from client, and implement different client for http vs. tcp origins
+
 2021.2.4
 - 2021-02-22 TUN-3948: Log error when retrying connection
 - 2021-02-23 TUN-3964: Revert "TUN-3922: Repoint urfave/cli/v2 library at patched branch at github.com/ipostelnik/cli/v2@fixed which correctly handles reading flags declared at multiple levels of subcommands."

carrier/carrier.go

@@ -4,6 +4,7 @@
 package carrier
 
 import (
+	"crypto/tls"
 	"io"
 	"net"
 	"net/http"
@@ -20,8 +21,10 @@ import (
 const LogFieldOriginURL = "originURL"
 
 type StartOptions struct {
-	OriginURL string
-	Headers   http.Header
+	OriginURL       string
+	Headers         http.Header
+	Host            string
+	TLSClientConfig *tls.Config
 }
 
 // Connection wraps up all the needed functions to forward over the tunnel

carrier/carrier_test.go

@@ -44,7 +44,7 @@ func (s *testStreamer) Write(p []byte) (int, error) {
 func TestStartClient(t *testing.T) {
 	message := "Good morning Austin! Time for another sunny day in the great state of Texas."
 	log := zerolog.Nop()
-	wsConn := NewWSConnection(&log, false)
+	wsConn := NewWSConnection(&log)
 	ts := newTestWebSocketServer()
 	defer ts.Close()
 
@@ -70,7 +70,7 @@ func TestStartServer(t *testing.T) {
 	message := "Good morning Austin! Time for another sunny day in the great state of Texas."
 	log := zerolog.Nop()
 	shutdownC := make(chan struct{})
-	wsConn := NewWSConnection(&log, false)
+	wsConn := NewWSConnection(&log)
 	ts := newTestWebSocketServer()
 	defer ts.Close()
 	options := &StartOptions{

carrier/websocket.go

@@ -8,6 +8,7 @@ import (
 	"net/http/httputil"
 
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/token"
+	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/socks"
 	cfwebsocket "github.com/cloudflare/cloudflared/websocket"
 
@@ -23,7 +24,7 @@ type Websocket struct {
 }
 
 type wsdialer struct {
-	conn *cfwebsocket.Conn
+	conn *cfwebsocket.GorillaConn
 }
 
 func (d *wsdialer) Dial(address string) (io.ReadWriteCloser, *socks.AddrSpec, error) {
@@ -37,10 +38,9 @@ func (d *wsdialer) Dial(address string) (io.ReadWriteCloser, *socks.AddrSpec, er
 }
 
 // NewWSConnection returns a new connection object
-func NewWSConnection(log *zerolog.Logger, isSocks bool) Connection {
+func NewWSConnection(log *zerolog.Logger) Connection {
 	return &Websocket{
-		log:     log,
-		isSocks: isSocks,
+		log: log,
 	}
 }
 
@@ -54,38 +54,36 @@ func (ws *Websocket) ServeStream(options *StartOptions, conn io.ReadWriter) erro
 	}
 	defer wsConn.Close()
 
-	if ws.isSocks {
-		dialer := &wsdialer{conn: wsConn}
-		requestHandler := socks.NewRequestHandler(dialer)
-		socksServer := socks.NewConnectionHandler(requestHandler)
-
-		_ = socksServer.Serve(conn)
-	} else {
-		cfwebsocket.Stream(wsConn, conn)
-	}
+	ingress.Stream(wsConn, conn, ws.log)
 	return nil
 }
 
 // StartServer creates a Websocket server to listen for connections.
 // This is used on the origin (tunnel) side to take data from the muxer and send it to the origin
 func (ws *Websocket) StartServer(listener net.Listener, remote string, shutdownC <-chan struct{}) error {
-	return cfwebsocket.StartProxyServer(ws.log, listener, remote, shutdownC, cfwebsocket.DefaultStreamHandler)
+	return cfwebsocket.StartProxyServer(ws.log, listener, remote, shutdownC, ingress.DefaultStreamHandler)
 }
 
 // createWebsocketStream will create a WebSocket connection to stream data over
 // It also handles redirects from Access and will present that flow if
 // the token is not present on the request
-func createWebsocketStream(options *StartOptions, log *zerolog.Logger) (*cfwebsocket.Conn, error) {
+func createWebsocketStream(options *StartOptions, log *zerolog.Logger) (*cfwebsocket.GorillaConn, error) {
 	req, err := http.NewRequest(http.MethodGet, options.OriginURL, nil)
 	if err != nil {
 		return nil, err
 	}
 	req.Header = options.Headers
+	if options.Host != "" {
+		req.Host = options.Host
+	}
 
 	dump, err := httputil.DumpRequest(req, false)
 	log.Debug().Msgf("Websocket request: %s", string(dump))
 
-	wsConn, resp, err := cfwebsocket.ClientConnect(req, nil)
+	dialer := &websocket.Dialer{
+		TLSClientConfig: options.TLSClientConfig,
+	}
+	wsConn, resp, err := cfwebsocket.ClientConnect(req, dialer)
 	defer closeRespBody(resp)
 
 	if err != nil && IsAccessResponse(resp) {
@@ -97,7 +95,7 @@ func createWebsocketStream(options *StartOptions, log *zerolog.Logger) (*cfwebso
 		return nil, err
 	}
 
-	return &cfwebsocket.Conn{Conn: wsConn}, nil
+	return &cfwebsocket.GorillaConn{Conn: wsConn}, nil
 }
 
 // createAccessAuthenticatedStream will try load a token from storage and make

cmd/cloudflared/access/carrier.go

@@ -1,6 +1,8 @@
 package access
 
 import (
+	"crypto/tls"
+	"fmt"
 	"net/http"
 	"strings"
 
@@ -46,7 +48,7 @@ func StartForwarder(forwarder config.Forwarder, shutdown <-chan struct{}, log *z
 	}
 
 	// we could add a cmd line variable for this bool if we want the SOCK5 server to be on the client side
-	wsConn := carrier.NewWSConnection(log, false)
+	wsConn := carrier.NewWSConnection(log)
 
 	log.Info().Str(LogFieldHost, validURL.Host).Msg("Start Websocket listener")
 	return carrier.StartForwarder(wsConn, validURL.Host, shutdown, options)
@@ -84,10 +86,30 @@ func ssh(c *cli.Context) error {
 	options := &carrier.StartOptions{
 		OriginURL: originURL,
 		Headers:   headers,
+		Host:      hostname,
+	}
+
+	if connectTo := c.String(sshConnectTo); connectTo != "" {
+		parts := strings.Split(connectTo, ":")
+		switch len(parts) {
+		case 1:
+			options.OriginURL = fmt.Sprintf("https://%s", parts[0])
+		case 2:
+			options.OriginURL = fmt.Sprintf("https://%s:%s", parts[0], parts[1])
+		case 3:
+			options.OriginURL = fmt.Sprintf("https://%s:%s", parts[2], parts[1])
+			options.TLSClientConfig = &tls.Config{
+				InsecureSkipVerify: true,
+				ServerName:         parts[0],
+			}
+			log.Warn().Msgf("Using insecure SSL connection because SNI overridden to %s", parts[0])
+		default:
+			return fmt.Errorf("invalid connection override: %s", connectTo)
+		}
 	}
 
 	// we could add a cmd line variable for this bool if we want the SOCK5 server to be on the client side
-	wsConn := carrier.NewWSConnection(log, false)
+	wsConn := carrier.NewWSConnection(log)
 
 	if c.NArg() > 0 || c.IsSet(sshURLFlag) {
 		forwarder, err := config.ValidateUrl(c, true)
@@ -95,7 +117,6 @@ func ssh(c *cli.Context) error {
 			log.Err(err).Msg("Error validating origin URL")
 			return errors.Wrap(err, "error validating origin URL")
 		}
-
 		log.Info().Str(LogFieldHost, forwarder.Host).Msg("Start Websocket listener")
 		err = carrier.StartForwarder(wsConn, forwarder.Host, shutdownC, options)
 		if err != nil {

cmd/cloudflared/access/cmd.go

@@ -33,6 +33,7 @@ const (
 	sshTokenIDFlag     = "service-token-id"
 	sshTokenSecretFlag = "service-token-secret"
 	sshGenCertFlag     = "short-lived-cert"
+	sshConnectTo       = "connect-to"
 	sshConfigTemplate  = `
 Add to your {{.Home}}/.ssh/config:
 
@@ -54,7 +55,7 @@ Host cfpipe-{{.Hostname}}
 const sentryDSN = "https://56a9c9fa5c364ab28f34b14f35ea0f1b@sentry.io/189878"
 
 var (
-	shutdownC      chan struct{}
+	shutdownC chan struct{}
 )
 
 // Init will initialize and store vars from the main program
@@ -164,6 +165,11 @@ func Commands() []*cli.Command {
 							Aliases: []string{"loglevel"}, //added to match the tunnel side
 							Usage:   "Application logging level {fatal, error, info, debug}. ",
 						},
+						&cli.StringFlag{
+							Name:  sshConnectTo,
+							Hidden: true,
+							Usage: "Connect to alternate location for testing, value is host, host:port, or sni:port:host",
+						},
 					},
 				},
 				{

cmd/cloudflared/config/configuration.go

@@ -226,10 +226,15 @@ type OriginRequestConfig struct {
 type Configuration struct {
 	TunnelID      string `yaml:"tunnel"`
 	Ingress       []UnvalidatedIngressRule
+	WarpRouting   WarpRoutingConfig   `yaml:"warp-routing"`
 	OriginRequest OriginRequestConfig `yaml:"originRequest"`
 	sourceFile    string
 }
 
+type WarpRoutingConfig struct {
+	Enabled bool `yaml:"enabled"`
+}
+
 type configFileSettings struct {
 	Configuration `yaml:",inline"`
 	// older settings will be aggregated into the generic map, should be read via cli.Context

cmd/cloudflared/config/configuration_test.go

@@ -20,6 +20,9 @@ func TestConfigFileSettings(t *testing.T) {
 			Path:     "",
 			Service:  "https://localhost:8001",
 		}
+		warpRouting = WarpRoutingConfig{
+			Enabled: true,
+		}
 	)
 	rawYAML := `
 tunnel: config-file-test
@@ -29,6 +32,8 @@ ingress:
    service: https://localhost:8000
  - hostname: "*"
    service: https://localhost:8001
+warp-routing: 
+  enabled: true
 retries: 5
 grace-period: 30s
 percentage: 3.14
@@ -47,6 +52,7 @@ counters:
 	assert.Equal(t, "config-file-test", config.TunnelID)
 	assert.Equal(t, firstIngress, config.Ingress[0])
 	assert.Equal(t, secondIngress, config.Ingress[1])
+	assert.Equal(t, warpRouting, config.WarpRouting)
 
 	retries, err := config.Int("retries")
 	assert.NoError(t, err)
@@ -73,4 +79,5 @@ counters:
 	assert.NoError(t, err)
 	assert.Equal(t, 123, counters[0])
 	assert.Equal(t, 456, counters[1])
+
 }

cmd/cloudflared/tunnel/configuration.go

@@ -195,6 +195,7 @@ func prepareTunnelConfig(
 		ingressRules  ingress.Ingress
 		classicTunnel *connection.ClassicTunnelConfig
 	)
+	cfg := config.GetConfiguration()
 	if isNamedTunnel {
 		clientUUID, err := uuid.NewRandom()
 		if err != nil {
@@ -206,7 +207,7 @@ func prepareTunnelConfig(
 			Version:  version,
 			Arch:     fmt.Sprintf("%s_%s", buildInfo.GoOS, buildInfo.GoArch),
 		}
-		ingressRules, err = ingress.ParseIngress(config.GetConfiguration())
+		ingressRules, err = ingress.ParseIngress(cfg)
 		if err != nil && err != ingress.ErrNoIngressRules {
 			return nil, ingress.Ingress{}, err
 		}
@@ -230,7 +231,14 @@ func prepareTunnelConfig(
 		}
 	}
 
-	protocolSelector, err := connection.NewProtocolSelector(c.String("protocol"), namedTunnel, edgediscovery.HTTP2Percentage, origin.ResolveTTL, log)
+	var warpRoutingService *ingress.WarpRoutingService
+	warpRoutingEnabled := isWarpRoutingEnabled(cfg.WarpRouting, isNamedTunnel)
+	if warpRoutingEnabled {
+		warpRoutingService = ingress.NewWarpRoutingService()
+		log.Info().Msgf("Warp-routing is enabled")
+	}
+
+	protocolSelector, err := connection.NewProtocolSelector(c.String("protocol"), warpRoutingEnabled, namedTunnel, edgediscovery.HTTP2Percentage, origin.ResolveTTL, log)
 	if err != nil {
 		return nil, ingress.Ingress{}, err
 	}
@@ -245,9 +253,9 @@ func prepareTunnelConfig(
 		edgeTLSConfigs[p] = edgeTLSConfig
 	}
 
-	originClient := origin.NewClient(ingressRules, tags, log)
+	originProxy := origin.NewOriginProxy(ingressRules, warpRoutingService, tags, log)
 	connectionConfig := &connection.Config{
-		OriginClient:    originClient,
+		OriginProxy:     originProxy,
 		GracePeriod:     c.Duration("grace-period"),
 		ReplaceExisting: c.Bool("force"),
 	}
@@ -286,6 +294,10 @@ func prepareTunnelConfig(
 	}, ingressRules, nil
 }
 
+func isWarpRoutingEnabled(warpConfig config.WarpRoutingConfig, isNamedTunnel bool) bool {
+	return warpConfig.Enabled && isNamedTunnel
+}
+
 func isRunningFromTerminal() bool {
 	return terminal.IsTerminal(int(os.Stdout.Fd()))
 }

cmd/cloudflared/tunnel/teamnet_subcommands.go

@@ -41,7 +41,7 @@ func buildRouteIPSubcommand() *cli.Command {
 				Usage:       "Show the routing table",
 				UsageText:   "cloudflared tunnel [--config FILEPATH] route ip show [flags]",
 				Description: `Shows your organization's private route table. You can use flags to filter the results.`,
-				Flags:       teamnet.FilterFlags,
+				Flags:       showRoutesFlags(),
 			},
 			{
 				Name:        "delete",
@@ -62,6 +62,13 @@ func buildRouteIPSubcommand() *cli.Command {
 	}
 }
 
+func showRoutesFlags() []cli.Flag {
+	flags := make([]cli.Flag, 0)
+	flags = append(flags, teamnet.FilterFlags...)
+	flags = append(flags, outputFormatFlag)
+	return flags
+}
+
 func showRoutesCommand(c *cli.Context) error {
 	sc, err := newSubcommandContext(c)
 	if err != nil {

connection/connection.go

@@ -1,6 +1,7 @@
 package connection
 
 import (
+	"fmt"
 	"io"
 	"net/http"
 	"strconv"
@@ -14,7 +15,7 @@ import (
 const LogFieldConnIndex = "connIndex"
 
 type Config struct {
-	OriginClient    OriginClient
+	OriginProxy     OriginProxy
 	GracePeriod     time.Duration
 	ReplaceExisting bool
 }
@@ -50,14 +51,49 @@ func (c *ClassicTunnelConfig) IsTrialZone() bool {
 	return c.Hostname == ""
 }
 
-type OriginClient interface {
-	Proxy(w ResponseWriter, req *http.Request, isWebsocket bool) error
+// Type indicates the connection type of the  connection.
+type Type int
+
+const (
+	TypeWebsocket Type = iota
+	TypeTCP
+	TypeControlStream
+	TypeHTTP
+)
+
+// ShouldFlush returns whether this kind of connection should actively flush data
+func (t Type) shouldFlush() bool {
+	switch t {
+	case TypeWebsocket, TypeTCP, TypeControlStream:
+		return true
+	default:
+		return false
+	}
+}
+
+func (t Type) String() string {
+	switch t {
+	case TypeWebsocket:
+		return "websocket"
+	case TypeTCP:
+		return "tcp"
+	case TypeControlStream:
+		return "control stream"
+	case TypeHTTP:
+		return "http"
+	default:
+		return fmt.Sprintf("Unknown Type %d", t)
+	}
+}
+
+type OriginProxy interface {
+	// If Proxy returns an error, the caller is responsible for writing the error status to ResponseWriter
+	Proxy(w ResponseWriter, req *http.Request, sourceConnectionType Type) error
 }
 
 type ResponseWriter interface {
-	WriteRespHeaders(*http.Response) error
-	WriteErrorResponse()
-	io.ReadWriter
+	WriteRespHeaders(status int, header http.Header) error
+	io.Writer
 }
 
 type ConnectedFuse interface {

connection/connection_test.go

@@ -19,8 +19,8 @@ const (
 
 var (
 	testConfig = &Config{
-		OriginClient: &mockOriginClient{},
-		GracePeriod:  time.Millisecond * 100,
+		OriginProxy: &mockOriginProxy{},
+		GracePeriod: time.Millisecond * 100,
 	}
 	log           = zerolog.Nop()
 	testOriginURL = &url.URL{
@@ -38,11 +38,11 @@ type testRequest struct {
 	isProxyError   bool
 }
 
-type mockOriginClient struct {
+type mockOriginProxy struct {
 }
 
-func (moc *mockOriginClient) Proxy(w ResponseWriter, r *http.Request, isWebsocket bool) error {
-	if isWebsocket {
+func (moc *mockOriginProxy) Proxy(w ResponseWriter, r *http.Request, sourceConnectionType Type) error {
+	if sourceConnectionType == TypeWebsocket {
 		return wsEndpoint(w, r)
 	}
 	switch r.URL.Path {
@@ -74,7 +74,7 @@ func wsEndpoint(w ResponseWriter, r *http.Request) error {
 	resp := &http.Response{
 		StatusCode: http.StatusSwitchingProtocols,
 	}
-	_ = w.WriteRespHeaders(resp)
+	_ = w.WriteRespHeaders(resp.StatusCode, resp.Header)
 	clientReader := nowriter{r.Body}
 	go func() {
 		for {
@@ -95,7 +95,7 @@ func originRespEndpoint(w ResponseWriter, status int, data []byte) {
 	resp := &http.Response{
 		StatusCode: status,
 	}
-	_ = w.WriteRespHeaders(resp)
+	_ = w.WriteRespHeaders(resp.StatusCode, resp.Header)
 	_, _ = w.Write(data)
 }
 

connection/h2mux.go

@@ -216,7 +216,12 @@ func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
 		return reqErr
 	}
 
-	err := h.config.OriginClient.Proxy(respWriter, req, websocket.IsWebSocketUpgrade(req))
+	var sourceConnectionType = TypeHTTP
+	if websocket.IsWebSocketUpgrade(req) {
+		sourceConnectionType = TypeWebsocket
+	}
+
+	err := h.config.OriginProxy.Proxy(respWriter, req, sourceConnectionType)
 	if err != nil {
 		respWriter.WriteErrorResponse()
 		return err
@@ -240,8 +245,8 @@ type h2muxRespWriter struct {
 	*h2mux.MuxedStream
 }
 
-func (rp *h2muxRespWriter) WriteRespHeaders(resp *http.Response) error {
-	headers := h2mux.H1ResponseToH2ResponseHeaders(resp)
+func (rp *h2muxRespWriter) WriteRespHeaders(status int, header http.Header) error {
+	headers := h2mux.H1ResponseToH2ResponseHeaders(status, header)
 	headers = append(headers, h2mux.Header{Name: ResponseMetaHeaderField, Value: responseMetaHeaderOrigin})
 	return rp.WriteHeaders(headers)
 }

connection/http2.go

@@ -19,6 +19,7 @@ import (
 
 const (
 	internalUpgradeHeader = "Cf-Cloudflared-Proxy-Connection-Upgrade"
+	tcpStreamHeader       = "Cf-Cloudflared-Proxy-Src"
 	websocketUpgrade      = "websocket"
 	controlStreamUpgrade  = "control-stream"
 )
@@ -96,31 +97,24 @@ func (c *http2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	c.activeRequestsWG.Add(1)
 	defer c.activeRequestsWG.Done()
 
-	respWriter := &http2RespWriter{
-		r: r.Body,
-		w: w,
-	}
-	flusher, isFlusher := w.(http.Flusher)
-	if !isFlusher {
-		c.observer.log.Error().Msgf("%T doesn't implement http.Flusher", w)
-		respWriter.WriteErrorResponse()
+	connType := determineHTTP2Type(r)
+	respWriter, err := newHTTP2RespWriter(r, w, connType)
+	if err != nil {
+		c.observer.log.Error().Msg(err.Error())
 		return
 	}
-	respWriter.flusher = flusher
-	var err error
-	if isControlStreamUpgrade(r) {
-		respWriter.shouldFlush = true
-		err = c.serveControlStream(r.Context(), respWriter)
-		c.controlStreamErr = err
-	} else if isWebsocketUpgrade(r) {
-		respWriter.shouldFlush = true
-		stripWebsocketUpgradeHeader(r)
-		err = c.config.OriginClient.Proxy(respWriter, r, true)
-	} else {
-		err = c.config.OriginClient.Proxy(respWriter, r, false)
-	}
 
-	if err != nil {
+	var proxyErr error
+	switch connType {
+	case TypeControlStream:
+		proxyErr = c.serveControlStream(r.Context(), respWriter)
+	case TypeWebsocket:
+		stripWebsocketUpgradeHeader(r)
+		proxyErr = c.config.OriginProxy.Proxy(respWriter, r, TypeWebsocket)
+	default:
+		proxyErr = c.config.OriginProxy.Proxy(respWriter, r, connType)
+	}
+	if proxyErr != nil {
 		respWriter.WriteErrorResponse()
 	}
 }
@@ -161,10 +155,29 @@ type http2RespWriter struct {
 	shouldFlush bool
 }
 
-func (rp *http2RespWriter) WriteRespHeaders(resp *http.Response) error {
+func newHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type) (*http2RespWriter, error) {
+	flusher, isFlusher := w.(http.Flusher)
+	if !isFlusher {
+		respWriter := &http2RespWriter{
+			r: r.Body,
+			w: w,
+		}
+		respWriter.WriteErrorResponse()
+		return nil, fmt.Errorf("%T doesn't implement http.Flusher", w)
+	}
+
+	return &http2RespWriter{
+		r:           r.Body,
+		w:           w,
+		flusher:     flusher,
+		shouldFlush: connType.shouldFlush(),
+	}, nil
+}
+
+func (rp *http2RespWriter) WriteRespHeaders(status int, header http.Header) error {
 	dest := rp.w.Header()
-	userHeaders := make(http.Header, len(resp.Header))
-	for header, values := range resp.Header {
+	userHeaders := make(http.Header, len(header))
+	for header, values := range header {
 		// Since these are http2 headers, they're required to be lowercase
 		h2name := strings.ToLower(header)
 		for _, v := range values {
@@ -184,13 +197,12 @@ func (rp *http2RespWriter) WriteRespHeaders(resp *http.Response) error {
 	// Perform user header serialization and set them in the single header
 	dest.Set(canonicalResponseUserHeadersField, h2mux.SerializeHeaders(userHeaders))
 	rp.setResponseMetaHeader(responseMetaHeaderOrigin)
-	status := resp.StatusCode
 	// HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
 	if status == http.StatusSwitchingProtocols {
 		status = http.StatusOK
 	}
 	rp.w.WriteHeader(status)
-	if IsServerSentEvent(resp.Header) {
+	if IsServerSentEvent(header) {
 		rp.shouldFlush = true
 	}
 	if rp.shouldFlush {
@@ -231,12 +243,30 @@ func (rp *http2RespWriter) Close() error {
 	return nil
 }
 
+func determineHTTP2Type(r *http.Request) Type {
+	switch {
+	case isWebsocketUpgrade(r):
+		return TypeWebsocket
+	case IsTCPStream(r):
+		return TypeTCP
+	case isControlStreamUpgrade(r):
+		return TypeControlStream
+	default:
+		return TypeHTTP
+	}
+}
+
 func isControlStreamUpgrade(r *http.Request) bool {
-	return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == controlStreamUpgrade
+	return r.Header.Get(internalUpgradeHeader) == controlStreamUpgrade
 }
 
 func isWebsocketUpgrade(r *http.Request) bool {
-	return strings.ToLower(r.Header.Get(internalUpgradeHeader)) == websocketUpgrade
+	return r.Header.Get(internalUpgradeHeader) == websocketUpgrade
+}
+
+// IsTCPStream discerns if the connection request needs a tcp stream proxy.
+func IsTCPStream(r *http.Request) bool {
+	return r.Header.Get(tcpStreamHeader) != ""
 }
 
 func stripWebsocketUpgradeHeader(r *http.Request) {

connection/protocol.go

@@ -141,16 +141,29 @@ type PercentageFetcher func() (int32, error)
 
 func NewProtocolSelector(
 	protocolFlag string,
+	warpRoutingEnabled bool,
 	namedTunnel *NamedTunnelConfig,
 	fetchFunc PercentageFetcher,
 	ttl time.Duration,
 	log *zerolog.Logger,
 ) (ProtocolSelector, error) {
+	// Classic tunnel is only supported with h2mux
 	if namedTunnel == nil {
 		return &staticProtocolSelector{
 			current: H2mux,
 		}, nil
 	}
+
+	// warp routing can only be served over http2 connections
+	if warpRoutingEnabled {
+		if protocolFlag == H2mux.String() {
+			log.Warn().Msg("Warp routing is only supported by http2 protocol. Upgrading protocol to http2")
+		}
+		return &staticProtocolSelector{
+			current: HTTP2,
+		}, nil
+	}
+
 	if protocolFlag == H2mux.String() {
 		return &staticProtocolSelector{
 			current: H2mux,

connection/protocol_test.go

@@ -9,7 +9,8 @@ import (
 )
 
 const (
-	testNoTTL = 0
+	testNoTTL            = 0
+	noWarpRoutingEnabled = false
 )
 
 var (
@@ -48,14 +49,15 @@ func (dmf *dynamicMockFetcher) fetch() PercentageFetcher {
 
 func TestNewProtocolSelector(t *testing.T) {
 	tests := []struct {
-		name              string
-		protocol          string
-		expectedProtocol  Protocol
-		hasFallback       bool
-		expectedFallback  Protocol
-		namedTunnelConfig *NamedTunnelConfig
-		fetchFunc         PercentageFetcher
-		wantErr           bool
+		name               string
+		protocol           string
+		expectedProtocol   Protocol
+		hasFallback        bool
+		expectedFallback   Protocol
+		warpRoutingEnabled bool
+		namedTunnelConfig  *NamedTunnelConfig
+		fetchFunc          PercentageFetcher
+		wantErr            bool
 	}{
 		{
 			name:              "classic tunnel",
@@ -108,6 +110,36 @@ func TestNewProtocolSelector(t *testing.T) {
 			fetchFunc:         mockFetcher(100),
 			namedTunnelConfig: testNamedTunnelConfig,
 		},
+		{
+			name:               "warp routing requesting h2mux",
+			protocol:           "h2mux",
+			expectedProtocol:   HTTP2,
+			hasFallback:        false,
+			expectedFallback:   H2mux,
+			fetchFunc:          mockFetcher(100),
+			warpRoutingEnabled: true,
+			namedTunnelConfig:  testNamedTunnelConfig,
+		},
+		{
+			name:               "warp routing http2",
+			protocol:           "http2",
+			expectedProtocol:   HTTP2,
+			hasFallback:        false,
+			expectedFallback:   H2mux,
+			fetchFunc:          mockFetcher(100),
+			warpRoutingEnabled: true,
+			namedTunnelConfig:  testNamedTunnelConfig,
+		},
+		{
+			name:               "warp routing auto",
+			protocol:           "auto",
+			expectedProtocol:   HTTP2,
+			hasFallback:        false,
+			expectedFallback:   H2mux,
+			fetchFunc:          mockFetcher(100),
+			warpRoutingEnabled: true,
+			namedTunnelConfig:  testNamedTunnelConfig,
+		},
 		{
 			// None named tunnel can only use h2mux, so specifying an unknown protocol is not an error
 			name:             "classic tunnel unknown protocol",
@@ -131,7 +163,7 @@ func TestNewProtocolSelector(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		selector, err := NewProtocolSelector(test.protocol, test.namedTunnelConfig, test.fetchFunc, testNoTTL, &log)
+		selector, err := NewProtocolSelector(test.protocol, test.warpRoutingEnabled, test.namedTunnelConfig, test.fetchFunc, testNoTTL, &log)
 		if test.wantErr {
 			assert.Error(t, err, fmt.Sprintf("test %s failed", test.name))
 		} else {
@@ -148,7 +180,7 @@ func TestNewProtocolSelector(t *testing.T) {
 
 func TestAutoProtocolSelectorRefresh(t *testing.T) {
 	fetcher := dynamicMockFetcher{}
-	selector, err := NewProtocolSelector("auto", testNamedTunnelConfig, fetcher.fetch(), testNoTTL, &log)
+	selector, err := NewProtocolSelector("auto", noWarpRoutingEnabled, testNamedTunnelConfig, fetcher.fetch(), testNoTTL, &log)
 	assert.NoError(t, err)
 	assert.Equal(t, H2mux, selector.Current())
 
@@ -177,7 +209,7 @@ func TestAutoProtocolSelectorRefresh(t *testing.T) {
 
 func TestHTTP2ProtocolSelectorRefresh(t *testing.T) {
 	fetcher := dynamicMockFetcher{}
-	selector, err := NewProtocolSelector("http2", testNamedTunnelConfig, fetcher.fetch(), testNoTTL, &log)
+	selector, err := NewProtocolSelector("http2", noWarpRoutingEnabled, testNamedTunnelConfig, fetcher.fetch(), testNoTTL, &log)
 	assert.NoError(t, err)
 	assert.Equal(t, HTTP2, selector.Current())
 
@@ -206,7 +238,7 @@ func TestHTTP2ProtocolSelectorRefresh(t *testing.T) {
 
 func TestProtocolSelectorRefreshTTL(t *testing.T) {
 	fetcher := dynamicMockFetcher{percentage: 100}
-	selector, err := NewProtocolSelector("auto", testNamedTunnelConfig, fetcher.fetch(), time.Hour, &log)
+	selector, err := NewProtocolSelector("auto", noWarpRoutingEnabled, testNamedTunnelConfig, fetcher.fetch(), time.Hour, &log)
 	assert.NoError(t, err)
 	assert.Equal(t, HTTP2, selector.Current())
 

h2mux/header.go

@@ -125,12 +125,12 @@ func IsWebsocketClientHeader(headerName string) bool {
 		headerName == "upgrade"
 }
 
-func H1ResponseToH2ResponseHeaders(h1 *http.Response) (h2 []Header) {
+func H1ResponseToH2ResponseHeaders(status int, h1 http.Header) (h2 []Header) {
 	h2 = []Header{
-		{Name: ":status", Value: strconv.Itoa(h1.StatusCode)},
+		{Name: ":status", Value: strconv.Itoa(status)},
 	}
-	userHeaders := make(http.Header, len(h1.Header))
-	for header, values := range h1.Header {
+	userHeaders := make(http.Header, len(h1))
+	for header, values := range h1 {
 		h2name := strings.ToLower(header)
 		if h2name == "content-length" {
 			// This header has meaning in HTTP/2 and will be used by the edge,

h2mux/header_test.go

@@ -579,7 +579,7 @@ func TestH1ResponseToH2ResponseHeaders(t *testing.T) {
 		Header:     mockHeaders,
 	}
 
-	headers := H1ResponseToH2ResponseHeaders(&mockResponse)
+	headers := H1ResponseToH2ResponseHeaders(mockResponse.StatusCode, mockResponse.Header)
 
 	serializedHeadersIndex := -1
 	for i, header := range headers {
@@ -622,7 +622,7 @@ func TestHeaderSize(t *testing.T) {
 		Header:     largeHeaders,
 	}
 
-	serializedHeaders := H1ResponseToH2ResponseHeaders(&mockResponse)
+	serializedHeaders := H1ResponseToH2ResponseHeaders(mockResponse.StatusCode, mockResponse.Header)
 	request, err := http.NewRequest(http.MethodGet, "https://example.com/", nil)
 	assert.NoError(t, err)
 	for _, header := range serializedHeaders {
@@ -669,6 +669,6 @@ func BenchmarkH1ResponseToH2ResponseHeaders(b *testing.B) {
 	b.ReportAllocs()
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
-		_ = H1ResponseToH2ResponseHeaders(h1resp)
+		_ = H1ResponseToH2ResponseHeaders(h1resp.StatusCode, h1resp.Header)
 	}
 }

ingress/ingress.go

@@ -24,6 +24,11 @@ var (
 	ErrURLIncompatibleWithIngress = errors.New("You can't set the --url flag (or $TUNNEL_URL) when using multiple-origin ingress rules")
 )
 
+const (
+	ServiceBastion     = "bastion"
+	ServiceWarpRouting = "warp-routing"
+)
+
 // FindMatchingRule returns the index of the Ingress Rule which matches the given
 // hostname and path. This function assumes the last rule matches everything,
 // which is the case if the rules were instantiated via the ingress#Validate method
@@ -38,6 +43,7 @@ func (ing Ingress) FindMatchingRule(hostname, path string) (*Rule, int) {
 			return &rule, i
 		}
 	}
+
 	i := len(ing.Rules) - 1
 	return &ing.Rules[i], i
 }
@@ -84,17 +90,35 @@ func NewSingleOrigin(c *cli.Context, allowURLFromArgs bool) (Ingress, error) {
 	return ing, err
 }
 
+// WarpRoutingService starts a tcp stream between the origin and requests from
+// warp clients.
+type WarpRoutingService struct {
+	Proxy StreamBasedOriginProxy
+}
+
+func NewWarpRoutingService() *WarpRoutingService {
+	return &WarpRoutingService{Proxy: &rawTCPService{name: ServiceWarpRouting}}
+}
+
 // Get a single origin service from the CLI/config.
-func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (OriginService, error) {
+func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (originService, error) {
 	if c.IsSet("hello-world") {
 		return new(helloWorld), nil
 	}
-	if c.IsSet("url") || c.IsSet(config.BastionFlag) {
+	if c.IsSet(config.BastionFlag) {
+		return newBastionService(), nil
+	}
+	if c.IsSet("url") {
 		originURL, err := config.ValidateUrl(c, allowURLFromArgs)
 		if err != nil {
 			return nil, errors.Wrap(err, "Error validating origin URL")
 		}
-		return &localService{URL: originURL, RootURL: originURL}, nil
+		if isHTTPService(originURL) {
+			return &httpService{
+				url: originURL,
+			}, nil
+		}
+		return newTCPOverWSService(originURL), nil
 	}
 	if c.IsSet("unix-socket") {
 		path, err := config.ValidateUnixSocket(c)
@@ -104,7 +128,7 @@ func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (OriginServ
 		return &unixSocketPath{path: path}, nil
 	}
 	u, err := url.Parse("http://localhost:8080")
-	return &localService{URL: u, RootURL: u}, err
+	return &httpService{url: u}, err
 }
 
 // IsEmpty checks if there are any ingress rules.
@@ -136,7 +160,7 @@ func validate(ingress []config.UnvalidatedIngressRule, defaults OriginRequestCon
 	rules := make([]Rule, len(ingress))
 	for i, r := range ingress {
 		cfg := setConfig(defaults, r.OriginRequest)
-		var service OriginService
+		var service originService
 
 		if prefix := "unix:"; strings.HasPrefix(r.Service, prefix) {
 			// No validation necessary for unix socket filepath services
@@ -151,12 +175,12 @@ func validate(ingress []config.UnvalidatedIngressRule, defaults OriginRequestCon
 			service = &srv
 		} else if r.Service == "hello_world" || r.Service == "hello-world" || r.Service == "helloworld" {
 			service = new(helloWorld)
-		} else if r.Service == "bastion" || cfg.BastionMode {
+		} else if r.Service == ServiceBastion || cfg.BastionMode {
 			// Bastion mode will always start a Websocket proxy server, which will
 			// overwrite the localService.URL field when `start` is called. So,
 			// leave the URL field empty for now.
 			cfg.BastionMode = true
-			service = new(localService)
+			service = newBastionService()
 		} else {
 			// Validate URL services
 			u, err := url.Parse(r.Service)
@@ -171,8 +195,11 @@ func validate(ingress []config.UnvalidatedIngressRule, defaults OriginRequestCon
 			if u.Path != "" {
 				return Ingress{}, fmt.Errorf("%s is an invalid address, ingress rules don't support proxying to a different path on the origin service. The path will be the same as the eyeball request's path", r.Service)
 			}
-			serviceURL := localService{URL: u}
-			service = &serviceURL
+			if isHTTPService(u) {
+				service = &httpService{url: u}
+			} else {
+				service = newTCPOverWSService(u)
+			}
 		}
 
 		if err := validateHostname(r, i, len(ingress)); err != nil {
@@ -241,3 +268,7 @@ func ParseIngress(conf *config.Configuration) (Ingress, error) {
 	}
 	return validate(conf.Ingress, originRequestFromYAML(conf.OriginRequest))
 }
+
+func isHTTPService(url *url.URL) bool {
+	return url.Scheme == "http" || url.Scheme == "https" || url.Scheme == "ws" || url.Scheme == "wss"
+}

ingress/ingress_test.go

@@ -3,6 +3,7 @@ package ingress
 import (
 	"flag"
 	"fmt"
+	"net/http"
 	"net/url"
 	"regexp"
 	"testing"
@@ -61,12 +62,12 @@ ingress:
 			want: []Rule{
 				{
 					Hostname: "tunnel1.example.com",
-					Service:  &localService{URL: localhost8000},
+					Service:  &httpService{url: localhost8000},
 					Config:   defaultConfig,
 				},
 				{
 					Hostname: "*",
-					Service:  &localService{URL: localhost8001},
+					Service:  &httpService{url: localhost8001},
 					Config:   defaultConfig,
 				},
 			},
@@ -82,7 +83,22 @@ extraKey: extraValue
 			want: []Rule{
 				{
 					Hostname: "*",
-					Service:  &localService{URL: localhost8000},
+					Service:  &httpService{url: localhost8000},
+					Config:   defaultConfig,
+				},
+			},
+		},
+		{
+			name: "ws service",
+			args: args{rawYAML: `
+ingress:
+ - hostname: "*"
+   service: wss://localhost:8000
+`},
+			want: []Rule{
+				{
+					Hostname: "*",
+					Service:  &httpService{url: MustParseURL(t, "wss://localhost:8000")},
 					Config:   defaultConfig,
 				},
 			},
@@ -95,7 +111,7 @@ ingress:
 `},
 			want: []Rule{
 				{
-					Service: &localService{URL: localhost8000},
+					Service: &httpService{url: localhost8000},
 					Config:  defaultConfig,
 				},
 			},
@@ -209,6 +225,85 @@ ingress:
 				},
 			},
 		},
+		{
+			name: "TCP services",
+			args: args{rawYAML: `
+ingress:
+- hostname: tcp.foo.com
+  service: tcp://127.0.0.1
+- hostname: tcp2.foo.com
+  service: tcp://localhost:8000
+- service: http_status:404
+`},
+			want: []Rule{
+				{
+					Hostname: "tcp.foo.com",
+					Service:  newTCPOverWSService(MustParseURL(t, "tcp://127.0.0.1:7864")),
+					Config:   defaultConfig,
+				},
+				{
+					Hostname: "tcp2.foo.com",
+					Service:  newTCPOverWSService(MustParseURL(t, "tcp://localhost:8000")),
+					Config:   defaultConfig,
+				},
+				{
+					Service: &fourOhFour,
+					Config:  defaultConfig,
+				},
+			},
+		},
+		{
+			name: "SSH services",
+			args: args{rawYAML: `
+ingress:
+- service: ssh://127.0.0.1
+`},
+			want: []Rule{
+				{
+					Service: newTCPOverWSService(MustParseURL(t, "ssh://127.0.0.1:22")),
+					Config:  defaultConfig,
+				},
+			},
+		},
+		{
+			name: "RDP services",
+			args: args{rawYAML: `
+ingress:
+- service: rdp://127.0.0.1
+`},
+			want: []Rule{
+				{
+					Service: newTCPOverWSService(MustParseURL(t, "rdp://127.0.0.1:3389")),
+					Config:  defaultConfig,
+				},
+			},
+		},
+		{
+			name: "SMB services",
+			args: args{rawYAML: `
+ingress:
+- service: smb://127.0.0.1
+`},
+			want: []Rule{
+				{
+					Service: newTCPOverWSService(MustParseURL(t, "smb://127.0.0.1:445")),
+					Config:  defaultConfig,
+				},
+			},
+		},
+		{
+			name: "Other TCP services",
+			args: args{rawYAML: `
+ingress:
+- service: ftp://127.0.0.1
+`},
+			want: []Rule{
+				{
+					Service: newTCPOverWSService(MustParseURL(t, "ftp://127.0.0.1")),
+					Config:  defaultConfig,
+				},
+			},
+		},
 		{
 			name: "URL isn't necessary if using bastion",
 			args: args{rawYAML: `
@@ -221,7 +316,7 @@ ingress:
 			want: []Rule{
 				{
 					Hostname: "bastion.foo.com",
-					Service:  &localService{},
+					Service:  newBastionService(),
 					Config:   setConfig(originRequestFromYAML(config.OriginRequestConfig{}), config.OriginRequestConfig{BastionMode: &tr}),
 				},
 				{
@@ -241,7 +336,7 @@ ingress:
 			want: []Rule{
 				{
 					Hostname: "bastion.foo.com",
-					Service:  &localService{},
+					Service:  newBastionService(),
 					Config:   setConfig(originRequestFromYAML(config.OriginRequestConfig{}), config.OriginRequestConfig{BastionMode: &tr}),
 				},
 				{
@@ -369,6 +464,7 @@ func TestFindMatchingRule(t *testing.T) {
 	tests := []struct {
 		host          string
 		path          string
+		req           *http.Request
 		wantRuleIndex int
 	}{
 		{
@@ -403,9 +499,40 @@ func TestFindMatchingRule(t *testing.T) {
 		},
 	}
 
-	for i, test := range tests {
+	for _, test := range tests {
 		_, ruleIndex := ingress.FindMatchingRule(test.host, test.path)
-		assert.Equal(t, test.wantRuleIndex, ruleIndex, fmt.Sprintf("Expect host=%s, path=%s to match rule %d, got %d", test.host, test.path, test.wantRuleIndex, i))
+		assert.Equal(t, test.wantRuleIndex, ruleIndex, fmt.Sprintf("Expect host=%s, path=%s to match rule %d, got %d", test.host, test.path, test.wantRuleIndex, ruleIndex))
+	}
+}
+
+func TestIsHTTPService(t *testing.T) {
+	tests := []struct {
+		url    *url.URL
+		isHTTP bool
+	}{
+		{
+			url:    MustParseURL(t, "http://localhost"),
+			isHTTP: true,
+		},
+		{
+			url:    MustParseURL(t, "https://127.0.0.1:8000"),
+			isHTTP: true,
+		},
+		{
+			url:    MustParseURL(t, "ws://localhost"),
+			isHTTP: true,
+		},
+		{
+			url:    MustParseURL(t, "wss://localhost:8000"),
+			isHTTP: true,
+		},
+		{
+			url:    MustParseURL(t, "tcp://localhost:9000"),
+			isHTTP: false,
+		},
+	}
+	for _, test := range tests {
+		assert.Equal(t, test.isHTTP, isHTTPService(test.url))
 	}
 }
 
@@ -436,6 +563,7 @@ ingress:
 	if err != nil {
 		b.Error(err)
 	}
+
 	for n := 0; n < b.N; n++ {
 		ing.FindMatchingRule("tunnel1.example.com", "")
 		ing.FindMatchingRule("tunnel2.example.com", "")

ingress/origin_connection.go

@@ -0,0 +1,109 @@
+package ingress
+
+import (
+	"context"
+	"crypto/tls"
+	"io"
+	"net"
+	"net/http"
+
+	"github.com/cloudflare/cloudflared/websocket"
+	gws "github.com/gorilla/websocket"
+	"github.com/rs/zerolog"
+)
+
+// OriginConnection is a way to stream to a service running on the user's origin.
+// Different concrete implementations will stream different protocols as long as they are io.ReadWriters.
+type OriginConnection interface {
+	// Stream should generally be implemented as a bidirectional io.Copy.
+	Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger)
+	Close()
+}
+
+type streamHandlerFunc func(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger)
+
+// Stream copies copy data to & from provided io.ReadWriters.
+func Stream(conn, backendConn io.ReadWriter, log *zerolog.Logger) {
+	proxyDone := make(chan struct{}, 2)
+
+	go func() {
+		_, err := io.Copy(conn, backendConn)
+		if err != nil {
+			log.Debug().Msgf("conn to backendConn copy: %v", err)
+		}
+		proxyDone <- struct{}{}
+	}()
+
+	go func() {
+		_, err := io.Copy(backendConn, conn)
+		if err != nil {
+			log.Debug().Msgf("backendConn to conn copy: %v", err)
+		}
+		proxyDone <- struct{}{}
+	}()
+
+	// If one side is done, we are done.
+	<-proxyDone
+}
+
+// DefaultStreamHandler is an implementation of streamHandlerFunc that
+// performs a two way io.Copy between originConn and remoteConn.
+func DefaultStreamHandler(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger) {
+	Stream(originConn, remoteConn, log)
+}
+
+// tcpConnection is an OriginConnection that directly streams to raw TCP.
+type tcpConnection struct {
+	conn net.Conn
+}
+
+func (tc *tcpConnection) Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger) {
+	Stream(tunnelConn, tc.conn, log)
+}
+
+func (tc *tcpConnection) Close() {
+	tc.conn.Close()
+}
+
+// tcpOverWSConnection is an OriginConnection that streams to TCP over WS.
+type tcpOverWSConnection struct {
+	conn          net.Conn
+	streamHandler streamHandlerFunc
+}
+
+func (wc *tcpOverWSConnection) Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger) {
+	wc.streamHandler(websocket.NewConn(ctx, tunnelConn, log), wc.conn, log)
+}
+
+func (wc *tcpOverWSConnection) Close() {
+	wc.conn.Close()
+}
+
+// wsConnection is an OriginConnection that streams WS between eyeball and origin.
+type wsConnection struct {
+	wsConn *gws.Conn
+	resp   *http.Response
+}
+
+func (wsc *wsConnection) Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger) {
+	Stream(tunnelConn, wsc.wsConn.UnderlyingConn(), log)
+}
+
+func (wsc *wsConnection) Close() {
+	wsc.resp.Body.Close()
+	wsc.wsConn.Close()
+}
+
+func newWSConnection(clientTLSConfig *tls.Config, r *http.Request) (OriginConnection, *http.Response, error) {
+	d := &gws.Dialer{
+		TLSClientConfig: clientTLSConfig,
+	}
+	wsConn, resp, err := websocket.ClientConnect(r, d)
+	if err != nil {
+		return nil, nil, err
+	}
+	return &wsConnection{
+		wsConn,
+		resp,
+	}, resp, nil
+}

ingress/origin_connection_test.go

@@ -0,0 +1,294 @@
+package ingress
+
+import (
+	"bytes"
+	"context"
+	"crypto/tls"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"testing"
+	"time"
+
+	"github.com/cloudflare/cloudflared/logger"
+	"github.com/cloudflare/cloudflared/socks"
+	"github.com/gobwas/ws/wsutil"
+	gorillaWS "github.com/gorilla/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/proxy"
+	"golang.org/x/sync/errgroup"
+)
+
+const (
+	testStreamTimeout = time.Second * 3
+	echoHeaderName    = "Test-Cloudflared-Echo"
+)
+
+var (
+	testLogger   = logger.Create(nil)
+	testMessage  = []byte("TestStreamOriginConnection")
+	testResponse = []byte(fmt.Sprintf("echo-%s", testMessage))
+)
+
+func TestStreamTCPConnection(t *testing.T) {
+	cfdConn, originConn := net.Pipe()
+	tcpConn := tcpConnection{
+		conn: cfdConn,
+	}
+
+	eyeballConn, edgeConn := net.Pipe()
+
+	ctx, cancel := context.WithTimeout(context.Background(), testStreamTimeout)
+	defer cancel()
+
+	errGroup, ctx := errgroup.WithContext(ctx)
+	errGroup.Go(func() error {
+		_, err := eyeballConn.Write(testMessage)
+
+		readBuffer := make([]byte, len(testResponse))
+		_, err = eyeballConn.Read(readBuffer)
+		require.NoError(t, err)
+
+		require.Equal(t, testResponse, readBuffer)
+
+		return nil
+	})
+	errGroup.Go(func() error {
+		echoTCPOrigin(t, originConn)
+		originConn.Close()
+		return nil
+	})
+
+	tcpConn.Stream(ctx, edgeConn, testLogger)
+	require.NoError(t, errGroup.Wait())
+}
+
+func TestDefaultStreamWSOverTCPConnection(t *testing.T) {
+	cfdConn, originConn := net.Pipe()
+	tcpOverWSConn := tcpOverWSConnection{
+		conn:          cfdConn,
+		streamHandler: DefaultStreamHandler,
+	}
+
+	eyeballConn, edgeConn := net.Pipe()
+
+	ctx, cancel := context.WithTimeout(context.Background(), testStreamTimeout)
+	defer cancel()
+
+	errGroup, ctx := errgroup.WithContext(ctx)
+	errGroup.Go(func() error {
+		echoWSEyeball(t, eyeballConn)
+		return nil
+	})
+	errGroup.Go(func() error {
+		echoTCPOrigin(t, originConn)
+		originConn.Close()
+		return nil
+	})
+
+	tcpOverWSConn.Stream(ctx, edgeConn, testLogger)
+	require.NoError(t, errGroup.Wait())
+}
+
+// TestSocksStreamWSOverTCPConnection simulates proxying in socks mode.
+// Eyeball side runs cloudflared accesss tcp with --url flag to start a websocket forwarder which
+// wraps SOCKS5 traffic in websocket
+// Origin side runs a tcpOverWSConnection with socks.StreamHandler
+func TestSocksStreamWSOverTCPConnection(t *testing.T) {
+	var (
+		sendMessage             = t.Name()
+		echoHeaderIncomingValue = fmt.Sprintf("header-%s", sendMessage)
+		echoMessage             = fmt.Sprintf("echo-%s", sendMessage)
+		echoHeaderReturnValue   = fmt.Sprintf("echo-%s", echoHeaderIncomingValue)
+	)
+
+	statusCodes := []int{
+		http.StatusOK,
+		http.StatusTemporaryRedirect,
+		http.StatusBadRequest,
+		http.StatusInternalServerError,
+	}
+	for _, status := range statusCodes {
+		handler := func(w http.ResponseWriter, r *http.Request) {
+			body, err := ioutil.ReadAll(r.Body)
+			require.NoError(t, err)
+			require.Equal(t, []byte(sendMessage), body)
+
+			require.Equal(t, echoHeaderIncomingValue, r.Header.Get(echoHeaderName))
+			w.Header().Set(echoHeaderName, echoHeaderReturnValue)
+
+			w.WriteHeader(status)
+			w.Write([]byte(echoMessage))
+		}
+		origin := httptest.NewServer(http.HandlerFunc(handler))
+		defer origin.Close()
+
+		originURL, err := url.Parse(origin.URL)
+		require.NoError(t, err)
+
+		originConn, err := net.Dial("tcp", originURL.Host)
+		require.NoError(t, err)
+
+		tcpOverWSConn := tcpOverWSConnection{
+			conn:          originConn,
+			streamHandler: socks.StreamHandler,
+		}
+
+		wsForwarderOutConn, edgeConn := net.Pipe()
+		ctx, cancel := context.WithTimeout(context.Background(), testStreamTimeout)
+		defer cancel()
+
+		errGroup, ctx := errgroup.WithContext(ctx)
+		errGroup.Go(func() error {
+			tcpOverWSConn.Stream(ctx, edgeConn, testLogger)
+			return nil
+		})
+
+		wsForwarderListener, err := net.Listen("tcp", "127.0.0.1:0")
+		require.NoError(t, err)
+
+		errGroup.Go(func() error {
+			wsForwarderInConn, err := wsForwarderListener.Accept()
+			require.NoError(t, err)
+			defer wsForwarderInConn.Close()
+
+			Stream(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, testLogger)
+			return nil
+		})
+
+		eyeballDialer, err := proxy.SOCKS5("tcp", wsForwarderListener.Addr().String(), nil, proxy.Direct)
+		require.NoError(t, err)
+
+		transport := &http.Transport{
+			Dial: eyeballDialer.Dial,
+		}
+
+		// Request URL doesn't matter because the transport is using eyeballDialer to connectq
+		req, err := http.NewRequestWithContext(ctx, "GET", "http://test-socks-stream.com", bytes.NewBuffer([]byte(sendMessage)))
+		assert.NoError(t, err)
+		req.Header.Set(echoHeaderName, echoHeaderIncomingValue)
+
+		resp, err := transport.RoundTrip(req)
+		assert.NoError(t, err)
+		assert.Equal(t, status, resp.StatusCode)
+		require.Equal(t, echoHeaderReturnValue, resp.Header.Get(echoHeaderName))
+		body, err := ioutil.ReadAll(resp.Body)
+		require.NoError(t, err)
+		require.Equal(t, []byte(echoMessage), body)
+
+		wsForwarderOutConn.Close()
+		edgeConn.Close()
+		tcpOverWSConn.Close()
+
+		require.NoError(t, errGroup.Wait())
+	}
+}
+
+func TestStreamWSConnection(t *testing.T) {
+	eyeballConn, edgeConn := net.Pipe()
+
+	origin := echoWSOrigin(t)
+	defer origin.Close()
+
+	req, err := http.NewRequest(http.MethodGet, origin.URL, nil)
+	require.NoError(t, err)
+	req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+
+	clientTLSConfig := &tls.Config{
+		InsecureSkipVerify: true,
+	}
+	wsConn, resp, err := newWSConnection(clientTLSConfig, req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
+	require.Equal(t, "Upgrade", resp.Header.Get("Connection"))
+	require.Equal(t, "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=", resp.Header.Get("Sec-Websocket-Accept"))
+	require.Equal(t, "websocket", resp.Header.Get("Upgrade"))
+
+	ctx, cancel := context.WithTimeout(context.Background(), testStreamTimeout)
+	defer cancel()
+
+	errGroup, ctx := errgroup.WithContext(ctx)
+	errGroup.Go(func() error {
+		echoWSEyeball(t, eyeballConn)
+		return nil
+	})
+
+	wsConn.Stream(ctx, edgeConn, testLogger)
+	require.NoError(t, errGroup.Wait())
+}
+
+type wsEyeball struct {
+	conn net.Conn
+}
+
+func (wse *wsEyeball) Read(p []byte) (int, error) {
+	data, err := wsutil.ReadServerBinary(wse.conn)
+	if err != nil {
+		return 0, err
+	}
+	return copy(p, data), nil
+}
+
+func (wse *wsEyeball) Write(p []byte) (int, error) {
+	err := wsutil.WriteClientBinary(wse.conn, p)
+	return len(p), err
+}
+
+func echoWSEyeball(t *testing.T, conn net.Conn) {
+	require.NoError(t, wsutil.WriteClientBinary(conn, testMessage))
+
+	readMsg, err := wsutil.ReadServerBinary(conn)
+	require.NoError(t, err)
+
+	require.Equal(t, testResponse, readMsg)
+
+	require.NoError(t, conn.Close())
+}
+
+func echoWSOrigin(t *testing.T) *httptest.Server {
+	var upgrader = gorillaWS.Upgrader{
+		ReadBufferSize:  10,
+		WriteBufferSize: 10,
+	}
+
+	ws := func(w http.ResponseWriter, r *http.Request) {
+		header := make(http.Header)
+		for k, vs := range r.Header {
+			if k == "Test-Cloudflared-Echo" {
+				header[k] = vs
+			}
+		}
+		conn, err := upgrader.Upgrade(w, r, header)
+		require.NoError(t, err)
+		defer conn.Close()
+
+		for {
+			messageType, p, err := conn.ReadMessage()
+			if err != nil {
+				return
+			}
+			require.Equal(t, testMessage, p)
+			if err := conn.WriteMessage(messageType, testResponse); err != nil {
+				return
+			}
+		}
+	}
+
+	// NewTLSServer starts the server in another thread
+	return httptest.NewTLSServer(http.HandlerFunc(ws))
+}
+
+func echoTCPOrigin(t *testing.T, conn net.Conn) {
+	readBuffer := make([]byte, len(testMessage))
+	_, err := conn.Read(readBuffer)
+	assert.NoError(t, err)
+
+	assert.Equal(t, testMessage, readBuffer)
+
+	_, err = conn.Write(testResponse)
+	assert.NoError(t, err)
+}

ingress/origin_proxy.go

@@ -0,0 +1,147 @@
+package ingress
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"strings"
+
+	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/cloudflare/cloudflared/websocket"
+	"github.com/pkg/errors"
+)
+
+var (
+	switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
+)
+
+// HTTPOriginProxy can be implemented by origin services that want to proxy http requests.
+type HTTPOriginProxy interface {
+	// RoundTrip is how cloudflared proxies eyeball requests to the actual origin services
+	http.RoundTripper
+}
+
+// StreamBasedOriginProxy can be implemented by origin services that want to proxy ws/TCP.
+type StreamBasedOriginProxy interface {
+	EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error)
+}
+
+func (o *unixSocketPath) RoundTrip(req *http.Request) (*http.Response, error) {
+	return o.transport.RoundTrip(req)
+}
+
+func (o *httpService) RoundTrip(req *http.Request) (*http.Response, error) {
+	// Rewrite the request URL so that it goes to the origin service.
+	req.URL.Host = o.url.Host
+	req.URL.Scheme = o.url.Scheme
+	if o.hostHeader != "" {
+		// For incoming requests, the Host header is promoted to the Request.Host field and removed from the Header map.
+		req.Host = o.hostHeader
+	}
+	return o.transport.RoundTrip(req)
+}
+
+func (o *httpService) EstablishConnection(req *http.Request) (OriginConnection, *http.Response, error) {
+	req.URL.Host = o.url.Host
+	req.URL.Scheme = websocket.ChangeRequestScheme(o.url)
+	if o.hostHeader != "" {
+		// For incoming requests, the Host header is promoted to the Request.Host field and removed from the Header map.
+		req.Host = o.hostHeader
+	}
+	return newWSConnection(o.transport.TLSClientConfig, req)
+}
+
+func (o *helloWorld) RoundTrip(req *http.Request) (*http.Response, error) {
+	// Rewrite the request URL so that it goes to the Hello World server.
+	req.URL.Host = o.server.Addr().String()
+	req.URL.Scheme = "https"
+	return o.transport.RoundTrip(req)
+}
+
+func (o *helloWorld) EstablishConnection(req *http.Request) (OriginConnection, *http.Response, error) {
+	req.URL.Host = o.server.Addr().String()
+	req.URL.Scheme = "wss"
+	return newWSConnection(o.transport.TLSClientConfig, req)
+}
+
+func (o *statusCode) RoundTrip(_ *http.Request) (*http.Response, error) {
+	return o.resp, nil
+}
+
+func (o *rawTCPService) EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error) {
+	dest, err := getRequestHost(r)
+	if err != nil {
+		return nil, nil, err
+	}
+	conn, err := net.Dial("tcp", dest)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	originConn := &tcpConnection{
+		conn: conn,
+	}
+	resp := &http.Response{
+		Status:        switchingProtocolText,
+		StatusCode:    http.StatusSwitchingProtocols,
+		ContentLength: -1,
+	}
+	return originConn, resp, nil
+}
+
+// getRequestHost returns the host of the http.Request.
+func getRequestHost(r *http.Request) (string, error) {
+	if r.Host != "" {
+		return r.Host, nil
+	}
+	if r.URL != nil {
+		return r.URL.Host, nil
+	}
+	return "", errors.New("host not found")
+}
+
+func (o *tcpOverWSService) EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error) {
+	var err error
+	dest := o.dest
+	if o.isBastion {
+		dest, err = o.bastionDest(r)
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+
+	conn, err := net.Dial("tcp", dest)
+	if err != nil {
+		return nil, nil, err
+	}
+	originConn := &tcpOverWSConnection{
+		conn:          conn,
+		streamHandler: o.streamHandler,
+	}
+	resp := &http.Response{
+		Status:        switchingProtocolText,
+		StatusCode:    http.StatusSwitchingProtocols,
+		Header:        websocket.NewResponseHeader(r),
+		ContentLength: -1,
+	}
+	return originConn, resp, nil
+
+}
+
+func (o *tcpOverWSService) bastionDest(r *http.Request) (string, error) {
+	jumpDestination := r.Header.Get(h2mux.CFJumpDestinationHeader)
+	if jumpDestination == "" {
+		return "", fmt.Errorf("Did not receive final destination from client. The --destination flag is likely not set on the client side")
+	}
+	// Strip scheme and path set by client. Without a scheme
+	// Parsing a hostname and path without scheme might not return an error due to parsing ambiguities
+	if jumpURL, err := url.Parse(jumpDestination); err == nil && jumpURL.Host != "" {
+		return removePath(jumpURL.Host), nil
+	}
+	return removePath(jumpDestination), nil
+}
+
+func removePath(dest string) string {
+	return strings.SplitN(dest, "/", 2)[0]
+}

ingress/origin_proxy_test.go

@@ -0,0 +1,330 @@
+package ingress
+
+import (
+	"context"
+	"crypto/tls"
+	"fmt"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"sync"
+	"testing"
+
+	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/cloudflare/cloudflared/websocket"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestEstablishConnectionResponse ensures each implementation of StreamBasedOriginProxy returns
+// the expected response
+func assertEstablishConnectionResponse(t *testing.T,
+	originProxy StreamBasedOriginProxy,
+	req *http.Request,
+	expectHeader http.Header,
+) {
+	_, resp, err := originProxy.EstablishConnection(req)
+	assert.NoError(t, err)
+	assert.Equal(t, switchingProtocolText, resp.Status)
+	assert.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
+	assert.Equal(t, expectHeader, resp.Header)
+}
+
+func TestHTTPServiceEstablishConnection(t *testing.T) {
+	origin := echoWSOrigin(t)
+	defer origin.Close()
+	originURL, err := url.Parse(origin.URL)
+	require.NoError(t, err)
+
+	httpService := &httpService{
+		url:        originURL,
+		hostHeader: origin.URL,
+		transport: &http.Transport{
+			TLSClientConfig: &tls.Config{
+				InsecureSkipVerify: true,
+			},
+		},
+	}
+	req, err := http.NewRequest(http.MethodGet, origin.URL, nil)
+	require.NoError(t, err)
+	req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+	req.Header.Set("Test-Cloudflared-Echo", t.Name())
+
+	expectHeader := http.Header{
+		"Connection":            {"Upgrade"},
+		"Sec-Websocket-Accept":  {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
+		"Upgrade":               {"websocket"},
+		"Test-Cloudflared-Echo": {t.Name()},
+	}
+	assertEstablishConnectionResponse(t, httpService, req, expectHeader)
+}
+
+func TestHelloWorldEstablishConnection(t *testing.T) {
+	var wg sync.WaitGroup
+	shutdownC := make(chan struct{})
+	errC := make(chan error)
+	helloWorldSerivce := &helloWorld{}
+	helloWorldSerivce.start(&wg, testLogger, shutdownC, errC, OriginRequestConfig{})
+
+	// Scheme and Host of URL will be override by the Scheme and Host of the helloWorld service
+	req, err := http.NewRequest(http.MethodGet, "https://place-holder/ws", nil)
+	require.NoError(t, err)
+
+	expectHeader := http.Header{
+		"Connection": {"Upgrade"},
+		// Accept key when Sec-Websocket-Key is not specified
+		"Sec-Websocket-Accept": {"Kfh9QIsMVZcl6xEPYxPHzW8SZ8w="},
+		"Upgrade":              {"websocket"},
+	}
+	assertEstablishConnectionResponse(t, helloWorldSerivce, req, expectHeader)
+
+	close(shutdownC)
+}
+
+func TestRawTCPServiceEstablishConnection(t *testing.T) {
+	originListener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	listenerClosed := make(chan struct{})
+	tcpListenRoutine(originListener, listenerClosed)
+
+	rawTCPService := &rawTCPService{name: ServiceWarpRouting}
+
+	req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://%s", originListener.Addr()), nil)
+	require.NoError(t, err)
+
+	assertEstablishConnectionResponse(t, rawTCPService, req, nil)
+
+	originListener.Close()
+	<-listenerClosed
+
+	req, err = http.NewRequest(http.MethodGet, fmt.Sprintf("http://%s", originListener.Addr()), nil)
+	require.NoError(t, err)
+
+	// Origin not listening for new connection, should return an error
+	_, resp, err := rawTCPService.EstablishConnection(req)
+	require.Error(t, err)
+	require.Nil(t, resp)
+}
+
+func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
+	originListener, err := net.Listen("tcp", "127.0.0.1:0")
+	require.NoError(t, err)
+
+	listenerClosed := make(chan struct{})
+	tcpListenRoutine(originListener, listenerClosed)
+
+	originURL := &url.URL{
+		Scheme: "tcp",
+		Host:   originListener.Addr().String(),
+	}
+
+	baseReq, err := http.NewRequest(http.MethodGet, "https://place-holder", nil)
+	require.NoError(t, err)
+	baseReq.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+
+	bastionReq := baseReq.Clone(context.Background())
+	bastionReq.Header.Set(h2mux.CFJumpDestinationHeader, originListener.Addr().String())
+
+	expectHeader := http.Header{
+		"Connection":           {"Upgrade"},
+		"Sec-Websocket-Accept": {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
+		"Upgrade":              {"websocket"},
+	}
+
+	tests := []struct {
+		service   *tcpOverWSService
+		req       *http.Request
+		expectErr bool
+	}{
+		{
+			service: newTCPOverWSService(originURL),
+			req:     baseReq,
+		},
+		{
+			service: newBastionService(),
+			req:     bastionReq,
+		},
+		{
+			service:   newBastionService(),
+			req:       baseReq,
+			expectErr: true,
+		},
+	}
+
+	for _, test := range tests {
+		if test.expectErr {
+			_, resp, err := test.service.EstablishConnection(test.req)
+			assert.Error(t, err)
+			assert.Nil(t, resp)
+		} else {
+			assertEstablishConnectionResponse(t, test.service, test.req, expectHeader)
+		}
+	}
+
+	originListener.Close()
+	<-listenerClosed
+
+	for _, service := range []*tcpOverWSService{newTCPOverWSService(originURL), newBastionService()} {
+		// Origin not listening for new connection, should return an error
+		_, resp, err := service.EstablishConnection(bastionReq)
+		assert.Error(t, err)
+		assert.Nil(t, resp)
+	}
+}
+
+func TestBastionDestination(t *testing.T) {
+	canonicalJumpDestHeader := http.CanonicalHeaderKey(h2mux.CFJumpDestinationHeader)
+	tests := []struct {
+		name         string
+		header       http.Header
+		expectedDest string
+		wantErr      bool
+	}{
+		{
+			name: "hostname destination",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"localhost"},
+			},
+			expectedDest: "localhost",
+		},
+		{
+			name: "hostname destination with port",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"localhost:9000"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "hostname destination with scheme and port",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"ssh://localhost:9000"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "full hostname url",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"ssh://localhost:9000/metrics"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "hostname destination with port and path",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"localhost:9000/metrics"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "ip destination",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"127.0.0.1"},
+			},
+			expectedDest: "127.0.0.1",
+		},
+		{
+			name: "ip destination with port",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"127.0.0.1:9000"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name: "ip destination with port and path",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"127.0.0.1:9000/metrics"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name: "ip destination with schem and port",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"tcp://127.0.0.1:9000"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name: "full ip url",
+			header: http.Header{
+				canonicalJumpDestHeader: []string{"ssh://127.0.0.1:9000/metrics"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name:    "no destination",
+			wantErr: true,
+		},
+	}
+	s := newBastionService()
+	for _, test := range tests {
+		r := &http.Request{
+			Header: test.header,
+		}
+		dest, err := s.bastionDest(r)
+		if test.wantErr {
+			assert.Error(t, err, "Test %s expects error", test.name)
+		} else {
+			assert.NoError(t, err, "Test %s expects no error, got error %v", test.name, err)
+			assert.Equal(t, test.expectedDest, dest, "Test %s expect dest %s, got %s", test.name, test.expectedDest, dest)
+		}
+	}
+}
+
+func TestHTTPServiceHostHeaderOverride(t *testing.T) {
+	cfg := OriginRequestConfig{
+		HTTPHostHeader: t.Name(),
+	}
+	handler := func(w http.ResponseWriter, r *http.Request) {
+		require.Equal(t, r.Host, t.Name())
+		if websocket.IsWebSocketUpgrade(r) {
+			respHeaders := websocket.NewResponseHeader(r)
+			for k, v := range respHeaders {
+				w.Header().Set(k, v[0])
+			}
+			w.WriteHeader(http.StatusSwitchingProtocols)
+			return
+		}
+		w.Write([]byte("ok"))
+	}
+	origin := httptest.NewServer(http.HandlerFunc(handler))
+	defer origin.Close()
+
+	originURL, err := url.Parse(origin.URL)
+	require.NoError(t, err)
+
+	httpService := &httpService{
+		url: originURL,
+	}
+	var wg sync.WaitGroup
+	shutdownC := make(chan struct{})
+	errC := make(chan error)
+	require.NoError(t, httpService.start(&wg, testLogger, shutdownC, errC, cfg))
+
+	req, err := http.NewRequest(http.MethodGet, originURL.String(), nil)
+	require.NoError(t, err)
+
+	resp, err := httpService.RoundTrip(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, resp.StatusCode)
+
+	req = req.Clone(context.Background())
+	_, resp, err = httpService.EstablishConnection(req)
+	require.NoError(t, err)
+	require.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
+}
+
+func tcpListenRoutine(listener net.Listener, closeChan chan struct{}) {
+	go func() {
+		for {
+			conn, err := listener.Accept()
+			if err != nil {
+				close(closeChan)
+				return
+			}
+			// Close immediately, this test is not about testing read/write on connection
+			conn.Close()
+		}
+	}()
+}

ingress/origin_service.go

@@ -8,7 +8,6 @@ import (
 	"net"
 	"net/http"
 	"net/url"
-	"strconv"
 	"sync"
 	"time"
 
@@ -21,10 +20,8 @@ import (
 	"github.com/rs/zerolog"
 )
 
-// OriginService is something a tunnel can proxy traffic to.
-type OriginService interface {
-	// RoundTrip is how cloudflared proxies eyeball requests to the actual origin services
-	http.RoundTripper
+// originService is something a tunnel can proxy traffic to.
+type originService interface {
 	String() string
 	// Start the origin service if it's managed by cloudflared, e.g. proxy servers or Hello World.
 	// If it's not managed by cloudflared, this is a no-op because the user is responsible for
@@ -51,10 +48,6 @@ func (o *unixSocketPath) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdown
 	return nil
 }
 
-func (o *unixSocketPath) RoundTrip(req *http.Request) (*http.Response, error) {
-	return o.transport.RoundTrip(req)
-}
-
 func (o *unixSocketPath) Dial(reqURL *url.URL, headers http.Header) (*gws.Conn, *http.Response, error) {
 	d := &gws.Dialer{
 		NetDial:         o.transport.Dial,
@@ -65,130 +58,90 @@ func (o *unixSocketPath) Dial(reqURL *url.URL, headers http.Header) (*gws.Conn,
 	return d.Dial(reqURL.String(), headers)
 }
 
-// localService is an OriginService listening on a TCP/IP address the user's origin can route to.
-type localService struct {
-	// The URL for the user's origin service
-	RootURL *url.URL
-	// The URL that cloudflared should send requests to.
-	// If this origin requires starting a proxy, this is the proxy's address,
-	// and that proxy points to RootURL. Otherwise, this is equal to RootURL.
-	URL       *url.URL
-	transport *http.Transport
+type httpService struct {
+	url        *url.URL
+	hostHeader string
+	transport  *http.Transport
 }
 
-func (o *localService) Dial(reqURL *url.URL, headers http.Header) (*gws.Conn, *http.Response, error) {
-	d := &gws.Dialer{TLSClientConfig: o.transport.TLSClientConfig}
-	// Rewrite the request URL so that it goes to the origin service.
-	reqURL.Host = o.URL.Host
-	reqURL.Scheme = websocket.ChangeRequestScheme(o.URL)
-	return d.Dial(reqURL.String(), headers)
-}
-
-func (o *localService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
+func (o *httpService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
 	transport, err := newHTTPTransport(o, cfg, log)
 	if err != nil {
 		return err
 	}
+	o.hostHeader = cfg.HTTPHostHeader
 	o.transport = transport
-
-	// Start a proxy if one is needed
-	if staticHost := o.staticHost(); originRequiresProxy(staticHost, cfg) {
-		if err := o.startProxy(staticHost, wg, log, shutdownC, errC, cfg); err != nil {
-			return err
-		}
-	}
-
 	return nil
 }
 
-func (o *localService) startProxy(staticHost string, wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
+func (o *httpService) String() string {
+	return o.url.String()
+}
 
-	// Start a listener for the proxy
-	proxyAddress := net.JoinHostPort(cfg.ProxyAddress, strconv.Itoa(int(cfg.ProxyPort)))
-	listener, err := net.Listen("tcp", proxyAddress)
-	if err != nil {
-		log.Error().Msgf("Cannot start Websocket Proxy Server: %s", err)
-		return errors.Wrap(err, "Cannot start Websocket Proxy Server")
-	}
+// rawTCPService dials TCP to the destination specified by the client
+// It's used by warp routing
+type rawTCPService struct {
+	name string
+}
 
-	// Start the proxy itself
-	wg.Add(1)
-	go func() {
-		defer wg.Done()
-		streamHandler := websocket.DefaultStreamHandler
-		// This origin's config specifies what type of proxy to start.
-		switch cfg.ProxyType {
-		case socksProxy:
-			log.Info().Msg("SOCKS5 server started")
-			streamHandler = func(wsConn *websocket.Conn, remoteConn net.Conn, _ http.Header) {
-				dialer := socks.NewConnDialer(remoteConn)
-				requestHandler := socks.NewRequestHandler(dialer)
-				socksServer := socks.NewConnectionHandler(requestHandler)
+func (o *rawTCPService) String() string {
+	return o.name
+}
 
-				_ = socksServer.Serve(wsConn)
-			}
-		case "":
-			log.Debug().Msg("Not starting any websocket proxy")
-		default:
-			log.Error().Msgf("%s isn't a valid proxy (valid options are {%s})", cfg.ProxyType, socksProxy)
-		}
-
-		errC <- websocket.StartProxyServer(log, listener, staticHost, shutdownC, streamHandler)
-	}()
-
-	// Modify this origin, so that it no longer points at the origin service directly.
-	// Instead, it points at the proxy to the origin service.
-	newURL, err := url.Parse("http://" + listener.Addr().String())
-	if err != nil {
-		return err
-	}
-	o.URL = newURL
+func (o *rawTCPService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
 	return nil
 }
 
-func (o *localService) String() string {
-	if o.isBastion() {
-		return "Bastion"
-	}
-	return o.URL.String()
+// tcpOverWSService models TCP origins serving eyeballs connecting over websocket, such as
+// cloudflared access commands.
+type tcpOverWSService struct {
+	dest          string
+	isBastion     bool
+	streamHandler streamHandlerFunc
 }
 
-func (o *localService) isBastion() bool {
-	return o.URL == nil
-}
-
-func (o *localService) RoundTrip(req *http.Request) (*http.Response, error) {
-	// Rewrite the request URL so that it goes to the origin service.
-	req.URL.Host = o.URL.Host
-	req.URL.Scheme = o.URL.Scheme
-	return o.transport.RoundTrip(req)
-}
-
-func (o *localService) staticHost() string {
-
-	if o.URL == nil {
-		return ""
-	}
-
-	addPortIfMissing := func(uri *url.URL, port int) string {
-		if uri.Port() != "" {
-			return uri.Host
-		}
-		return fmt.Sprintf("%s:%d", uri.Hostname(), port)
-	}
-
-	switch o.URL.Scheme {
+func newTCPOverWSService(url *url.URL) *tcpOverWSService {
+	switch url.Scheme {
 	case "ssh":
-		return addPortIfMissing(o.URL, 22)
+		addPortIfMissing(url, 22)
 	case "rdp":
-		return addPortIfMissing(o.URL, 3389)
+		addPortIfMissing(url, 3389)
 	case "smb":
-		return addPortIfMissing(o.URL, 445)
+		addPortIfMissing(url, 445)
 	case "tcp":
-		return addPortIfMissing(o.URL, 7864) // just a random port since there isn't a default in this case
+		addPortIfMissing(url, 7864) // just a random port since there isn't a default in this case
 	}
-	return ""
+	return &tcpOverWSService{
+		dest: url.Host,
+	}
+}
 
+func newBastionService() *tcpOverWSService {
+	return &tcpOverWSService{
+		isBastion: true,
+	}
+}
+
+func addPortIfMissing(uri *url.URL, port int) {
+	if uri.Port() == "" {
+		uri.Host = fmt.Sprintf("%s:%d", uri.Hostname(), port)
+	}
+}
+
+func (o *tcpOverWSService) String() string {
+	if o.isBastion {
+		return ServiceBastion
+	}
+	return o.dest
+}
+
+func (o *tcpOverWSService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
+	if cfg.ProxyType == socksProxy {
+		o.streamHandler = socks.StreamHandler
+	} else {
+		o.streamHandler = DefaultStreamHandler
+	}
+	return nil
 }
 
 // HelloWorld is an OriginService for the built-in Hello World server.
@@ -228,26 +181,6 @@ func (o *helloWorld) start(
 	return nil
 }
 
-func (o *helloWorld) RoundTrip(req *http.Request) (*http.Response, error) {
-	// Rewrite the request URL so that it goes to the Hello World server.
-	req.URL.Host = o.server.Addr().String()
-	req.URL.Scheme = "https"
-	return o.transport.RoundTrip(req)
-}
-
-func (o *helloWorld) Dial(reqURL *url.URL, headers http.Header) (*gws.Conn, *http.Response, error) {
-	d := &gws.Dialer{
-		TLSClientConfig: o.transport.TLSClientConfig,
-	}
-	reqURL.Host = o.server.Addr().String()
-	reqURL.Scheme = "wss"
-	return d.Dial(reqURL.String(), headers)
-}
-
-func originRequiresProxy(staticHost string, cfg OriginRequestConfig) bool {
-	return staticHost != "" || cfg.BastionMode
-}
-
 // statusCode is an OriginService that just responds with a given HTTP status.
 // Typical use-case is "user wants the catch-all rule to just respond 404".
 type statusCode struct {
@@ -277,10 +210,6 @@ func (o *statusCode) start(
 	return nil
 }
 
-func (o *statusCode) RoundTrip(_ *http.Request) (*http.Response, error) {
-	return o.resp, nil
-}
-
 type NopReadCloser struct{}
 
 // Read always returns EOF to signal end of input
@@ -292,7 +221,7 @@ func (nrc *NopReadCloser) Close() error {
 	return nil
 }
 
-func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerolog.Logger) (*http.Transport, error) {
+func newHTTPTransport(service originService, cfg OriginRequestConfig, log *zerolog.Logger) (*http.Transport, error) {
 	originCertPool, err := tlsconfig.LoadOriginCA(cfg.CAPool, log)
 	if err != nil {
 		return nil, errors.Wrap(err, "Error loading cert pool")
@@ -337,19 +266,19 @@ func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerol
 	return &httpTransport, nil
 }
 
-// MockOriginService should only be used by other packages to mock OriginService. Set Transport to configure desired RoundTripper behavior.
-type MockOriginService struct {
+// MockOriginHTTPService should only be used by other packages to mock OriginService. Set Transport to configure desired RoundTripper behavior.
+type MockOriginHTTPService struct {
 	Transport http.RoundTripper
 }
 
-func (mos MockOriginService) RoundTrip(req *http.Request) (*http.Response, error) {
+func (mos MockOriginHTTPService) RoundTrip(req *http.Request) (*http.Response, error) {
 	return mos.Transport.RoundTrip(req)
 }
 
-func (mos MockOriginService) String() string {
+func (mos MockOriginHTTPService) String() string {
 	return "MockOriginService"
 }
 
-func (mos MockOriginService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
+func (mos MockOriginHTTPService) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdownC <-chan struct{}, errC chan error, cfg OriginRequestConfig) error {
 	return nil
 }

ingress/rule.go

@@ -17,7 +17,7 @@ type Rule struct {
 	// A (probably local) address. Requests for a hostname which matches this
 	// rule's hostname pattern will be proxied to the service running on this
 	// address.
-	Service OriginService
+	Service originService
 
 	// Configure the request cloudflared sends to this specific origin.
 	Config OriginRequestConfig

ingress/rule_test.go

@@ -14,7 +14,7 @@ func Test_rule_matches(t *testing.T) {
 	type fields struct {
 		Hostname string
 		Path     *regexp.Regexp
-		Service  OriginService
+		Service  originService
 	}
 	type args struct {
 		requestURL *url.URL

origin/proxy.go

@@ -5,7 +5,6 @@ import (
 	"context"
 	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"strconv"
 	"strings"
@@ -14,8 +13,6 @@ import (
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/ingress"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
-	"github.com/cloudflare/cloudflared/websocket"
-
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 )
@@ -24,52 +21,89 @@ const (
 	TagHeaderNamePrefix = "Cf-Warp-Tag-"
 )
 
-type client struct {
+type proxy struct {
 	ingressRules ingress.Ingress
+	warpRouting  *ingress.WarpRoutingService
 	tags         []tunnelpogs.Tag
 	log          *zerolog.Logger
 	bufferPool   *buffer.Pool
 }
 
-func NewClient(ingressRules ingress.Ingress, tags []tunnelpogs.Tag, log *zerolog.Logger) connection.OriginClient {
-	return &client{
+func NewOriginProxy(
+	ingressRules ingress.Ingress,
+	warpRouting *ingress.WarpRoutingService,
+	tags []tunnelpogs.Tag,
+	log *zerolog.Logger) connection.OriginProxy {
+
+	return &proxy{
 		ingressRules: ingressRules,
+		warpRouting:  warpRouting,
 		tags:         tags,
 		log:          log,
 		bufferPool:   buffer.NewPool(512 * 1024),
 	}
 }
 
-func (c *client) Proxy(w connection.ResponseWriter, req *http.Request, isWebsocket bool) error {
+// Caller is responsible for writing any error to ResponseWriter
+func (p *proxy) Proxy(w connection.ResponseWriter, req *http.Request, sourceConnectionType connection.Type) error {
 	incrementRequests()
 	defer decrementConcurrentRequests()
 
 	cfRay := findCfRayHeader(req)
 	lbProbe := isLBProbeRequest(req)
 
-	c.appendTagHeaders(req)
-	rule, ruleNum := c.ingressRules.FindMatchingRule(req.Host, req.URL.Path)
-	c.logRequest(req, cfRay, lbProbe, ruleNum)
+	serveCtx, cancel := context.WithCancel(req.Context())
+	defer cancel()
 
-	var (
-		resp *http.Response
-		err  error
-	)
-	if isWebsocket {
-		resp, err = c.proxyWebsocket(w, req, rule)
-	} else {
-		resp, err = c.proxyHTTP(w, req, rule)
+	p.appendTagHeaders(req)
+	if sourceConnectionType == connection.TypeTCP {
+		if p.warpRouting == nil {
+			err := errors.New(`cloudflared received a request from Warp client, but your configuration has disabled ingress from Warp clients. To enable this, set "warp-routing:\n\t enabled: true" in your config.yaml`)
+			p.log.Error().Msg(err.Error())
+			return err
+		}
+		logFields := logFields{
+			cfRay:   cfRay,
+			lbProbe: lbProbe,
+			rule:    ingress.ServiceWarpRouting,
+		}
+		if err := p.proxyStreamRequest(serveCtx, w, req, sourceConnectionType, p.warpRouting.Proxy, logFields); err != nil {
+			p.logRequestError(err, cfRay, ingress.ServiceWarpRouting)
+			return err
+		}
+		return nil
 	}
-	if err != nil {
-		c.logRequestError(err, cfRay, ruleNum)
-		w.WriteErrorResponse()
+
+	rule, ruleNum := p.ingressRules.FindMatchingRule(req.Host, req.URL.Path)
+	logFields := logFields{
+		cfRay:   cfRay,
+		lbProbe: lbProbe,
+		rule:    ruleNum,
+	}
+	p.logRequest(req, logFields)
+
+	if sourceConnectionType == connection.TypeHTTP {
+		if err := p.proxyHTTPRequest(w, req, rule, logFields); err != nil {
+			p.logRequestError(err, cfRay, ruleNum)
+			return err
+		}
+		return nil
+	}
+
+	connectionProxy, ok := rule.Service.(ingress.StreamBasedOriginProxy)
+	if !ok {
+		p.log.Error().Msgf("%s is not a connection-oriented service", rule.Service)
+		return fmt.Errorf("Not a connection-oriented service")
+	}
+
+	if err := p.proxyStreamRequest(serveCtx, w, req, sourceConnectionType, connectionProxy, logFields); err != nil {
+		p.logRequestError(err, cfRay, ruleNum)
 		return err
 	}
-	c.logOriginResponse(resp, cfRay, lbProbe, ruleNum)
 	return nil
 }
 
-func (c *client) proxyHTTP(w connection.ResponseWriter, req *http.Request, rule *ingress.Rule) (*http.Response, error) {
+func (p *proxy) proxyHTTPRequest(w connection.ResponseWriter, req *http.Request, rule *ingress.Rule, fields logFields) error {
 	// Support for WSGI Servers by switching transfer encoding from chunked to gzip/deflate
 	if rule.Config.DisableChunkedEncoding {
 		req.TransferEncoding = []string{"gzip", "deflate"}
@@ -82,78 +116,90 @@ func (c *client) proxyHTTP(w connection.ResponseWriter, req *http.Request, rule
 	// Request origin to keep connection alive to improve performance
 	req.Header.Set("Connection", "keep-alive")
 
-	if hostHeader := rule.Config.HTTPHostHeader; hostHeader != "" {
-		req.Header.Set("Host", hostHeader)
-		req.Host = hostHeader
+	httpService, ok := rule.Service.(ingress.HTTPOriginProxy)
+	if !ok {
+		p.log.Error().Msgf("%s is not a http service", rule.Service)
+		return fmt.Errorf("Not a http service")
 	}
 
-	resp, err := rule.Service.RoundTrip(req)
+	resp, err := httpService.RoundTrip(req)
 	if err != nil {
-		return nil, errors.Wrap(err, "Error proxying request to origin")
+		return errors.Wrap(err, "Error proxying request to origin")
 	}
 	defer resp.Body.Close()
 
-	err = w.WriteRespHeaders(resp)
+	err = w.WriteRespHeaders(resp.StatusCode, resp.Header)
 	if err != nil {
-		return nil, errors.Wrap(err, "Error writing response header")
+		return errors.Wrap(err, "Error writing response header")
 	}
 	if connection.IsServerSentEvent(resp.Header) {
-		c.log.Debug().Msg("Detected Server-Side Events from Origin")
-		c.writeEventStream(w, resp.Body)
+		p.log.Debug().Msg("Detected Server-Side Events from Origin")
+		p.writeEventStream(w, resp.Body)
 	} else {
 		// Use CopyBuffer, because Copy only allocates a 32KiB buffer, and cross-stream
 		// compression generates dictionary on first write
-		buf := c.bufferPool.Get()
-		defer c.bufferPool.Put(buf)
+		buf := p.bufferPool.Get()
+		defer p.bufferPool.Put(buf)
 		_, _ = io.CopyBuffer(w, resp.Body, buf)
 	}
-	return resp, nil
-}
-
-func (c *client) proxyWebsocket(w connection.ResponseWriter, req *http.Request, rule *ingress.Rule) (*http.Response, error) {
-	if hostHeader := rule.Config.HTTPHostHeader; hostHeader != "" {
-		req.Header.Set("Host", hostHeader)
-		req.Host = hostHeader
-	}
-
-	dialler, ok := rule.Service.(websocket.Dialler)
-	if !ok {
-		return nil, fmt.Errorf("Websockets aren't supported by the origin service '%s'", rule.Service)
-	}
-	conn, resp, err := websocket.ClientConnect(req, dialler)
-	if err != nil {
-		return nil, err
-	}
-
-	serveCtx, cancel := context.WithCancel(req.Context())
-	connClosedChan := make(chan struct{})
-	go func() {
-		// serveCtx is done if req is cancelled, or streamWebsocket returns
-		<-serveCtx.Done()
-		_ = conn.Close()
-		close(connClosedChan)
-	}()
-
-	// Copy to/from stream to the undelying connection. Use the underlying
-	// connection because cloudflared doesn't operate on the message themselves
-	err = c.streamWebsocket(w, conn.UnderlyingConn(), resp)
-	cancel()
-
-	// We need to make sure conn is closed before returning, otherwise we might write to conn after Proxy returns
-	<-connClosedChan
-	return resp, err
-}
-
-func (c *client) streamWebsocket(w connection.ResponseWriter, conn net.Conn, resp *http.Response) error {
-	err := w.WriteRespHeaders(resp)
-	if err != nil {
-		return errors.Wrap(err, "Error writing websocket response header")
-	}
-	websocket.Stream(conn, w)
+	p.logOriginResponse(resp, fields)
 	return nil
 }
 
-func (c *client) writeEventStream(w connection.ResponseWriter, respBody io.ReadCloser) {
+// proxyStreamRequest first establish a connection with origin, then it writes the status code and headers, and finally it streams data between
+// eyeball and origin.
+func (p *proxy) proxyStreamRequest(
+	serveCtx context.Context,
+	w connection.ResponseWriter,
+	req *http.Request,
+	sourceConnectionType connection.Type,
+	connectionProxy ingress.StreamBasedOriginProxy,
+	fields logFields,
+) error {
+	originConn, resp, err := connectionProxy.EstablishConnection(req)
+	if err != nil {
+		return err
+	}
+	if resp.Body != nil {
+		defer resp.Body.Close()
+	}
+
+	if err = w.WriteRespHeaders(resp.StatusCode, resp.Header); err != nil {
+		return err
+	}
+
+	streamCtx, cancel := context.WithCancel(serveCtx)
+	defer cancel()
+
+	go func() {
+		// streamCtx is done if req is cancelled or if Stream returns
+		<-streamCtx.Done()
+		originConn.Close()
+	}()
+
+	eyeballStream := &bidirectionalStream{
+		writer: w,
+		reader: req.Body,
+	}
+	originConn.Stream(serveCtx, eyeballStream, p.log)
+	p.logOriginResponse(resp, fields)
+	return nil
+}
+
+type bidirectionalStream struct {
+	reader io.Reader
+	writer io.Writer
+}
+
+func (wr *bidirectionalStream) Read(p []byte) (n int, err error) {
+	return wr.reader.Read(p)
+}
+
+func (wr *bidirectionalStream) Write(p []byte) (n int, err error) {
+	return wr.writer.Write(p)
+}
+
+func (p *proxy) writeEventStream(w connection.ResponseWriter, respBody io.ReadCloser) {
 	reader := bufio.NewReader(respBody)
 	for {
 		line, err := reader.ReadBytes('\n')
@@ -164,56 +210,61 @@ func (c *client) writeEventStream(w connection.ResponseWriter, respBody io.ReadC
 	}
 }
 
-func (c *client) appendTagHeaders(r *http.Request) {
-	for _, tag := range c.tags {
+func (p *proxy) appendTagHeaders(r *http.Request) {
+	for _, tag := range p.tags {
 		r.Header.Add(TagHeaderNamePrefix+tag.Name, tag.Value)
 	}
 }
 
-func (c *client) logRequest(r *http.Request, cfRay string, lbProbe bool, ruleNum int) {
-	if cfRay != "" {
-		c.log.Debug().Msgf("CF-RAY: %s %s %s %s", cfRay, r.Method, r.URL, r.Proto)
-	} else if lbProbe {
-		c.log.Debug().Msgf("CF-RAY: %s Load Balancer health check %s %s %s", cfRay, r.Method, r.URL, r.Proto)
+type logFields struct {
+	cfRay   string
+	lbProbe bool
+	rule    interface{}
+}
+
+func (p *proxy) logRequest(r *http.Request, fields logFields) {
+	if fields.cfRay != "" {
+		p.log.Debug().Msgf("CF-RAY: %s %s %s %s", fields.cfRay, r.Method, r.URL, r.Proto)
+	} else if fields.lbProbe {
+		p.log.Debug().Msgf("CF-RAY: %s Load Balancer health check %s %s %s", fields.cfRay, r.Method, r.URL, r.Proto)
 	} else {
-		c.log.Debug().Msgf("All requests should have a CF-RAY header. Please open a support ticket with Cloudflare. %s %s %s ", r.Method, r.URL, r.Proto)
+		p.log.Debug().Msgf("All requests should have a CF-RAY header. Please open a support ticket with Cloudflare. %s %s %s ", r.Method, r.URL, r.Proto)
 	}
-	c.log.Debug().Msgf("CF-RAY: %s Request Headers %+v", cfRay, r.Header)
-	c.log.Debug().Msgf("CF-RAY: %s Serving with ingress rule %d", cfRay, ruleNum)
+	p.log.Debug().Msgf("CF-RAY: %s Request Headers %+v", fields.cfRay, r.Header)
+	p.log.Debug().Msgf("CF-RAY: %s Serving with ingress rule %v", fields.cfRay, fields.rule)
 
 	if contentLen := r.ContentLength; contentLen == -1 {
-		c.log.Debug().Msgf("CF-RAY: %s Request Content length unknown", cfRay)
+		p.log.Debug().Msgf("CF-RAY: %s Request Content length unknown", fields.cfRay)
 	} else {
-		c.log.Debug().Msgf("CF-RAY: %s Request content length %d", cfRay, contentLen)
+		p.log.Debug().Msgf("CF-RAY: %s Request content length %d", fields.cfRay, contentLen)
 	}
 }
 
-func (c *client) logOriginResponse(r *http.Response, cfRay string, lbProbe bool, ruleNum int) {
-	responseByCode.WithLabelValues(strconv.Itoa(r.StatusCode)).Inc()
-	if cfRay != "" {
-		c.log.Debug().Msgf("CF-RAY: %s Status: %s served by ingress %d", cfRay, r.Status, ruleNum)
-	} else if lbProbe {
-		c.log.Debug().Msgf("Response to Load Balancer health check %s", r.Status)
+func (p *proxy) logOriginResponse(resp *http.Response, fields logFields) {
+	responseByCode.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
+	if fields.cfRay != "" {
+		p.log.Debug().Msgf("CF-RAY: %s Status: %s served by ingress %d", fields.cfRay, resp.Status, fields.rule)
+	} else if fields.lbProbe {
+		p.log.Debug().Msgf("Response to Load Balancer health check %s", resp.Status)
 	} else {
-		c.log.Debug().Msgf("Status: %s served by ingress %d", r.Status, ruleNum)
+		p.log.Debug().Msgf("Status: %s served by ingress %v", resp.Status, fields.rule)
 	}
-	c.log.Debug().Msgf("CF-RAY: %s Response Headers %+v", cfRay, r.Header)
+	p.log.Debug().Msgf("CF-RAY: %s Response Headers %+v", fields.cfRay, resp.Header)
 
-	if contentLen := r.ContentLength; contentLen == -1 {
-		c.log.Debug().Msgf("CF-RAY: %s Response content length unknown", cfRay)
+	if contentLen := resp.ContentLength; contentLen == -1 {
+		p.log.Debug().Msgf("CF-RAY: %s Response content length unknown", fields.cfRay)
 	} else {
-		c.log.Debug().Msgf("CF-RAY: %s Response content length %d", cfRay, contentLen)
+		p.log.Debug().Msgf("CF-RAY: %s Response content length %d", fields.cfRay, contentLen)
 	}
 }
 
-func (c *client) logRequestError(err error, cfRay string, ruleNum int) {
+func (p *proxy) logRequestError(err error, cfRay string, rule interface{}) {
 	requestErrors.Inc()
 	if cfRay != "" {
-		c.log.Error().Msgf("CF-RAY: %s Proxying to ingress %d error: %v", cfRay, ruleNum, err)
+		p.log.Error().Msgf("CF-RAY: %s Proxying to ingress %v error: %v", cfRay, rule, err)
 	} else {
-		c.log.Error().Msgf("Proxying to ingress %d error: %v", ruleNum, err)
+		p.log.Error().Msgf("Proxying to ingress %v error: %v", rule, err)
 	}
-
 }
 
 func findCfRayHeader(req *http.Request) string {

origin/proxy_test.go

@@ -6,17 +6,21 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"sync"
 	"testing"
 	"time"
 
+	"github.com/cloudflare/cloudflared/logger"
+
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/config"
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/hello"
 	"github.com/cloudflare/cloudflared/ingress"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	gorillaWS "github.com/gorilla/websocket"
 	"github.com/urfave/cli/v2"
 
 	"github.com/gobwas/ws/wsutil"
@@ -26,7 +30,8 @@ import (
 )
 
 var (
-	testTags = []tunnelpogs.Tag(nil)
+	testTags                 = []tunnelpogs.Tag(nil)
+	unusedWarpRoutingService = (*ingress.WarpRoutingService)(nil)
 )
 
 type mockHTTPRespWriter struct {
@@ -39,19 +44,14 @@ func newMockHTTPRespWriter() *mockHTTPRespWriter {
 	}
 }
 
-func (w *mockHTTPRespWriter) WriteRespHeaders(resp *http.Response) error {
-	w.WriteHeader(resp.StatusCode)
-	for header, val := range resp.Header {
+func (w *mockHTTPRespWriter) WriteRespHeaders(status int, header http.Header) error {
+	w.WriteHeader(status)
+	for header, val := range header {
 		w.Header()[header] = val
 	}
 	return nil
 }
 
-func (w *mockHTTPRespWriter) WriteErrorResponse() {
-	w.WriteHeader(http.StatusBadGateway)
-	_, _ = w.Write([]byte("http response error"))
-}
-
 func (w *mockHTTPRespWriter) Read(data []byte) (int, error) {
 	return 0, fmt.Errorf("mockHTTPRespWriter doesn't implement io.Reader")
 }
@@ -125,59 +125,58 @@ func TestProxySingleOrigin(t *testing.T) {
 	errC := make(chan error)
 	require.NoError(t, ingressRule.StartOrigins(&wg, &log, ctx.Done(), errC))
 
-	client := NewClient(ingressRule, testTags, &log)
-	t.Run("testProxyHTTP", testProxyHTTP(t, client))
-	t.Run("testProxyWebsocket", testProxyWebsocket(t, client))
-	t.Run("testProxySSE", testProxySSE(t, client))
+	proxy := NewOriginProxy(ingressRule, unusedWarpRoutingService, testTags, &log)
+	t.Run("testProxyHTTP", testProxyHTTP(t, proxy))
+	t.Run("testProxyWebsocket", testProxyWebsocket(t, proxy))
+	t.Run("testProxySSE", testProxySSE(t, proxy))
 	cancel()
 	wg.Wait()
 }
 
-func testProxyHTTP(t *testing.T, client connection.OriginClient) func(t *testing.T) {
+func testProxyHTTP(t *testing.T, proxy connection.OriginProxy) func(t *testing.T) {
 	return func(t *testing.T) {
-		respWriter := newMockHTTPRespWriter()
+		responseWriter := newMockHTTPRespWriter()
 		req, err := http.NewRequest(http.MethodGet, "http://localhost:8080", nil)
 		require.NoError(t, err)
 
-		err = client.Proxy(respWriter, req, false)
+		err = proxy.Proxy(responseWriter, req, connection.TypeHTTP)
 		require.NoError(t, err)
 
-		assert.Equal(t, http.StatusOK, respWriter.Code)
+		assert.Equal(t, http.StatusOK, responseWriter.Code)
 	}
 }
 
-func testProxyWebsocket(t *testing.T, client connection.OriginClient) func(t *testing.T) {
+func testProxyWebsocket(t *testing.T, proxy connection.OriginProxy) func(t *testing.T) {
 	return func(t *testing.T) {
 		// WSRoute is a websocket echo handler
 		ctx, cancel := context.WithCancel(context.Background())
-		req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://localhost:8080%s", hello.WSRoute), nil)
-
 		readPipe, writePipe := io.Pipe()
-		respWriter := newMockWSRespWriter(readPipe)
+		req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://localhost:8080%s", hello.WSRoute), readPipe)
+		responseWriter := newMockWSRespWriter(readPipe)
 
 		var wg sync.WaitGroup
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			err = client.Proxy(respWriter, req, true)
+			err = proxy.Proxy(responseWriter, req, connection.TypeWebsocket)
 			require.NoError(t, err)
 
-			require.Equal(t, http.StatusSwitchingProtocols, respWriter.Code)
+			require.Equal(t, http.StatusSwitchingProtocols, responseWriter.Code)
 		}()
 
 		msg := []byte("test websocket")
 		err = wsutil.WriteClientText(writePipe, msg)
 		require.NoError(t, err)
 
-		// ReadServerText reads next data message from rw, considering that caller represents client side.
-		returnedMsg, err := wsutil.ReadServerText(respWriter.respBody())
+		// ReadServerText reads next data message from rw, considering that caller represents proxy side.
+		returnedMsg, err := wsutil.ReadServerText(responseWriter.respBody())
 		require.NoError(t, err)
 		require.Equal(t, msg, returnedMsg)
 
 		err = wsutil.WriteClientBinary(writePipe, msg)
 		require.NoError(t, err)
 
-		returnedMsg, err = wsutil.ReadServerBinary(respWriter.respBody())
+		returnedMsg, err = wsutil.ReadServerBinary(responseWriter.respBody())
 		require.NoError(t, err)
 		require.Equal(t, msg, returnedMsg)
 
@@ -186,13 +185,13 @@ func testProxyWebsocket(t *testing.T, client connection.OriginClient) func(t *te
 	}
 }
 
-func testProxySSE(t *testing.T, client connection.OriginClient) func(t *testing.T) {
+func testProxySSE(t *testing.T, proxy connection.OriginProxy) func(t *testing.T) {
 	return func(t *testing.T) {
 		var (
 			pushCount = 50
 			pushFreq  = time.Millisecond * 10
 		)
-		respWriter := newMockSSERespWriter()
+		responseWriter := newMockSSERespWriter()
 		ctx, cancel := context.WithCancel(context.Background())
 		req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://localhost:8080%s?freq=%s", hello.SSERoute, pushFreq), nil)
 		require.NoError(t, err)
@@ -201,18 +200,18 @@ func testProxySSE(t *testing.T, client connection.OriginClient) func(t *testing.
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			err = client.Proxy(respWriter, req, false)
+			err = proxy.Proxy(responseWriter, req, connection.TypeHTTP)
 			require.NoError(t, err)
 
-			require.Equal(t, http.StatusOK, respWriter.Code)
+			require.Equal(t, http.StatusOK, responseWriter.Code)
 		}()
 
 		for i := 0; i < pushCount; i++ {
-			line := respWriter.ReadBytes()
+			line := responseWriter.ReadBytes()
 			expect := fmt.Sprintf("%d\n", i)
 			require.Equal(t, []byte(expect), line, fmt.Sprintf("Expect to read %v, got %v", expect, line))
 
-			line = respWriter.ReadBytes()
+			line = responseWriter.ReadBytes()
 			require.Equal(t, []byte("\n"), line, fmt.Sprintf("Expect to read '\n', got %v", line))
 		}
 
@@ -258,7 +257,7 @@ func TestProxyMultipleOrigins(t *testing.T) {
 	var wg sync.WaitGroup
 	require.NoError(t, ingress.StartOrigins(&wg, &log, ctx.Done(), errC))
 
-	client := NewClient(ingress, testTags, &log)
+	proxy := NewOriginProxy(ingress, unusedWarpRoutingService, testTags, &log)
 
 	tests := []struct {
 		url            string
@@ -290,18 +289,18 @@ func TestProxyMultipleOrigins(t *testing.T) {
 	}
 
 	for _, test := range tests {
-		respWriter := newMockHTTPRespWriter()
+		responseWriter := newMockHTTPRespWriter()
 		req, err := http.NewRequest(http.MethodGet, test.url, nil)
 		require.NoError(t, err)
 
-		err = client.Proxy(respWriter, req, false)
+		err = proxy.Proxy(responseWriter, req, connection.TypeHTTP)
 		require.NoError(t, err)
 
-		assert.Equal(t, test.expectedStatus, respWriter.Code)
+		assert.Equal(t, test.expectedStatus, responseWriter.Code)
 		if test.expectedBody != nil {
-			assert.Equal(t, test.expectedBody, respWriter.Body.Bytes())
+			assert.Equal(t, test.expectedBody, responseWriter.Body.Bytes())
 		} else {
-			assert.Equal(t, 0, respWriter.Body.Len())
+			assert.Equal(t, 0, responseWriter.Body.Len())
 		}
 	}
 	cancel()
@@ -327,7 +326,7 @@ func TestProxyError(t *testing.T) {
 			{
 				Hostname: "*",
 				Path:     nil,
-				Service: ingress.MockOriginService{
+				Service: ingress.MockOriginHTTPService{
 					Transport: errorOriginTransport{},
 				},
 			},
@@ -336,14 +335,509 @@ func TestProxyError(t *testing.T) {
 
 	log := zerolog.Nop()
 
-	client := NewClient(ingress, testTags, &log)
+	proxy := NewOriginProxy(ingress, unusedWarpRoutingService, testTags, &log)
 
-	respWriter := newMockHTTPRespWriter()
+	responseWriter := newMockHTTPRespWriter()
 	req, err := http.NewRequest(http.MethodGet, "http://127.0.0.1", nil)
 	assert.NoError(t, err)
 
-	err = client.Proxy(respWriter, req, false)
-	assert.Error(t, err)
-	assert.Equal(t, http.StatusBadGateway, respWriter.Code)
-	assert.Equal(t, "http response error", respWriter.Body.String())
+	assert.Error(t, proxy.Proxy(responseWriter, req, connection.TypeHTTP))
+}
+
+type replayer struct {
+	sync.RWMutex
+	writeDone chan struct{}
+	rw        *bytes.Buffer
+}
+
+func newReplayer(buffer *bytes.Buffer) {
+
+}
+
+func (r *replayer) Read(p []byte) (int, error) {
+	r.RLock()
+	defer r.RUnlock()
+	return r.rw.Read(p)
+}
+
+func (r *replayer) Write(p []byte) (int, error) {
+	r.Lock()
+	defer r.Unlock()
+	n, err := r.rw.Write(p)
+	return n, err
+}
+
+func (r *replayer) String() string {
+	r.Lock()
+	defer r.Unlock()
+	return r.rw.String()
+}
+
+func (r *replayer) Bytes() []byte {
+	r.Lock()
+	defer r.Unlock()
+	return r.rw.Bytes()
+}
+
+// TestConnections tests every possible permutation of connection protocols
+// proxied by cloudflared.
+//
+// WS - WS : When a websocket based ingress is configured on the origin and
+// the eyeball is also a websocket client streaming data.
+// TCP - TCP : When teamnet is enabled and an http or tcp service is running
+// on the origin.
+// TCP - WS: When teamnet is enabled and a websocket based service is running
+// on the origin.
+// WS - TCP: When a tcp based ingress is configured on the origin and the
+// eyeball sends tcp packets wrapped in websockets. (E.g: cloudflared access).
+func TestConnections(t *testing.T) {
+	logger := logger.Create(nil)
+	replayer := &replayer{rw: &bytes.Buffer{}}
+	type args struct {
+		ingressServiceScheme  string
+		originService         func(*testing.T, net.Listener)
+		eyeballResponseWriter connection.ResponseWriter
+		eyeballRequestBody    io.ReadCloser
+
+		// Can be set to nil to show warp routing is not enabled.
+		warpRoutingService *ingress.WarpRoutingService
+
+		// eyeball connection type.
+		connectionType connection.Type
+
+		//requestheaders to be sent in the call to proxy.Proxy
+		requestHeaders http.Header
+	}
+
+	type want struct {
+		message []byte
+		headers http.Header
+		err     bool
+	}
+
+	var tests = []struct {
+		name string
+		args args
+		want want
+	}{
+		{
+			name: "ws-ws proxy",
+			args: args{
+				ingressServiceScheme:  "ws://",
+				originService:         runEchoWSService,
+				eyeballResponseWriter: newWSRespWriter(replayer),
+				eyeballRequestBody:    newWSRequestBody([]byte("test1")),
+				connectionType:        connection.TypeWebsocket,
+				requestHeaders: map[string][]string{
+					// Example key from https://tools.ietf.org/html/rfc6455#section-1.2
+					"Sec-Websocket-Key":     {"dGhlIHNhbXBsZSBub25jZQ=="},
+					"Test-Cloudflared-Echo": {"Echo"},
+				},
+			},
+			want: want{
+				message: []byte("echo-test1"),
+				headers: map[string][]string{
+					"Connection":            {"Upgrade"},
+					"Sec-Websocket-Accept":  {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
+					"Upgrade":               {"websocket"},
+					"Test-Cloudflared-Echo": {"Echo"},
+				},
+			},
+		},
+		{
+			name: "tcp-tcp proxy",
+			args: args{
+				ingressServiceScheme:  "tcp://",
+				originService:         runEchoTCPService,
+				eyeballResponseWriter: newTCPRespWriter(replayer),
+				eyeballRequestBody:    newTCPRequestBody([]byte("test2")),
+				warpRoutingService:    ingress.NewWarpRoutingService(),
+				connectionType:        connection.TypeTCP,
+				requestHeaders: map[string][]string{
+					"Cf-Cloudflared-Proxy-Src": {"non-blank-value"},
+				},
+			},
+			want: want{
+				message: []byte("echo-test2"),
+			},
+		},
+		{
+			name: "tcp-ws proxy",
+			args: args{
+				ingressServiceScheme: "ws://",
+				originService:        runEchoWSService,
+				//eyeballResponseWriter gets set after roundtrip dial.
+				eyeballRequestBody: newPipedWSRequestBody([]byte("test3")),
+				warpRoutingService: ingress.NewWarpRoutingService(),
+				requestHeaders: map[string][]string{
+					"Cf-Cloudflared-Proxy-Src": {"non-blank-value"},
+				},
+				connectionType: connection.TypeTCP,
+			},
+			want: want{
+				message: []byte("echo-test3"),
+				// We expect no headers here because they are sent back via
+				// the stream.
+			},
+		},
+		{
+			name: "ws-tcp proxy",
+			args: args{
+				ingressServiceScheme:  "tcp://",
+				originService:         runEchoTCPService,
+				eyeballResponseWriter: newWSRespWriter(replayer),
+				eyeballRequestBody:    newWSRequestBody([]byte("test4")),
+				connectionType:        connection.TypeWebsocket,
+				requestHeaders: map[string][]string{
+					// Example key from https://tools.ietf.org/html/rfc6455#section-1.2
+					"Sec-Websocket-Key": {"dGhlIHNhbXBsZSBub25jZQ=="},
+				},
+			},
+			want: want{
+				message: []byte("echo-test4"),
+				headers: map[string][]string{
+					"Connection":           {"Upgrade"},
+					"Sec-Websocket-Accept": {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
+					"Upgrade":              {"websocket"},
+				},
+			},
+		},
+		{
+			name: "tcp-tcp proxy without warpRoutingService enabled",
+			args: args{
+				ingressServiceScheme:  "tcp://",
+				originService:         runEchoTCPService,
+				eyeballResponseWriter: newTCPRespWriter(replayer),
+				eyeballRequestBody:    newTCPRequestBody([]byte("test2")),
+				connectionType:        connection.TypeTCP,
+				requestHeaders: map[string][]string{
+					"Cf-Cloudflared-Proxy-Src": {"non-blank-value"},
+				},
+			},
+			want: want{
+				message: []byte{},
+				err:     true,
+			},
+		},
+		{
+			name: "ws-ws proxy when origin is different",
+			args: args{
+				ingressServiceScheme:  "ws://",
+				originService:         runEchoWSService,
+				eyeballResponseWriter: newWSRespWriter(replayer),
+				eyeballRequestBody:    newWSRequestBody([]byte("test1")),
+				connectionType:        connection.TypeWebsocket,
+				requestHeaders: map[string][]string{
+					// Example key from https://tools.ietf.org/html/rfc6455#section-1.2
+					"Sec-Websocket-Key": {"dGhlIHNhbXBsZSBub25jZQ=="},
+					"Origin":            {"Different origin"},
+				},
+			},
+			want: want{
+				message: []byte{},
+				err:     true,
+			},
+		},
+		{
+			name: "tcp-* proxy when origin service has already closed the connection/ is no longer running",
+			args: args{
+				ingressServiceScheme: "tcp://",
+				originService: func(t *testing.T, ln net.Listener) {
+					// closing the listener created by the test.
+					ln.Close()
+				},
+				eyeballResponseWriter: newTCPRespWriter(replayer),
+				eyeballRequestBody:    newTCPRequestBody([]byte("test2")),
+				connectionType:        connection.TypeTCP,
+				requestHeaders: map[string][]string{
+					"Cf-Cloudflared-Proxy-Src": {"non-blank-value"},
+				},
+			},
+			want: want{
+				message: []byte{},
+				err:     true,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			ln, err := net.Listen("tcp", "127.0.0.1:0")
+			require.NoError(t, err)
+			// Starts origin service
+			test.args.originService(t, ln)
+
+			ingressRule := createSingleIngressConfig(t, test.args.ingressServiceScheme+ln.Addr().String())
+			var wg sync.WaitGroup
+			errC := make(chan error)
+			ingressRule.StartOrigins(&wg, logger, ctx.Done(), errC)
+			proxy := NewOriginProxy(ingressRule, test.args.warpRoutingService, testTags, logger)
+
+			req, err := http.NewRequest(
+				http.MethodGet,
+				test.args.ingressServiceScheme+ln.Addr().String(),
+				test.args.eyeballRequestBody,
+			)
+			require.NoError(t, err)
+
+			req.Header = test.args.requestHeaders
+			respWriter := test.args.eyeballResponseWriter
+
+			if pipedReqBody, ok := test.args.eyeballRequestBody.(*pipedRequestBody); ok {
+				respWriter = newTCPRespWriter(pipedReqBody.pipedConn)
+				go func() {
+					resp := pipedReqBody.roundtrip(test.args.ingressServiceScheme + ln.Addr().String())
+					replayer.Write(resp)
+				}()
+			}
+
+			err = proxy.Proxy(respWriter, req, test.args.connectionType)
+
+			cancel()
+			assert.Equal(t, test.want.err, err != nil)
+			assert.Equal(t, test.want.message, replayer.Bytes())
+			respPrinter := respWriter.(responsePrinter)
+			assert.Equal(t, test.want.headers, respPrinter.headers())
+			replayer.rw.Reset()
+		})
+	}
+}
+
+type requestBody struct {
+	pw *io.PipeWriter
+	pr *io.PipeReader
+}
+
+func newWSRequestBody(data []byte) *requestBody {
+	pr, pw := io.Pipe()
+	go wsutil.WriteClientBinary(pw, data)
+	return &requestBody{
+		pr: pr,
+		pw: pw,
+	}
+}
+func newTCPRequestBody(data []byte) *requestBody {
+	pr, pw := io.Pipe()
+	go pw.Write(data)
+	return &requestBody{
+		pr: pr,
+		pw: pw,
+	}
+}
+
+func (r *requestBody) Read(p []byte) (n int, err error) {
+	return r.pr.Read(p)
+}
+
+func (r *requestBody) Close() error {
+	r.pw.Close()
+	r.pr.Close()
+	return nil
+}
+
+type pipedRequestBody struct {
+	dialer         gorillaWS.Dialer
+	pipedConn      net.Conn
+	wsConn         net.Conn
+	messageToWrite []byte
+}
+
+func newPipedWSRequestBody(data []byte) *pipedRequestBody {
+	conn1, conn2 := net.Pipe()
+	dialer := gorillaWS.Dialer{
+		NetDial: func(network, addr string) (net.Conn, error) {
+			return conn2, nil
+		},
+	}
+	return &pipedRequestBody{
+		dialer:         dialer,
+		pipedConn:      conn1,
+		wsConn:         conn2,
+		messageToWrite: data,
+	}
+}
+
+func (p *pipedRequestBody) roundtrip(addr string) []byte {
+	header := http.Header{}
+	conn, resp, err := p.dialer.Dial(addr, header)
+	if err != nil {
+		panic(err)
+	}
+	defer conn.Close()
+
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		panic(fmt.Errorf("resp returned status code: %d", resp.StatusCode))
+	}
+
+	err = conn.WriteMessage(gorillaWS.TextMessage, p.messageToWrite)
+	if err != nil {
+		panic(err)
+	}
+
+	_, data, err := conn.ReadMessage()
+	if err != nil {
+		panic(err)
+	}
+
+	return data
+}
+
+func (p *pipedRequestBody) Read(data []byte) (n int, err error) {
+	return p.pipedConn.Read(data)
+}
+
+func (p *pipedRequestBody) Close() error {
+	return nil
+}
+
+type responsePrinter interface {
+	headers() http.Header
+}
+
+type wsRespWriter struct {
+	w               io.Writer
+	responseHeaders http.Header
+	code            int
+}
+
+// newWSRespWriter uses wsutil.WriteClientText to generate websocket frames.
+// and wsutil.ReadClientText to translate frames from server to byte data.
+// In essence, this acts as a wsClient.
+func newWSRespWriter(w io.Writer) *wsRespWriter {
+	return &wsRespWriter{
+		w: w,
+	}
+}
+
+// Write is written to by ingress.Stream and serves as the output to the client.
+func (w *wsRespWriter) Write(p []byte) (int, error) {
+	returnedMsg, err := wsutil.ReadServerBinary(bytes.NewBuffer(p))
+	if err != nil {
+		// The data was not returned by a websocket connecton.
+		if err != io.ErrUnexpectedEOF {
+			return w.w.Write(p)
+		}
+	}
+	return w.w.Write(returnedMsg)
+}
+
+func (w *wsRespWriter) WriteRespHeaders(status int, header http.Header) error {
+	w.responseHeaders = header
+	w.code = status
+	return nil
+}
+
+// respHeaders is a test function to read respHeaders
+func (w *wsRespWriter) headers() http.Header {
+	return w.responseHeaders
+}
+
+type mockTCPRespWriter struct {
+	w               io.Writer
+	responseHeaders http.Header
+	code            int
+}
+
+func newTCPRespWriter(w io.Writer) *mockTCPRespWriter {
+	return &mockTCPRespWriter{
+		w: w,
+	}
+}
+
+func (m *mockTCPRespWriter) Write(p []byte) (n int, err error) {
+	return m.w.Write(p)
+}
+
+func (m *mockTCPRespWriter) WriteRespHeaders(status int, header http.Header) error {
+	m.responseHeaders = header
+	m.code = status
+	return nil
+}
+
+// respHeaders is a test function to read respHeaders
+func (m *mockTCPRespWriter) headers() http.Header {
+	return m.responseHeaders
+}
+
+func createSingleIngressConfig(t *testing.T, service string) ingress.Ingress {
+	ingressConfig := &config.Configuration{
+		Ingress: []config.UnvalidatedIngressRule{
+			{
+				Hostname: "*",
+				Service:  service,
+			},
+		},
+	}
+	ingressRule, err := ingress.ParseIngress(ingressConfig)
+	require.NoError(t, err)
+	return ingressRule
+}
+
+func runEchoTCPService(t *testing.T, l net.Listener) {
+	go func() {
+		for {
+			conn, err := l.Accept()
+			require.NoError(t, err)
+			defer conn.Close()
+
+			for {
+				buf := make([]byte, 1024)
+				size, err := conn.Read(buf)
+				if err == io.EOF {
+					return
+				}
+				data := []byte("echo-")
+				data = append(data, buf[:size]...)
+				_, err = conn.Write(data)
+				if err != nil {
+					t.Log(err)
+				}
+				return
+			}
+		}
+	}()
+}
+
+func runEchoWSService(t *testing.T, l net.Listener) {
+	var upgrader = gorillaWS.Upgrader{
+		ReadBufferSize:  10,
+		WriteBufferSize: 10,
+	}
+
+	var ws = func(w http.ResponseWriter, r *http.Request) {
+		header := make(http.Header)
+		for k, vs := range r.Header {
+			if k == "Test-Cloudflared-Echo" {
+				header[k] = vs
+			}
+		}
+		conn, err := upgrader.Upgrade(w, r, header)
+		if err != nil {
+			t.Log(err)
+			return
+		}
+		defer conn.Close()
+
+		for {
+			messageType, p, err := conn.ReadMessage()
+			if err != nil {
+				return
+			}
+			data := []byte("echo-")
+			data = append(data, p...)
+			if err := conn.WriteMessage(messageType, data); err != nil {
+				return
+			}
+		}
+	}
+
+	server := http.Server{
+		Handler: http.HandlerFunc(ws),
+	}
+
+	go func() {
+		err := server.Serve(l)
+		require.NoError(t, err)
+	}()
 }

origin/tunnel_test.go

@@ -40,8 +40,10 @@ func TestWaitForBackoffFallback(t *testing.T) {
 	mockFetcher := dynamicMockFetcher{
 		percentage: 0,
 	}
+	warpRoutingEnabled := false
 	protocolSelector, err := connection.NewProtocolSelector(
 		connection.HTTP2.String(),
+		warpRoutingEnabled,
 		namedTunnel,
 		mockFetcher.fetch(),
 		resolveTTL,

socks/request_handler.go

@@ -3,7 +3,10 @@ package socks
 import (
 	"fmt"
 	"io"
+	"net"
 	"strings"
+
+	"github.com/rs/zerolog"
 )
 
 // RequestHandler is the functions needed to handle a SOCKS5 command
@@ -104,3 +107,13 @@ func (h *StandardRequestHandler) handleAssociate(conn io.ReadWriter, req *Reques
 	}
 	return nil
 }
+
+func StreamHandler(tunnelConn io.ReadWriter, originConn net.Conn, log *zerolog.Logger) {
+	dialer := NewConnDialer(originConn)
+	requestHandler := NewRequestHandler(dialer)
+	socksServer := NewConnectionHandler(requestHandler)
+
+	if err := socksServer.Serve(tunnelConn); err != nil {
+		log.Debug().Err(err).Msg("Socks stream handler error")
+	}
+}

teamnet/api.go

@@ -16,9 +16,9 @@ import (
 // network, and says that eyeballs can reach that route using the corresponding
 // tunnel.
 type Route struct {
-	Network   CIDR
+	Network   CIDR      `json:"network"`
 	TunnelID  uuid.UUID `json:"tunnel_id"`
-	Comment   string
+	Comment   string    `json:"comment"`
 	CreatedAt time.Time `json:"created_at"`
 	DeletedAt time.Time `json:"deleted_at"`
 }
@@ -26,11 +26,20 @@ type Route struct {
 // CIDR is just a newtype wrapper around net.IPNet. It adds JSON unmarshalling.
 type CIDR net.IPNet
 
-func (c *CIDR) String() string {
-	n := net.IPNet(*c)
+func (c CIDR) String() string {
+	n := net.IPNet(c)
 	return n.String()
 }
 
+func (c CIDR) MarshalJSON() ([]byte, error) {
+	str := c.String()
+	json, err := json.Marshal(str)
+	if err != nil {
+		return nil, errors.Wrap(err, "error serializing CIDR into JSON")
+	}
+	return json, nil
+}
+
 // UnmarshalJSON parses a JSON string into net.IPNet
 func (c *CIDR) UnmarshalJSON(data []byte) error {
 	var s string
@@ -68,9 +77,9 @@ func (r NewRoute) MarshalJSON() ([]byte, error) {
 
 // DetailedRoute is just a Route with some extra fields, e.g. TunnelName.
 type DetailedRoute struct {
-	Network    CIDR
+	Network    CIDR      `json:"network"`
 	TunnelID   uuid.UUID `json:"tunnel_id"`
-	Comment    string
+	Comment    string    `json:"comment"`
 	CreatedAt  time.Time `json:"created_at"`
 	DeletedAt  time.Time `json:"deleted_at"`
 	TunnelName string    `json:"tunnel_name"`

teamnet/api_test.go

@@ -33,15 +33,15 @@ func TestUnmarshalRoute(t *testing.T) {
 	require.Equal(t, "test", r.Comment)
 }
 
-func TestUnmarshalDetailedRoute(t *testing.T) {
+func TestDetailedRouteJsonRoundtrip(t *testing.T) {
 	// Response from the teamnet route backend
 	data := `{
 		"network":"10.1.2.40/29",
 		"tunnel_id":"fba6ffea-807f-4e7a-a740-4184ee1b82c8",
-		"tunnel_name":"Mr. Tun",
 		"comment":"test",
 		"created_at":"2020-12-22T02:00:15.587008Z",
-		"deleted_at":null
+		"deleted_at":"2021-01-14T05:01:42.183002Z",
+		"tunnel_name":"Mr. Tun"
 	}`
 	var r DetailedRoute
 	err := json.Unmarshal([]byte(data), &r)
@@ -55,6 +55,13 @@ func TestUnmarshalDetailedRoute(t *testing.T) {
 	require.Equal(t, CIDR(*cidr), r.Network)
 	require.Equal(t, "test", r.Comment)
 	require.Equal(t, "Mr. Tun", r.TunnelName)
+
+	bytes, err := json.Marshal(r)
+	require.NoError(t, err)
+	obtainedJson := string(bytes)
+	data = strings.Replace(data, "\t", "", -1)
+	data = strings.Replace(data, "\n", "", -1)
+	require.Equal(t, data, obtainedJson)
 }
 
 func TestMarshalNewRoute(t *testing.T) {

vendor/github.com/rs/zerolog/go.mod

@@ -1,5 +1,7 @@
 module github.com/rs/zerolog
 
+go 1.15
+
 require (
 	github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e
 	github.com/pkg/errors v0.8.1

websocket/connection.go

@@ -0,0 +1,119 @@
+package websocket
+
+import (
+	"context"
+	"io"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	gobwas "github.com/gobwas/ws"
+	"github.com/gobwas/ws/wsutil"
+	"github.com/gorilla/websocket"
+)
+
+const (
+	// Time allowed to write a message to the peer.
+	writeWait = 10 * time.Second
+
+	// Time allowed to read the next pong message from the peer.
+	pongWait = 60 * time.Second
+
+	// Send pings to peer with this period. Must be less than pongWait.
+	pingPeriod = (pongWait * 9) / 10
+)
+
+// GorillaConn is a wrapper around the standard gorilla websocket but implements a ReadWriter
+// This is still used by access carrier
+type GorillaConn struct {
+	*websocket.Conn
+	log *zerolog.Logger
+}
+
+// Read will read messages from the websocket connection
+func (c *GorillaConn) Read(p []byte) (int, error) {
+	_, message, err := c.Conn.ReadMessage()
+	if err != nil {
+		return 0, err
+	}
+	return copy(p, message), nil
+
+}
+
+// Write will write messages to the websocket connection
+func (c *GorillaConn) Write(p []byte) (int, error) {
+	if err := c.Conn.WriteMessage(websocket.BinaryMessage, p); err != nil {
+		return 0, err
+	}
+
+	return len(p), nil
+}
+
+// pinger simulates the websocket connection to keep it alive
+func (c *GorillaConn) pinger(ctx context.Context) {
+	ticker := time.NewTicker(pingPeriod)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			if err := c.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait)); err != nil {
+				c.log.Debug().Msgf("failed to send ping message: %s", err)
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+type Conn struct {
+	rw  io.ReadWriter
+	log *zerolog.Logger
+}
+
+func NewConn(ctx context.Context, rw io.ReadWriter, log *zerolog.Logger) *Conn {
+	c := &Conn{
+		rw:  rw,
+		log: log,
+	}
+	go c.pinger(ctx)
+	return c
+}
+
+// Read will read messages from the websocket connection
+func (c *Conn) Read(reader []byte) (int, error) {
+	data, err := wsutil.ReadClientBinary(c.rw)
+	if err != nil {
+		return 0, err
+	}
+	return copy(reader, data), nil
+}
+
+// Write will write messages to the websocket connection
+func (c *Conn) Write(p []byte) (int, error) {
+	if err := wsutil.WriteServerBinary(c.rw, p); err != nil {
+		return 0, err
+	}
+	return len(p), nil
+}
+
+func (c *Conn) pinger(ctx context.Context) {
+	pongMessge := wsutil.Message{
+		OpCode:  gobwas.OpPong,
+		Payload: []byte{},
+	}
+	ticker := time.NewTicker(pingPeriod)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			if err := wsutil.WriteServerMessage(c.rw, gobwas.OpPing, []byte{}); err != nil {
+				c.log.Err(err).Msgf("failed to write ping message")
+			}
+			if err := wsutil.HandleClientControlMessage(c.rw, pongMessge); err != nil {
+				c.log.Err(err).Msgf("failed to write pong message")
+			}
+		case <-ctx.Done():
+			return
+		}
+	}
+}

websocket/websocket.go

@@ -2,7 +2,6 @@ package websocket
 
 import (
 	"crypto/sha1"
-	"crypto/tls"
 	"encoding/base64"
 	"io"
 	"net"
@@ -16,17 +15,6 @@ import (
 	"github.com/rs/zerolog"
 )
 
-const (
-	// Time allowed to write a message to the peer.
-	writeWait = 10 * time.Second
-
-	// Time allowed to read the next pong message from the peer.
-	pongWait = 60 * time.Second
-
-	// Send pings to peer with this period. Must be less than pongWait.
-	pingPeriod = (pongWait * 9) / 10
-)
-
 var stripWebsocketHeaders = []string{
 	"Upgrade",
 	"Connection",
@@ -35,94 +23,28 @@ var stripWebsocketHeaders = []string{
 	"Sec-Websocket-Extensions",
 }
 
-// Conn is a wrapper around the standard gorilla websocket
-// but implements a ReadWriter
-type Conn struct {
-	*websocket.Conn
-}
-
-// Read will read messages from the websocket connection
-func (c *Conn) Read(p []byte) (int, error) {
-	_, message, err := c.Conn.ReadMessage()
-	if err != nil {
-		return 0, err
-	}
-
-	return copy(p, message), nil
-
-}
-
-// Write will write messages to the websocket connection
-func (c *Conn) Write(p []byte) (int, error) {
-	if err := c.Conn.WriteMessage(websocket.BinaryMessage, p); err != nil {
-		return 0, err
-	}
-
-	return len(p), nil
-}
-
 // IsWebSocketUpgrade checks to see if the request is a WebSocket connection.
 func IsWebSocketUpgrade(req *http.Request) bool {
 	return websocket.IsWebSocketUpgrade(req)
 }
 
-// Dialler is something that can proxy websocket requests.
-type Dialler interface {
-	Dial(url *url.URL, headers http.Header) (*websocket.Conn, *http.Response, error)
-}
-
-type defaultDialler struct {
-	tlsConfig *tls.Config
-}
-
-func (dd *defaultDialler) Dial(url *url.URL, header http.Header) (*websocket.Conn, *http.Response, error) {
-	d := &websocket.Dialer{
-		TLSClientConfig: dd.tlsConfig,
-		Proxy:           http.ProxyFromEnvironment,
-	}
-	return d.Dial(url.String(), header)
-}
-
 // ClientConnect creates a WebSocket client connection for provided request. Caller is responsible for closing
 // the connection. The response body may not contain the entire response and does
 // not need to be closed by the application.
-func ClientConnect(req *http.Request, dialler Dialler) (*websocket.Conn, *http.Response, error) {
+func ClientConnect(req *http.Request, dialler *websocket.Dialer) (*websocket.Conn, *http.Response, error) {
 	req.URL.Scheme = ChangeRequestScheme(req.URL)
 	wsHeaders := websocketHeaders(req)
-
 	if dialler == nil {
-		dialler = new(defaultDialler)
+		dialler = &websocket.Dialer{
+			Proxy: http.ProxyFromEnvironment,
+		}
 	}
-	conn, response, err := dialler.Dial(req.URL, wsHeaders)
+	conn, response, err := dialler.Dial(req.URL.String(), wsHeaders)
 	if err != nil {
 		return nil, response, err
 	}
 	response.Header.Set("Sec-WebSocket-Accept", generateAcceptKey(req))
-	return conn, response, err
-}
-
-// Stream copies copy data to & from provided io.ReadWriters.
-func Stream(conn, backendConn io.ReadWriter) {
-	proxyDone := make(chan struct{}, 2)
-
-	go func() {
-		_, _ = io.Copy(conn, backendConn)
-		proxyDone <- struct{}{}
-	}()
-
-	go func() {
-		_, _ = io.Copy(backendConn, conn)
-		proxyDone <- struct{}{}
-	}()
-
-	// If one side is done, we are done.
-	<-proxyDone
-}
-
-// DefaultStreamHandler is provided to the the standard websocket to origin stream
-// This exist to allow SOCKS to deframe data before it gets to the origin
-func DefaultStreamHandler(wsConn *Conn, remoteConn net.Conn, _ http.Header) {
-	Stream(wsConn, remoteConn)
+	return conn, response, nil
 }
 
 // StartProxyServer will start a websocket server that will decode
@@ -132,7 +54,7 @@ func StartProxyServer(
 	listener net.Listener,
 	staticHost string,
 	shutdownC <-chan struct{},
-	streamHandler func(wsConn *Conn, remoteConn net.Conn, requestHeaders http.Header),
+	streamHandler func(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger),
 ) error {
 	upgrader := websocket.Upgrader{
 		ReadBufferSize:  1024,
@@ -159,7 +81,7 @@ type handler struct {
 	log           *zerolog.Logger
 	staticHost    string
 	upgrader      websocket.Upgrader
-	streamHandler func(wsConn *Conn, remoteConn net.Conn, requestHeaders http.Header)
+	streamHandler func(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger)
 }
 
 func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -192,14 +114,20 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 	_ = conn.SetReadDeadline(time.Now().Add(pongWait))
 	conn.SetPongHandler(func(string) error { _ = conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
-	done := make(chan struct{})
-	go pinger(h.log, conn, done)
-	defer func() {
-		done <- struct{}{}
-		_ = conn.Close()
-	}()
+	gorillaConn := &GorillaConn{conn, h.log}
+	go gorillaConn.pinger(r.Context())
+	defer conn.Close()
 
-	h.streamHandler(&Conn{conn}, stream, r.Header)
+	h.streamHandler(gorillaConn, stream, h.log)
+}
+
+// NewResponseHeader returns headers needed to return to origin for completing handshake
+func NewResponseHeader(req *http.Request) http.Header {
+	header := http.Header{}
+	header.Add("Connection", "Upgrade")
+	header.Add("Sec-Websocket-Accept", generateAcceptKey(req))
+	header.Add("Upgrade", "websocket")
+	return header
 }
 
 // the gorilla websocket library sets its own Upgrade, Connection, Sec-WebSocket-Key,
@@ -246,19 +174,3 @@ func ChangeRequestScheme(reqURL *url.URL) string {
 		return reqURL.Scheme
 	}
 }
-
-// pinger simulates the websocket connection to keep it alive
-func pinger(logger *zerolog.Logger, ws *websocket.Conn, done chan struct{}) {
-	ticker := time.NewTicker(pingPeriod)
-	defer ticker.Stop()
-	for {
-		select {
-		case <-ticker.C:
-			if err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait)); err != nil {
-				logger.Debug().Msgf("failed to send ping message: %s", err)
-			}
-		case <-done:
-			return
-		}
-	}
-}

websocket/websocket_test.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/cloudflare/cloudflared/hello"
 	"github.com/cloudflare/cloudflared/tlsconfig"
-
+	gws "github.com/gorilla/websocket"
 	"github.com/stretchr/testify/assert"
 	"golang.org/x/net/websocket"
 )
@@ -78,7 +78,7 @@ func TestServe(t *testing.T) {
 
 	tlsConfig := websocketClientTLSConfig(t)
 	assert.NotNil(t, tlsConfig)
-	d := defaultDialler{tlsConfig: tlsConfig}
+	d := gws.Dialer{TLSClientConfig: tlsConfig}
 	conn, resp, err := ClientConnect(req, &d)
 	assert.NoError(t, err)
 	assert.Equal(t, testSecWebsocketAccept, resp.Header.Get("Sec-WebSocket-Accept"))