Release 2021.7.0

This change extracts the need for EstablishConnection to know about a
request's entire context. It also removes the concern of populating the
http.Response from EstablishConnection's responsibilities.

RELEASE_NOTES

@@ -1,3 +1,10 @@
+2021.7.0
+- 2021-07-01 TUN-4626: Proxy non-stream based origin websockets with http Roundtrip.
+- 2021-07-01 TUN-4655: ingress.StreamBasedProxy.EstablishConnection takes dest input
+- 2021-07-09 TUN-4698: Add cloudflared metrics endpoint to serve quick tunnel hostname
+- 2021-06-21 TUN-4521: Modify cloudflared to use zoneless-tunnels-worker for free tunnels
+- 2021-04-05 AUTH-3475: Updated GetAppInfo error message
+
 2021.6.0
 - 2021-06-21 TUN-4571: Changelog for 2021.6.0
 - 2021-06-18 TUN-4571: Fix proxying to unix sockets when using HTTP2 transport to Cloudflare Edge

cmd/cloudflared/proxydns/cmd.go

@@ -73,7 +73,7 @@ func Run(c *cli.Context) error {
 		log.Fatal().Err(err).Msg("Failed to open the metrics listener")
 	}
 
-	go metrics.ServeMetrics(metricsListener, nil, nil, log)
+	go metrics.ServeMetrics(metricsListener, nil, nil, "", log)
 
 	listener, err := tunneldns.CreateListener(
 		c.String("address"),

cmd/cloudflared/tunnel/cmd.go

@@ -163,6 +163,12 @@ func TunnelCommand(c *cli.Context) error {
 		return fmt.Errorf("Use `cloudflared tunnel run` to start tunnel %s", ref)
 	}
 
+	// Unauthenticated named tunnel on <random>.<quick-tunnels-service>.com
+	// For now, default to legacy setup unless quick-service is specified
+	if c.String("hostname") == "" && c.String("quick-service") != "" {
+		return RunQuickTunnel(sc)
+	}
+
 	// Start a classic tunnel
 	return runClassicTunnel(sc)
 }
@@ -200,7 +206,7 @@ func runAdhocNamedTunnel(sc *subcommandContext, name, credentialsOutputPath stri
 
 // runClassicTunnel creates a "classic" non-named tunnel
 func runClassicTunnel(sc *subcommandContext) error {
-	return StartServer(sc.c, version, nil, sc.log, sc.isUIEnabled)
+	return StartServer(sc.c, version, nil, sc.log, sc.isUIEnabled, "")
 }
 
 func routeFromFlag(c *cli.Context) (route tunnelstore.Route, ok bool) {
@@ -219,6 +225,7 @@ func StartServer(
 	namedTunnel *connection.NamedTunnelConfig,
 	log *zerolog.Logger,
 	isUIEnabled bool,
+	quickTunnelHostname string,
 ) error {
 	_ = raven.SetDSN(sentryDSN)
 	var wg sync.WaitGroup
@@ -335,7 +342,7 @@ func StartServer(
 		defer wg.Done()
 		readinessServer := metrics.NewReadyServer(log)
 		observer.RegisterSink(readinessServer)
-		errC <- metrics.ServeMetrics(metricsListener, ctx.Done(), readinessServer, log)
+		errC <- metrics.ServeMetrics(metricsListener, ctx.Done(), readinessServer, quickTunnelHostname, log)
 	}()
 
 	if err := ingressRules.StartOrigins(&wg, log, ctx.Done(), errC); err != nil {
@@ -616,6 +623,11 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 			Value:  false,
 			Hidden: shouldHide,
 		}),
+		altsrc.NewStringFlag(&cli.StringFlag{
+			Name:   "quick-service",
+			Usage:  "URL for a service which manages unauthenticated 'quick' tunnels.",
+			Hidden: true,
+		}),
 		selectProtocolFlag,
 		overwriteDNSFlag,
 	}...)

cmd/cloudflared/tunnel/quick_tunnel.go

@@ -0,0 +1,88 @@
+package tunnel
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+
+	"github.com/cloudflare/cloudflared/connection"
+)
+
+const httpTimeout = 15 * time.Second
+
+// RunQuickTunnel requests a tunnel from the specified service.
+// We use this to power quick tunnels on trycloudflare.com, but the
+// service is open-source and could be used by anyone.
+func RunQuickTunnel(sc *subcommandContext) error {
+	sc.log.Info().Msg("Requesting new Quick Tunnel...")
+
+	client := http.Client{
+		Transport: &http.Transport{
+			TLSHandshakeTimeout:   httpTimeout,
+			ResponseHeaderTimeout: httpTimeout,
+		},
+		Timeout: httpTimeout,
+	}
+
+	resp, err := client.Post(fmt.Sprintf("%s/tunnel", sc.c.String("quick-service")), "application/json", nil)
+	if err != nil {
+		return errors.Wrap(err, "failed to request quick tunnel")
+	}
+	defer resp.Body.Close()
+
+	var data QuickTunnelResponse
+	if err := json.NewDecoder(resp.Body).Decode(&data); err != nil {
+		return errors.Wrap(err, "failed to unmarshal quick tunnel")
+	}
+
+	tunnelID, err := uuid.Parse(data.Result.ID)
+	if err != nil {
+		return errors.Wrap(err, "failed to parse quick tunnel ID")
+	}
+
+	credentials := connection.Credentials{
+		AccountTag:   data.Result.AccountTag,
+		TunnelSecret: data.Result.Secret,
+		TunnelID:     tunnelID,
+		TunnelName:   data.Result.Name,
+	}
+
+	for _, line := range connection.AsciiBox([]string{
+		"Your Quick Tunnel has been created! Visit it at:",
+		data.Result.Hostname,
+	}, 2) {
+		sc.log.Info().Msg(line)
+	}
+
+	return StartServer(
+		sc.c,
+		version,
+		&connection.NamedTunnelConfig{Credentials: credentials},
+		sc.log,
+		sc.isUIEnabled,
+		data.Result.Hostname,
+	)
+}
+
+type QuickTunnelResponse struct {
+	Success bool
+	Result  QuickTunnel
+	Errors  []QuickTunnelError
+}
+
+type QuickTunnelError struct {
+	Code    int
+	Message string
+}
+
+type QuickTunnel struct {
+	ID         string `json:"id"`
+	Name       string `json:"name"`
+	Hostname   string `json:"hostname"`
+	AccountTag string `json:"account_tag"`
+	Secret     []byte `json:"secret"`
+}

cmd/cloudflared/tunnel/subcommand_context.go

@@ -286,6 +286,7 @@ func (sc *subcommandContext) run(tunnelID uuid.UUID) error {
 		&connection.NamedTunnelConfig{Credentials: credentials},
 		sc.log,
 		sc.isUIEnabled,
+		"",
 	)
 }
 

connection/observer.go

@@ -58,7 +58,7 @@ func (o *Observer) logTrialHostname(registration *tunnelpogs.TunnelRegistration)
 	// Print out the user's trial zone URL in a nice box (if they requested and got one and UI flag is not set)
 	if !o.uiEnabled {
 		if registrationURL, err := url.Parse(registration.Url); err == nil {
-			for _, line := range asciiBox(trialZoneMsg(registrationURL.String()), 2) {
+			for _, line := range AsciiBox(TrialZoneMsg(registrationURL.String()), 2) {
 				o.log.Info().Msg(line)
 			}
 		} else {
@@ -70,7 +70,7 @@ func (o *Observer) logTrialHostname(registration *tunnelpogs.TunnelRegistration)
 }
 
 // Print out the given lines in a nice ASCII box.
-func asciiBox(lines []string, padding int) (box []string) {
+func AsciiBox(lines []string, padding int) (box []string) {
 	maxLen := maxLen(lines)
 	spacer := strings.Repeat(" ", padding)
 
@@ -94,7 +94,7 @@ func maxLen(lines []string) int {
 	return max
 }
 
-func trialZoneMsg(url string) []string {
+func TrialZoneMsg(url string) []string {
 	return []string{
 		"Your free tunnel has started! Visit it:",
 		"  " + url,

ingress/ingress.go

@@ -103,7 +103,7 @@ func NewWarpRoutingService() *WarpRoutingService {
 }
 
 // Get a single origin service from the CLI/config.
-func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (originService, error) {
+func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (OriginService, error) {
 	if c.IsSet("hello-world") {
 		return new(helloWorld), nil
 	}
@@ -167,7 +167,7 @@ func validate(ingress []config.UnvalidatedIngressRule, defaults OriginRequestCon
 	rules := make([]Rule, len(ingress))
 	for i, r := range ingress {
 		cfg := setConfig(defaults, r.OriginRequest)
-		var service originService
+		var service OriginService
 
 		if prefix := "unix:"; strings.HasPrefix(r.Service, prefix) {
 			// No validation necessary for unix socket filepath services

ingress/origin_connection_test.go

@@ -9,7 +9,6 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
-	"sync"
 	"testing"
 	"time"
 
@@ -190,71 +189,6 @@ func TestSocksStreamWSOverTCPConnection(t *testing.T) {
 	}
 }
 
-func TestStreamWSConnection(t *testing.T) {
-	eyeballConn, edgeConn := net.Pipe()
-
-	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)
-
-	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"))
-	require.Equal(t, "websocket", resp.Header.Get("Upgrade"))
-
-	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
-	})
-
-	require.NoError(t, errGroup.Wait())
-}
-
 type wsEyeball struct {
 	conn net.Conn
 }

ingress/origin_proxy.go

@@ -2,15 +2,10 @@ package ingress
 
 import (
 	"fmt"
-	"io"
 	"net"
 	"net/http"
-	"strings"
 
 	"github.com/pkg/errors"
-
-	"github.com/cloudflare/cloudflared/carrier"
-	"github.com/cloudflare/cloudflared/websocket"
 )
 
 var (
@@ -26,7 +21,7 @@ type HTTPOriginProxy interface {
 
 // StreamBasedOriginProxy can be implemented by origin services that want to proxy ws/TCP.
 type StreamBasedOriginProxy interface {
-	EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error)
+	EstablishConnection(dest string) (OriginConnection, error)
 }
 
 func (o *unixSocketPath) RoundTrip(req *http.Request) (*http.Response, error) {
@@ -36,7 +31,15 @@ func (o *unixSocketPath) RoundTrip(req *http.Request) (*http.Response, error) {
 func (o *httpService) RoundTrip(req *http.Request) (*http.Response, error) {
 	// Rewrite the request URL so that it goes to the origin service.
 	req.URL.Host = o.url.Host
-	req.URL.Scheme = o.url.Scheme
+	switch o.url.Scheme {
+	case "ws":
+		req.URL.Scheme = "http"
+	case "wss":
+		req.URL.Scheme = "https"
+	default:
+		req.URL.Scheme = o.url.Scheme
+	}
+
 	if o.hostHeader != "" {
 		// For incoming requests, the Host header is promoted to the Request.Host field and removed from the Header map.
 		req.Host = o.hostHeader
@@ -44,138 +47,40 @@ func (o *httpService) RoundTrip(req *http.Request) (*http.Response, error) {
 	return o.transport.RoundTrip(req)
 }
 
-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 = 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 o.newWebsocketProxyConnection(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")
-
-	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) {
 	return o.resp, nil
 }
 
-func (o *rawTCPService) EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error) {
-	dest, err := getRequestHost(r)
-	if err != nil {
-		return nil, nil, err
-	}
+func (o *rawTCPService) EstablishConnection(dest string) (OriginConnection, error) {
 	conn, err := net.Dial("tcp", dest)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	originConn := &tcpConnection{
 		conn: conn,
 	}
-	resp := &http.Response{
-		Status:        switchingProtocolText,
-		StatusCode:    http.StatusSwitchingProtocols,
-		ContentLength: -1,
-	}
-	return originConn, resp, nil
+	return originConn, nil
 }
 
-// getRequestHost returns the host of the http.Request.
-func getRequestHost(r *http.Request) (string, error) {
-	if r.Host != "" {
-		return r.Host, nil
-	}
-	if r.URL != nil {
-		return r.URL.Host, nil
-	}
-	return "", errors.New("host not found")
-}
-
-func (o *tcpOverWSService) EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error) {
+func (o *tcpOverWSService) EstablishConnection(dest string) (OriginConnection, error) {
 	var err error
-	dest := o.dest
-	if o.isBastion {
-		dest, err = carrier.ResolveBastionDest(r)
-		if err != nil {
-			return nil, nil, err
-		}
+	if !o.isBastion {
+		dest = o.dest
 	}
 
 	conn, err := net.Dial("tcp", dest)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	originConn := &tcpOverWSConnection{
 		conn:          conn,
 		streamHandler: o.streamHandler,
 	}
-	resp := &http.Response{
-		Status:        switchingProtocolText,
-		StatusCode:    http.StatusSwitchingProtocols,
-		Header:        websocket.NewResponseHeader(r),
-		ContentLength: -1,
-	}
-	return originConn, resp, nil
+	return originConn, nil
 
 }
 
-func (o *socksProxyOverWSService) EstablishConnection(r *http.Request) (OriginConnection, *http.Response, error) {
-	originConn := o.conn
-	resp := &http.Response{
-		Status:        switchingProtocolText,
-		StatusCode:    http.StatusSwitchingProtocols,
-		Header:        websocket.NewResponseHeader(r),
-		ContentLength: -1,
-	}
-	return originConn, resp, nil
+func (o *socksProxyOverWSService) EstablishConnection(dest string) (OriginConnection, error) {
+	return o.conn, nil
 }

ingress/origin_proxy_test.go

@@ -2,7 +2,6 @@ package ingress
 
 import (
 	"context"
-	"crypto/tls"
 	"fmt"
 	"net"
 	"net/http"
@@ -18,71 +17,6 @@ import (
 	"github.com/cloudflare/cloudflared/websocket"
 )
 
-// TestEstablishConnectionResponse ensures each implementation of StreamBasedOriginProxy returns
-// the expected response
-func assertEstablishConnectionResponse(t *testing.T,
-	originProxy StreamBasedOriginProxy,
-	req *http.Request,
-	expectHeader http.Header,
-) {
-	_, resp, err := originProxy.EstablishConnection(req)
-	assert.NoError(t, err)
-	assert.Equal(t, switchingProtocolText, resp.Status)
-	assert.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
-	assert.Equal(t, expectHeader, resp.Header)
-}
-
-func TestHTTPServiceEstablishConnection(t *testing.T) {
-	origin := echoWSOrigin(t, false)
-	defer origin.Close()
-	originURL, err := url.Parse(origin.URL)
-	require.NoError(t, err)
-
-	httpService := &httpService{
-		url:        originURL,
-		hostHeader: origin.URL,
-		transport: &http.Transport{
-			TLSClientConfig: &tls.Config{
-				InsecureSkipVerify: true,
-			},
-		},
-	}
-	req, err := http.NewRequest(http.MethodGet, origin.URL, nil)
-	require.NoError(t, err)
-	req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
-	req.Header.Set("Test-Cloudflared-Echo", t.Name())
-
-	expectHeader := http.Header{
-		"Connection":            {"Upgrade"},
-		"Sec-Websocket-Accept":  {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
-		"Upgrade":               {"websocket"},
-		"Test-Cloudflared-Echo": {t.Name()},
-	}
-	assertEstablishConnectionResponse(t, httpService, req, expectHeader)
-}
-
-func TestHelloWorldEstablishConnection(t *testing.T) {
-	var wg sync.WaitGroup
-	shutdownC := make(chan struct{})
-	errC := make(chan error)
-	helloWorldSerivce := &helloWorld{}
-	helloWorldSerivce.start(&wg, testLogger, shutdownC, errC, OriginRequestConfig{})
-
-	// Scheme and Host of URL will be override by the Scheme and Host of the helloWorld service
-	req, err := http.NewRequest(http.MethodGet, "https://place-holder/ws", nil)
-	require.NoError(t, err)
-	req.Header.Set("Sec-Websocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
-
-	expectHeader := http.Header{
-		"Connection":           {"Upgrade"},
-		"Sec-Websocket-Accept": {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
-		"Upgrade":              {"websocket"},
-	}
-	assertEstablishConnectionResponse(t, helloWorldSerivce, req, expectHeader)
-
-	close(shutdownC)
-}
-
 func TestRawTCPServiceEstablishConnection(t *testing.T) {
 	originListener, err := net.Listen("tcp", "127.0.0.1:0")
 	require.NoError(t, err)
@@ -95,8 +29,6 @@ func TestRawTCPServiceEstablishConnection(t *testing.T) {
 	req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("http://%s", originListener.Addr()), nil)
 	require.NoError(t, err)
 
-	assertEstablishConnectionResponse(t, rawTCPService, req, nil)
-
 	originListener.Close()
 	<-listenerClosed
 
@@ -104,9 +36,8 @@ func TestRawTCPServiceEstablishConnection(t *testing.T) {
 	require.NoError(t, err)
 
 	// Origin not listening for new connection, should return an error
-	_, resp, err := rawTCPService.EstablishConnection(req)
+	_, err = rawTCPService.EstablishConnection(req.URL.String())
 	require.Error(t, err)
-	require.Nil(t, resp)
 }
 
 func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
@@ -128,12 +59,6 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 	bastionReq := baseReq.Clone(context.Background())
 	carrier.SetBastionDest(bastionReq.Header, originListener.Addr().String())
 
-	expectHeader := http.Header{
-		"Connection":           {"Upgrade"},
-		"Sec-Websocket-Accept": {"s3pPLMBiTxaQ9kYGzzhZRbK+xOo="},
-		"Upgrade":              {"websocket"},
-	}
-
 	tests := []struct {
 		testCase  string
 		service   *tcpOverWSService
@@ -161,11 +86,9 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.testCase, func(t *testing.T) {
 			if test.expectErr {
-				_, resp, err := test.service.EstablishConnection(test.req)
+				bastionHost, _ := carrier.ResolveBastionDest(test.req)
+				_, err := test.service.EstablishConnection(bastionHost)
 				assert.Error(t, err)
-				assert.Nil(t, resp)
-			} else {
-				assertEstablishConnectionResponse(t, test.service, test.req, expectHeader)
 			}
 		})
 	}
@@ -175,9 +98,9 @@ func TestTCPOverWSServiceEstablishConnection(t *testing.T) {
 
 	for _, service := range []*tcpOverWSService{newTCPOverWSService(originURL), newBastionService()} {
 		// Origin not listening for new connection, should return an error
-		_, resp, err := service.EstablishConnection(bastionReq)
+		bastionHost, _ := carrier.ResolveBastionDest(bastionReq)
+		_, err := service.EstablishConnection(bastionHost)
 		assert.Error(t, err)
-		assert.Nil(t, resp)
 	}
 }
 
@@ -218,10 +141,6 @@ func TestHTTPServiceHostHeaderOverride(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, http.StatusOK, resp.StatusCode)
 
-	req = req.Clone(context.Background())
-	_, resp, err = httpService.EstablishConnection(req)
-	require.NoError(t, err)
-	require.Equal(t, http.StatusSwitchingProtocols, resp.StatusCode)
 }
 
 func tcpListenRoutine(listener net.Listener, closeChan chan struct{}) {

ingress/origin_service.go

@@ -20,8 +20,8 @@ import (
 	"github.com/cloudflare/cloudflared/tlsconfig"
 )
 
-// originService is something a tunnel can proxy traffic to.
-type originService interface {
+// OriginService is something a tunnel can proxy traffic to.
+type OriginService interface {
 	String() string
 	// Start the origin service if it's managed by cloudflared, e.g. proxy servers or Hello World.
 	// If it's not managed by cloudflared, this is a no-op because the user is responsible for
@@ -238,7 +238,7 @@ func (nrc *NopReadCloser) Close() error {
 	return nil
 }
 
-func newHTTPTransport(service originService, cfg OriginRequestConfig, log *zerolog.Logger) (*http.Transport, error) {
+func newHTTPTransport(service OriginService, cfg OriginRequestConfig, log *zerolog.Logger) (*http.Transport, error) {
 	originCertPool, err := tlsconfig.LoadOriginCA(cfg.CAPool, log)
 	if err != nil {
 		return nil, errors.Wrap(err, "Error loading cert pool")

ingress/rule.go

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

ingress/rule_test.go

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

metrics/metrics.go

@@ -22,7 +22,7 @@ const (
 	startupTime     = time.Millisecond * 500
 )
 
-func newMetricsHandler(readyServer *ReadyServer) *mux.Router {
+func newMetricsHandler(readyServer *ReadyServer, quickTunnelHostname string) *mux.Router {
 	router := mux.NewRouter()
 	router.PathPrefix("/debug/").Handler(http.DefaultServeMux)
 
@@ -33,6 +33,9 @@ func newMetricsHandler(readyServer *ReadyServer) *mux.Router {
 	if readyServer != nil {
 		router.Handle("/ready", readyServer)
 	}
+	router.HandleFunc("/quicktunnel", func(w http.ResponseWriter, r *http.Request) {
+		_, _ = fmt.Fprintf(w, `{"hostname":"%s"}`, quickTunnelHostname)
+	})
 
 	return router
 }
@@ -41,6 +44,7 @@ func ServeMetrics(
 	l net.Listener,
 	shutdownC <-chan struct{},
 	readyServer *ReadyServer,
+	quickTunnelHostname string,
 	log *zerolog.Logger,
 ) (err error) {
 	var wg sync.WaitGroup
@@ -48,7 +52,7 @@ func ServeMetrics(
 	trace.AuthRequest = func(*http.Request) (bool, bool) { return true, true }
 	// TODO: parameterize ReadTimeout and WriteTimeout. The maximum time we can
 	// profile CPU usage depends on WriteTimeout
-	h := newMetricsHandler(readyServer)
+	h := newMetricsHandler(readyServer, quickTunnelHostname)
 	server := &http.Server{
 		ReadTimeout:  10 * time.Second,
 		WriteTimeout: 10 * time.Second,

origin/proxy.go

@@ -12,9 +12,11 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 
+	"github.com/cloudflare/cloudflared/carrier"
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/ingress"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
+	"github.com/cloudflare/cloudflared/websocket"
 )
 
 const (
@@ -32,6 +34,8 @@ type proxy struct {
 	bufferPool   *bufferPool
 }
 
+var switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols, http.StatusText(http.StatusSwitchingProtocols))
+
 func NewOriginProxy(
 	ingressRules ingress.Ingress,
 	warpRouting *ingress.WarpRoutingService,
@@ -70,7 +74,13 @@ func (p *proxy) Proxy(w connection.ResponseWriter, req *http.Request, sourceConn
 			lbProbe: lbProbe,
 			rule:    ingress.ServiceWarpRouting,
 		}
-		if err := p.proxyStreamRequest(serveCtx, w, req, p.warpRouting.Proxy, logFields); err != nil {
+
+		host, err := getRequestHost(req)
+		if err != nil {
+			err = fmt.Errorf(`cloudflared recieved a warp-routing request with an empty host value: %v`, err)
+			return err
+		}
+		if err := p.proxyStreamRequest(serveCtx, w, host, req, p.warpRouting.Proxy, logFields); err != nil {
 			p.logRequestError(err, cfRay, "", ingress.ServiceWarpRouting)
 			return err
 		}
@@ -85,27 +95,50 @@ func (p *proxy) Proxy(w connection.ResponseWriter, req *http.Request, sourceConn
 	}
 	p.logRequest(req, logFields)
 
-	if sourceConnectionType == connection.TypeHTTP {
-		if err := p.proxyHTTPRequest(w, req, rule, logFields); err != nil {
+	switch originProxy := rule.Service.(type) {
+	case ingress.HTTPOriginProxy:
+		if err := p.proxyHTTPRequest(w, req, originProxy, sourceConnectionType == connection.TypeWebsocket,
+			rule.Config.DisableChunkedEncoding, logFields); err != nil {
 			rule, srv := ruleField(p.ingressRules, ruleNum)
 			p.logRequestError(err, cfRay, rule, srv)
 			return err
 		}
 		return nil
-	}
 
-	connectionProxy, ok := rule.Service.(ingress.StreamBasedOriginProxy)
-	if !ok {
-		p.log.Error().Msgf("%s is not a connection-oriented service", rule.Service)
-		return fmt.Errorf("Not a connection-oriented service")
+	case ingress.StreamBasedOriginProxy:
+		dest, err := getDestFromRule(rule, req)
+		if err != nil {
+			return err
+		}
+		if err := p.proxyStreamRequest(serveCtx, w, dest, req, originProxy, logFields); err != nil {
+			rule, srv := ruleField(p.ingressRules, ruleNum)
+			p.logRequestError(err, cfRay, rule, srv)
+			return err
+		}
+		return nil
+	default:
+		return fmt.Errorf("Unrecognized service: %s, %t", rule.Service, originProxy)
 	}
+}
 
-	if err := p.proxyStreamRequest(serveCtx, w, req, connectionProxy, logFields); err != nil {
-		rule, srv := ruleField(p.ingressRules, ruleNum)
-		p.logRequestError(err, cfRay, rule, srv)
-		return err
+func getDestFromRule(rule *ingress.Rule, req *http.Request) (string, error) {
+	switch rule.Service.String() {
+	case ingress.ServiceBastion:
+		return carrier.ResolveBastionDest(req)
+	default:
+		return rule.Service.String(), nil
 	}
-	return nil
+}
+
+// getRequestHost returns the host of the http.Request.
+func getRequestHost(r *http.Request) (string, error) {
+	if r.Host != "" {
+		return r.Host, nil
+	}
+	if r.URL != nil {
+		return r.URL.Host, nil
+	}
+	return "", errors.New("host not set in incoming request")
 }
 
 func ruleField(ing ingress.Ingress, ruleNum int) (ruleID string, srv string) {
@@ -116,26 +149,35 @@ func ruleField(ing ingress.Ingress, ruleNum int) (ruleID string, srv string) {
 	return fmt.Sprintf("%d", ruleNum), srv
 }
 
-func (p *proxy) proxyHTTPRequest(w connection.ResponseWriter, req *http.Request, rule *ingress.Rule, fields logFields) error {
-	// Support for WSGI Servers by switching transfer encoding from chunked to gzip/deflate
-	if rule.Config.DisableChunkedEncoding {
-		req.TransferEncoding = []string{"gzip", "deflate"}
-		cLength, err := strconv.Atoi(req.Header.Get("Content-Length"))
-		if err == nil {
-			req.ContentLength = int64(cLength)
+func (p *proxy) proxyHTTPRequest(
+	w connection.ResponseWriter,
+	req *http.Request,
+	httpService ingress.HTTPOriginProxy,
+	isWebsocket bool,
+	disableChunkedEncoding bool,
+	fields logFields) error {
+	roundTripReq := req
+	if isWebsocket {
+		roundTripReq = req.Clone(req.Context())
+		roundTripReq.Header.Set("Connection", "Upgrade")
+		roundTripReq.Header.Set("Upgrade", "websocket")
+		roundTripReq.Header.Set("Sec-Websocket-Version", "13")
+		roundTripReq.ContentLength = 0
+		roundTripReq.Body = nil
+	} else {
+		// Support for WSGI Servers by switching transfer encoding from chunked to gzip/deflate
+		if disableChunkedEncoding {
+			roundTripReq.TransferEncoding = []string{"gzip", "deflate"}
+			cLength, err := strconv.Atoi(req.Header.Get("Content-Length"))
+			if err == nil {
+				roundTripReq.ContentLength = int64(cLength)
+			}
 		}
+		// Request origin to keep connection alive to improve performance
+		roundTripReq.Header.Set("Connection", "keep-alive")
 	}
 
-	// Request origin to keep connection alive to improve performance
-	req.Header.Set("Connection", "keep-alive")
-
-	httpService, ok := rule.Service.(ingress.HTTPOriginProxy)
-	if !ok {
-		p.log.Error().Msgf("%s is not a http service", rule.Service)
-		return fmt.Errorf("Not a http service")
-	}
-
-	resp, err := httpService.RoundTrip(req)
+	resp, err := httpService.RoundTrip(roundTripReq)
 	if err != nil {
 		return errors.Wrap(err, "Unable to reach the origin service. The service may be down or it may not be responding to traffic from cloudflared")
 	}
@@ -145,6 +187,23 @@ func (p *proxy) proxyHTTPRequest(w connection.ResponseWriter, req *http.Request,
 	if err != nil {
 		return errors.Wrap(err, "Error writing response header")
 	}
+
+	if resp.StatusCode == http.StatusSwitchingProtocols {
+		rwc, ok := resp.Body.(io.ReadWriteCloser)
+		if !ok {
+			return errors.New("internal error: unsupported connection type")
+		}
+		defer rwc.Close()
+
+		eyeballStream := &bidirectionalStream{
+			writer: w,
+			reader: req.Body,
+		}
+
+		websocket.Stream(eyeballStream, rwc, p.log)
+		return nil
+	}
+
 	if connection.IsServerSentEvent(resp.Header) {
 		p.log.Debug().Msg("Detected Server-Side Events from Origin")
 		p.writeEventStream(w, resp.Body)
@@ -164,16 +223,24 @@ func (p *proxy) proxyHTTPRequest(w connection.ResponseWriter, req *http.Request,
 func (p *proxy) proxyStreamRequest(
 	serveCtx context.Context,
 	w connection.ResponseWriter,
+	dest string,
 	req *http.Request,
 	connectionProxy ingress.StreamBasedOriginProxy,
 	fields logFields,
 ) error {
-	originConn, resp, err := connectionProxy.EstablishConnection(req)
+	originConn, err := connectionProxy.EstablishConnection(dest)
 	if err != nil {
 		return err
 	}
-	if resp.Body != nil {
-		defer resp.Body.Close()
+
+	resp := &http.Response{
+		Status:        switchingProtocolText,
+		StatusCode:    http.StatusSwitchingProtocols,
+		ContentLength: -1,
+	}
+
+	if secWebsocketKey := req.Header.Get("Sec-WebSocket-Key"); secWebsocketKey != "" {
+		resp.Header = websocket.NewResponseHeader(req)
 	}
 
 	if err = w.WriteRespHeaders(resp.StatusCode, resp.Header); err != nil {

origin/proxy_test.go

@@ -571,8 +571,14 @@ func TestConnections(t *testing.T) {
 				},
 			},
 			want: want{
-				message: []byte{},
-				err:     true,
+				message: []byte("Forbidden\n"),
+				err:     false,
+				headers: map[string][]string{
+					"Content-Length":         {"10"},
+					"Content-Type":           {"text/plain; charset=utf-8"},
+					"Sec-Websocket-Version":  {"13"},
+					"X-Content-Type-Options": {"nosniff"},
+				},
 			},
 		},
 		{
@@ -806,6 +812,8 @@ func (w *wsRespWriter) WriteRespHeaders(status int, header http.Header) error {
 
 // respHeaders is a test function to read respHeaders
 func (w *wsRespWriter) headers() http.Header {
+	// Removing indeterminstic header because it cannot be asserted.
+	w.responseHeaders.Del("Date")
 	return w.responseHeaders
 }
 

token/token.go

@@ -278,7 +278,7 @@ func GetAppInfo(reqURL *url.URL) (*AppInfo, error) {
 		// 403/401 from the edge will have aud in a header
 		aud = audHeader
 	} else {
-		return nil, fmt.Errorf("failed to get Access app info for %s", reqURL.String())
+		return nil, fmt.Errorf("failed to find Access application at %s", reqURL.String())
 	}
 
 	domain := resp.Header.Get(appDomainHeader)