TUN-3863: Consolidate header handling logic in the connection package; move headers definitions from h2mux to packages that manage them; cleanup header conversions

All header transformation code from h2mux has been consolidated in the connection package since it's used by both h2mux and http2 logic.
Exported headers used by proxying between edge and cloudflared so then can be shared by tunnel service on the edge.
Moved access-related headers to corresponding packages that have the code that sets/uses these headers.
Removed tunnel hostname tracking from h2mux since it wasn't used by anything. We will continue to set the tunnel hostname header from the edge for backward compatibilty, but it's no longer used by cloudflared.
Move bastion-related logic into carrier package, untangled dependencies between carrier, origin, and websocket packages.

Author: ipostelnik

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,13 @@
 package carrier
 
 import (
-	"fmt"
 	"io"
-	"net"
 	"net/http"
 	"net/http/httputil"
 
 	"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 +19,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 +36,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

cmd/cloudflared/access/carrier.go

@@ -6,19 +6,20 @@ import (
 	"net/http"
 	"strings"
 
+	"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/h2mux"
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/validation"
-
-	"github.com/pkg/errors"
-	"github.com/rs/zerolog"
-	"github.com/urfave/cli/v2"
 )
 
 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}
 

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)