Release 2021.4.0

This commit reverts part of 1cf6ae37eb.

CHANGES.md

@@ -1,5 +1,31 @@
 **Experimental**: This is a new format for release notes. The format and availability is subject to change.
 
+## 2021.4.0
+
+### Bug Fixes
+
+- Fixed proxying of websocket requests to avoid possibility of losing initial frames that were sent in the same TCP
+  packet as response headers [#345](https://github.com/cloudflare/cloudflared/issues/345).
+- `proxy-dns` option now works in conjunction with running a named tunnel [#346](https://github.com/cloudflare/cloudflared/issues/346).
+
+## 2021.3.6
+
+### Bug Fixes
+
+- Reverted 2021.3.5 improvement to use HTTP/2 in a best-effort manner between cloudflared and origin services because
+  it was found to break in some cases.
+
+## 2021.3.5
+
+### Improvements
+
+ - HTTP/2 transport is now always chosen if origin server supports it and the service url scheme is HTTPS.
+   This was previously done in a best attempt manner.
+
+### Bug Fixes
+
+ - The MacOS binaries were not successfully released in 2021.3.3 and 2021.3.4. This release is aimed at addressing that.
+
 ## 2021.3.3
 
 ### Improvements

RELEASE_NOTES

@@ -1,3 +1,24 @@
+2021.4.0
+- 2021-04-05 TUN-4178: Fix component test for running as a service in MacOS to not assume a named tunnel
+- 2021-04-05 TUN-4177: Running with proxy-dns should not prevent running Named Tunnels
+- 2021-04-02 TUN-4168: Transparently proxy websocket connections using stdlib HTTP client instead of gorilla/websocket; move websocket client code into carrier package since it's only used by access subcommands now (#345).
+- 2021-04-07 Publish change log for 2021.4.0
+
+2021.3.6
+- 2021-03-30 TUN-4150: Only show the connector table in 'tunnel info' if there are connectors. Don't show rows with zero connections.
+- 2021-03-31 TUN-4153: Revert best-effort HTTP2 usage when talking to origins
+- 2021-03-26 TUN-4141: Better error messages for tunnel info subcommand.
+- 2021-03-29 TUN-4146: Unhide and document grace-period
+- 2021-03-25 TUN-3863: Consolidate header handling logic in the connection package; move headers definitions from h2mux to packages that manage them; cleanup header conversions
+
+2021.3.5
+- 2021-03-26 TUN-3896: http-service and tunnelstore client use http2 transport.
+- 2021-03-25 TUN-4125: Change component tests to run in CI with its own dedicated resources
+- 2021-03-26 Publish change log for 2021.3.5
+
+2021.3.4
+
+
 2021.3.3
 - 2021-03-23 TUN-4111: Warn the user if both properties "tunnel" and "hostname" are used
 - 2021-03-23 TUN-4082: Test logging when running as a service

carrier/carrier.go

@@ -5,20 +5,25 @@ package carrier
 
 import (
 	"crypto/tls"
+	"fmt"
 	"io"
 	"net"
 	"net/http"
+	"net/url"
 	"os"
 	"strings"
 
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 
-	"github.com/cloudflare/cloudflared/h2mux"
 	"github.com/cloudflare/cloudflared/token"
 )
 
-const LogFieldOriginURL = "originURL"
+const (
+	LogFieldOriginURL       = "originURL"
+	CFAccessTokenHeader     = "Cf-Access-Token"
+	cfJumpDestinationHeader = "Cf-Access-Jump-Destination"
+)
 
 type StartOptions struct {
 	AppInfo         *token.AppInfo
@@ -32,15 +37,11 @@ type StartOptions struct {
 type Connection interface {
 	// ServeStream is used to forward data from the client to the edge
 	ServeStream(*StartOptions, io.ReadWriter) error
-
-	// StartServer is used to listen for incoming connections from the edge to the origin
-	StartServer(net.Listener, string, <-chan struct{}) error
 }
 
 // StdinoutStream is empty struct for wrapping stdin/stdout
 // into a single ReadWriter
-type StdinoutStream struct {
-}
+type StdinoutStream struct{}
 
 // Read will read from Stdin
 func (c *StdinoutStream) Read(p []byte) (int, error) {
@@ -149,7 +150,7 @@ func BuildAccessRequest(options *StartOptions, log *zerolog.Logger) (*http.Reque
 	if err != nil {
 		return nil, err
 	}
-	originRequest.Header.Set(h2mux.CFAccessTokenHeader, token)
+	originRequest.Header.Set(CFAccessTokenHeader, token)
 
 	for k, v := range options.Headers {
 		if len(v) >= 1 {
@@ -159,3 +160,26 @@ func BuildAccessRequest(options *StartOptions, log *zerolog.Logger) (*http.Reque
 
 	return originRequest, nil
 }
+
+func SetBastionDest(header http.Header, destination string) {
+	if destination != "" {
+		header.Set(cfJumpDestinationHeader, destination)
+	}
+}
+
+func ResolveBastionDest(r *http.Request) (string, error) {
+	jumpDestination := r.Header.Get(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]
+}

carrier/carrier_test.go

@@ -156,3 +156,99 @@ func testRequest(t *testing.T, url string, stream io.ReadWriter) *http.Request {
 
 	return req
 }
+
+func TestBastionDestination(t *testing.T) {
+	tests := []struct {
+		name         string
+		header       http.Header
+		expectedDest string
+		wantErr      bool
+	}{
+		{
+			name: "hostname destination",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"localhost"},
+			},
+			expectedDest: "localhost",
+		},
+		{
+			name: "hostname destination with port",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"localhost:9000"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "hostname destination with scheme and port",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"ssh://localhost:9000"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "full hostname url",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"ssh://localhost:9000/metrics"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "hostname destination with port and path",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"localhost:9000/metrics"},
+			},
+			expectedDest: "localhost:9000",
+		},
+		{
+			name: "ip destination",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"127.0.0.1"},
+			},
+			expectedDest: "127.0.0.1",
+		},
+		{
+			name: "ip destination with port",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"127.0.0.1:9000"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name: "ip destination with port and path",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"127.0.0.1:9000/metrics"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name: "ip destination with schem and port",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"tcp://127.0.0.1:9000"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name: "full ip url",
+			header: http.Header{
+				cfJumpDestinationHeader: []string{"ssh://127.0.0.1:9000/metrics"},
+			},
+			expectedDest: "127.0.0.1:9000",
+		},
+		{
+			name:    "no destination",
+			wantErr: true,
+		},
+	}
+	for _, test := range tests {
+		r := &http.Request{
+			Header: test.header,
+		}
+		dest, err := ResolveBastionDest(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)
+		}
+	}
+}

carrier/websocket.go

@@ -1,17 +1,14 @@
 package carrier
 
 import (
-	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"net/http/httputil"
+	"net/url"
 
 	"github.com/gorilla/websocket"
 	"github.com/rs/zerolog"
 
-	"github.com/cloudflare/cloudflared/ingress"
-	"github.com/cloudflare/cloudflared/socks"
 	"github.com/cloudflare/cloudflared/token"
 	cfwebsocket "github.com/cloudflare/cloudflared/websocket"
 )
@@ -23,20 +20,6 @@ type Websocket struct {
 	isSocks bool
 }
 
-type wsdialer struct {
-	conn *cfwebsocket.GorillaConn
-}
-
-func (d *wsdialer) Dial(address string) (io.ReadWriteCloser, *socks.AddrSpec, error) {
-	local, ok := d.conn.LocalAddr().(*net.TCPAddr)
-	if !ok {
-		return nil, nil, fmt.Errorf("not a tcp connection")
-	}
-
-	addr := socks.AddrSpec{IP: local.IP, Port: local.Port}
-	return d.conn, &addr, nil
-}
-
 // NewWSConnection returns a new connection object
 func NewWSConnection(log *zerolog.Logger) Connection {
 	return &Websocket{
@@ -54,16 +37,10 @@ func (ws *Websocket) ServeStream(options *StartOptions, conn io.ReadWriter) erro
 	}
 	defer wsConn.Close()
 
-	ingress.Stream(wsConn, conn, ws.log)
+	cfwebsocket.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, 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
@@ -84,7 +61,7 @@ func createWebsocketStream(options *StartOptions, log *zerolog.Logger) (*cfwebso
 		TLSClientConfig: options.TLSClientConfig,
 		Proxy:           http.ProxyFromEnvironment,
 	}
-	wsConn, resp, err := cfwebsocket.ClientConnect(req, dialer)
+	wsConn, resp, err := clientConnect(req, dialer)
 	defer closeRespBody(resp)
 
 	if err != nil && IsAccessResponse(resp) {
@@ -111,6 +88,63 @@ func createWebsocketStream(options *StartOptions, log *zerolog.Logger) (*cfwebso
 	return &cfwebsocket.GorillaConn{Conn: wsConn}, nil
 }
 
+var stripWebsocketHeaders = []string{
+	"Upgrade",
+	"Connection",
+	"Sec-Websocket-Key",
+	"Sec-Websocket-Version",
+	"Sec-Websocket-Extensions",
+}
+
+// the gorilla websocket library sets its own Upgrade, Connection, Sec-WebSocket-Key,
+// Sec-WebSocket-Version and Sec-Websocket-Extensions headers.
+// https://github.com/gorilla/websocket/blob/master/client.go#L189-L194.
+func websocketHeaders(req *http.Request) http.Header {
+	wsHeaders := make(http.Header)
+	for key, val := range req.Header {
+		wsHeaders[key] = val
+	}
+	// Assume the header keys are in canonical format.
+	for _, header := range stripWebsocketHeaders {
+		wsHeaders.Del(header)
+	}
+	wsHeaders.Set("Host", req.Host) // See TUN-1097
+	return wsHeaders
+}
+
+// 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 *websocket.Dialer) (*websocket.Conn, *http.Response, error) {
+	req.URL.Scheme = changeRequestScheme(req.URL)
+	wsHeaders := websocketHeaders(req)
+	if dialler == nil {
+		dialler = &websocket.Dialer{
+			Proxy: http.ProxyFromEnvironment,
+		}
+	}
+	conn, response, err := dialler.Dial(req.URL.String(), wsHeaders)
+	if err != nil {
+		return nil, response, err
+	}
+	return conn, response, nil
+}
+
+// changeRequestScheme is needed as the gorilla websocket library requires the ws scheme.
+// (even though it changes it back to http/https, but ¯\_(ツ)_/¯.)
+func changeRequestScheme(reqURL *url.URL) string {
+	switch reqURL.Scheme {
+	case "https":
+		return "wss"
+	case "http":
+		return "ws"
+	case "":
+		return "ws"
+	default:
+		return reqURL.Scheme
+	}
+}
+
 // createAccessAuthenticatedStream will try load a token from storage and make
 // a connection with the token set on the request. If it still get redirect,
 // this probably means the token in storage is invalid (expired/revoked). If that
@@ -150,7 +184,7 @@ func createAccessWebSocketStream(options *StartOptions, log *zerolog.Logger) (*w
 	dump, err := httputil.DumpRequest(req, false)
 	log.Debug().Msgf("Access Websocket request: %s", string(dump))
 
-	conn, resp, err := cfwebsocket.ClientConnect(req, nil)
+	conn, resp, err := clientConnect(req, nil)
 
 	if resp != nil {
 		r, err := httputil.DumpResponse(resp, true)

carrier/websocket_test.go

@@ -0,0 +1,123 @@
+package carrier
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	gws "github.com/gorilla/websocket"
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/net/websocket"
+
+	"github.com/cloudflare/cloudflared/hello"
+	"github.com/cloudflare/cloudflared/tlsconfig"
+	cfwebsocket "github.com/cloudflare/cloudflared/websocket"
+)
+
+func websocketClientTLSConfig(t *testing.T) *tls.Config {
+	certPool := x509.NewCertPool()
+	helloCert, err := tlsconfig.GetHelloCertificateX509()
+	assert.NoError(t, err)
+	certPool.AddCert(helloCert)
+	assert.NotNil(t, certPool)
+	return &tls.Config{RootCAs: certPool}
+}
+
+func TestWebsocketHeaders(t *testing.T) {
+	req := testRequest(t, "http://example.com", nil)
+	wsHeaders := websocketHeaders(req)
+	for _, header := range stripWebsocketHeaders {
+		assert.Empty(t, wsHeaders[header])
+	}
+	assert.Equal(t, "curl/7.59.0", wsHeaders.Get("User-Agent"))
+}
+
+func TestServe(t *testing.T) {
+	log := zerolog.Nop()
+	shutdownC := make(chan struct{})
+	errC := make(chan error)
+	listener, err := hello.CreateTLSListener("localhost:1111")
+	assert.NoError(t, err)
+	defer listener.Close()
+
+	go func() {
+		errC <- hello.StartHelloWorldServer(&log, listener, shutdownC)
+	}()
+
+	req := testRequest(t, "https://localhost:1111/ws", nil)
+
+	tlsConfig := websocketClientTLSConfig(t)
+	assert.NotNil(t, tlsConfig)
+	d := gws.Dialer{TLSClientConfig: tlsConfig}
+	conn, resp, err := clientConnect(req, &d)
+	assert.NoError(t, err)
+	assert.Equal(t, "websocket", resp.Header.Get("Upgrade"))
+
+	for i := 0; i < 1000; i++ {
+		messageSize := rand.Int()%2048 + 1
+		clientMessage := make([]byte, messageSize)
+		// rand.Read always returns len(clientMessage) and a nil error
+		rand.Read(clientMessage)
+		err = conn.WriteMessage(websocket.BinaryFrame, clientMessage)
+		assert.NoError(t, err)
+
+		messageType, message, err := conn.ReadMessage()
+		assert.NoError(t, err)
+		assert.Equal(t, websocket.BinaryFrame, messageType)
+		assert.Equal(t, clientMessage, message)
+	}
+
+	_ = conn.Close()
+	close(shutdownC)
+	<-errC
+}
+
+func TestWebsocketWrapper(t *testing.T) {
+	listener, err := hello.CreateTLSListener("localhost:0")
+	require.NoError(t, err)
+
+	serverErrorChan := make(chan error)
+	helloSvrCtx, cancelHelloSvr := context.WithCancel(context.Background())
+	defer func() { <-serverErrorChan }()
+	defer cancelHelloSvr()
+	go func() {
+		log := zerolog.Nop()
+		serverErrorChan <- hello.StartHelloWorldServer(&log, listener, helloSvrCtx.Done())
+	}()
+
+	tlsConfig := websocketClientTLSConfig(t)
+	d := gws.Dialer{TLSClientConfig: tlsConfig, HandshakeTimeout: time.Minute}
+	testAddr := fmt.Sprintf("https://%s/ws", listener.Addr().String())
+	req := testRequest(t, testAddr, nil)
+	conn, resp, err := clientConnect(req, &d)
+	require.NoError(t, err)
+	assert.Equal(t, "websocket", resp.Header.Get("Upgrade"))
+
+	// Websocket now connected to test server so lets check our wrapper
+	wrapper := cfwebsocket.GorillaConn{Conn: conn}
+	buf := make([]byte, 100)
+	wrapper.Write([]byte("abc"))
+	n, err := wrapper.Read(buf)
+	require.NoError(t, err)
+	require.Equal(t, n, 3)
+	require.Equal(t, "abc", string(buf[:n]))
+
+	// Test partial read, read 1 of 3 bytes in one read and the other 2 in another read
+	wrapper.Write([]byte("abc"))
+	buf = buf[:1]
+	n, err = wrapper.Read(buf)
+	require.NoError(t, err)
+	require.Equal(t, n, 1)
+	require.Equal(t, "a", string(buf[:n]))
+	buf = buf[:cap(buf)]
+	n, err = wrapper.Read(buf)
+	require.NoError(t, err)
+	require.Equal(t, n, 2)
+	require.Equal(t, "bc", string(buf[:n]))
+}

cfsetup.yaml

@@ -220,9 +220,11 @@ stretch: &stretch
     pre-cache:
       - sudo pip3 install --upgrade -r component-tests/requirements.txt
     post-cache:
-      # Constructs config file from env vars
-      - python3 component-tests/config.py
+      # Creates and routes a Named Tunnel for this build. Also constructs config file from env vars.
+      - python3 component-tests/setup.py --type create
       - pytest component-tests
+      # The Named Tunnel is deleted and its route unprovisioned here.
+      - python3 component-tests/setup.py --type cleanup
   update-homebrew:
     builddeps:
       - openssh-client

cmd/cloudflared/access/carrier.go

@@ -6,19 +6,20 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/cloudflare/cloudflared/carrier"
-	"github.com/cloudflare/cloudflared/config"
-	"github.com/cloudflare/cloudflared/h2mux"
-	"github.com/cloudflare/cloudflared/logger"
-	"github.com/cloudflare/cloudflared/validation"
-
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/urfave/cli/v2"
+
+	"github.com/cloudflare/cloudflared/carrier"
+	"github.com/cloudflare/cloudflared/config"
+	"github.com/cloudflare/cloudflared/logger"
+	"github.com/cloudflare/cloudflared/validation"
 )
 
 const (
-	LogFieldHost = "host"
+	LogFieldHost               = "host"
+	cfAccessClientIDHeader     = "Cf-Access-Client-Id"
+	cfAccessClientSecretHeader = "Cf-Access-Client-Secret"
 )
 
 // StartForwarder starts a client side websocket forward
@@ -31,16 +32,14 @@ func StartForwarder(forwarder config.Forwarder, shutdown <-chan struct{}, log *z
 	// get the headers from the config file and add to the request
 	headers := make(http.Header)
 	if forwarder.TokenClientID != "" {
-		headers.Set(h2mux.CFAccessClientIDHeader, forwarder.TokenClientID)
+		headers.Set(cfAccessClientIDHeader, forwarder.TokenClientID)
 	}
 
 	if forwarder.TokenSecret != "" {
-		headers.Set(h2mux.CFAccessClientSecretHeader, forwarder.TokenSecret)
+		headers.Set(cfAccessClientSecretHeader, forwarder.TokenSecret)
 	}
 
-	if forwarder.Destination != "" {
-		headers.Add(h2mux.CFJumpDestinationHeader, forwarder.Destination)
-	}
+	carrier.SetBastionDest(headers, forwarder.Destination)
 
 	options := &carrier.StartOptions{
 		OriginURL: forwarder.URL,
@@ -72,16 +71,13 @@ func ssh(c *cli.Context) error {
 	// get the headers from the cmdline and add them
 	headers := buildRequestHeaders(c.StringSlice(sshHeaderFlag))
 	if c.IsSet(sshTokenIDFlag) {
-		headers.Set(h2mux.CFAccessClientIDHeader, c.String(sshTokenIDFlag))
+		headers.Set(cfAccessClientIDHeader, c.String(sshTokenIDFlag))
 	}
 	if c.IsSet(sshTokenSecretFlag) {
-		headers.Set(h2mux.CFAccessClientSecretHeader, c.String(sshTokenSecretFlag))
+		headers.Set(cfAccessClientSecretHeader, c.String(sshTokenSecretFlag))
 	}
 
-	destination := c.String(sshDestinationFlag)
-	if destination != "" {
-		headers.Add(h2mux.CFJumpDestinationHeader, destination)
-	}
+	carrier.SetBastionDest(headers, c.String(sshDestinationFlag))
 
 	options := &carrier.StartOptions{
 		OriginURL: originURL,

cmd/cloudflared/access/cmd.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/cloudflare/cloudflared/carrier"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/cliutil"
-	"github.com/cloudflare/cloudflared/h2mux"
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/sshgen"
 	"github.com/cloudflare/cloudflared/token"
@@ -286,7 +285,7 @@ func curl(c *cli.Context) error {
 	}
 
 	cmdArgs = append(cmdArgs, "-H")
-	cmdArgs = append(cmdArgs, fmt.Sprintf("%s: %s", h2mux.CFAccessTokenHeader, tok))
+	cmdArgs = append(cmdArgs, fmt.Sprintf("%s: %s", carrier.CFAccessTokenHeader, tok))
 	return run("curl", cmdArgs...)
 }
 
@@ -472,10 +471,10 @@ func isFileThere(candidate string) bool {
 func verifyTokenAtEdge(appUrl *url.URL, appInfo *token.AppInfo, c *cli.Context, log *zerolog.Logger) error {
 	headers := buildRequestHeaders(c.StringSlice(sshHeaderFlag))
 	if c.IsSet(sshTokenIDFlag) {
-		headers.Add(h2mux.CFAccessClientIDHeader, c.String(sshTokenIDFlag))
+		headers.Add(cfAccessClientIDHeader, c.String(sshTokenIDFlag))
 	}
 	if c.IsSet(sshTokenSecretFlag) {
-		headers.Add(h2mux.CFAccessClientSecretHeader, c.String(sshTokenSecretFlag))
+		headers.Add(cfAccessClientSecretHeader, c.String(sshTokenSecretFlag))
 	}
 	options := &carrier.StartOptions{AppInfo: appInfo, OriginURL: appUrl.String(), Headers: headers}
 

cmd/cloudflared/tunnel/cmd.go

@@ -296,7 +296,7 @@ func StartServer(
 	}()
 
 	// Serve DNS proxy stand-alone if no hostname or tag or app is going to run
-	if dnsProxyStandAlone(c) {
+	if dnsProxyStandAlone(c, namedTunnel) {
 		connectedSignal.Notify()
 		// no grace period, handle SIGINT/SIGTERM immediately
 		return waitToShutdown(&wg, cancel, errC, graceShutdownC, 0, log)
@@ -565,10 +565,10 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 		}),
 		altsrc.NewDurationFlag(&cli.DurationFlag{
 			Name:    "grace-period",
-			Usage:   "Duration to accept new requests after cloudflared receives first SIGINT/SIGTERM. A second SIGINT/SIGTERM will force cloudflared to shutdown immediately.",
+			Usage:   "When cloudflared receives SIGINT/SIGTERM it will stop accepting new requests, wait for in-progress requests to terminate, then shutdown. Waiting for in-progress requests will timeout after this grace period, or when a second SIGTERM/SIGINT is received.",
 			Value:   time.Second * 30,
 			EnvVars: []string{"TUNNEL_GRACE_PERIOD"},
-			Hidden:  true,
+			Hidden:  shouldHide,
 		}),
 		// Note TUN-3758 , we use Int because UInt is not supported with altsrc
 		altsrc.NewIntFlag(&cli.IntFlag{

cmd/cloudflared/tunnel/configuration.go

@@ -85,8 +85,8 @@ func logClientOptions(c *cli.Context, log *zerolog.Logger) {
 	}
 }
 
-func dnsProxyStandAlone(c *cli.Context) bool {
-	return c.IsSet("proxy-dns") && (!c.IsSet("hostname") && !c.IsSet("tag") && !c.IsSet("hello-world"))
+func dnsProxyStandAlone(c *cli.Context, namedTunnel *connection.NamedTunnelConfig) bool {
+	return c.IsSet("proxy-dns") && (!c.IsSet("hostname") && !c.IsSet("tag") && !c.IsSet("hello-world") && namedTunnel == nil)
 }
 
 func findOriginCert(originCertPath string, log *zerolog.Logger) (string, error) {

cmd/cloudflared/tunnel/subcommands.go

@@ -368,8 +368,8 @@ func tunnelInfo(c *cli.Context) error {
 	warningChecker := updater.StartWarningCheck(c)
 	defer warningChecker.LogWarningIfAny(sc.log)
 
-	if c.NArg() > 1 {
-		return cliutil.UsageError(`"cloudflared tunnel info" accepts only one argument, the ID or name of the tunnel to run.`)
+	if c.NArg() != 1 {
+		return cliutil.UsageError(`"cloudflared tunnel info" accepts exactly one argument, the ID or name of the tunnel to get info about.`)
 	}
 	tunnelID, err := sc.findID(c.Args().First())
 	if err != nil {
@@ -454,14 +454,30 @@ func formatAndPrintConnectionsList(tunnelInfo Info, showRecentlyDisconnected boo
 	writer := tabWriter()
 	defer writer.Flush()
 
+	// Print the general tunnel info table
 	_, _ = fmt.Fprintf(writer, "NAME:     %s\nID:       %s\nCREATED:  %s\n\n", tunnelInfo.Name, tunnelInfo.ID, tunnelInfo.CreatedAt)
 
+	// Determine whether to print the connector table
+	shouldDisplayTable := false
+	for _, c := range tunnelInfo.Connectors {
+		conns := fmtConnections(c.Connections, showRecentlyDisconnected)
+		if len(conns) > 0 {
+			shouldDisplayTable = true
+		}
+	}
+	if !shouldDisplayTable {
+		fmt.Println("This tunnel has no active connectors.")
+		return
+	}
+
+	// Print the connector table
 	_, _ = fmt.Fprintln(writer, "CONNECTOR ID\tCREATED\tARCHITECTURE\tVERSION\tORIGIN IP\tEDGE\t")
 	for _, c := range tunnelInfo.Connectors {
-		var originIp = ""
-		if len(c.Connections) > 0 {
-			originIp = c.Connections[0].OriginIP.String()
+		conns := fmtConnections(c.Connections, showRecentlyDisconnected)
+		if len(conns) == 0 {
+			continue
 		}
+		originIp := c.Connections[0].OriginIP.String()
 		formattedStr := fmt.Sprintf(
 			"%s\t%s\t%s\t%s\t%s\t%s\t",
 			c.ID,
@@ -469,7 +485,7 @@ func formatAndPrintConnectionsList(tunnelInfo Info, showRecentlyDisconnected boo
 			c.Arch,
 			c.Version,
 			originIp,
-			fmtConnections(c.Connections, showRecentlyDisconnected),
+			conns,
 		)
 		_, _ = fmt.Fprintln(writer, formattedStr)
 	}

component-tests/config.py

@@ -1,15 +1,10 @@
 #!/usr/bin/env python
-import base64
 import copy
-import os
-import yaml
 
 from dataclasses import dataclass, InitVar
 
 from constants import METRICS_PORT
 
-from util import LOGGER
-
 # frozen=True raises exception when assigning to fields. This emulates immutability
 
 
@@ -98,35 +93,3 @@ class ClassicTunnelConfig(ClassicTunnelBaseConfig):
 
     def get_url(self):
         return "https://" + self.hostname
-
-
-def build_config_from_env():
-    config_path = get_env("COMPONENT_TESTS_CONFIG")
-    config_content = base64.b64decode(
-        get_env("COMPONENT_TESTS_CONFIG_CONTENT")).decode('utf-8')
-    config_yaml = yaml.safe_load(config_content)
-
-    credentials_file = get_env("COMPONENT_TESTS_CREDENTIALS_FILE")
-    write_file(credentials_file, config_yaml["credentials_file"])
-
-    origincert = get_env("COMPONENT_TESTS_ORIGINCERT")
-    write_file(origincert, config_yaml["origincert"])
-
-    write_file(config_content, config_path)
-
-
-def write_file(content, path):
-    with open(path, 'w') as outfile:
-        outfile.write(content)
-        outfile.close
-
-
-def get_env(env_name):
-    val = os.getenv(env_name)
-    if val is None:
-        raise Exception(f"{env_name} is not set")
-    return val
-
-
-if __name__ == '__main__':
-    build_config_from_env()

component-tests/conftest.py

@@ -20,11 +20,21 @@ def component_tests_config():
         LOGGER.info(f"component tests base config {config}")
 
         def _component_tests_config(additional_config={}, named_tunnel=True):
+
+            # Regression test for TUN-4177, running with proxy-dns should not prevent tunnels from running
+            additional_config["proxy-dns"] = True
+            additional_config["proxy-dns-port"] = 9053
+
             if named_tunnel:
                 return NamedTunnelConfig(additional_config=additional_config,
-                                         cloudflared_binary=config['cloudflared_binary'], tunnel=config['tunnel'], credentials_file=config['credentials_file'],  ingress=config['ingress'])
+                                         cloudflared_binary=config['cloudflared_binary'],
+                                         tunnel=config['tunnel'],
+                                         credentials_file=config['credentials_file'],
+                                         ingress=config['ingress'])
+
             return ClassicTunnelConfig(
-                additional_config=additional_config, cloudflared_binary=config['cloudflared_binary'], hostname=config['classic_hostname'], origincert=config['origincert'])
+                additional_config=additional_config, cloudflared_binary=config['cloudflared_binary'],
+                hostname=config['classic_hostname'], origincert=config['origincert'])
 
         return _component_tests_config
 

component-tests/requirements.txt

@@ -1,3 +1,4 @@
+cloudflare==2.8.15
 flaky==3.7.0
 pytest==6.2.2
 pyyaml==5.4.1

component-tests/setup.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python
+import argparse
+import base64
+import json
+import os
+import subprocess
+import uuid
+
+import CloudFlare
+import yaml
+from retrying import retry
+
+from constants import MAX_RETRIES, BACKOFF_SECS
+from util import LOGGER
+
+
+def get_config_from_env():
+    config_content = base64.b64decode(get_env("COMPONENT_TESTS_CONFIG_CONTENT")).decode('utf-8')
+    return yaml.safe_load(config_content)
+
+
+def get_config_from_file():
+    config_path = get_env("COMPONENT_TESTS_CONFIG")
+    with open(config_path, 'r') as infile:
+        return yaml.safe_load(infile)
+
+
+def persist_config(config):
+    config_path = get_env("COMPONENT_TESTS_CONFIG")
+    with open(config_path, 'w') as outfile:
+        yaml.safe_dump(config, outfile)
+
+
+def persist_origin_cert(config):
+    origincert = get_env("COMPONENT_TESTS_ORIGINCERT")
+    path = config["origincert"]
+    with open(path, 'w') as outfile:
+        outfile.write(origincert)
+    return path
+
+
+@retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
+def create_tunnel(config, origincert_path, random_uuid):
+    # Delete any previous existing credentials file. If the agent keeps files around (that's the case in Windows) then
+    # cloudflared tunnel create will refuse to create the tunnel because it does not want to overwrite credentials
+    # files.
+    credentials_path = config["credentials_file"]
+    try:
+        os.remove(credentials_path)
+    except OSError:
+        pass
+
+    tunnel_name = "cfd_component_test-" + random_uuid
+    create_cmd = [config["cloudflared_binary"], "tunnel", "--origincert", origincert_path, "create",
+                  "--credentials-file", credentials_path, tunnel_name]
+    LOGGER.info(f"Creating tunnel with {create_cmd}")
+    subprocess.run(create_cmd, check=True)
+
+    list_cmd = [config["cloudflared_binary"], "tunnel", "--origincert", origincert_path, "list", "--name",
+                tunnel_name, "--output", "json"]
+    LOGGER.info(f"Listing tunnel with {list_cmd}")
+    cloudflared = subprocess.run(list_cmd, check=True, capture_output=True)
+    return json.loads(cloudflared.stdout)[0]["id"]
+
+
+@retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
+def delete_tunnel(config):
+    credentials_path = config["credentials_file"]
+    delete_cmd = [config["cloudflared_binary"], "tunnel", "--origincert", config["origincert"], "delete",
+                  "--credentials-file", credentials_path, "-f", config["tunnel"]]
+    LOGGER.info(f"Deleting tunnel with {delete_cmd}")
+    subprocess.run(delete_cmd, check=True)
+
+
+@retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
+def create_dns(config, hostname, type, content):
+    cf = CloudFlare.CloudFlare(debug=True, token=get_env("DNS_API_TOKEN"))
+    cf.zones.dns_records.post(
+        config["zone_tag"],
+        data={'name': hostname, 'type': type, 'content': content, 'proxied': True}
+    )
+
+
+def create_classic_dns(config, random_uuid):
+    classic_hostname = "classic-" + random_uuid + "." + config["zone_domain"]
+    create_dns(config, classic_hostname, "AAAA", "fd10:aec2:5dae::")
+    return classic_hostname
+
+
+def create_named_dns(config, random_uuid):
+    hostname = "named-" + random_uuid + "." + config["zone_domain"]
+    create_dns(config, hostname, "CNAME", config["tunnel"] + ".cfargotunnel.com")
+    return hostname
+
+
+@retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
+def delete_dns(config, hostname):
+    cf = CloudFlare.CloudFlare(debug=True, token=get_env("DNS_API_TOKEN"))
+    zone_tag = config["zone_tag"]
+    dns_records = cf.zones.dns_records.get(zone_tag, params={'name': hostname})
+    if len(dns_records) > 0:
+        cf.zones.dns_records.delete(zone_tag, dns_records[0]['id'])
+
+
+def write_file(content, path):
+    with open(path, 'w') as outfile:
+        outfile.write(content)
+
+
+def get_env(env_name):
+    val = os.getenv(env_name)
+    if val is None:
+        raise Exception(f"{env_name} is not set")
+    return val
+
+
+def create():
+    """
+    Creates the necessary resources for the components test to run.
+     - Creates a named tunnel with a random name.
+     - Creates a random CNAME DNS entry for that tunnel.
+     - Creates a random AAAA DNS entry for a classic tunnel.
+
+    Those created resources are added to the config (obtained from an environment variable).
+    The resulting configuration is persisted for the tests to use.
+    """
+    config = get_config_from_env()
+    origincert_path = persist_origin_cert(config)
+
+    random_uuid = str(uuid.uuid4())
+    config["tunnel"] = create_tunnel(config, origincert_path, random_uuid)
+    config["classic_hostname"] = create_classic_dns(config, random_uuid)
+    config["ingress"] = [
+        {
+            "hostname": create_named_dns(config, random_uuid),
+            "service": "hello_world"
+        },
+        {
+            "service": "http_status:404"
+        }
+    ]
+
+    persist_config(config)
+
+
+def cleanup():
+    """
+    Reads the persisted configuration that was created previously.
+    Deletes the resources that were created there.
+    """
+    config = get_config_from_file()
+    delete_tunnel(config)
+    delete_dns(config, config["classic_hostname"])
+    delete_dns(config, config["ingress"][0]["hostname"])
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description='setup component tests')
+    parser.add_argument('--type', choices=['create', 'cleanup'], default='create')
+    args = parser.parse_args()
+
+    if args.type == 'create':
+        create()
+    else:
+        cleanup()

component-tests/test_logging.py

@@ -1,28 +1,23 @@
 #!/usr/bin/env python
 import json
 import os
-import subprocess
 
 from util import start_cloudflared, wait_tunnel_ready, send_requests
 
 # Rolling logger rotate log files after 1 MB
 rotate_after_size = 1000 * 1000
 default_log_file = "cloudflared.log"
-expect_message = "Starting tunnel"
+expect_message = "Starting Hello World server"
 
 
 def assert_log_to_terminal(cloudflared):
-    stderr = cloudflared.stderr.read(200)
-    # cloudflared logs the following when it first starts
-    # 2021-03-10T12:30:39Z INF Starting tunnel tunnelID=<tunnel ID>
+    stderr = cloudflared.stderr.read(1500)
     assert expect_message.encode() in stderr, f"{stderr} doesn't contain {expect_message}"
 
 
 def assert_log_in_file(file):
     with open(file, "r") as f:
-        log = f.read(200)
-        # cloudflared logs the following when it first starts
-        # {"level":"info","tunnelID":"<tunnel ID>","time":"2021-03-10T12:21:13Z","message":"Starting tunnel"}
+        log = f.read(1850)
         assert expect_message in log, f"{log} doesn't contain {expect_message}"
 
 

connection/h2mux.go

@@ -234,7 +234,7 @@ func (h *h2muxConnection) newRequest(stream *h2mux.MuxedStream) (*http.Request,
 	if err != nil {
 		return nil, errors.Wrap(err, "Unexpected error from http.NewRequest")
 	}
-	err = h2mux.H2RequestHeadersToH1Request(stream.Headers, req)
+	err = H2RequestHeadersToH1Request(stream.Headers, req)
 	if err != nil {
 		return nil, errors.Wrap(err, "invalid request received")
 	}
@@ -246,15 +246,15 @@ type h2muxRespWriter struct {
 }
 
 func (rp *h2muxRespWriter) WriteRespHeaders(status int, header http.Header) error {
-	headers := h2mux.H1ResponseToH2ResponseHeaders(status, header)
-	headers = append(headers, h2mux.Header{Name: ResponseMetaHeaderField, Value: responseMetaHeaderOrigin})
+	headers := H1ResponseToH2ResponseHeaders(status, header)
+	headers = append(headers, h2mux.Header{Name: ResponseMetaHeader, Value: responseMetaHeaderOrigin})
 	return rp.WriteHeaders(headers)
 }
 
 func (rp *h2muxRespWriter) WriteErrorResponse() {
 	_ = rp.WriteHeaders([]h2mux.Header{
 		{Name: ":status", Value: "502"},
-		{Name: ResponseMetaHeaderField, Value: responseMetaHeaderCfd},
+		{Name: ResponseMetaHeader, Value: responseMetaHeaderCfd},
 	})
 	_, _ = rp.Write([]byte("502 Bad Gateway"))
 }

connection/h2mux_test.go

@@ -115,9 +115,9 @@ func TestServeStreamHTTP(t *testing.T) {
 		require.True(t, hasHeader(stream, ":status", strconv.Itoa(test.expectedStatus)))
 
 		if test.isProxyError {
-			assert.True(t, hasHeader(stream, ResponseMetaHeaderField, responseMetaHeaderCfd))
+			assert.True(t, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderCfd))
 		} else {
-			assert.True(t, hasHeader(stream, ResponseMetaHeaderField, responseMetaHeaderOrigin))
+			assert.True(t, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderOrigin))
 			body := make([]byte, len(test.expectedBody))
 			_, err = stream.Read(body)
 			require.NoError(t, err)
@@ -164,7 +164,7 @@ func TestServeStreamWS(t *testing.T) {
 	require.NoError(t, err)
 
 	require.True(t, hasHeader(stream, ":status", strconv.Itoa(http.StatusSwitchingProtocols)))
-	assert.True(t, hasHeader(stream, ResponseMetaHeaderField, responseMetaHeaderOrigin))
+	assert.True(t, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderOrigin))
 
 	data := []byte("test websocket")
 	err = wsutil.WriteClientText(writePipe, data)
@@ -268,7 +268,7 @@ func benchmarkServeStreamHTTPSimple(b *testing.B, test testRequest) {
 		b.StopTimer()
 
 		require.NoError(b, openstreamErr)
-		assert.True(b, hasHeader(stream, ResponseMetaHeaderField, responseMetaHeaderOrigin))
+		assert.True(b, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderOrigin))
 		require.True(b, hasHeader(stream, ":status", strconv.Itoa(http.StatusOK)))
 		require.NoError(b, readBodyErr)
 		require.Equal(b, test.expectedBody, body)

connection/header.go

@@ -1,21 +1,33 @@
 package connection
 
 import (
+	"encoding/base64"
 	"fmt"
 	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+
+	"github.com/pkg/errors"
 
 	"github.com/cloudflare/cloudflared/h2mux"
 )
 
-const (
-	ResponseMetaHeaderField = "cf-cloudflared-response-meta"
+var (
+	// h2mux-style special headers
+	RequestUserHeaders  = "cf-cloudflared-request-headers"
+	ResponseUserHeaders = "cf-cloudflared-response-headers"
+	ResponseMetaHeader  = "cf-cloudflared-response-meta"
+
+	// h2mux-style special headers
+	CanonicalResponseUserHeaders = http.CanonicalHeaderKey(ResponseUserHeaders)
+	CanonicalResponseMetaHeader  = http.CanonicalHeaderKey(ResponseMetaHeader)
 )
 
 var (
-	canonicalResponseUserHeadersField = http.CanonicalHeaderKey(h2mux.ResponseUserHeadersField)
-	canonicalResponseMetaHeaderField  = http.CanonicalHeaderKey(ResponseMetaHeaderField)
-	responseMetaHeaderCfd             = mustInitRespMetaHeader("cloudflared")
-	responseMetaHeaderOrigin          = mustInitRespMetaHeader("origin")
+	// pre-generate possible values for res
+	responseMetaHeaderCfd    = mustInitRespMetaHeader("cloudflared")
+	responseMetaHeaderOrigin = mustInitRespMetaHeader("origin")
 )
 
 type responseMetaHeader struct {
@@ -29,3 +41,204 @@ func mustInitRespMetaHeader(src string) string {
 	}
 	return string(header)
 }
+
+var headerEncoding = base64.RawStdEncoding
+
+// note: all h2mux headers should be lower-case (http/2 style)
+const ()
+
+// H2RequestHeadersToH1Request converts the HTTP/2 headers coming from origintunneld
+// to an HTTP/1 Request object destined for the local origin web service.
+// This operation includes conversion of the pseudo-headers into their closest
+// HTTP/1 equivalents. See https://tools.ietf.org/html/rfc7540#section-8.1.2.3
+func H2RequestHeadersToH1Request(h2 []h2mux.Header, h1 *http.Request) error {
+	for _, header := range h2 {
+		name := strings.ToLower(header.Name)
+		if !IsControlHeader(name) {
+			continue
+		}
+
+		switch name {
+		case ":method":
+			h1.Method = header.Value
+		case ":scheme":
+			// noop - use the preexisting scheme from h1.URL
+		case ":authority":
+			// Otherwise the host header will be based on the origin URL
+			h1.Host = header.Value
+		case ":path":
+			// We don't want to be an "opinionated" proxy, so ideally we would use :path as-is.
+			// However, this HTTP/1 Request object belongs to the Go standard library,
+			// whose URL package makes some opinionated decisions about the encoding of
+			// URL characters: see the docs of https://godoc.org/net/url#URL,
+			// in particular the EscapedPath method https://godoc.org/net/url#URL.EscapedPath,
+			// which is always used when computing url.URL.String(), whether we'd like it or not.
+			//
+			// Well, not *always*. We could circumvent this by using url.URL.Opaque. But
+			// that would present unusual difficulties when using an HTTP proxy: url.URL.Opaque
+			// is treated differently when HTTP_PROXY is set!
+			// See https://github.com/golang/go/issues/5684#issuecomment-66080888
+			//
+			// This means we are subject to the behavior of net/url's function `shouldEscape`
+			// (as invoked with mode=encodePath): https://github.com/golang/go/blob/go1.12.7/src/net/url/url.go#L101
+
+			if header.Value == "*" {
+				h1.URL.Path = "*"
+				continue
+			}
+			// Due to the behavior of validation.ValidateUrl, h1.URL may
+			// already have a partial value, with or without a trailing slash.
+			base := h1.URL.String()
+			base = strings.TrimRight(base, "/")
+			// But we know :path begins with '/', because we handled '*' above - see RFC7540
+			requestURL, err := url.Parse(base + header.Value)
+			if err != nil {
+				return errors.Wrap(err, fmt.Sprintf("invalid path '%v'", header.Value))
+			}
+			h1.URL = requestURL
+		case "content-length":
+			contentLength, err := strconv.ParseInt(header.Value, 10, 64)
+			if err != nil {
+				return fmt.Errorf("unparseable content length")
+			}
+			h1.ContentLength = contentLength
+		case RequestUserHeaders:
+			// Do not forward the serialized headers to the origin -- deserialize them, and ditch the serialized version
+			// Find and parse user headers serialized into a single one
+			userHeaders, err := DeserializeHeaders(header.Value)
+			if err != nil {
+				return errors.Wrap(err, "Unable to parse user headers")
+			}
+			for _, userHeader := range userHeaders {
+				h1.Header.Add(userHeader.Name, userHeader.Value)
+			}
+		default:
+			// All other control headers shall just be proxied transparently
+			h1.Header.Add(header.Name, header.Value)
+		}
+	}
+
+	return nil
+}
+
+func IsControlHeader(headerName string) bool {
+	return headerName == "content-length" ||
+		headerName == "connection" || headerName == "upgrade" || // Websocket headers
+		strings.HasPrefix(headerName, ":") ||
+		strings.HasPrefix(headerName, "cf-")
+}
+
+// isWebsocketClientHeader returns true if the header name is required by the client to upgrade properly
+func IsWebsocketClientHeader(headerName string) bool {
+	return headerName == "sec-websocket-accept" ||
+		headerName == "connection" ||
+		headerName == "upgrade"
+}
+
+func H1ResponseToH2ResponseHeaders(status int, h1 http.Header) (h2 []h2mux.Header) {
+	h2 = []h2mux.Header{
+		{Name: ":status", Value: strconv.Itoa(status)},
+	}
+	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,
+			// so it should be sent as an HTTP/2 response header.
+
+			// Since these are http2 headers, they're required to be lowercase
+			h2 = append(h2, h2mux.Header{Name: "content-length", Value: values[0]})
+		} else if !IsControlHeader(h2name) || IsWebsocketClientHeader(h2name) {
+			// User headers, on the other hand, must all be serialized so that
+			// HTTP/2 header validation won't be applied to HTTP/1 header values
+			userHeaders[header] = values
+		}
+	}
+
+	// Perform user header serialization and set them in the single header
+	h2 = append(h2, h2mux.Header{Name: ResponseUserHeaders, Value: SerializeHeaders(userHeaders)})
+	return h2
+}
+
+// Serialize HTTP1.x headers by base64-encoding each header name and value,
+// and then joining them in the format of [key:value;]
+func SerializeHeaders(h1Headers http.Header) string {
+	// compute size of the fully serialized value and largest temp buffer we will need
+	serializedLen := 0
+	maxTempLen := 0
+	for headerName, headerValues := range h1Headers {
+		for _, headerValue := range headerValues {
+			nameLen := headerEncoding.EncodedLen(len(headerName))
+			valueLen := headerEncoding.EncodedLen(len(headerValue))
+			const delims = 2
+			serializedLen += delims + nameLen + valueLen
+			if nameLen > maxTempLen {
+				maxTempLen = nameLen
+			}
+			if valueLen > maxTempLen {
+				maxTempLen = valueLen
+			}
+		}
+	}
+	var buf strings.Builder
+	buf.Grow(serializedLen)
+
+	temp := make([]byte, maxTempLen)
+	writeB64 := func(s string) {
+		n := headerEncoding.EncodedLen(len(s))
+		if n > len(temp) {
+			temp = make([]byte, n)
+		}
+		headerEncoding.Encode(temp[:n], []byte(s))
+		buf.Write(temp[:n])
+	}
+
+	for headerName, headerValues := range h1Headers {
+		for _, headerValue := range headerValues {
+			if buf.Len() > 0 {
+				buf.WriteByte(';')
+			}
+			writeB64(headerName)
+			buf.WriteByte(':')
+			writeB64(headerValue)
+		}
+	}
+
+	return buf.String()
+}
+
+// Deserialize headers serialized by `SerializeHeader`
+func DeserializeHeaders(serializedHeaders string) ([]h2mux.Header, error) {
+	const unableToDeserializeErr = "Unable to deserialize headers"
+
+	var deserialized []h2mux.Header
+	for _, serializedPair := range strings.Split(serializedHeaders, ";") {
+		if len(serializedPair) == 0 {
+			continue
+		}
+
+		serializedHeaderParts := strings.Split(serializedPair, ":")
+		if len(serializedHeaderParts) != 2 {
+			return nil, errors.New(unableToDeserializeErr)
+		}
+
+		serializedName := serializedHeaderParts[0]
+		serializedValue := serializedHeaderParts[1]
+		deserializedName := make([]byte, headerEncoding.DecodedLen(len(serializedName)))
+		deserializedValue := make([]byte, headerEncoding.DecodedLen(len(serializedValue)))
+
+		if _, err := headerEncoding.Decode(deserializedName, []byte(serializedName)); err != nil {
+			return nil, errors.Wrap(err, unableToDeserializeErr)
+		}
+		if _, err := headerEncoding.Decode(deserializedValue, []byte(serializedValue)); err != nil {
+			return nil, errors.Wrap(err, unableToDeserializeErr)
+		}
+
+		deserialized = append(deserialized, h2mux.Header{
+			Name:  string(deserializedName),
+			Value: string(deserializedValue),
+		})
+	}
+
+	return deserialized, nil
+}

connection/header_test.go

@@ -1,4 +1,4 @@
-package h2mux
+package connection
 
 import (
 	"fmt"
@@ -14,9 +14,11 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/cloudflare/cloudflared/h2mux"
 )
 
-type ByName []Header
+type ByName []h2mux.Header
 
 func (a ByName) Len() int      { return len(a) }
 func (a ByName) Swap(i, j int) { a[i], a[j] = a[j], a[i] }
@@ -37,16 +39,16 @@ func TestH2RequestHeadersToH1Request_RegularHeaders(t *testing.T) {
 		"Mock header 2": {"Mock value 2"},
 	}
 
-	headersConversionErr := H2RequestHeadersToH1Request(createSerializedHeaders(RequestUserHeadersField, mockHeaders), request)
+	headersConversionErr := H2RequestHeadersToH1Request(createSerializedHeaders(RequestUserHeaders, mockHeaders), request)
 
 	assert.True(t, reflect.DeepEqual(mockHeaders, request.Header))
 	assert.NoError(t, headersConversionErr)
 }
 
-func createSerializedHeaders(headersField string, headers http.Header) []Header {
-	return []Header{{
-		headersField,
-		SerializeHeaders(headers),
+func createSerializedHeaders(headersField string, headers http.Header) []h2mux.Header {
+	return []h2mux.Header{{
+		Name:  headersField,
+		Value: SerializeHeaders(headers),
 	}}
 }
 
@@ -54,15 +56,16 @@ func TestH2RequestHeadersToH1Request_NoHeaders(t *testing.T) {
 	request, err := http.NewRequest(http.MethodGet, "http://example.com", nil)
 	assert.NoError(t, err)
 
+	emptyHeaders := make(http.Header)
 	headersConversionErr := H2RequestHeadersToH1Request(
-		[]Header{{
-			RequestUserHeadersField,
-			SerializeHeaders(http.Header{}),
+		[]h2mux.Header{{
+			Name:  RequestUserHeaders,
+			Value: SerializeHeaders(emptyHeaders),
 		}},
 		request,
 	)
 
-	assert.True(t, reflect.DeepEqual(http.Header{}, request.Header))
+	assert.True(t, reflect.DeepEqual(emptyHeaders, request.Header))
 	assert.NoError(t, headersConversionErr)
 }
 
@@ -70,9 +73,9 @@ func TestH2RequestHeadersToH1Request_InvalidHostPath(t *testing.T) {
 	request, err := http.NewRequest(http.MethodGet, "http://example.com", nil)
 	assert.NoError(t, err)
 
-	mockRequestHeaders := []Header{
+	mockRequestHeaders := []h2mux.Header{
 		{Name: ":path", Value: "//bad_path/"},
-		{Name: RequestUserHeadersField, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
+		{Name: RequestUserHeaders, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
 	}
 
 	headersConversionErr := H2RequestHeadersToH1Request(mockRequestHeaders, request)
@@ -90,9 +93,9 @@ func TestH2RequestHeadersToH1Request_HostPathWithQuery(t *testing.T) {
 	request, err := http.NewRequest(http.MethodGet, "http://example.com/", nil)
 	assert.NoError(t, err)
 
-	mockRequestHeaders := []Header{
+	mockRequestHeaders := []h2mux.Header{
 		{Name: ":path", Value: "/?query=mock%20value"},
-		{Name: RequestUserHeadersField, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
+		{Name: RequestUserHeaders, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
 	}
 
 	headersConversionErr := H2RequestHeadersToH1Request(mockRequestHeaders, request)
@@ -110,9 +113,9 @@ func TestH2RequestHeadersToH1Request_HostPathWithURLEncoding(t *testing.T) {
 	request, err := http.NewRequest(http.MethodGet, "http://example.com/", nil)
 	assert.NoError(t, err)
 
-	mockRequestHeaders := []Header{
+	mockRequestHeaders := []h2mux.Header{
 		{Name: ":path", Value: "/mock%20path"},
-		{Name: RequestUserHeadersField, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
+		{Name: RequestUserHeaders, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
 	}
 
 	headersConversionErr := H2RequestHeadersToH1Request(mockRequestHeaders, request)
@@ -267,9 +270,9 @@ func TestH2RequestHeadersToH1Request_WeirdURLs(t *testing.T) {
 		request, err := http.NewRequest(http.MethodGet, requestURL, nil)
 		assert.NoError(t, err)
 
-		mockRequestHeaders := []Header{
+		mockRequestHeaders := []h2mux.Header{
 			{Name: ":path", Value: testCase.path},
-			{Name: RequestUserHeadersField, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
+			{Name: RequestUserHeaders, Value: SerializeHeaders(http.Header{"Mock header": {"Mock value"}})},
 		}
 
 		headersConversionErr := H2RequestHeadersToH1Request(mockRequestHeaders, request)
@@ -337,12 +340,12 @@ func TestH2RequestHeadersToH1Request_QuickCheck(t *testing.T) {
 				const expectedMethod = "POST"
 				const expectedHostname = "request.hostname.example.com"
 
-				h2 := []Header{
+				h2 := []h2mux.Header{
 					{Name: ":method", Value: expectedMethod},
 					{Name: ":scheme", Value: testScheme},
 					{Name: ":authority", Value: expectedHostname},
 					{Name: ":path", Value: testPath},
-					{Name: RequestUserHeadersField, Value: ""},
+					{Name: RequestUserHeaders, Value: ""},
 				}
 				h1, err := http.NewRequest("GET", testOrigin.url, nil)
 				require.NoError(t, err)
@@ -424,10 +427,10 @@ func randomHTTP2Path(t *testing.T, rand *rand.Rand) string {
 	return result
 }
 
-func stdlibHeaderToH2muxHeader(headers http.Header) (h2muxHeaders []Header) {
+func stdlibHeaderToH2muxHeader(headers http.Header) (h2muxHeaders []h2mux.Header) {
 	for name, values := range headers {
 		for _, value := range values {
-			h2muxHeaders = append(h2muxHeaders, Header{name, value})
+			h2muxHeaders = append(h2muxHeaders, h2mux.Header{Name: name, Value: value})
 		}
 	}
 
@@ -515,14 +518,14 @@ func TestParseHeaders(t *testing.T) {
 		"Mock-Header-Three": {"3"},
 	}
 
-	mockHeaders := []Header{
+	mockHeaders := []h2mux.Header{
 		{Name: "One", Value: "1"}, // will be dropped
 		{Name: "Cf-Two", Value: "cf-value-1"},
 		{Name: "Cf-Two", Value: "cf-value-2"},
-		{Name: RequestUserHeadersField, Value: SerializeHeaders(mockUserHeadersToSerialize)},
+		{Name: RequestUserHeaders, Value: SerializeHeaders(mockUserHeadersToSerialize)},
 	}
 
-	expectedHeaders := []Header{
+	expectedHeaders := []h2mux.Header{
 		{Name: "Cf-Two", Value: "cf-value-1"},
 		{Name: "Cf-Two", Value: "cf-value-2"},
 		{Name: "Mock-Header-One", Value: "1"},
@@ -583,7 +586,7 @@ func TestH1ResponseToH2ResponseHeaders(t *testing.T) {
 
 	serializedHeadersIndex := -1
 	for i, header := range headers {
-		if header.Name == ResponseUserHeadersField {
+		if header.Name == ResponseUserHeaders {
 			serializedHeadersIndex = i
 			break
 		}
@@ -593,7 +596,7 @@ func TestH1ResponseToH2ResponseHeaders(t *testing.T) {
 		headers[:serializedHeadersIndex],
 		headers[serializedHeadersIndex+1:]...,
 	)
-	expectedControlHeaders := []Header{
+	expectedControlHeaders := []h2mux.Header{
 		{Name: ":status", Value: "200"},
 		{Name: "content-length", Value: "123"},
 	}
@@ -601,7 +604,7 @@ func TestH1ResponseToH2ResponseHeaders(t *testing.T) {
 	assert.ElementsMatch(t, expectedControlHeaders, actualControlHeaders)
 
 	actualUserHeaders, err := DeserializeHeaders(headers[serializedHeadersIndex].Value)
-	expectedUserHeaders := []Header{
+	expectedUserHeaders := []h2mux.Header{
 		{Name: "User-header-one", Value: ""},
 		{Name: "User-header-two", Value: "1"},
 		{Name: "User-header-two", Value: "2"},
@@ -630,7 +633,7 @@ func TestHeaderSize(t *testing.T) {
 	}
 
 	for _, header := range serializedHeaders {
-		if header.Name != ResponseUserHeadersField {
+		if header.Name != ResponseUserHeaders {
 			continue
 		}
 

connection/http2.go

@@ -13,15 +13,15 @@ import (
 	"github.com/rs/zerolog"
 	"golang.org/x/net/http2"
 
-	"github.com/cloudflare/cloudflared/h2mux"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
+// note: these constants are exported so we can reuse them in the edge-side code
 const (
-	internalUpgradeHeader = "Cf-Cloudflared-Proxy-Connection-Upgrade"
-	tcpStreamHeader       = "Cf-Cloudflared-Proxy-Src"
-	websocketUpgrade      = "websocket"
-	controlStreamUpgrade  = "control-stream"
+	InternalUpgradeHeader     = "Cf-Cloudflared-Proxy-Connection-Upgrade"
+	InternalTCPProxySrcHeader = "Cf-Cloudflared-Proxy-Src"
+	WebsocketUpgrade          = "websocket"
+	ControlStreamUpgrade      = "control-stream"
 )
 
 var errEdgeConnectionClosed = fmt.Errorf("connection with edge closed")
@@ -178,25 +178,23 @@ func newHTTP2RespWriter(r *http.Request, w http.ResponseWriter, connType Type) (
 func (rp *http2RespWriter) WriteRespHeaders(status int, header http.Header) error {
 	dest := rp.w.Header()
 	userHeaders := make(http.Header, len(header))
-	for header, values := range header {
+	for name, values := range header {
 		// Since these are http2 headers, they're required to be lowercase
-		h2name := strings.ToLower(header)
-		for _, v := range values {
-			if h2name == "content-length" {
-				// This header has meaning in HTTP/2 and will be used by the edge,
-				// so it should be sent as an HTTP/2 response header.
-				dest.Add(h2name, v)
-				// Since these are http2 headers, they're required to be lowercase
-			} else if !h2mux.IsControlHeader(h2name) || h2mux.IsWebsocketClientHeader(h2name) {
-				// User headers, on the other hand, must all be serialized so that
-				// HTTP/2 header validation won't be applied to HTTP/1 header values
-				userHeaders.Add(h2name, v)
-			}
+		h2name := strings.ToLower(name)
+		if h2name == "content-length" {
+			// This header has meaning in HTTP/2 and will be used by the edge,
+			// so it should be sent as an HTTP/2 response header.
+			dest[name] = values
+			// Since these are http2 headers, they're required to be lowercase
+		} else if !IsControlHeader(h2name) || IsWebsocketClientHeader(h2name) {
+			// User headers, on the other hand, must all be serialized so that
+			// HTTP/2 header validation won't be applied to HTTP/1 header values
+			userHeaders[name] = values
 		}
 	}
 
 	// Perform user header serialization and set them in the single header
-	dest.Set(canonicalResponseUserHeadersField, h2mux.SerializeHeaders(userHeaders))
+	dest.Set(CanonicalResponseUserHeaders, SerializeHeaders(userHeaders))
 	rp.setResponseMetaHeader(responseMetaHeaderOrigin)
 	// HTTP2 removes support for 101 Switching Protocols https://tools.ietf.org/html/rfc7540#section-8.1.1
 	if status == http.StatusSwitchingProtocols {
@@ -218,7 +216,7 @@ func (rp *http2RespWriter) WriteErrorResponse() {
 }
 
 func (rp *http2RespWriter) setResponseMetaHeader(value string) {
-	rp.w.Header().Set(canonicalResponseMetaHeaderField, value)
+	rp.w.Header().Set(CanonicalResponseMetaHeader, value)
 }
 
 func (rp *http2RespWriter) Read(p []byte) (n int, err error) {
@@ -258,18 +256,18 @@ func determineHTTP2Type(r *http.Request) Type {
 }
 
 func isControlStreamUpgrade(r *http.Request) bool {
-	return r.Header.Get(internalUpgradeHeader) == controlStreamUpgrade
+	return r.Header.Get(InternalUpgradeHeader) == ControlStreamUpgrade
 }
 
 func isWebsocketUpgrade(r *http.Request) bool {
-	return 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) != ""
+	return r.Header.Get(InternalTCPProxySrcHeader) != ""
 }
 
 func stripWebsocketUpgradeHeader(r *http.Request) {
-	r.Header.Del(internalUpgradeHeader)
+	r.Header.Del(InternalUpgradeHeader)
 }

connection/http2_test.go

@@ -103,9 +103,9 @@ func TestServeHTTP(t *testing.T) {
 			require.Equal(t, test.expectedBody, respBody)
 		}
 		if test.isProxyError {
-			require.Equal(t, responseMetaHeaderCfd, resp.Header.Get(ResponseMetaHeaderField))
+			require.Equal(t, responseMetaHeaderCfd, resp.Header.Get(ResponseMetaHeader))
 		} else {
-			require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeaderField))
+			require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeader))
 		}
 	}
 	cancel()
@@ -191,7 +191,7 @@ func TestServeWS(t *testing.T) {
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/ws", readPipe)
 	require.NoError(t, err)
-	req.Header.Set(internalUpgradeHeader, websocketUpgrade)
+	req.Header.Set(InternalUpgradeHeader, WebsocketUpgrade)
 
 	wg.Add(1)
 	go func() {
@@ -211,7 +211,7 @@ func TestServeWS(t *testing.T) {
 	resp := respWriter.Result()
 	// http2RespWriter should rewrite status 101 to 200
 	require.Equal(t, http.StatusOK, resp.StatusCode)
-	require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeaderField))
+	require.Equal(t, responseMetaHeaderOrigin, resp.Header.Get(ResponseMetaHeader))
 
 	wg.Wait()
 }
@@ -235,7 +235,7 @@ func TestServeControlStream(t *testing.T) {
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
 	require.NoError(t, err)
-	req.Header.Set(internalUpgradeHeader, controlStreamUpgrade)
+	req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
 
 	edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
 	require.NoError(t, err)
@@ -274,7 +274,7 @@ func TestFailRegistration(t *testing.T) {
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
 	require.NoError(t, err)
-	req.Header.Set(internalUpgradeHeader, controlStreamUpgrade)
+	req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
 
 	edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
 	require.NoError(t, err)
@@ -310,7 +310,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "http://localhost:8080/", nil)
 	require.NoError(t, err)
-	req.Header.Set(internalUpgradeHeader, controlStreamUpgrade)
+	req.Header.Set(InternalUpgradeHeader, ControlStreamUpgrade)
 
 	edgeHTTP2Conn, err := testTransport.NewClientConn(edgeConn)
 	require.NoError(t, err)

h2mux/header.go

@@ -1,234 +0,0 @@
-package h2mux
-
-import (
-	"encoding/base64"
-	"fmt"
-	"net/http"
-	"net/url"
-	"strconv"
-	"strings"
-
-	"github.com/pkg/errors"
-)
-
-type Header struct {
-	Name, Value string
-}
-
-var headerEncoding = base64.RawStdEncoding
-
-const (
-	RequestUserHeadersField  = "cf-cloudflared-request-headers"
-	ResponseUserHeadersField = "cf-cloudflared-response-headers"
-
-	CFAccessTokenHeader        = "cf-access-token"
-	CFJumpDestinationHeader    = "CF-Access-Jump-Destination"
-	CFAccessClientIDHeader     = "CF-Access-Client-Id"
-	CFAccessClientSecretHeader = "CF-Access-Client-Secret"
-)
-
-// H2RequestHeadersToH1Request converts the HTTP/2 headers coming from origintunneld
-// to an HTTP/1 Request object destined for the local origin web service.
-// This operation includes conversion of the pseudo-headers into their closest
-// HTTP/1 equivalents. See https://tools.ietf.org/html/rfc7540#section-8.1.2.3
-func H2RequestHeadersToH1Request(h2 []Header, h1 *http.Request) error {
-	for _, header := range h2 {
-		name := strings.ToLower(header.Name)
-		if !IsControlHeader(name) {
-			continue
-		}
-
-		switch name {
-		case ":method":
-			h1.Method = header.Value
-		case ":scheme":
-			// noop - use the preexisting scheme from h1.URL
-		case ":authority":
-			// Otherwise the host header will be based on the origin URL
-			h1.Host = header.Value
-		case ":path":
-			// We don't want to be an "opinionated" proxy, so ideally we would use :path as-is.
-			// However, this HTTP/1 Request object belongs to the Go standard library,
-			// whose URL package makes some opinionated decisions about the encoding of
-			// URL characters: see the docs of https://godoc.org/net/url#URL,
-			// in particular the EscapedPath method https://godoc.org/net/url#URL.EscapedPath,
-			// which is always used when computing url.URL.String(), whether we'd like it or not.
-			//
-			// Well, not *always*. We could circumvent this by using url.URL.Opaque. But
-			// that would present unusual difficulties when using an HTTP proxy: url.URL.Opaque
-			// is treated differently when HTTP_PROXY is set!
-			// See https://github.com/golang/go/issues/5684#issuecomment-66080888
-			//
-			// This means we are subject to the behavior of net/url's function `shouldEscape`
-			// (as invoked with mode=encodePath): https://github.com/golang/go/blob/go1.12.7/src/net/url/url.go#L101
-
-			if header.Value == "*" {
-				h1.URL.Path = "*"
-				continue
-			}
-			// Due to the behavior of validation.ValidateUrl, h1.URL may
-			// already have a partial value, with or without a trailing slash.
-			base := h1.URL.String()
-			base = strings.TrimRight(base, "/")
-			// But we know :path begins with '/', because we handled '*' above - see RFC7540
-			requestURL, err := url.Parse(base + header.Value)
-			if err != nil {
-				return errors.Wrap(err, fmt.Sprintf("invalid path '%v'", header.Value))
-			}
-			h1.URL = requestURL
-		case "content-length":
-			contentLength, err := strconv.ParseInt(header.Value, 10, 64)
-			if err != nil {
-				return fmt.Errorf("unparseable content length")
-			}
-			h1.ContentLength = contentLength
-		case RequestUserHeadersField:
-			// Do not forward the serialized headers to the origin -- deserialize them, and ditch the serialized version
-			// Find and parse user headers serialized into a single one
-			userHeaders, err := ParseUserHeaders(RequestUserHeadersField, h2)
-			if err != nil {
-				return errors.Wrap(err, "Unable to parse user headers")
-			}
-			for _, userHeader := range userHeaders {
-				h1.Header.Add(http.CanonicalHeaderKey(userHeader.Name), userHeader.Value)
-			}
-		default:
-			// All other control headers shall just be proxied transparently
-			h1.Header.Add(http.CanonicalHeaderKey(header.Name), header.Value)
-		}
-	}
-
-	return nil
-}
-
-func ParseUserHeaders(headerNameToParseFrom string, headers []Header) ([]Header, error) {
-	for _, header := range headers {
-		if header.Name == headerNameToParseFrom {
-			return DeserializeHeaders(header.Value)
-		}
-	}
-
-	return nil, fmt.Errorf("%v header not found", RequestUserHeadersField)
-}
-
-func IsControlHeader(headerName string) bool {
-	return headerName == "content-length" ||
-		headerName == "connection" || headerName == "upgrade" || // Websocket headers
-		strings.HasPrefix(headerName, ":") ||
-		strings.HasPrefix(headerName, "cf-")
-}
-
-// isWebsocketClientHeader returns true if the header name is required by the client to upgrade properly
-func IsWebsocketClientHeader(headerName string) bool {
-	return headerName == "sec-websocket-accept" ||
-		headerName == "connection" ||
-		headerName == "upgrade"
-}
-
-func H1ResponseToH2ResponseHeaders(status int, h1 http.Header) (h2 []Header) {
-	h2 = []Header{
-		{Name: ":status", Value: strconv.Itoa(status)},
-	}
-	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,
-			// so it should be sent as an HTTP/2 response header.
-
-			// Since these are http2 headers, they're required to be lowercase
-			h2 = append(h2, Header{Name: "content-length", Value: values[0]})
-		} else if !IsControlHeader(h2name) || IsWebsocketClientHeader(h2name) {
-			// User headers, on the other hand, must all be serialized so that
-			// HTTP/2 header validation won't be applied to HTTP/1 header values
-			userHeaders[header] = values
-		}
-	}
-
-	// Perform user header serialization and set them in the single header
-	h2 = append(h2, Header{ResponseUserHeadersField, SerializeHeaders(userHeaders)})
-	return h2
-}
-
-// Serialize HTTP1.x headers by base64-encoding each header name and value,
-// and then joining them in the format of [key:value;]
-func SerializeHeaders(h1Headers http.Header) string {
-	// compute size of the fully serialized value and largest temp buffer we will need
-	serializedLen := 0
-	maxTempLen := 0
-	for headerName, headerValues := range h1Headers {
-		for _, headerValue := range headerValues {
-			nameLen := headerEncoding.EncodedLen(len(headerName))
-			valueLen := headerEncoding.EncodedLen(len(headerValue))
-			const delims = 2
-			serializedLen += delims + nameLen + valueLen
-			if nameLen > maxTempLen {
-				maxTempLen = nameLen
-			}
-			if valueLen > maxTempLen {
-				maxTempLen = valueLen
-			}
-		}
-	}
-	var buf strings.Builder
-	buf.Grow(serializedLen)
-
-	temp := make([]byte, maxTempLen)
-	writeB64 := func(s string) {
-		n := headerEncoding.EncodedLen(len(s))
-		if n > len(temp) {
-			temp = make([]byte, n)
-		}
-		headerEncoding.Encode(temp[:n], []byte(s))
-		buf.Write(temp[:n])
-	}
-
-	for headerName, headerValues := range h1Headers {
-		for _, headerValue := range headerValues {
-			if buf.Len() > 0 {
-				buf.WriteByte(';')
-			}
-			writeB64(headerName)
-			buf.WriteByte(':')
-			writeB64(headerValue)
-		}
-	}
-
-	return buf.String()
-}
-
-// Deserialize headers serialized by `SerializeHeader`
-func DeserializeHeaders(serializedHeaders string) ([]Header, error) {
-	const unableToDeserializeErr = "Unable to deserialize headers"
-
-	var deserialized []Header
-	for _, serializedPair := range strings.Split(serializedHeaders, ";") {
-		if len(serializedPair) == 0 {
-			continue
-		}
-
-		serializedHeaderParts := strings.Split(serializedPair, ":")
-		if len(serializedHeaderParts) != 2 {
-			return nil, errors.New(unableToDeserializeErr)
-		}
-
-		serializedName := serializedHeaderParts[0]
-		serializedValue := serializedHeaderParts[1]
-		deserializedName := make([]byte, headerEncoding.DecodedLen(len(serializedName)))
-		deserializedValue := make([]byte, headerEncoding.DecodedLen(len(serializedValue)))
-
-		if _, err := headerEncoding.Decode(deserializedName, []byte(serializedName)); err != nil {
-			return nil, errors.Wrap(err, unableToDeserializeErr)
-		}
-		if _, err := headerEncoding.Decode(deserializedValue, []byte(serializedValue)); err != nil {
-			return nil, errors.Wrap(err, unableToDeserializeErr)
-		}
-
-		deserialized = append(deserialized, Header{
-			Name:  string(deserializedName),
-			Value: string(deserializedValue),
-		})
-	}
-
-	return deserialized, nil
-}

h2mux/muxedstream.go

@@ -23,6 +23,10 @@ type MuxedStreamDataSignaller interface {
 	Signal(ID uint32)
 }
 
+type Header struct {
+	Name, Value string
+}
+
 // MuxedStream is logically an HTTP/2 stream, with an additional buffer for outgoing data.
 type MuxedStream struct {
 	streamID uint32
@@ -74,8 +78,6 @@ type MuxedStream struct {
 	sentEOF bool
 	// true if the peer sent us an EOF
 	receivedEOF bool
-	// If valid, tunnelHostname is used to identify which origin service is the intended recipient of the request
-	tunnelHostname TunnelHostname
 	// Compression-related fields
 	receivedUseDict bool
 	method          string
@@ -252,10 +254,6 @@ func (s *MuxedStream) IsRPCStream() bool {
 	return true
 }
 
-func (s *MuxedStream) TunnelHostname() TunnelHostname {
-	return s.tunnelHostname
-}
-
 // Block until a value is sent on writeBufferHasSpace.
 // Must be called while holding writeLock
 func (s *MuxedStream) awaitWriteBufferHasSpace() {

h2mux/muxreader.go

@@ -12,10 +12,6 @@ import (
 	"golang.org/x/net/http2"
 )
 
-const (
-	CloudflaredProxyTunnelHostnameHeader = "cf-cloudflared-proxy-tunnel-hostname"
-)
-
 type MuxReader struct {
 	// f is used to read HTTP2 frames.
 	f *http2.Framer
@@ -252,8 +248,6 @@ func (r *MuxReader) receiveHeaderData(frame *http2.MetaHeadersFrame) error {
 			if r.dictionaries.write != nil {
 				continue
 			}
-		case CloudflaredProxyTunnelHostnameHeader:
-			stream.tunnelHostname = TunnelHostname(header.Value)
 		}
 		headers = append(headers, Header{Name: header.Name, Value: header.Value})
 	}

h2mux/muxreader_test.go

@@ -21,10 +21,6 @@ var (
 		Name:  ":path",
 		Value: "/api/tunnels",
 	}
-	tunnelHostnameHeader = Header{
-		Name:  CloudflaredProxyTunnelHostnameHeader,
-		Value: "tunnel.example.com",
-	}
 	respStatusHeader = Header{
 		Name:  ":status",
 		Value: "200",
@@ -42,15 +38,6 @@ func (mosh *mockOriginStreamHandler) ServeStream(stream *MuxedStream) error {
 	return nil
 }
 
-func getCloudflaredProxyTunnelHostnameHeader(stream *MuxedStream) string {
-	for _, header := range stream.Headers {
-		if header.Name == CloudflaredProxyTunnelHostnameHeader {
-			return header.Value
-		}
-	}
-	return ""
-}
-
 func assertOpenStreamSucceed(t *testing.T, stream *MuxedStream, err error) {
 	assert.NoError(t, err)
 	assert.Len(t, stream.Headers, 1)
@@ -72,13 +59,11 @@ func TestMissingHeaders(t *testing.T) {
 		},
 	}
 
-	// Request doesn't contain CloudflaredProxyTunnelHostnameHeader
 	stream, err := muxPair.EdgeMux.OpenStream(ctx, reqHeaders, nil)
 	assertOpenStreamSucceed(t, stream, err)
 
 	assert.Empty(t, originHandler.stream.method)
 	assert.Empty(t, originHandler.stream.path)
-	assert.False(t, originHandler.stream.TunnelHostname().IsSet())
 }
 
 func TestReceiveHeaderData(t *testing.T) {
@@ -90,18 +75,14 @@ func TestReceiveHeaderData(t *testing.T) {
 		methodHeader,
 		schemeHeader,
 		pathHeader,
-		tunnelHostnameHeader,
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	defer cancel()
 
-	reqHeaders = append(reqHeaders, tunnelHostnameHeader)
 	stream, err := muxPair.EdgeMux.OpenStream(ctx, reqHeaders, nil)
 	assertOpenStreamSucceed(t, stream, err)
 
 	assert.Equal(t, methodHeader.Value, originHandler.stream.method)
 	assert.Equal(t, pathHeader.Value, originHandler.stream.path)
-	assert.True(t, originHandler.stream.TunnelHostname().IsSet())
-	assert.Equal(t, tunnelHostnameHeader.Value, originHandler.stream.TunnelHostname().String())
 }

ingress/origin_connection.go

@@ -2,12 +2,9 @@ package ingress
 
 import (
 	"context"
-	"crypto/tls"
 	"io"
 	"net"
-	"net/http"
 
-	gws "github.com/gorilla/websocket"
 	"github.com/rs/zerolog"
 
 	"github.com/cloudflare/cloudflared/ipaccess"
@@ -25,34 +22,10 @@ type OriginConnection interface {
 
 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)
+	websocket.Stream(originConn, remoteConn, log)
 }
 
 // tcpConnection is an OriginConnection that directly streams to raw TCP.
@@ -61,7 +34,7 @@ type tcpConnection struct {
 }
 
 func (tc *tcpConnection) Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger) {
-	Stream(tunnelConn, tc.conn, log)
+	websocket.Stream(tunnelConn, tc.conn, log)
 }
 
 func (tc *tcpConnection) Close() {
@@ -82,35 +55,6 @@ 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
-}
-
 // socksProxyOverWSConnection is an OriginConnection that streams SOCKS connections over WS.
 // The connection to the origin happens inside the SOCKS code as the client specifies the origin
 // details in the packet.
@@ -124,3 +68,16 @@ func (sp *socksProxyOverWSConnection) Stream(ctx context.Context, tunnelConn io.
 
 func (sp *socksProxyOverWSConnection) Close() {
 }
+
+// wsProxyConnection represents a bidirectional stream for a websocket connection to the origin
+type wsProxyConnection struct {
+	rwc io.ReadWriteCloser
+}
+
+func (conn *wsProxyConnection) Stream(ctx context.Context, tunnelConn io.ReadWriter, log *zerolog.Logger) {
+	websocket.Stream(tunnelConn, conn.rwc, log)
+}
+
+func (conn *wsProxyConnection) Close() {
+	conn.rwc.Close()
+}

ingress/origin_connection_test.go

@@ -3,13 +3,13 @@ package ingress
 import (
 	"bytes"
 	"context"
-	"crypto/tls"
 	"fmt"
 	"io/ioutil"
 	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"sync"
 	"testing"
 	"time"
 
@@ -22,6 +22,7 @@ import (
 
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/socks"
+	"github.com/cloudflare/cloudflared/websocket"
 )
 
 const (
@@ -157,7 +158,7 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) {
 			require.NoError(t, err)
 			defer wsForwarderInConn.Close()
 
-			Stream(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, testLogger)
+			websocket.Stream(wsForwarderInConn, &wsEyeball{wsForwarderOutConn}, testLogger)
 			return nil
 		})
 
@@ -192,18 +193,26 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) {
 func TestStreamWSConnection(t *testing.T) {
 	eyeballConn, edgeConn := net.Pipe()
 
-	origin := echoWSOrigin(t)
+	origin := echoWSOrigin(t, true)
 	defer origin.Close()
 
+	var svc httpService
+	err := svc.start(&sync.WaitGroup{}, testLogger, nil, nil, OriginRequestConfig{
+		NoTLSVerify: true,
+	})
+	require.NoError(t, err)
+
 	req, err := http.NewRequest(http.MethodGet, origin.URL, nil)
 	require.NoError(t, err)
 	req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+	req.Header.Set("Connection", "Upgrade")
+	req.Header.Set("Upgrade", "websocket")
+
+	conn, resp, err := svc.newWebsocketProxyConnection(req)
 
-	clientTLSConfig := &tls.Config{
-		InsecureSkipVerify: true,
-	}
-	wsConn, resp, err := newWSConnection(clientTLSConfig, req)
 	require.NoError(t, err)
+	defer conn.Close()
+
 	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"))
@@ -212,13 +221,37 @@ func TestStreamWSConnection(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), testStreamTimeout)
 	defer cancel()
 
+	connClosed := make(chan struct{})
+
 	errGroup, ctx := errgroup.WithContext(ctx)
+	errGroup.Go(func() error {
+		select {
+		case <-connClosed:
+		case <-ctx.Done():
+		}
+		if ctx.Err() == context.DeadlineExceeded {
+			eyeballConn.Close()
+			edgeConn.Close()
+			conn.Close()
+		}
+
+		return ctx.Err()
+	})
+
 	errGroup.Go(func() error {
 		echoWSEyeball(t, eyeballConn)
+		fmt.Println("closing pipe")
+		edgeConn.Close()
+		return eyeballConn.Close()
+	})
+
+	errGroup.Go(func() error {
+		defer conn.Close()
+		conn.Stream(ctx, edgeConn, testLogger)
+		close(connClosed)
 		return nil
 	})
 
-	wsConn.Stream(ctx, edgeConn, testLogger)
 	require.NoError(t, errGroup.Wait())
 }
 
@@ -240,17 +273,23 @@ func (wse *wsEyeball) Write(p []byte) (int, error) {
 }
 
 func echoWSEyeball(t *testing.T, conn net.Conn) {
-	require.NoError(t, wsutil.WriteClientBinary(conn, testMessage))
+	defer func() {
+		assert.NoError(t, conn.Close())
+	}()
+
+	if !assert.NoError(t, wsutil.WriteClientBinary(conn, testMessage)) {
+		return
+	}
 
 	readMsg, err := wsutil.ReadServerBinary(conn)
-	require.NoError(t, err)
+	if !assert.NoError(t, err) {
+		return
+	}
 
-	require.Equal(t, testResponse, readMsg)
-
-	require.NoError(t, conn.Close())
+	assert.Equal(t, testResponse, readMsg)
 }
 
-func echoWSOrigin(t *testing.T) *httptest.Server {
+func echoWSOrigin(t *testing.T, expectMessages bool) *httptest.Server {
 	var upgrader = gorillaWS.Upgrader{
 		ReadBufferSize:  10,
 		WriteBufferSize: 10,
@@ -267,12 +306,17 @@ func echoWSOrigin(t *testing.T) *httptest.Server {
 		require.NoError(t, err)
 		defer conn.Close()
 
+		sawMessage := false
 		for {
 			messageType, p, err := conn.ReadMessage()
 			if err != nil {
+				if expectMessages && !sawMessage {
+					t.Errorf("unexpected error: %v", err)
+				}
 				return
 			}
-			require.Equal(t, testMessage, p)
+			assert.Equal(t, testMessage, p)
+			sawMessage = true
 			if err := conn.WriteMessage(messageType, testResponse); err != nil {
 				return
 			}

ingress/origin_proxy.go

@@ -2,19 +2,20 @@ package ingress
 
 import (
 	"fmt"
+	"io"
 	"net"
 	"net/http"
-	"net/url"
 	"strings"
 
 	"github.com/pkg/errors"
 
-	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/cloudflare/cloudflared/carrier"
 	"github.com/cloudflare/cloudflared/websocket"
 )
 
 var (
-	switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
+	switchingProtocolText        = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
+	errUnsupportedConnectionType = errors.New("internal error: unsupported connection type")
 )
 
 // HTTPOriginProxy can be implemented by origin services that want to proxy http requests.
@@ -44,26 +45,64 @@ func (o *httpService) RoundTrip(req *http.Request) (*http.Response, error) {
 }
 
 func (o *httpService) EstablishConnection(req *http.Request) (OriginConnection, *http.Response, error) {
+	req = req.Clone(req.Context())
+
 	req.URL.Host = o.url.Host
-	req.URL.Scheme = websocket.ChangeRequestScheme(o.url)
+	req.URL.Scheme = o.url.Scheme
+	// allow ws(s) scheme for websocket-only origins, normal http(s) requests will fail
+	switch req.URL.Scheme {
+	case "ws":
+		req.URL.Scheme = "http"
+	case "wss":
+		req.URL.Scheme = "https"
+	}
+
 	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)
+
+	return o.newWebsocketProxyConnection(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 *httpService) newWebsocketProxyConnection(req *http.Request) (OriginConnection, *http.Response, error) {
+	req.Header.Set("Connection", "Upgrade")
+	req.Header.Set("Upgrade", "websocket")
+	req.Header.Set("Sec-WebSocket-Version", "13")
 
-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)
+	req.ContentLength = 0
+	req.Body = nil
+
+	resp, err := o.transport.RoundTrip(req)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	toClose := resp.Body
+	defer func() {
+		if toClose != nil {
+			_ = toClose.Close()
+		}
+	}()
+
+	if resp.StatusCode != http.StatusSwitchingProtocols {
+		return nil, nil, fmt.Errorf("unexpected origin response: %s", resp.Status)
+	}
+	if strings.ToLower(resp.Header.Get("Upgrade")) != "websocket" {
+		return nil, nil, fmt.Errorf("unexpected upgrade: %q", resp.Header.Get("Upgrade"))
+	}
+
+	rwc, ok := resp.Body.(io.ReadWriteCloser)
+	if !ok {
+		return nil, nil, errUnsupportedConnectionType
+	}
+	conn := wsProxyConnection{
+		rwc: rwc,
+	}
+	// clear to prevent defer from closing
+	toClose = nil
+
+	return &conn, resp, nil
 }
 
 func (o *statusCode) RoundTrip(_ *http.Request) (*http.Response, error) {
@@ -106,7 +145,7 @@ func (o *tcpOverWSService) EstablishConnection(r *http.Request) (OriginConnectio
 	var err error
 	dest := o.dest
 	if o.isBastion {
-		dest, err = o.bastionDest(r)
+		dest, err = carrier.ResolveBastionDest(r)
 		if err != nil {
 			return nil, nil, err
 		}
@@ -130,23 +169,6 @@ func (o *tcpOverWSService) EstablishConnection(r *http.Request) (OriginConnectio
 
 }
 
-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]
-}
-
 func (o *socksProxyOverWSService) EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error) {
 	originConn := o.conn
 	resp := &http.Response{

ingress/origin_proxy_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
-	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/cloudflare/cloudflared/carrier"
 	"github.com/cloudflare/cloudflared/websocket"
 )
 
@@ -33,7 +33,7 @@ func assertEstablishConnectionResponse(t *testing.T,
 }
 
 func TestHTTPServiceEstablishConnection(t *testing.T) {
-	origin := echoWSOrigin(t)
+	origin := echoWSOrigin(t, false)
 	defer origin.Close()
 	originURL, err := url.Parse(origin.URL)
 	require.NoError(t, err)
@@ -71,11 +71,11 @@ func TestHelloWorldEstablishConnection(t *testing.T) {
 	// 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)
+	req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
 
 	expectHeader := http.Header{
-		"Connection": {"Upgrade"},
-		// Accept key when Sec-Websocket-Key is not specified
-		"Sec-Websocket-Accept": {"Kfh9QIsMVZcl6xEPYxPHzW8SZ8w="},
+		"Connection":           {"Upgrade"},
+		"Sec-Websocket-Accept": {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
 		"Upgrade":              {"websocket"},
 	}
 	assertEstablishConnectionResponse(t, helloWorldSerivce, req, expectHeader)
@@ -126,7 +126,7 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 	baseReq.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
 
 	bastionReq := baseReq.Clone(context.Background())
-	bastionReq.Header.Set(h2mux.CFJumpDestinationHeader, originListener.Addr().String())
+	carrier.SetBastionDest(bastionReq.Header, originListener.Addr().String())
 
 	expectHeader := http.Header{
 		"Connection":           {"Upgrade"},
@@ -135,19 +135,23 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 	}
 
 	tests := []struct {
+		testCase  string
 		service   *tcpOverWSService
 		req       *http.Request
 		expectErr bool
 	}{
 		{
-			service: newTCPOverWSService(originURL),
-			req:     baseReq,
+			testCase: "specific TCP service",
+			service:  newTCPOverWSService(originURL),
+			req:      baseReq,
 		},
 		{
-			service: newBastionService(),
-			req:     bastionReq,
+			testCase: "bastion service",
+			service:  newBastionService(),
+			req:      bastionReq,
 		},
 		{
+			testCase:  "invalid bastion request",
 			service:   newBastionService(),
 			req:       baseReq,
 			expectErr: true,
@@ -155,13 +159,15 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 	}
 
 	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)
-		}
+		t.Run(test.testCase, func(t *testing.T) {
+			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()
@@ -175,104 +181,6 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 	}
 }
 
-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(),

ingress/origin_service.go

@@ -11,7 +11,6 @@ import (
 	"sync"
 	"time"
 
-	gws "github.com/gorilla/websocket"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 
@@ -19,7 +18,6 @@ import (
 	"github.com/cloudflare/cloudflared/ipaccess"
 	"github.com/cloudflare/cloudflared/socks"
 	"github.com/cloudflare/cloudflared/tlsconfig"
-	"github.com/cloudflare/cloudflared/websocket"
 )
 
 // originService is something a tunnel can proxy traffic to.
@@ -50,16 +48,6 @@ func (o *unixSocketPath) start(wg *sync.WaitGroup, log *zerolog.Logger, shutdown
 	return nil
 }
 
-func (o *unixSocketPath) Dial(reqURL *url.URL, headers http.Header) (*gws.Conn, *http.Response, error) {
-	d := &gws.Dialer{
-		NetDial:         o.transport.Dial,
-		NetDialContext:  o.transport.DialContext,
-		TLSClientConfig: o.transport.TLSClientConfig,
-	}
-	reqURL.Scheme = websocket.ChangeRequestScheme(reqURL)
-	return d.Dial(reqURL.String(), headers)
-}
-
 type httpService struct {
 	url        *url.URL
 	hostHeader string
@@ -171,8 +159,8 @@ func (o *socksProxyOverWSService) String() string {
 // HelloWorld is an OriginService for the built-in Hello World server.
 // Users only use this for testing and experimenting with cloudflared.
 type helloWorld struct {
-	server    net.Listener
-	transport *http.Transport
+	httpService
+	server net.Listener
 }
 
 func (o *helloWorld) String() string {
@@ -187,11 +175,10 @@ func (o *helloWorld) start(
 	errC chan error,
 	cfg OriginRequestConfig,
 ) error {
-	transport, err := newHTTPTransport(o, cfg, log)
-	if err != nil {
+	if err := o.httpService.start(wg, log, shutdownC, errC, cfg); err != nil {
 		return err
 	}
-	o.transport = transport
+
 	helloListener, err := hello.CreateTLSListener("127.0.0.1:")
 	if err != nil {
 		return errors.Wrap(err, "Cannot start Hello World Server")
@@ -202,6 +189,12 @@ func (o *helloWorld) start(
 		_ = hello.StartHelloWorldServer(log, helloListener, shutdownC)
 	}()
 	o.server = helloListener
+
+	o.httpService.url = &url.URL{
+		Scheme: "https",
+		Host:   o.server.Addr().String(),
+	}
+
 	return nil
 }
 

origin/proxy.go

@@ -67,7 +67,7 @@ func (p *proxy) Proxy(w connection.ResponseWriter, req *http.Request, sourceConn
 			lbProbe: lbProbe,
 			rule:    ingress.ServiceWarpRouting,
 		}
-		if err := p.proxyStreamRequest(serveCtx, w, req, sourceConnectionType, p.warpRouting.Proxy, logFields); err != nil {
+		if err := p.proxyStreamRequest(serveCtx, w, req, p.warpRouting.Proxy, logFields); err != nil {
 			p.logRequestError(err, cfRay, ingress.ServiceWarpRouting)
 			return err
 		}
@@ -96,7 +96,7 @@ func (p *proxy) Proxy(w connection.ResponseWriter, req *http.Request, sourceConn
 		return fmt.Errorf("Not a connection-oriented service")
 	}
 
-	if err := p.proxyStreamRequest(serveCtx, w, req, sourceConnectionType, connectionProxy, logFields); err != nil {
+	if err := p.proxyStreamRequest(serveCtx, w, req, connectionProxy, logFields); err != nil {
 		p.logRequestError(err, cfRay, ruleNum)
 		return err
 	}
@@ -152,7 +152,6 @@ func (p *proxy) proxyStreamRequest(
 	serveCtx context.Context,
 	w connection.ResponseWriter,
 	req *http.Request,
-	sourceConnectionType connection.Type,
 	connectionProxy ingress.StreamBasedOriginProxy,
 	fields logFields,
 ) error {

origin/proxy_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/urfave/cli/v2"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/cloudflare/cloudflared/config"
 	"github.com/cloudflare/cloudflared/connection"
@@ -79,6 +80,11 @@ func (w *mockWSRespWriter) respBody() io.ReadWriter {
 	return bytes.NewBuffer(data)
 }
 
+func (w *mockWSRespWriter) Close() error {
+	close(w.writeNotification)
+	return nil
+}
+
 func (w *mockWSRespWriter) Read(data []byte) (int, error) {
 	return w.reader.Read(data)
 }
@@ -125,14 +131,14 @@ func TestProxySingleOrigin(t *testing.T) {
 	require.NoError(t, ingressRule.StartOrigins(&wg, &log, ctx.Done(), errC))
 
 	proxy := NewOriginProxy(ingressRule, unusedWarpRoutingService, testTags, &log)
-	t.Run("testProxyHTTP", testProxyHTTP(t, proxy))
-	t.Run("testProxyWebsocket", testProxyWebsocket(t, proxy))
-	t.Run("testProxySSE", testProxySSE(t, proxy))
+	t.Run("testProxyHTTP", testProxyHTTP(proxy))
+	t.Run("testProxyWebsocket", testProxyWebsocket(proxy))
+	t.Run("testProxySSE", testProxySSE(proxy))
 	cancel()
 	wg.Wait()
 }
 
-func testProxyHTTP(t *testing.T, proxy connection.OriginProxy) func(t *testing.T) {
+func testProxyHTTP(proxy connection.OriginProxy) func(t *testing.T) {
 	return func(t *testing.T) {
 		responseWriter := newMockHTTPRespWriter()
 		req, err := http.NewRequest(http.MethodGet, "http://localhost:8080", nil)
@@ -145,23 +151,43 @@ func testProxyHTTP(t *testing.T, proxy connection.OriginProxy) func(t *testing.T
 	}
 }
 
-func testProxyWebsocket(t *testing.T, proxy connection.OriginProxy) func(t *testing.T) {
+func testProxyWebsocket(proxy connection.OriginProxy) func(t *testing.T) {
 	return func(t *testing.T) {
 		// WSRoute is a websocket echo handler
-		ctx, cancel := context.WithCancel(context.Background())
+		const testTimeout = 5 * time.Second * 1000
+		ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
+		defer cancel()
 		readPipe, writePipe := io.Pipe()
 		req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("http://localhost:8080%s", hello.WSRoute), readPipe)
-		responseWriter := newMockWSRespWriter(readPipe)
+		req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+		req.Header.Set("Connection", "Upgrade")
+		req.Header.Set("Upgrade", "websocket")
+		responseWriter := newMockWSRespWriter(nil)
 
-		var wg sync.WaitGroup
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
+		finished := make(chan struct{})
+
+		errGroup, ctx := errgroup.WithContext(ctx)
+		errGroup.Go(func() error {
 			err = proxy.Proxy(responseWriter, req, connection.TypeWebsocket)
 			require.NoError(t, err)
 
 			require.Equal(t, http.StatusSwitchingProtocols, responseWriter.Code)
-		}()
+			return nil
+		})
+
+		errGroup.Go(func() error {
+			select {
+			case <-finished:
+			case <-ctx.Done():
+			}
+			if ctx.Err() == context.DeadlineExceeded {
+				t.Errorf("Test timed out")
+				readPipe.Close()
+				writePipe.Close()
+				responseWriter.Close()
+			}
+			return nil
+		})
 
 		msg := []byte("test websocket")
 		err = wsutil.WriteClientText(writePipe, msg)
@@ -179,12 +205,16 @@ func testProxyWebsocket(t *testing.T, proxy connection.OriginProxy) func(t *test
 		require.NoError(t, err)
 		require.Equal(t, msg, returnedMsg)
 
-		cancel()
-		wg.Wait()
+		_ = readPipe.Close()
+		_ = writePipe.Close()
+		_ = responseWriter.Close()
+
+		close(finished)
+		errGroup.Wait()
 	}
 }
 
-func testProxySSE(t *testing.T, proxy connection.OriginProxy) func(t *testing.T) {
+func testProxySSE(proxy connection.OriginProxy) func(t *testing.T) {
 	return func(t *testing.T) {
 		var (
 			pushCount = 50

origin/reconnect.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/cloudflare/cloudflared/retry"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
@@ -103,7 +104,7 @@ func (cm *reconnectCredentialManager) SetConnDigest(connID uint8, digest []byte)
 
 func (cm *reconnectCredentialManager) RefreshAuth(
 	ctx context.Context,
-	backoff *BackoffHandler,
+	backoff *retry.BackoffHandler,
 	authenticate func(ctx context.Context, numPreviousAttempts int) (tunnelpogs.AuthOutcome, error),
 ) (retryTimer <-chan time.Time, err error) {
 	authOutcome, err := authenticate(ctx, backoff.Retries())
@@ -121,11 +122,11 @@ func (cm *reconnectCredentialManager) RefreshAuth(
 	case tunnelpogs.AuthSuccess:
 		cm.SetReconnectToken(outcome.JWT())
 		cm.authSuccess.Inc()
-		return timeAfter(outcome.RefreshAfter()), nil
+		return retry.Clock.After(outcome.RefreshAfter()), nil
 	case tunnelpogs.AuthUnknown:
 		duration := outcome.RefreshAfter()
 		cm.authFail.WithLabelValues(outcome.Error()).Inc()
-		return timeAfter(duration), nil
+		return retry.Clock.After(duration), nil
 	case tunnelpogs.AuthFail:
 		cm.authFail.WithLabelValues(outcome.Error()).Inc()
 		return nil, outcome

origin/reconnect_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/cloudflare/cloudflared/retry"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
@@ -17,11 +18,11 @@ func TestRefreshAuthBackoff(t *testing.T) {
 	rcm := newReconnectCredentialManager(t.Name(), t.Name(), 4)
 
 	var wait time.Duration
-	timeAfter = func(d time.Duration) <-chan time.Time {
+	retry.Clock.After = func(d time.Duration) <-chan time.Time {
 		wait = d
 		return time.After(d)
 	}
-	backoff := &BackoffHandler{MaxRetries: 3}
+	backoff := &retry.BackoffHandler{MaxRetries: 3}
 	auth := func(ctx context.Context, n int) (tunnelpogs.AuthOutcome, error) {
 		return nil, fmt.Errorf("authentication failure")
 	}
@@ -45,7 +46,7 @@ func TestRefreshAuthBackoff(t *testing.T) {
 
 	// The backoff timer should have been reset. To confirm this, make timeNow
 	// return a value after the backoff timer's grace period
-	timeNow = func() time.Time {
+	retry.Clock.Now = func() time.Time {
 		expectedGracePeriod := time.Duration(time.Second * 2 << backoff.MaxRetries)
 		return time.Now().Add(expectedGracePeriod * 2)
 	}
@@ -57,12 +58,12 @@ func TestRefreshAuthSuccess(t *testing.T) {
 	rcm := newReconnectCredentialManager(t.Name(), t.Name(), 4)
 
 	var wait time.Duration
-	timeAfter = func(d time.Duration) <-chan time.Time {
+	retry.Clock.After = func(d time.Duration) <-chan time.Time {
 		wait = d
 		return time.After(d)
 	}
 
-	backoff := &BackoffHandler{MaxRetries: 3}
+	backoff := &retry.BackoffHandler{MaxRetries: 3}
 	auth := func(ctx context.Context, n int) (tunnelpogs.AuthOutcome, error) {
 		return tunnelpogs.NewAuthSuccess([]byte("jwt"), 19), nil
 	}
@@ -81,12 +82,12 @@ func TestRefreshAuthUnknown(t *testing.T) {
 	rcm := newReconnectCredentialManager(t.Name(), t.Name(), 4)
 
 	var wait time.Duration
-	timeAfter = func(d time.Duration) <-chan time.Time {
+	retry.Clock.After = func(d time.Duration) <-chan time.Time {
 		wait = d
 		return time.After(d)
 	}
 
-	backoff := &BackoffHandler{MaxRetries: 3}
+	backoff := &retry.BackoffHandler{MaxRetries: 3}
 	auth := func(ctx context.Context, n int) (tunnelpogs.AuthOutcome, error) {
 		return tunnelpogs.NewAuthUnknown(errors.New("auth unknown"), 19), nil
 	}
@@ -104,7 +105,7 @@ func TestRefreshAuthUnknown(t *testing.T) {
 func TestRefreshAuthFail(t *testing.T) {
 	rcm := newReconnectCredentialManager(t.Name(), t.Name(), 4)
 
-	backoff := &BackoffHandler{MaxRetries: 3}
+	backoff := &retry.BackoffHandler{MaxRetries: 3}
 	auth := func(ctx context.Context, n int) (tunnelpogs.AuthOutcome, error) {
 		return tunnelpogs.NewAuthFail(errors.New("auth fail")), nil
 	}

origin/supervisor.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/edgediscovery"
 	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/cloudflare/cloudflared/retry"
 	"github.com/cloudflare/cloudflared/signal"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
@@ -112,10 +113,10 @@ func (s *Supervisor) Run(
 	var tunnelsWaiting []int
 	tunnelsActive := s.config.HAConnections
 
-	backoff := BackoffHandler{MaxRetries: s.config.Retries, BaseTime: tunnelRetryDuration, RetryForever: true}
+	backoff := retry.BackoffHandler{MaxRetries: s.config.Retries, BaseTime: tunnelRetryDuration, RetryForever: true}
 	var backoffTimer <-chan time.Time
 
-	refreshAuthBackoff := &BackoffHandler{MaxRetries: refreshAuthMaxBackoff, BaseTime: refreshAuthRetryDuration, RetryForever: true}
+	refreshAuthBackoff := &retry.BackoffHandler{MaxRetries: refreshAuthMaxBackoff, BaseTime: refreshAuthRetryDuration, RetryForever: true}
 	var refreshAuthBackoffTimer <-chan time.Time
 
 	if s.useReconnectToken {

origin/tunnel.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/edgediscovery"
 	"github.com/cloudflare/cloudflared/h2mux"
+	"github.com/cloudflare/cloudflared/retry"
 	"github.com/cloudflare/cloudflared/signal"
 	"github.com/cloudflare/cloudflared/tunnelrpc"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
@@ -138,7 +139,7 @@ func ServeTunnelLoop(
 	connLog := config.Log.With().Uint8(connection.LogFieldConnIndex, connIndex).Logger()
 
 	protocolFallback := &protocolFallback{
-		BackoffHandler{MaxRetries: config.Retries},
+		retry.BackoffHandler{MaxRetries: config.Retries},
 		config.ProtocolSelector.Current(),
 		false,
 	}
@@ -195,18 +196,18 @@ func ServeTunnelLoop(
 // protocolFallback is a wrapper around backoffHandler that will try fallback option when backoff reaches
 // max retries
 type protocolFallback struct {
-	BackoffHandler
+	retry.BackoffHandler
 	protocol   connection.Protocol
 	inFallback bool
 }
 
 func (pf *protocolFallback) reset() {
-	pf.resetNow()
+	pf.ResetNow()
 	pf.inFallback = false
 }
 
 func (pf *protocolFallback) fallback(fallback connection.Protocol) {
-	pf.resetNow()
+	pf.ResetNow()
 	pf.protocol = fallback
 	pf.inFallback = true
 }
@@ -281,7 +282,7 @@ func ServeTunnel(
 	}
 
 	if protocol == connection.HTTP2 {
-		connOptions := config.ConnectionOptions(edgeConn.LocalAddr().String(), uint8(backoff.retries))
+		connOptions := config.ConnectionOptions(edgeConn.LocalAddr().String(), uint8(backoff.Retries()))
 		err = ServeHTTP2(
 			ctx,
 			connLog,
@@ -382,7 +383,7 @@ func ServeH2mux(
 
 	errGroup.Go(func() error {
 		if config.NamedTunnel != nil {
-			connOptions := config.ConnectionOptions(edgeConn.LocalAddr().String(), uint8(connectedFuse.backoff.retries))
+			connOptions := config.ConnectionOptions(edgeConn.LocalAddr().String(), uint8(connectedFuse.backoff.Retries()))
 			return handler.ServeNamedTunnel(serveCtx, config.NamedTunnel, connOptions, connectedFuse)
 		}
 		registrationOptions := config.RegistrationOptions(connIndex, edgeConn.LocalAddr().String(), cloudflaredUUID)

origin/tunnel_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/stretchr/testify/assert"
 
 	"github.com/cloudflare/cloudflared/connection"
+	"github.com/cloudflare/cloudflared/retry"
 )
 
 type dynamicMockFetcher struct {
@@ -26,7 +27,7 @@ func (dmf *dynamicMockFetcher) fetch() connection.PercentageFetcher {
 
 func TestWaitForBackoffFallback(t *testing.T) {
 	maxRetries := uint(3)
-	backoff := BackoffHandler{
+	backoff := retry.BackoffHandler{
 		MaxRetries: maxRetries,
 		BaseTime:   time.Millisecond * 10,
 	}

retry/backoffhandler.go

@@ -1,4 +1,4 @@
-package origin
+package retry
 
 import (
 	"context"
@@ -7,10 +7,15 @@ import (
 )
 
 // Redeclare time functions so they can be overridden in tests.
-var (
-	timeNow   = time.Now
-	timeAfter = time.After
-)
+type clock struct {
+	Now   func() time.Time
+	After func(d time.Duration) <-chan time.Time
+}
+
+var Clock = clock{
+	Now:   time.Now,
+	After: time.After,
+}
 
 // BackoffHandler manages exponential backoff and limits the maximum number of retries.
 // The base time period is 1 second, doubling with each retry.
@@ -39,7 +44,7 @@ func (b BackoffHandler) GetMaxBackoffDuration(ctx context.Context) (time.Duratio
 		return time.Duration(0), false
 	default:
 	}
-	if !b.resetDeadline.IsZero() && timeNow().After(b.resetDeadline) {
+	if !b.resetDeadline.IsZero() && Clock.Now().After(b.resetDeadline) {
 		// b.retries would be set to 0 at this point
 		return time.Second, true
 	}
@@ -53,7 +58,7 @@ func (b BackoffHandler) GetMaxBackoffDuration(ctx context.Context) (time.Duratio
 // BackoffTimer returns a channel that sends the current time when the exponential backoff timeout expires.
 // Returns nil if the maximum number of retries have been used.
 func (b *BackoffHandler) BackoffTimer() <-chan time.Time {
-	if !b.resetDeadline.IsZero() && timeNow().After(b.resetDeadline) {
+	if !b.resetDeadline.IsZero() && Clock.Now().After(b.resetDeadline) {
 		b.retries = 0
 		b.resetDeadline = time.Time{}
 	}
@@ -66,7 +71,7 @@ func (b *BackoffHandler) BackoffTimer() <-chan time.Time {
 	}
 	maxTimeToWait := time.Duration(b.GetBaseTime() * 1 << (b.retries))
 	timeToWait := time.Duration(rand.Int63n(maxTimeToWait.Nanoseconds()))
-	return timeAfter(timeToWait)
+	return Clock.After(timeToWait)
 }
 
 // Backoff is used to wait according to exponential backoff. Returns false if the
@@ -89,7 +94,7 @@ func (b *BackoffHandler) Backoff(ctx context.Context) bool {
 func (b *BackoffHandler) SetGracePeriod() {
 	maxTimeToWait := b.GetBaseTime() * 2 << (b.retries + 1)
 	timeToWait := time.Duration(rand.Int63n(maxTimeToWait.Nanoseconds()))
-	b.resetDeadline = timeNow().Add(timeToWait)
+	b.resetDeadline = Clock.Now().Add(timeToWait)
 }
 
 func (b BackoffHandler) GetBaseTime() time.Duration {
@@ -108,6 +113,6 @@ func (b *BackoffHandler) ReachedMaxRetries() bool {
 	return b.retries == b.MaxRetries
 }
 
-func (b *BackoffHandler) resetNow() {
+func (b *BackoffHandler) ResetNow() {
 	b.resetDeadline = time.Now()
 }

retry/backoffhandler_test.go

@@ -1,4 +1,4 @@
-package origin
+package retry
 
 import (
 	"context"
@@ -14,7 +14,7 @@ func immediateTimeAfter(time.Duration) <-chan time.Time {
 
 func TestBackoffRetries(t *testing.T) {
 	// make backoff return immediately
-	timeAfter = immediateTimeAfter
+	Clock.After = immediateTimeAfter
 	ctx := context.Background()
 	backoff := BackoffHandler{MaxRetries: 3}
 	if !backoff.Backoff(ctx) {
@@ -33,7 +33,7 @@ func TestBackoffRetries(t *testing.T) {
 
 func TestBackoffCancel(t *testing.T) {
 	// prevent backoff from returning normally
-	timeAfter = func(time.Duration) <-chan time.Time { return make(chan time.Time) }
+	Clock.After = func(time.Duration) <-chan time.Time { return make(chan time.Time) }
 	ctx, cancelFunc := context.WithCancel(context.Background())
 	backoff := BackoffHandler{MaxRetries: 3}
 	cancelFunc()
@@ -47,10 +47,10 @@ func TestBackoffCancel(t *testing.T) {
 
 func TestBackoffGracePeriod(t *testing.T) {
 	currentTime := time.Now()
-	// make timeNow return whatever we like
-	timeNow = func() time.Time { return currentTime }
+	// make Clock.Now return whatever we like
+	Clock.Now = func() time.Time { return currentTime }
 	// make backoff return immediately
-	timeAfter = immediateTimeAfter
+	Clock.After = immediateTimeAfter
 	ctx := context.Background()
 	backoff := BackoffHandler{MaxRetries: 1}
 	if !backoff.Backoff(ctx) {
@@ -71,7 +71,7 @@ func TestBackoffGracePeriod(t *testing.T) {
 
 func TestGetMaxBackoffDurationRetries(t *testing.T) {
 	// make backoff return immediately
-	timeAfter = immediateTimeAfter
+	Clock.After = immediateTimeAfter
 	ctx := context.Background()
 	backoff := BackoffHandler{MaxRetries: 3}
 	if _, ok := backoff.GetMaxBackoffDuration(ctx); !ok {
@@ -96,7 +96,7 @@ func TestGetMaxBackoffDurationRetries(t *testing.T) {
 
 func TestGetMaxBackoffDuration(t *testing.T) {
 	// make backoff return immediately
-	timeAfter = immediateTimeAfter
+	Clock.After = immediateTimeAfter
 	ctx := context.Background()
 	backoff := BackoffHandler{MaxRetries: 3}
 	if duration, ok := backoff.GetMaxBackoffDuration(ctx); !ok || duration > time.Second*2 {
@@ -118,7 +118,7 @@ func TestGetMaxBackoffDuration(t *testing.T) {
 
 func TestBackoffRetryForever(t *testing.T) {
 	// make backoff return immediately
-	timeAfter = immediateTimeAfter
+	Clock.After = immediateTimeAfter
 	ctx := context.Background()
 	backoff := BackoffHandler{MaxRetries: 3, RetryForever: true}
 	if duration, ok := backoff.GetMaxBackoffDuration(ctx); !ok || duration > time.Second*2 {

token/token.go

@@ -18,7 +18,7 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/cloudflare/cloudflared/config"
-	"github.com/cloudflare/cloudflared/origin"
+	"github.com/cloudflare/cloudflared/retry"
 )
 
 const (
@@ -36,7 +36,7 @@ type AppInfo struct {
 
 type lock struct {
 	lockFilePath string
-	backoff      *origin.BackoffHandler
+	backoff      *retry.BackoffHandler
 	sigHandler   *signalHandler
 }
 
@@ -94,7 +94,7 @@ func newLock(path string) *lock {
 	lockPath := path + ".lock"
 	return &lock{
 		lockFilePath: lockPath,
-		backoff:      &origin.BackoffHandler{MaxRetries: 7},
+		backoff:      &retry.BackoffHandler{MaxRetries: 7},
 		sigHandler: &signalHandler{
 			signals: []os.Signal{syscall.SIGINT, syscall.SIGTERM},
 		},

tunnelstore/client.go

@@ -16,6 +16,8 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 
+	"golang.org/x/net/http2"
+
 	"github.com/cloudflare/cloudflared/teamnet"
 )
 
@@ -247,6 +249,11 @@ func NewRESTClient(baseURL, accountTag, zoneTag, authToken, userAgent string, lo
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create account level endpoint")
 	}
+	httpTransport := http.Transport{
+		TLSHandshakeTimeout:   defaultTimeout,
+		ResponseHeaderTimeout: defaultTimeout,
+	}
+	http2.ConfigureTransport(&httpTransport)
 	return &RESTClient{
 		baseEndpoints: &baseEndpoints{
 			accountLevel:  *accountLevelEndpoint,
@@ -256,11 +263,8 @@ func NewRESTClient(baseURL, accountTag, zoneTag, authToken, userAgent string, lo
 		authToken: authToken,
 		userAgent: userAgent,
 		client: http.Client{
-			Transport: &http.Transport{
-				TLSHandshakeTimeout:   defaultTimeout,
-				ResponseHeaderTimeout: defaultTimeout,
-			},
-			Timeout: defaultTimeout,
+			Transport: &httpTransport,
+			Timeout:   defaultTimeout,
 		},
 		log: log,
 	}, nil

websocket/websocket.go

@@ -3,174 +3,108 @@ package websocket
 import (
 	"crypto/sha1"
 	"encoding/base64"
+	"encoding/hex"
+	"errors"
+	"fmt"
 	"io"
-	"net"
 	"net/http"
-	"net/url"
 	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/rs/zerolog"
-
-	"github.com/cloudflare/cloudflared/h2mux"
 )
 
-var stripWebsocketHeaders = []string{
-	"Upgrade",
-	"Connection",
-	"Sec-Websocket-Key",
-	"Sec-Websocket-Version",
-	"Sec-Websocket-Extensions",
-}
-
 // IsWebSocketUpgrade checks to see if the request is a WebSocket connection.
 func IsWebSocketUpgrade(req *http.Request) bool {
 	return websocket.IsWebSocketUpgrade(req)
 }
 
-// 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 *websocket.Dialer) (*websocket.Conn, *http.Response, error) {
-	req.URL.Scheme = ChangeRequestScheme(req.URL)
-	wsHeaders := websocketHeaders(req)
-	if dialler == nil {
-		dialler = &websocket.Dialer{
-			Proxy: http.ProxyFromEnvironment,
-		}
-	}
-	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, nil
-}
-
-// StartProxyServer will start a websocket server that will decode
-// the websocket data and write the resulting data to the provided
-func StartProxyServer(
-	log *zerolog.Logger,
-	listener net.Listener,
-	staticHost string,
-	shutdownC <-chan struct{},
-	streamHandler func(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger),
-) error {
-	upgrader := websocket.Upgrader{
-		ReadBufferSize:  1024,
-		WriteBufferSize: 1024,
-	}
-	h := handler{
-		upgrader:      upgrader,
-		log:           log,
-		staticHost:    staticHost,
-		streamHandler: streamHandler,
-	}
-
-	httpServer := &http.Server{Addr: listener.Addr().String(), Handler: &h}
-	go func() {
-		<-shutdownC
-		_ = httpServer.Close()
-	}()
-
-	return httpServer.Serve(listener)
-}
-
-// HTTP handler for the websocket proxy.
-type handler struct {
-	log           *zerolog.Logger
-	staticHost    string
-	upgrader      websocket.Upgrader
-	streamHandler func(originConn io.ReadWriter, remoteConn net.Conn, log *zerolog.Logger)
-}
-
-func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	// If remote is an empty string, get the destination from the client.
-	finalDestination := h.staticHost
-	if finalDestination == "" {
-		if jumpDestination := r.Header.Get(h2mux.CFJumpDestinationHeader); jumpDestination == "" {
-			h.log.Error().Msg("Did not receive final destination from client. The --destination flag is likely not set")
-			return
-		} else {
-			finalDestination = jumpDestination
-		}
-	}
-
-	stream, err := net.Dial("tcp", finalDestination)
-	if err != nil {
-		h.log.Err(err).Msg("Cannot connect to remote")
-		return
-	}
-	defer stream.Close()
-
-	if !websocket.IsWebSocketUpgrade(r) {
-		_, _ = w.Write(nonWebSocketRequestPage())
-		return
-	}
-	conn, err := h.upgrader.Upgrade(w, r, nil)
-	if err != nil {
-		h.log.Err(err).Msg("failed to upgrade")
-		return
-	}
-	_ = conn.SetReadDeadline(time.Now().Add(pongWait))
-	conn.SetPongHandler(func(string) error { _ = conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
-	gorillaConn := &GorillaConn{Conn: conn, log: h.log}
-	go gorillaConn.pinger(r.Context())
-	defer conn.Close()
-
-	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("Sec-Websocket-Accept", generateAcceptKey(req.Header.Get("Sec-WebSocket-Key")))
 	header.Add("Upgrade", "websocket")
 	return header
 }
 
-// the gorilla websocket library sets its own Upgrade, Connection, Sec-WebSocket-Key,
-// Sec-WebSocket-Version and Sec-Websocket-Extensions headers.
-// https://github.com/gorilla/websocket/blob/master/client.go#L189-L194.
-func websocketHeaders(req *http.Request) http.Header {
-	wsHeaders := make(http.Header)
-	for key, val := range req.Header {
-		wsHeaders[key] = val
-	}
-	// Assume the header keys are in canonical format.
-	for _, header := range stripWebsocketHeaders {
-		wsHeaders.Del(header)
-	}
-	wsHeaders.Set("Host", req.Host) // See TUN-1097
-	return wsHeaders
+// Stream copies copy data to & from provided io.ReadWriters.
+func Stream(tunnelConn, originConn io.ReadWriter, log *zerolog.Logger) {
+	proxyDone := make(chan struct{}, 2)
+
+	go func() {
+		_, err := copyData(tunnelConn, originConn, "origin->tunnel")
+		if err != nil {
+			log.Debug().Msgf("origin to tunnel copy: %v", err)
+		}
+		proxyDone <- struct{}{}
+	}()
+
+	go func() {
+		_, err := copyData(originConn, tunnelConn, "tunnel->origin")
+		if err != nil {
+			log.Debug().Msgf("tunnel to origin copy: %v", err)
+		}
+		proxyDone <- struct{}{}
+	}()
+
+	// If one side is done, we are done.
+	<-proxyDone
 }
 
-// sha1Base64 sha1 and then base64 encodes str.
-func sha1Base64(str string) string {
-	hasher := sha1.New()
-	_, _ = io.WriteString(hasher, str)
-	hash := hasher.Sum(nil)
-	return base64.StdEncoding.EncodeToString(hash)
-}
+// when set to true, enables logging of content copied to/from origin and tunnel
+const debugCopy = false
 
-// generateAcceptKey returns the string needed for the Sec-WebSocket-Accept header.
-// https://tools.ietf.org/html/rfc6455#section-1.3 describes this process in more detail.
-func generateAcceptKey(req *http.Request) string {
-	return sha1Base64(req.Header.Get("Sec-WebSocket-Key") + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11")
-}
-
-// ChangeRequestScheme is needed as the gorilla websocket library requires the ws scheme.
-// (even though it changes it back to http/https, but ¯\_(ツ)_/¯.)
-func ChangeRequestScheme(reqURL *url.URL) string {
-	switch reqURL.Scheme {
-	case "https":
-		return "wss"
-	case "http":
-		return "ws"
-	case "":
-		return "ws"
-	default:
-		return reqURL.Scheme
+func copyData(dst io.Writer, src io.Reader, dir string) (written int64, err error) {
+	if debugCopy {
+		// copyBuffer is based on stdio Copy implementation but shows copied data
+		copyBuffer := func(dst io.Writer, src io.Reader, dir string) (written int64, err error) {
+			var buf []byte
+			size := 32 * 1024
+			buf = make([]byte, size)
+			for {
+				t := time.Now()
+				nr, er := src.Read(buf)
+				if nr > 0 {
+					fmt.Println(dir, t.UnixNano(), "\n"+hex.Dump(buf[0:nr]))
+					nw, ew := dst.Write(buf[0:nr])
+					if nw < 0 || nr < nw {
+						nw = 0
+						if ew == nil {
+							ew = errors.New("invalid write")
+						}
+					}
+					written += int64(nw)
+					if ew != nil {
+						err = ew
+						break
+					}
+					if nr != nw {
+						err = io.ErrShortWrite
+						break
+					}
+				}
+				if er != nil {
+					if er != io.EOF {
+						err = er
+					}
+					break
+				}
+			}
+			return written, err
+		}
+		return copyBuffer(dst, src, dir)
+	} else {
+		return io.Copy(dst, src)
 	}
 }
+
+// from RFC-6455
+var keyGUID = []byte("258EAFA5-E914-47DA-95CA-C5AB0DC85B11")
+
+func generateAcceptKey(challengeKey string) string {
+	h := sha1.New()
+	h.Write([]byte(challengeKey))
+	h.Write(keyGUID)
+	return base64.StdEncoding.EncodeToString(h.Sum(nil))
+}

websocket/websocket_test.go

@@ -1,24 +1,9 @@
 package websocket
 
 import (
-	"context"
-	"crypto/tls"
-	"crypto/x509"
-	"fmt"
-	"io"
-	"math/rand"
-	"net/http"
 	"testing"
-	"time"
 
-	gws "github.com/gorilla/websocket"
-	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"golang.org/x/net/websocket"
-
-	"github.com/cloudflare/cloudflared/hello"
-	"github.com/cloudflare/cloudflared/tlsconfig"
 )
 
 const (
@@ -28,164 +13,6 @@ const (
 	testSecWebsocketAccept = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
 )
 
-func testRequest(t *testing.T, url string, stream io.ReadWriter) *http.Request {
-	req, err := http.NewRequest("GET", url, stream)
-	if err != nil {
-		t.Fatalf("testRequestHeader error")
-	}
-
-	req.Header.Add("Connection", "Upgrade")
-	req.Header.Add("Upgrade", "WebSocket")
-	req.Header.Add("Sec-Websocket-Key", testSecWebsocketKey)
-	req.Header.Add("Sec-Websocket-Protocol", "tunnel-protocol")
-	req.Header.Add("Sec-Websocket-Version", "13")
-	req.Header.Add("User-Agent", "curl/7.59.0")
-
-	return req
-}
-
-func websocketClientTLSConfig(t *testing.T) *tls.Config {
-	certPool := x509.NewCertPool()
-	helloCert, err := tlsconfig.GetHelloCertificateX509()
-	assert.NoError(t, err)
-	certPool.AddCert(helloCert)
-	assert.NotNil(t, certPool)
-	return &tls.Config{RootCAs: certPool}
-}
-
-func TestWebsocketHeaders(t *testing.T) {
-	req := testRequest(t, "http://example.com", nil)
-	wsHeaders := websocketHeaders(req)
-	for _, header := range stripWebsocketHeaders {
-		assert.Empty(t, wsHeaders[header])
-	}
-	assert.Equal(t, "curl/7.59.0", wsHeaders.Get("User-Agent"))
-}
-
 func TestGenerateAcceptKey(t *testing.T) {
-	req := testRequest(t, "http://example.com", nil)
-	assert.Equal(t, testSecWebsocketAccept, generateAcceptKey(req))
+	assert.Equal(t, testSecWebsocketAccept, generateAcceptKey(testSecWebsocketKey))
 }
-
-func TestServe(t *testing.T) {
-	log := zerolog.Nop()
-	shutdownC := make(chan struct{})
-	errC := make(chan error)
-	listener, err := hello.CreateTLSListener("localhost:1111")
-	assert.NoError(t, err)
-	defer listener.Close()
-
-	go func() {
-		errC <- hello.StartHelloWorldServer(&log, listener, shutdownC)
-	}()
-
-	req := testRequest(t, "https://localhost:1111/ws", nil)
-
-	tlsConfig := websocketClientTLSConfig(t)
-	assert.NotNil(t, 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"))
-
-	for i := 0; i < 1000; i++ {
-		messageSize := rand.Int()%2048 + 1
-		clientMessage := make([]byte, messageSize)
-		// rand.Read always returns len(clientMessage) and a nil error
-		rand.Read(clientMessage)
-		err = conn.WriteMessage(websocket.BinaryFrame, clientMessage)
-		assert.NoError(t, err)
-
-		messageType, message, err := conn.ReadMessage()
-		assert.NoError(t, err)
-		assert.Equal(t, websocket.BinaryFrame, messageType)
-		assert.Equal(t, clientMessage, message)
-	}
-
-	_ = conn.Close()
-	close(shutdownC)
-	<-errC
-}
-
-func TestWebsocketWrapper(t *testing.T) {
-
-	listener, err := hello.CreateTLSListener("localhost:0")
-	require.NoError(t, err)
-
-	serverErrorChan := make(chan error)
-	helloSvrCtx, cancelHelloSvr := context.WithCancel(context.Background())
-	defer func() { <-serverErrorChan }()
-	defer cancelHelloSvr()
-	go func() {
-		log := zerolog.Nop()
-		serverErrorChan <- hello.StartHelloWorldServer(&log, listener, helloSvrCtx.Done())
-	}()
-
-	tlsConfig := websocketClientTLSConfig(t)
-	d := gws.Dialer{TLSClientConfig: tlsConfig, HandshakeTimeout: time.Minute}
-	testAddr := fmt.Sprintf("https://%s/ws", listener.Addr().String())
-	req := testRequest(t, testAddr, nil)
-	conn, resp, err := ClientConnect(req, &d)
-	require.NoError(t, err)
-	require.Equal(t, testSecWebsocketAccept, resp.Header.Get("Sec-WebSocket-Accept"))
-
-	// Websocket now connected to test server so lets check our wrapper
-	wrapper := GorillaConn{Conn: conn}
-	buf := make([]byte, 100)
-	wrapper.Write([]byte("abc"))
-	n, err := wrapper.Read(buf)
-	require.NoError(t, err)
-	require.Equal(t, n, 3)
-	require.Equal(t, "abc", string(buf[:n]))
-
-	// Test partial read, read 1 of 3 bytes in one read and the other 2 in another read
-	wrapper.Write([]byte("abc"))
-	buf = buf[:1]
-	n, err = wrapper.Read(buf)
-	require.NoError(t, err)
-	require.Equal(t, n, 1)
-	require.Equal(t, "a", string(buf[:n]))
-	buf = buf[:cap(buf)]
-	n, err = wrapper.Read(buf)
-	require.NoError(t, err)
-	require.Equal(t, n, 2)
-	require.Equal(t, "bc", string(buf[:n]))
-}
-
-// func TestStartProxyServer(t *testing.T) {
-// 	var wg sync.WaitGroup
-// 	remoteAddress := "localhost:1113"
-// 	listenerAddress := "localhost:1112"
-// 	message := "Good morning Austin! Time for another sunny day in the great state of Texas."
-// 	logger := zerolog.Nop()
-// 	shutdownC := make(chan struct{})
-
-// 	listener, err := net.Listen("tcp", listenerAddress)
-// 	assert.NoError(t, err)
-// 	defer listener.Close()
-
-// 	remoteListener, err := net.Listen("tcp", remoteAddress)
-// 	assert.NoError(t, err)
-// 	defer remoteListener.Close()
-
-// 	wg.Add(1)
-// 	go func() {
-// 		defer wg.Done()
-// 		conn, err := remoteListener.Accept()
-// 		assert.NoError(t, err)
-// 		buf := make([]byte, len(message))
-// 		conn.Read(buf)
-// 		assert.Equal(t, string(buf), message)
-// 	}()
-
-// 	go func() {
-// 		StartProxyServer(logger, listener, remoteAddress, shutdownC)
-// 	}()
-
-// 	req := testRequest(t, fmt.Sprintf("http://%s/", listenerAddress), nil)
-// 	conn, _, err := ClientConnect(req, nil)
-// 	assert.NoError(t, err)
-// 	err = conn.WriteMessage(1, []byte(message))
-// 	assert.NoError(t, err)
-// 	wg.Wait()
-// }