Release 2018.10.5

TUN-1158: Fix segfault when carrier test case fails

Makefile

@@ -16,9 +16,17 @@ ifeq ($(EQUINOX_IS_DRAFT), true)
 	EQUINOX_FLAGS := --draft $(EQUINOX_FLAGS)
 endif
 
+ifeq ($(GOARCH),)
+	GOARCH := amd64
+endif
+
 .PHONY: all
 all: cloudflared test
 
+.PHONY: clean
+clean:
+	go clean
+
 .PHONY: cloudflared
 cloudflared:
 	go build -v $(VERSION_FLAGS) $(IMPORT_PATH)/cmd/cloudflared
@@ -32,7 +40,7 @@ cloudflared-deb: cloudflared
 	mkdir -p $(PACKAGE_DIR)
 	cp cloudflared $(PACKAGE_DIR)/cloudflared
 	fakeroot fpm -C $(PACKAGE_DIR) -s dir -t deb --deb-compression bzip2 \
-		-a $(GOARCH) -v $(VERSION) -n cloudflared
+		-a $(GOARCH) -v $(VERSION) -n cloudflared cloudflared=/usr/local/bin/
 
 .PHONY: cloudflared-darwin-amd64.tgz
 cloudflared-darwin-amd64.tgz: cloudflared

RELEASE_NOTES

@@ -1,3 +1,15 @@
+2018.10.5
+- 2018-10-18 TUN-968: Flow control for large requests/responses
+- 2018-10-26 TUN-1158: Windows: use process arguments rather than trivial service arguments
+- 2018-10-20 #30: Fix the Content-Length header for HTTP2->HTTP1
+- 2018-10-29 TUN-1160: pass Host header during origin url validation
+
+2018.10.4
+- 2018-09-21 AUTH-1070: added SSH/protocol forwarding
+- 2018-10-19 AUTH-1235: fixed packaging of deb dev file
+- 2018-10-19 TUN-1097: Host missing from WebSocket request
+- 2018-10-19 AUTH-1188: UX Review and Changes for CLI SSH Access
+
 2018.10.3
 - 2018-10-08 TUN-1099: Bring back changes in 2018.10.1
 - 2018-10-08 TUN-1098: removed deprecation error

carrier/carrier.go

@@ -0,0 +1,131 @@
+//Package carrier provides a WebSocket proxy to carry or proxy a connection
+//from the local client to the edge. See it as a wrapper around any protocol
+//that it packages up in a WebSocket connection to the edge.
+package carrier
+
+import (
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"strings"
+
+	"github.com/cloudflare/cloudflared/cmd/cloudflared/token"
+	"github.com/cloudflare/cloudflared/websocket"
+	"github.com/sirupsen/logrus"
+)
+
+// StdinoutStream is empty struct for wrapping stdin/stdout
+// into a single ReadWriter
+type StdinoutStream struct {
+}
+
+// Read will read from Stdin
+func (c *StdinoutStream) Read(p []byte) (int, error) {
+	return os.Stdin.Read(p)
+
+}
+
+// Write will write to Stdout
+func (c *StdinoutStream) Write(p []byte) (int, error) {
+	return os.Stdout.Write(p)
+}
+
+// StartClient will copy the data from stdin/stdout over a WebSocket connection
+// to the edge (originURL)
+func StartClient(logger *logrus.Logger, originURL string, stream io.ReadWriter) error {
+	return serveStream(logger, originURL, stream)
+}
+
+// StartServer will setup a server on a specified port and copy data over a WebSocket connection
+// to the edge (originURL)
+func StartServer(logger *logrus.Logger, address, originURL string, shutdownC <-chan struct{}) error {
+	listener, err := net.Listen("tcp", address)
+	if err != nil {
+		logger.WithError(err).Error("failed to start forwarding server")
+		return err
+	}
+	defer listener.Close()
+	for {
+		select {
+		case <-shutdownC:
+			return nil
+		default:
+			conn, err := listener.Accept()
+			if err != nil {
+				return err
+			}
+			go serveConnection(logger, conn, originURL)
+		}
+	}
+}
+
+// serveConnection handles connections for the StartServer call
+func serveConnection(logger *logrus.Logger, c net.Conn, originURL string) {
+	defer c.Close()
+	serveStream(logger, originURL, c)
+}
+
+// serveStream will serve the data over the WebSocket stream
+func serveStream(logger *logrus.Logger, originURL string, conn io.ReadWriter) error {
+	wsConn, err := createWebsocketStream(originURL)
+	if err != nil {
+		logger.WithError(err).Error("failed to create websocket stream")
+		return err
+	}
+	defer wsConn.Close()
+
+	websocket.Stream(wsConn, conn)
+
+	return nil
+}
+
+// createWebsocketStream will create a WebSocket connection to stream data over
+// It also handles redirects from Access and will present that flow if
+// the token is not present on the request
+func createWebsocketStream(originURL string) (*websocket.Conn, error) {
+	req, err := http.NewRequest(http.MethodGet, originURL, nil)
+	if err != nil {
+		return nil, err
+	}
+	wsConn, resp, err := websocket.ClientConnect(req, nil)
+	if err != nil && resp != nil && resp.StatusCode > 300 {
+		location, err := resp.Location()
+		if err != nil {
+			return nil, err
+		}
+		if !strings.Contains(location.String(), "cdn-cgi/access/login") {
+			return nil, errors.New("not an Access redirect")
+		}
+		req, err := buildAccessRequest(originURL)
+		if err != nil {
+			return nil, err
+		}
+
+		wsConn, _, err = websocket.ClientConnect(req, nil)
+		if err != nil {
+			return nil, err
+		}
+	} else if err != nil {
+		return nil, err
+	}
+
+	return &websocket.Conn{Conn: wsConn}, nil
+}
+
+// buildAccessRequest builds an HTTP request with the Access token set
+func buildAccessRequest(originURL string) (*http.Request, error) {
+	req, err := http.NewRequest(http.MethodGet, originURL, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	token, err := token.FetchToken(req.URL)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("cf-access-token", token)
+
+	return req, nil
+}

carrier/carrier_test.go

@@ -0,0 +1,122 @@
+package carrier
+
+import (
+	"bytes"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+
+	ws "github.com/gorilla/websocket"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	// example in Sec-Websocket-Key in rfc6455
+	testSecWebsocketKey = "dGhlIHNhbXBsZSBub25jZQ=="
+)
+
+type testStreamer struct {
+	buf *bytes.Buffer
+	l   sync.RWMutex
+}
+
+func newTestStream() *testStreamer {
+	return &testStreamer{buf: new(bytes.Buffer)}
+}
+
+func (s *testStreamer) Read(p []byte) (int, error) {
+	s.l.RLock()
+	defer s.l.RUnlock()
+	return s.buf.Read(p)
+
+}
+
+func (s *testStreamer) Write(p []byte) (int, error) {
+	s.l.Lock()
+	defer s.l.Unlock()
+	return s.buf.Write(p)
+}
+
+func TestStartClient(t *testing.T) {
+	message := "Good morning Austin! Time for another sunny day in the great state of Texas."
+	logger := logrus.New()
+	ts := newTestWebSocketServer()
+	defer ts.Close()
+
+	buf := newTestStream()
+	err := StartClient(logger, "http://"+ts.Listener.Addr().String(), buf)
+	assert.NoError(t, err)
+	buf.Write([]byte(message))
+
+	readBuffer := make([]byte, len(message))
+	buf.Read(readBuffer)
+	assert.Equal(t, message, string(readBuffer))
+}
+
+func TestStartServer(t *testing.T) {
+	listenerAddress := "localhost:1117"
+	message := "Good morning Austin! Time for another sunny day in the great state of Texas."
+	logger := logrus.New()
+	shutdownC := make(chan struct{})
+	ts := newTestWebSocketServer()
+	defer ts.Close()
+
+	go func() {
+		err := StartServer(logger, listenerAddress, "http://"+ts.Listener.Addr().String(), shutdownC)
+		if err != nil {
+			t.Fatalf("Error starting server: %v", err)
+		}
+	}()
+
+	conn, err := net.Dial("tcp", listenerAddress)
+	if err != nil {
+		t.Fatalf("Error connecting to server: %v", err)
+	}
+	conn.Write([]byte(message))
+
+	readBuffer := make([]byte, len(message))
+	conn.Read(readBuffer)
+	assert.Equal(t, string(readBuffer), message)
+}
+
+func newTestWebSocketServer() *httptest.Server {
+	upgrader := ws.Upgrader{
+		ReadBufferSize:  1024,
+		WriteBufferSize: 1024,
+	}
+
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, _ := upgrader.Upgrade(w, r, nil)
+		defer conn.Close()
+		for {
+			mt, message, err := conn.ReadMessage()
+			if err != nil {
+				break
+			}
+
+			if err := conn.WriteMessage(mt, []byte(message)); err != nil {
+				break
+			}
+		}
+	}))
+}
+
+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
+}

cfsetup.yaml

@@ -1,6 +1,6 @@
 pinned_go: &pinned_go go=1.9.3-1
 build_dir: &build_dir /cfsetup_build/src/github.com/cloudflare/cloudflared/
-stretch:
+stretch: &stretch
   build:
     build_dir: *build_dir
     builddeps:
@@ -87,3 +87,5 @@ stretch:
       - export GOOS=linux
       - export GOARCH=amd64
       - make test
+
+jessie: *stretch

cmd/cloudflared/access/carrier.go

@@ -0,0 +1,38 @@
+package access
+
+import (
+	"net/url"
+
+	"github.com/cloudflare/cloudflared/carrier"
+	"github.com/cloudflare/cloudflared/cmd/cloudflared/config"
+	"github.com/cloudflare/cloudflared/validation"
+	"github.com/pkg/errors"
+	cli "gopkg.in/urfave/cli.v2"
+)
+
+// ssh will start a WS proxy server for server mode
+// or copy from stdin/stdout for client mode
+// useful for proxying other protocols (like ssh) over websockets
+// (which you can put Access in front of)
+func ssh(c *cli.Context) error {
+	hostname, err := validation.ValidateHostname(c.String("hostname"))
+	if err != nil || c.String("hostname") == "" {
+		return cli.ShowCommandHelp(c, "ssh")
+	}
+
+	if c.NArg() > 0 || c.IsSet("url") {
+		localForwarder, err := config.ValidateUrl(c)
+		if err != nil {
+			logger.WithError(err).Error("Error validating origin URL")
+			return errors.Wrap(err, "error validating origin URL")
+		}
+		forwarder, err := url.Parse(localForwarder)
+		if err != nil {
+			logger.WithError(err).Error("Error validating origin URL")
+			return errors.Wrap(err, "error validating origin URL")
+		}
+		return carrier.StartServer(logger, forwarder.Host, "https://"+hostname, shutdownC)
+	}
+
+	return carrier.StartClient(logger, "https://"+hostname, &carrier.StdinoutStream{})
+}

cmd/cloudflared/access/cmd.go

@@ -7,6 +7,7 @@ import (
 	"os"
 
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/shell"
+	"github.com/cloudflare/cloudflared/cmd/cloudflared/token"
 	"golang.org/x/net/idna"
 
 	"github.com/cloudflare/cloudflared/log"
@@ -16,6 +17,17 @@ import (
 
 const sentryDSN = "https://56a9c9fa5c364ab28f34b14f35ea0f1b@sentry.io/189878"
 
+var (
+	logger         = log.CreateLogger()
+	shutdownC      chan struct{}
+	graceShutdownC chan struct{}
+)
+
+// Init will initialize and store vars from the main program
+func Init(s, g chan struct{}) {
+	shutdownC, graceShutdownC = s, g
+}
+
 // Flags return the global flags for Access related commands (hopefully none)
 func Flags() []cli.Flag {
 	return []cli.Flag{} // no flags yet.
@@ -61,7 +73,7 @@ func Commands() []*cli.Command {
 				},
 				{
 					Name:        "token",
-					Action:      token,
+					Action:      generateToken,
 					Usage:       "token -app=<url of access application>",
 					ArgsUsage:   "url of Access application",
 					Description: `The token subcommand produces a JWT which can be used to authenticate requests.`,
@@ -71,6 +83,30 @@ func Commands() []*cli.Command {
 						},
 					},
 				},
+				{
+					Name:        "ssh",
+					Action:      ssh,
+					Usage:       "",
+					ArgsUsage:   "",
+					Description: `The ssh subcommand sends data over a proxy to the Cloudflare edge.`,
+					Hidden:      true,
+					Flags: []cli.Flag{
+						&cli.StringFlag{
+							Name: "hostname",
+						},
+						&cli.StringFlag{
+							Name:   "url",
+							Hidden: true,
+						},
+					},
+				},
+				{
+					Name:        "ssh-config",
+					Action:      sshConfig,
+					Usage:       "ssh-config",
+					Description: `Prints an example configuration ~/.ssh/config`,
+					Hidden:      true,
+				},
 			},
 		},
 	}
@@ -86,10 +122,12 @@ func login(c *cli.Context) error {
 		logger.Errorf("Please provide the url of the Access application\n")
 		return err
 	}
-	if _, err := fetchToken(c, appURL); err != nil {
+	token, err := token.FetchToken(appURL)
+	if err != nil {
 		logger.Errorf("Failed to fetch token: %s\n", err)
 		return err
 	}
+	fmt.Fprintf(os.Stdout, "Successfully fetched your token:\n\n%s\n\n", string(token))
 
 	return nil
 }
@@ -109,13 +147,13 @@ func curl(c *cli.Context) error {
 		return err
 	}
 
-	token, err := getTokenIfExists(appURL)
-	if err != nil || token == "" {
+	tok, err := token.GetTokenIfExists(appURL)
+	if err != nil || tok == "" {
 		if allowRequest {
 			logger.Warn("You don't have an Access token set. Please run access token <access application> to fetch one.")
 			return shell.Run("curl", cmdArgs...)
 		}
-		token, err = fetchToken(c, appURL)
+		tok, err = token.FetchToken(appURL)
 		if err != nil {
 			logger.Error("Failed to refresh token: ", err)
 			return err
@@ -123,31 +161,39 @@ func curl(c *cli.Context) error {
 	}
 
 	cmdArgs = append(cmdArgs, "-H")
-	cmdArgs = append(cmdArgs, fmt.Sprintf("cf-access-token: %s", token))
+	cmdArgs = append(cmdArgs, fmt.Sprintf("cf-access-token: %s", tok))
 	return shell.Run("curl", cmdArgs...)
 }
 
 // token dumps provided token to stdout
-func token(c *cli.Context) error {
+func generateToken(c *cli.Context) error {
 	raven.SetDSN(sentryDSN)
 	appURL, err := url.Parse(c.String("app"))
 	if err != nil || c.NumFlags() < 1 {
 		fmt.Fprintln(os.Stderr, "Please provide a url.")
 		return err
 	}
-	token, err := getTokenIfExists(appURL)
-	if err != nil || token == "" {
+	tok, err := token.GetTokenIfExists(appURL)
+	if err != nil || tok == "" {
 		fmt.Fprintln(os.Stderr, "Unable to find token for provided application. Please run token command to generate token.")
 		return err
 	}
 
-	if _, err := fmt.Fprint(os.Stdout, token); err != nil {
+	if _, err := fmt.Fprint(os.Stdout, tok); err != nil {
 		fmt.Fprintln(os.Stderr, "Failed to write token to stdout.")
 		return err
 	}
 	return nil
 }
 
+// sshConfig prints an example SSH config to stdout
+func sshConfig(c *cli.Context) error {
+	_, err := os.Stdout.Write([]byte(`Add this configuration block to your $HOME/.ssh/config
+		Host <your hostname>
+		    ProxyCommand cloudflared access ssh --hostname %h` + "\n"))
+	return err
+}
+
 // processURL will preprocess the string (parse to a url, convert to punycode, etc).
 func processURL(s string) (*url.URL, error) {
 	u, err := url.ParseRequestURI(s)

cmd/cloudflared/config/configuration.go

@@ -1,9 +1,11 @@
 package config
 
 import (
+	"errors"
 	"os"
 	"path/filepath"
 
+	"github.com/cloudflare/cloudflared/validation"
 	homedir "github.com/mitchellh/go-homedir"
 	"gopkg.in/urfave/cli.v2"
 	"gopkg.in/urfave/cli.v2/altsrc"
@@ -60,3 +62,16 @@ func FindDefaultConfigPath() string {
 	}
 	return ""
 }
+
+// ValidateUrl will validate url flag correctness. It can be either from --url or argument
+func ValidateUrl(c *cli.Context) (string, error) {
+	var url = c.String("url")
+	if c.NArg() > 0 {
+		if c.IsSet("url") {
+			return "", errors.New("Specified origin urls using both --url and argument. Decide which one you want, I can only support one.")
+		}
+		url = c.Args().Get(0)
+	}
+	validUrl, err := validation.ValidateUrl(url)
+	return validUrl, err
+}

cmd/cloudflared/main.go

@@ -55,6 +55,7 @@ func main() {
 	app.Commands = commands()
 
 	tunnel.Init(Version, shutdownC, graceShutdownC) // we need this to support the tunnel sub command...
+	access.Init(shutdownC, graceShutdownC)
 	runApp(app, shutdownC, graceShutdownC)
 }
 

cmd/cloudflared/token/token.go

@@ -1,4 +1,4 @@
-package access
+package token
 
 import (
 	"fmt"
@@ -15,15 +15,13 @@ import (
 	"github.com/coreos/go-oidc/jose"
 	"github.com/coreos/go-oidc/oidc"
 	homedir "github.com/mitchellh/go-homedir"
-	cli "gopkg.in/urfave/cli.v2"
 )
 
 var logger = log.CreateLogger()
 
-// fetchToken will either load a stored token or generate a new one
-func fetchToken(c *cli.Context, appURL *url.URL) (string, error) {
-	if token, err := getTokenIfExists(appURL); token != "" && err == nil {
-		fmt.Fprintf(os.Stdout, "You have an existing token:\n\n%s\n\n", token)
+// FetchToken will either load a stored token or generate a new one
+func FetchToken(appURL *url.URL) (string, error) {
+	if token, err := GetTokenIfExists(appURL); token != "" && err == nil {
 		return token, nil
 	}
 
@@ -36,17 +34,16 @@ func fetchToken(c *cli.Context, appURL *url.URL) (string, error) {
 	// we want to send to the transfer service. the key is token and the value
 	// is blank (basically just the id generated in the transfer service)
 	const resourceName, key, value = "token", "token", ""
-	token, err := transfer.Run(c, appURL, resourceName, key, value, path, true)
+	token, err := transfer.Run(appURL, resourceName, key, value, path, true)
 	if err != nil {
 		return "", err
 	}
 
-	fmt.Fprintf(os.Stdout, "Successfully fetched your token:\n\n%s\n\n", string(token))
 	return string(token), nil
 }
 
-// getTokenIfExists will return the token from local storage if it exists
-func getTokenIfExists(url *url.URL) (string, error) {
+// GetTokenIfExists will return the token from local storage if it exists
+func GetTokenIfExists(url *url.URL) (string, error) {
 	path, err := generateFilePathForTokenURL(url)
 	if err != nil {
 		return "", err

cmd/cloudflared/transfer/transfer.go

@@ -16,7 +16,6 @@ import (
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/encrypter"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/shell"
 	"github.com/cloudflare/cloudflared/log"
-	cli "gopkg.in/urfave/cli.v2"
 )
 
 const (
@@ -32,7 +31,7 @@ var logger = log.CreateLogger()
 // The "dance" we refer to is building a HTTP request, opening that in a browser waiting for
 // the user to complete an action, while it long polls in the background waiting for an
 // action to be completed to download the resource.
-func Run(c *cli.Context, transferURL *url.URL, resourceName, key, value, path string, shouldEncrypt bool) ([]byte, error) {
+func Run(transferURL *url.URL, resourceName, key, value, path string, shouldEncrypt bool) ([]byte, error) {
 	encrypterClient, err := encrypter.New("cloudflared_priv.pem", "cloudflared_pub.pem")
 	if err != nil {
 		return nil, err
@@ -49,16 +48,10 @@ func Run(c *cli.Context, transferURL *url.URL, resourceName, key, value, path st
 		fmt.Fprintf(os.Stdout, "A browser window should have opened at the following URL:\n\n%s\n\nIf the browser failed to open, open it yourself and visit the URL above.\n", requestURL)
 	}
 
-	// for local debugging
-	baseURL := baseStoreURL
-	if c.IsSet("url") {
-		baseURL = c.String("url")
-	}
-
 	var resourceData []byte
 
 	if shouldEncrypt {
-		buf, key, err := transferRequest(baseURL + filepath.Join("transfer", encrypterClient.PublicKey()))
+		buf, key, err := transferRequest(baseStoreURL + filepath.Join("transfer", encrypterClient.PublicKey()))
 		if err != nil {
 			return nil, err
 		}
@@ -74,7 +67,7 @@ func Run(c *cli.Context, transferURL *url.URL, resourceName, key, value, path st
 
 		resourceData = decrypted
 	} else {
-		buf, _, err := transferRequest(baseURL + filepath.Join(encrypterClient.PublicKey()))
+		buf, _, err := transferRequest(baseStoreURL + filepath.Join(encrypterClient.PublicKey()))
 		if err != nil {
 			return nil, err
 		}

cmd/cloudflared/tunnel/cmd.go

@@ -3,6 +3,8 @@ package tunnel
 import (
 	"fmt"
 	"io/ioutil"
+	"net"
+	"net/url"
 	"os"
 	"runtime/trace"
 	"sync"
@@ -19,6 +21,7 @@ import (
 	"github.com/cloudflare/cloudflared/metrics"
 	"github.com/cloudflare/cloudflared/origin"
 	"github.com/cloudflare/cloudflared/tunneldns"
+	"github.com/cloudflare/cloudflared/websocket"
 	"github.com/coreos/go-systemd/daemon"
 	"github.com/facebookgo/grace/gracenet"
 	"github.com/pkg/errors"
@@ -308,6 +311,24 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
 		c.Set("url", "https://"+helloListener.Addr().String())
 	}
 
+	if uri, _ := url.Parse(c.String("url")); uri.Scheme == "ssh" {
+		host := uri.Host
+		if uri.Port() == "" { // default to 22
+			host = uri.Hostname() + ":22"
+		}
+		listener, err := net.Listen("tcp", "127.0.0.1:")
+		if err != nil {
+			logger.WithError(err).Error("Cannot start Websocket Proxy Server")
+			return errors.Wrap(err, "Cannot start Websocket Proxy Server")
+		}
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			errC <- websocket.StartProxyServer(logger, listener, host, shutdownC)
+		}()
+		c.Set("url", "http://"+listener.Addr().String())
+	}
+
 	tunnelConfig, err := prepareTunnelConfig(c, buildInfo, version, logger, protoLogger)
 	if err != nil {
 		return err
@@ -324,19 +345,19 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
 
 func Before(c *cli.Context) error {
 	if c.String("config") == "" {
-		logger.Warnf("Cannot determine default configuration path. No file %v in %v", config.DefaultConfigFiles, config.DefaultConfigDirs)
+		logger.Debugf("Cannot determine default configuration path. No file %v in %v", config.DefaultConfigFiles, config.DefaultConfigDirs)
 	}
 	inputSource, err := config.FindInputSourceContext(c)
 	if err != nil {
-		logger.WithError(err).Infof("Cannot load configuration from %s", c.String("config"))
+		logger.WithError(err).Debugf("Cannot load configuration from %s", c.String("config"))
 		return err
 	} else if inputSource != nil {
 		err := altsrc.ApplyInputSourceValues(c, inputSource, c.App.Flags)
 		if err != nil {
-			logger.WithError(err).Infof("Cannot apply configuration from %s", c.String("config"))
+			logger.WithError(err).Debugf("Cannot apply configuration from %s", c.String("config"))
 			return err
 		}
-		logger.Infof("Applied configuration from %s", c.String("config"))
+		logger.Debugf("Applied configuration from %s", c.String("config"))
 	}
 	return nil
 }

cmd/cloudflared/tunnel/configuration.go

@@ -64,19 +64,6 @@ func handleDeprecatedOptions(c *cli.Context) error {
 	return nil
 }
 
-// validate url. It can be either from --url or argument
-func validateUrl(c *cli.Context) (string, error) {
-	var url = c.String("url")
-	if c.NArg() > 0 {
-		if c.IsSet("url") {
-			return "", errors.New("Specified origin urls using both --url and argument. Decide which one you want, I can only support one.")
-		}
-		url = c.Args().Get(0)
-	}
-	validUrl, err := validation.ValidateUrl(url)
-	return validUrl, err
-}
-
 func logClientOptions(c *cli.Context) {
 	flags := make(map[string]interface{})
 	for _, flag := range c.LocalFlagNames() {
@@ -168,7 +155,7 @@ func prepareTunnelConfig(c *cli.Context, buildInfo *origin.BuildInfo, version st
 
 	tags = append(tags, tunnelpogs.Tag{Name: "ID", Value: clientID})
 
-	originURL, err := validateUrl(c)
+	originURL, err := config.ValidateUrl(c)
 	if err != nil {
 		logger.WithError(err).Error("Error validating origin URL")
 		return nil, errors.Wrap(err, "Error validating origin URL")
@@ -205,7 +192,7 @@ func prepareTunnelConfig(c *cli.Context, buildInfo *origin.BuildInfo, version st
 		httpTransport.TLSClientConfig.ServerName = c.String("origin-server-name")
 	}
 
-	err = validation.ValidateHTTPService(originURL, httpTransport)
+	err = validation.ValidateHTTPService(originURL, hostname, httpTransport)
 	if err != nil {
 		logger.WithError(err).Error("unable to connect to the origin")
 		return nil, errors.Wrap(err, "unable to connect to the origin")

cmd/cloudflared/tunnel/login.go

@@ -33,7 +33,7 @@ func login(c *cli.Context) error {
 		return err
 	}
 
-	_, err = transfer.Run(c, loginURL, "cert", "callback", callbackStoreURL, path, false)
+	_, err = transfer.Run(loginURL, "cert", "callback", callbackStoreURL, path, false)
 	if err != nil {
 		fmt.Fprintf(os.Stderr, "Failed to write the certificate due to the following error:\n%v\n\nYour browser will download the certificate instead. You will have to manually\ncopy it to the following path:\n\n%s\n", err, path)
 		return err

cmd/cloudflared/windows_service.go

@@ -87,7 +87,20 @@ type windowsService struct {
 }
 
 // called by the package code at the start of the service
-func (s *windowsService) Execute(args []string, r <-chan svc.ChangeRequest, statusChan chan<- svc.Status) (ssec bool, errno uint32) {
+func (s *windowsService) Execute(serviceArgs []string, r <-chan svc.ChangeRequest, statusChan chan<- svc.Status) (ssec bool, errno uint32) {
+	// the arguments passed here are only meaningful if they were manually
+	// specified by the user, e.g. using the Services console or `sc start`.
+	// https://docs.microsoft.com/en-us/windows/desktop/services/service-entry-point
+	// https://stackoverflow.com/a/6235139
+	var args []string
+	if len(serviceArgs) > 1 {
+		args = serviceArgs
+	} else {
+		// fall back to the arguments from ImagePath (or, as sc calls it, binPath)
+		args = os.Args
+	}
+	s.elog.Info(1, fmt.Sprintf("%s service arguments: %v", windowsServiceName, args))
+
 	const cmdsAccepted = svc.AcceptStop | svc.AcceptShutdown
 	statusChan <- svc.Status{State: svc.StartPending}
 	errC := make(chan error)

h2mux/h2mux.go

@@ -15,11 +15,12 @@ import (
 )
 
 const (
-	defaultFrameSize  uint32        = 1 << 14 // Minimum frame size in http2 spec
-	defaultWindowSize uint32        = 65535
-	maxWindowSize     uint32        = (1 << 31) - 1 // 2^31-1 = 2147483647, max window size specified in http2 spec
-	defaultTimeout    time.Duration = 5 * time.Second
-	defaultRetries    uint64        = 5
+	defaultFrameSize         uint32        = 1 << 14       // Minimum frame size in http2 spec
+	defaultWindowSize        uint32        = (1 << 16) - 1 // Minimum window size in http2 spec
+	maxWindowSize            uint32        = (1 << 31) - 1 // 2^31-1 = 2147483647, max window size in http2 spec
+	defaultTimeout           time.Duration = 5 * time.Second
+	defaultRetries           uint64        = 5
+	defaultWriteBufferMaxLen int           = 1024 * 1024 * 512 // 500mb
 
 	SettingMuxerMagic http2.SettingID = 0x42db
 	MuxerMagicOrigin  uint32          = 0xa2e43c8b
@@ -49,6 +50,12 @@ type MuxerConfig struct {
 	// Logger to use
 	Logger             *log.Entry
 	CompressionQuality CompressionSetting
+	// Initial size for HTTP2 flow control windows
+	DefaultWindowSize uint32
+	// Largest allowable size for HTTP2 flow control windows
+	MaxWindowSize uint32
+	// Largest allowable capacity for the buffer of data to be sent
+	StreamWriteBufferMaxLen int
 }
 
 type Muxer struct {
@@ -98,6 +105,15 @@ func Handshake(
 	if config.Timeout == 0 {
 		config.Timeout = defaultTimeout
 	}
+	if config.DefaultWindowSize == 0 {
+		config.DefaultWindowSize = defaultWindowSize
+	}
+	if config.MaxWindowSize == 0 {
+		config.MaxWindowSize = maxWindowSize
+	}
+	if config.StreamWriteBufferMaxLen == 0 {
+		config.StreamWriteBufferMaxLen = defaultWriteBufferMaxLen
+	}
 	// Initialise connection state fields
 	m := &Muxer{
 		f:             http2.NewFramer(w, r), // A framer that writes to w and reads from r
@@ -179,8 +195,9 @@ func Handshake(
 		abortChan:               m.abortChan,
 		pingTimestamp:           pingTimestamp,
 		connActive:              connActive,
-		initialStreamWindow:     defaultWindowSize,
-		streamWindowMax:         maxWindowSize,
+		initialStreamWindow:     m.config.DefaultWindowSize,
+		streamWindowMax:         m.config.MaxWindowSize,
+		streamWriteBufferMaxLen: m.config.StreamWriteBufferMaxLen,
 		r:                       m.r,
 		updateRTTChan:           updateRTTChan,
 		updateReceiveWindowChan: updateReceiveWindowChan,
@@ -375,10 +392,12 @@ func (m *Muxer) OpenStream(headers []Header, body io.Reader) (*MuxedStream, erro
 		responseHeadersReceived: make(chan struct{}),
 		readBuffer:              NewSharedBuffer(),
 		writeBuffer:             &bytes.Buffer{},
-		receiveWindow:           defaultWindowSize,
-		receiveWindowCurrentMax: defaultWindowSize, // Initial window size limit. exponentially increase it when receiveWindow is exhausted
-		receiveWindowMax:        maxWindowSize,
-		sendWindow:              defaultWindowSize,
+		writeBufferMaxLen:       m.config.StreamWriteBufferMaxLen,
+		writeBufferHasSpace:     make(chan struct{}, 1),
+		receiveWindow:           m.config.DefaultWindowSize,
+		receiveWindowCurrentMax: m.config.DefaultWindowSize,
+		receiveWindowMax:        m.config.MaxWindowSize,
+		sendWindow:              m.config.DefaultWindowSize,
 		readyList:               m.readyList,
 		writeHeaders:            headers,
 		dictionaries:            m.muxReader.dictionaries,

h2mux/h2mux_test.go

@@ -39,20 +39,26 @@ func NewDefaultMuxerPair() *DefaultMuxerPair {
 	origin, edge := net.Pipe()
 	return &DefaultMuxerPair{
 		OriginMuxConfig: MuxerConfig{
-			Timeout: time.Second,
-			IsClient: true,
-			Name: "origin",
-			Logger: log.NewEntry(log.New()),
+			Timeout:                 time.Second,
+			IsClient:                true,
+			Name:                    "origin",
+			Logger:                  log.NewEntry(log.New()),
+			DefaultWindowSize:       (1 << 8) - 1,
+			MaxWindowSize:           (1 << 15) - 1,
+			StreamWriteBufferMaxLen: 1024,
 		},
-		OriginConn:      origin,
-		EdgeMuxConfig:   MuxerConfig{
-			Timeout: time.Second,
-			IsClient: false,
-			Name: "edge",
-			Logger: log.NewEntry(log.New()),
+		OriginConn: origin,
+		EdgeMuxConfig: MuxerConfig{
+			Timeout:                 time.Second,
+			IsClient:                false,
+			Name:                    "edge",
+			Logger:                  log.NewEntry(log.New()),
+			DefaultWindowSize:       (1 << 8) - 1,
+			MaxWindowSize:           (1 << 15) - 1,
+			StreamWriteBufferMaxLen: 1024,
 		},
-		EdgeConn:        edge,
-		doneC:           make(chan struct{}),
+		EdgeConn: edge,
+		doneC:    make(chan struct{}),
 	}
 }
 
@@ -60,22 +66,22 @@ func NewCompressedMuxerPair(quality CompressionSetting) *DefaultMuxerPair {
 	origin, edge := net.Pipe()
 	return &DefaultMuxerPair{
 		OriginMuxConfig: MuxerConfig{
-			Timeout: time.Second,
-			IsClient: true,
-			Name: "origin",
+			Timeout:            time.Second,
+			IsClient:           true,
+			Name:               "origin",
 			CompressionQuality: quality,
-			Logger: log.NewEntry(log.New()),
+			Logger:             log.NewEntry(log.New()),
 		},
-		OriginConn:      origin,
-		EdgeMuxConfig:   MuxerConfig{
-			Timeout: time.Second,
-			IsClient: false,
-			Name: "edge",
+		OriginConn: origin,
+		EdgeMuxConfig: MuxerConfig{
+			Timeout:            time.Second,
+			IsClient:           false,
+			Name:               "edge",
 			CompressionQuality: quality,
-			Logger: log.NewEntry(log.New()),
+			Logger:             log.NewEntry(log.New()),
 		},
-		EdgeConn:        edge,
-		doneC:           make(chan struct{}),
+		EdgeConn: edge,
+		doneC:    make(chan struct{}),
 	}
 }
 
@@ -230,7 +236,6 @@ func TestSingleStream(t *testing.T) {
 func TestSingleStreamLargeResponseBody(t *testing.T) {
 	muxPair := NewDefaultMuxerPair()
 	bodySize := 1 << 24
-	streamReady := make(chan struct{})
 	muxPair.OriginMuxConfig.Handler = MuxedStreamFunc(func(stream *MuxedStream) error {
 		if len(stream.Headers) != 1 {
 			t.Fatalf("expected %d headers, got %d", 1, len(stream.Headers))
@@ -257,8 +262,6 @@ func TestSingleStreamLargeResponseBody(t *testing.T) {
 		if n != len(payload) {
 			t.Fatalf("origin short write: %d/%d bytes", n, len(payload))
 		}
-		t.Log("Payload written; signaling that the stream is ready")
-		streamReady <- struct{}{}
 
 		return nil
 	})
@@ -282,9 +285,6 @@ func TestSingleStreamLargeResponseBody(t *testing.T) {
 	}
 	responseBody := make([]byte, bodySize)
 
-	<-streamReady
-	t.Log("Received stream ready signal; resuming the test")
-
 	n, err := io.ReadFull(stream, responseBody)
 	if err != nil {
 		t.Fatalf("error from (*MuxedStream).Read: %s", err)
@@ -367,14 +367,13 @@ func TestMultipleStreams(t *testing.T) {
 		log.Error(err)
 	}
 	if testFail {
-		t.Fatalf("TestMultipleStreamsFlowControl failed")
+		t.Fatalf("TestMultipleStreams failed")
 	}
 }
 
 func TestMultipleStreamsFlowControl(t *testing.T) {
 	maxStreams := 32
 	errorsC := make(chan error, maxStreams)
-	streamReady := make(chan struct{})
 	responseSizes := make([]int32, maxStreams)
 	for i := 0; i < maxStreams; i++ {
 		responseSizes[i] = rand.Int31n(int32(defaultWindowSize << 4))
@@ -398,7 +397,6 @@ func TestMultipleStreamsFlowControl(t *testing.T) {
 			payload[i] = byte(i % 256)
 		}
 		n, err := stream.Write(payload)
-		streamReady <- struct{}{}
 		if err != nil {
 			t.Fatalf("origin write error: %s", err)
 		}
@@ -435,7 +433,6 @@ func TestMultipleStreamsFlowControl(t *testing.T) {
 				return
 			}
 
-			<-streamReady
 			responseBody := make([]byte, responseSizes[(stream.streamID-2)/2])
 			n, err := io.ReadFull(stream, responseBody)
 			if err != nil {
@@ -782,9 +779,11 @@ func TestMultipleStreamsWithDictionaries(t *testing.T) {
 		}
 
 		wg.Add(len(paths))
+		errorsC := make(chan error, len(paths))
 
 		for i, s := range paths {
 			go func(i int, path string) {
+				defer wg.Done()
 				stream, err := muxPair.EdgeMux.OpenStream(
 					[]Header{
 						{Name: ":method", Value: "GET"},
@@ -805,22 +804,30 @@ func TestMultipleStreamsWithDictionaries(t *testing.T) {
 				responseBody := make([]byte, len(expectBody)*2)
 				n, err := stream.Read(responseBody)
 				if err != nil {
-					log.Printf("error from (*MuxedStream).Read: %s", err)
-					t.Fatalf("error from (*MuxedStream).Read: %s", err)
+					errorsC <- fmt.Errorf("stream %d error from (*MuxedStream).Read: %s", stream.streamID, err)
+					return
 				}
 				if n != len(expectBody) {
-					log.Printf("expected response body to have %d bytes, got %d", len(expectBody), n)
-					t.Fatalf("expected response body to have %d bytes, got %d", len(expectBody), n)
+					errorsC <- fmt.Errorf("stream %d expected response body to have %d bytes, got %d", stream.streamID, len(expectBody), n)
+					return
 				}
 				if string(responseBody[:n]) != expectBody {
-					log.Printf("expected response body %s, got %s", expectBody, responseBody[:n])
-					t.Fatalf("expected response body %s, got %s", expectBody, responseBody[:n])
+					errorsC <- fmt.Errorf("stream %d expected response body %s, got %s", stream.streamID, expectBody, responseBody[:n])
+					return
 				}
-				wg.Done()
 			}(i, s)
-			time.Sleep(1 * time.Millisecond)
 		}
+
 		wg.Wait()
+		close(errorsC)
+		testFail := false
+		for err := range errorsC {
+			testFail = true
+			log.Error(err)
+		}
+		if testFail {
+			t.Fatalf("TestMultipleStreams failed")
+		}
 
 		if q > CompressionNone && muxPair.OriginMux.muxMetricsUpdater.compBytesBefore.Value() <= 10*muxPair.OriginMux.muxMetricsUpdater.compBytesAfter.Value() {
 			t.Fatalf("Cross-stream compression is expected to give a better compression ratio")

h2mux/muxedstream.go

@@ -17,32 +17,51 @@ type ReadWriteClosedCloser interface {
 	Closed() bool
 }
 
+// MuxedStream is logically an HTTP/2 stream, with an additional buffer for outgoing data.
 type MuxedStream struct {
-	Headers []Header
-
 	streamID uint32
 
+	// The "Receive" end of the stream
+	readBufferLock sync.RWMutex
+	readBuffer     ReadWriteClosedCloser
+	// This is the amount of bytes that are in our receive window
+	// (how much data we can receive into this stream).
+	receiveWindow uint32
+	// current receive window size limit. Exponentially increase it when it's exhausted
+	receiveWindowCurrentMax uint32
+	// hard limit set in http2 spec. 2^31-1
+	receiveWindowMax uint32
+	// The desired size increment for receiveWindow.
+	// If this is nonzero, a WINDOW_UPDATE frame needs to be sent.
+	windowUpdate uint32
+	// The headers that were most recently received.
+	// Particularly:
+	//     * for an eyeball-initiated stream (as passed to TunnelHandler::ServeStream),
+	//       these are the request headers
+	//     * for a cloudflared-initiated stream (as created by Register/UnregisterTunnel),
+	//       these are the response headers.
+	// They are useful in both of these contexts; hence `Headers` is public.
+	Headers []Header
+	// For use in the context of a cloudflared-initiated stream.
 	responseHeadersReceived chan struct{}
 
-	readBuffer    ReadWriteClosedCloser
-	receiveWindow uint32
-	// current window size limit. Exponentially increase it when it's exhausted
-	receiveWindowCurrentMax uint32
-	// limit set in http2 spec. 2^31-1
-	receiveWindowMax uint32
-
-	// nonzero if a WINDOW_UPDATE frame for a stream needs to be sent
-	windowUpdate uint32
-
-	writeLock sync.Mutex
-	// The zero value for Buffer is an empty buffer ready to use.
+	// The "Send" end of the stream
+	writeLock   sync.Mutex
 	writeBuffer ReadWriteLengther
-
+	// The maximum capacity that the send buffer should grow to.
+	writeBufferMaxLen int
+	// A channel to be notified when the send buffer is not full.
+	writeBufferHasSpace chan struct{}
+	// This is the amount of bytes that are in the peer's receive window
+	// (how much data we can send from this stream).
 	sendWindow uint32
-
-	readyList    *ReadyList
+	// Reference to the muxer's readyList; signal this for stream data to be sent.
+	readyList *ReadyList
+	// The headers that should be sent, and a flag so we only send them once.
 	headersSent  bool
 	writeHeaders []Header
+
+	// EOF-related fields
 	// true if the write end of this stream has been closed
 	writeEOF bool
 	// true if we have sent EOF to the peer
@@ -50,40 +69,63 @@ type MuxedStream struct {
 	// true if the peer sent us an EOF
 	receivedEOF bool
 
-	// dictionary that was used to compress the stream
+	// Compression-related fields
 	receivedUseDict bool
 	method          string
 	contentType     string
 	path            string
 	dictionaries    h2Dictionaries
-	readBufferLock  sync.RWMutex
 }
 
 func (s *MuxedStream) Read(p []byte) (n int, err error) {
+	var readBuffer ReadWriteClosedCloser
 	if s.dictionaries.read != nil {
 		s.readBufferLock.RLock()
-		b := s.readBuffer
+		readBuffer = s.readBuffer
 		s.readBufferLock.RUnlock()
-		return b.Read(p)
+	} else {
+		readBuffer = s.readBuffer
 	}
-	return s.readBuffer.Read(p)
+	n, err = readBuffer.Read(p)
+	s.replenishReceiveWindow(uint32(n))
+	return
 }
 
-func (s *MuxedStream) Write(p []byte) (n int, err error) {
+// Blocks until len(p) bytes have been written to the buffer
+func (s *MuxedStream) Write(p []byte) (int, error) {
+	// If assignDictToStream returns success, then it will have acquired the
+	// writeLock. Otherwise we must acquire it ourselves.
 	ok := assignDictToStream(s, p)
 	if !ok {
 		s.writeLock.Lock()
 	}
 	defer s.writeLock.Unlock()
+
 	if s.writeEOF {
 		return 0, io.EOF
 	}
-	n, err = s.writeBuffer.Write(p)
-	if n != len(p) || err != nil {
-		return n, err
+	totalWritten := 0
+	for totalWritten < len(p) {
+		// If the buffer is full, block till there is more room.
+		// Use a loop to recheck the buffer size after the lock is reacquired.
+		for s.writeBufferMaxLen <= s.writeBuffer.Len() {
+			s.writeLock.Unlock()
+			<-s.writeBufferHasSpace
+			s.writeLock.Lock()
+		}
+		amountToWrite := len(p) - totalWritten
+		spaceAvailable := s.writeBufferMaxLen - s.writeBuffer.Len()
+		if spaceAvailable < amountToWrite {
+			amountToWrite = spaceAvailable
+		}
+		amountWritten, err := s.writeBuffer.Write(p[totalWritten : totalWritten+amountToWrite])
+		totalWritten += amountWritten
+		if err != nil {
+			return totalWritten, err
+		}
+		s.writeNotify()
 	}
-	s.writeNotify()
-	return n, nil
+	return totalWritten, nil
 }
 
 func (s *MuxedStream) Close() error {
@@ -164,9 +206,9 @@ func (s *MuxedStream) writeNotify() {
 // receive window (how much data we can send).
 func (s *MuxedStream) replenishSendWindow(bytes uint32) {
 	s.writeLock.Lock()
+	defer s.writeLock.Unlock()
 	s.sendWindow += bytes
 	s.writeNotify()
-	s.writeLock.Unlock()
 }
 
 // Call by muxreader when it receives a data frame
@@ -178,17 +220,30 @@ func (s *MuxedStream) consumeReceiveWindow(bytes uint32) bool {
 		return false
 	}
 	s.receiveWindow -= bytes
-	if s.receiveWindow < s.receiveWindowCurrentMax/2 {
+	if s.receiveWindow < s.receiveWindowCurrentMax/2 && s.receiveWindowCurrentMax < s.receiveWindowMax {
 		// exhausting client send window (how much data client can send)
-		if s.receiveWindowCurrentMax < s.receiveWindowMax {
-			s.receiveWindowCurrentMax <<= 1
+		// and there is room to grow the receive window
+		newMax := s.receiveWindowCurrentMax << 1
+		if newMax > s.receiveWindowMax {
+			newMax = s.receiveWindowMax
 		}
-		s.windowUpdate += s.receiveWindowCurrentMax - s.receiveWindow
+		s.windowUpdate += newMax - s.receiveWindowCurrentMax
+		s.receiveWindowCurrentMax = newMax
+		// notify MuxWriter to write WINDOW_UPDATE frame
 		s.writeNotify()
 	}
 	return true
 }
 
+// Arranges for the MuxWriter to send a WINDOW_UPDATE
+// Called by MuxedStream::Read when data has left the read buffer.
+func (s *MuxedStream) replenishReceiveWindow(bytes uint32) {
+	s.writeLock.Lock()
+	defer s.writeLock.Unlock()
+	s.windowUpdate += bytes
+	s.writeNotify()
+}
+
 // receiveEOF should be called when the peer indicates no more data will be sent.
 // Returns true if the socket is now closed (i.e. the write side is already closed).
 func (s *MuxedStream) receiveEOF() (closed bool) {
@@ -226,7 +281,8 @@ type streamChunk struct {
 	// true if a HEADERS frame should be sent
 	sendHeaders bool
 	headers     []Header
-	// nonzero if a WINDOW_UPDATE frame should be sent
+	// nonzero if a WINDOW_UPDATE frame should be sent;
+	// in that case, it is the increment value to use
 	windowUpdate uint32
 	// true if data frames should be sent
 	sendData bool
@@ -249,11 +305,23 @@ func (s *MuxedStream) getChunk() *streamChunk {
 		eof:          s.writeEOF && uint32(s.writeBuffer.Len()) <= s.sendWindow,
 	}
 
-	// Copies at most s.sendWindow bytes
+	// Copy at most s.sendWindow bytes, adjust the sendWindow accordingly
 	writeLen, _ := io.CopyN(&chunk.buffer, s.writeBuffer, int64(s.sendWindow))
 	s.sendWindow -= uint32(writeLen)
+
+	// Non-blocking channel send. This will allow MuxedStream::Write() to continue, if needed
+	if s.writeBuffer.Len() < s.writeBufferMaxLen {
+		select {
+		case s.writeBufferHasSpace <- struct{}{}:
+		default:
+		}
+	}
+
+	// When we write the chunk, we'll write the WINDOW_UPDATE frame if needed
 	s.receiveWindow += s.windowUpdate
 	s.windowUpdate = 0
+
+	// When we write the chunk, we'll write the headers if needed
 	s.headersSent = true
 
 	// if this chunk contains the end of the stream, close the stream now

h2mux/muxedstream_test.go

@@ -23,47 +23,55 @@ func TestFlowControlSingleStream(t *testing.T) {
 		sendWindow:              testWindowSize,
 		readyList:               NewReadyList(),
 	}
+	var tempWindowUpdate uint32
+	var tempStreamChunk *streamChunk
+
 	assert.True(t, stream.consumeReceiveWindow(testWindowSize/2))
 	dataSent := testWindowSize / 2
 	assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
 	assert.Equal(t, testWindowSize, stream.receiveWindowCurrentMax)
-	assert.Equal(t, uint32(0), stream.windowUpdate)
-	tempWindowUpdate := stream.windowUpdate
-
-	streamChunk := stream.getChunk()
-	assert.Equal(t, tempWindowUpdate, streamChunk.windowUpdate)
-	assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
-	assert.Equal(t, uint32(0), stream.windowUpdate)
 	assert.Equal(t, testWindowSize, stream.sendWindow)
+	assert.Equal(t, uint32(0), stream.windowUpdate)
+
+	tempStreamChunk = stream.getChunk()
+	assert.Equal(t, uint32(0), tempStreamChunk.windowUpdate)
+	assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
+	assert.Equal(t, testWindowSize, stream.receiveWindowCurrentMax)
+	assert.Equal(t, testWindowSize, stream.sendWindow)
+	assert.Equal(t, uint32(0), stream.windowUpdate)
 
 	assert.True(t, stream.consumeReceiveWindow(2))
 	dataSent += 2
 	assert.Equal(t, testWindowSize-dataSent, stream.receiveWindow)
 	assert.Equal(t, testWindowSize<<1, stream.receiveWindowCurrentMax)
-	assert.Equal(t, (testWindowSize<<1)-stream.receiveWindow, stream.windowUpdate)
+	assert.Equal(t, testWindowSize, stream.sendWindow)
+	assert.Equal(t, testWindowSize, stream.windowUpdate)
 	tempWindowUpdate = stream.windowUpdate
 
-	streamChunk = stream.getChunk()
-	assert.Equal(t, tempWindowUpdate, streamChunk.windowUpdate)
-	assert.Equal(t, testWindowSize<<1, stream.receiveWindow)
-	assert.Equal(t, uint32(0), stream.windowUpdate)
+	tempStreamChunk = stream.getChunk()
+	assert.Equal(t, tempWindowUpdate, tempStreamChunk.windowUpdate)
+	assert.Equal(t, (testWindowSize<<1)-dataSent, stream.receiveWindow)
+	assert.Equal(t, testWindowSize<<1, stream.receiveWindowCurrentMax)
 	assert.Equal(t, testWindowSize, stream.sendWindow)
+	assert.Equal(t, uint32(0), stream.windowUpdate)
 
 	assert.True(t, stream.consumeReceiveWindow(testWindowSize+10))
-	dataSent = testWindowSize + 10
+	dataSent += testWindowSize + 10
 	assert.Equal(t, (testWindowSize<<1)-dataSent, stream.receiveWindow)
 	assert.Equal(t, testWindowSize<<2, stream.receiveWindowCurrentMax)
-	assert.Equal(t, (testWindowSize<<2)-stream.receiveWindow, stream.windowUpdate)
+	assert.Equal(t, testWindowSize, stream.sendWindow)
+	assert.Equal(t, testWindowSize<<1, stream.windowUpdate)
 	tempWindowUpdate = stream.windowUpdate
 
-	streamChunk = stream.getChunk()
-	assert.Equal(t, tempWindowUpdate, streamChunk.windowUpdate)
-	assert.Equal(t, testWindowSize<<2, stream.receiveWindow)
-	assert.Equal(t, uint32(0), stream.windowUpdate)
+	tempStreamChunk = stream.getChunk()
+	assert.Equal(t, tempWindowUpdate, tempStreamChunk.windowUpdate)
+	assert.Equal(t, (testWindowSize<<2)-dataSent, stream.receiveWindow)
+	assert.Equal(t, testWindowSize<<2, stream.receiveWindowCurrentMax)
 	assert.Equal(t, testWindowSize, stream.sendWindow)
+	assert.Equal(t, uint32(0), stream.windowUpdate)
 
 	assert.False(t, stream.consumeReceiveWindow(testMaxWindowSize+1))
-	assert.Equal(t, testWindowSize<<2, stream.receiveWindow)
+	assert.Equal(t, (testWindowSize<<2)-dataSent, stream.receiveWindow)
 	assert.Equal(t, testMaxWindowSize, stream.receiveWindowCurrentMax)
 }
 

h2mux/muxreader.go

@@ -35,6 +35,8 @@ type MuxReader struct {
 	initialStreamWindow uint32
 	// The max value for the send window of a stream.
 	streamWindowMax uint32
+	// The max size for the write buffer of a stream
+	streamWriteBufferMaxLen int
 	// r is a reference to the underlying connection used when shutting down.
 	r io.Closer
 	// updateRTTChan is the channel to send new RTT measurement to muxerMetricsUpdater
@@ -153,6 +155,8 @@ func (r *MuxReader) newMuxedStream(streamID uint32) *MuxedStream {
 		streamID:                streamID,
 		readBuffer:              NewSharedBuffer(),
 		writeBuffer:             &bytes.Buffer{},
+		writeBufferMaxLen:       r.streamWriteBufferMaxLen,
+		writeBufferHasSpace:     make(chan struct{}, 1),
 		receiveWindow:           r.initialStreamWindow,
 		receiveWindowCurrentMax: r.initialStreamWindow,
 		receiveWindowMax:        r.streamWindowMax,

origin/tunnel.go

@@ -431,6 +431,12 @@ func H2RequestHeadersToH1Request(h2 []h2mux.Header, h1 *http.Request) error {
 				return fmt.Errorf("invalid path")
 			}
 			h1.URL = resolved
+		case "content-length":
+			contentLength, err := strconv.ParseInt(header.Value, 10, 64)
+			if err != nil {
+				return fmt.Errorf("unparseable content length")
+			}
+			h1.ContentLength = contentLength
 		default:
 			h1.Header.Add(http.CanonicalHeaderKey(header.Name), header.Value)
 		}

validation/validation.go

@@ -6,9 +6,10 @@ import (
 	"net/url"
 	"strings"
 
+	"net/http"
+
 	"github.com/pkg/errors"
 	"golang.org/x/net/idna"
-	"net/http"
 )
 
 const defaultScheme = "http"
@@ -137,7 +138,7 @@ func validateIP(scheme, host, port string) (string, error) {
 	return fmt.Sprintf("%s://%s", scheme, host), nil
 }
 
-func ValidateHTTPService(originURL string, transport http.RoundTripper) error {
+func ValidateHTTPService(originURL string, hostname string, transport http.RoundTripper) error {
 	parsedURL, err := url.Parse(originURL)
 	if err != nil {
 		return err
@@ -145,18 +146,27 @@ func ValidateHTTPService(originURL string, transport http.RoundTripper) error {
 
 	client := &http.Client{Transport: transport}
 
-	initialResponse, initialErr := client.Get(parsedURL.String())
+	initialRequest, err := http.NewRequest("GET", parsedURL.String(), nil)
+	if err != nil {
+		return err
+	}
+	initialRequest.Host = hostname
+	initialResponse, initialErr := client.Do(initialRequest)
 	if initialErr != nil || initialResponse.StatusCode != http.StatusOK {
 		// Attempt the same endpoint via the other protocol (http/https); maybe we have better luck?
 		oldScheme := parsedURL.Scheme
 		parsedURL.Scheme = toggleProtocol(parsedURL.Scheme)
 
-		secondResponse, _ := client.Get(parsedURL.String())
-
+		secondRequest, err := http.NewRequest("GET", parsedURL.String(), nil)
+		if err != nil {
+			return err
+		}
+		secondRequest.Host = hostname
+		secondResponse, _ := client.Do(secondRequest)
 		if secondResponse != nil && secondResponse.StatusCode == http.StatusOK { // Worked this time--advise the user to switch protocols
 			return errors.Errorf(
 				"%s doesn't seem to work over %s, but does seem to work over %s. Consider changing the origin URL to %s",
-				parsedURL.Hostname(),
+				parsedURL.Host,
 				oldScheme,
 				parsedURL.Scheme,
 				parsedURL,

validation/validation_test.go

@@ -7,12 +7,13 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
-	"github.com/stretchr/testify/assert"
 	"net"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"strings"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestValidateHostname(t *testing.T) {
@@ -151,57 +152,88 @@ func TestToggleProtocol(t *testing.T) {
 }
 
 func TestValidateHTTPService_HTTP2HTTP(t *testing.T) {
+	originURL := "http://127.0.0.1/"
+	hostname := "example.com"
 	server, client, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, hostname, r.Host)
 		w.WriteHeader(200)
 	}))
 	assert.NoError(t, err)
 	defer server.Close()
 
-	assert.Equal(t, nil, ValidateHTTPService("http://example.com/", client.Transport))
+	assert.Equal(t, nil, ValidateHTTPService(originURL, hostname, client.Transport))
 }
 
 func TestValidateHTTPService_ServerNonOKResponse(t *testing.T) {
+	originURL := "http://127.0.0.1/"
+	hostname := "example.com"
 	server, client, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == "CONNECT" {
+			assert.Equal(t, "127.0.0.1:443", r.Host)
+		} else {
+			assert.Equal(t, hostname, r.Host)
+		}
 		w.WriteHeader(400)
 	}))
 	assert.NoError(t, err)
 	defer server.Close()
 
-	assert.Equal(t, nil, ValidateHTTPService("http://example.com/", client.Transport))
+	assert.Equal(t, nil, ValidateHTTPService(originURL, hostname, client.Transport))
 }
 
 func TestValidateHTTPService_HTTPS2HTTP(t *testing.T) {
+	originURL := "https://127.0.0.1:1234/"
+	hostname := "example.com"
 	server, client, err := createMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == "CONNECT" {
+			assert.Equal(t, "127.0.0.1:1234", r.Host)
+		} else {
+			assert.Equal(t, hostname, r.Host)
+		}
 		w.WriteHeader(200)
 	}))
 	assert.NoError(t, err)
 	defer server.Close()
 
 	assert.Equal(t,
-		"example.com doesn't seem to work over https, but does seem to work over http. Consider changing the origin URL to http://example.com:1234/",
-		ValidateHTTPService("https://example.com:1234/", client.Transport).Error())
+		"127.0.0.1:1234 doesn't seem to work over https, but does seem to work over http. Consider changing the origin URL to http://127.0.0.1:1234/",
+		ValidateHTTPService(originURL, hostname, client.Transport).Error())
 }
 
 func TestValidateHTTPService_HTTPS2HTTPS(t *testing.T) {
+	originURL := "https://127.0.0.1/"
+	hostname := "example.com"
 	server, client, err := createSecureMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == "CONNECT" {
+			assert.Equal(t, "127.0.0.1:443", r.Host)
+		} else {
+			assert.Equal(t, hostname, r.Host)
+		}
 		w.WriteHeader(200)
 	}))
 	assert.NoError(t, err)
 	defer server.Close()
 
-	assert.Equal(t, nil, ValidateHTTPService("https://example.com/", client.Transport))
+	assert.Equal(t, nil, ValidateHTTPService(originURL, hostname, client.Transport))
 }
 
 func TestValidateHTTPService_HTTP2HTTPS(t *testing.T) {
+	originURL := "http://127.0.0.1:1234/"
+	hostname := "example.com"
 	server, client, err := createSecureMockServerAndClient(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Method == "CONNECT" {
+			assert.Equal(t, "127.0.0.1:1234", r.Host)
+		} else {
+			assert.Equal(t, hostname, r.Host)
+		}
 		w.WriteHeader(200)
 	}))
 	assert.NoError(t, err)
 	defer server.Close()
 
 	assert.Equal(t,
-		"example.com doesn't seem to work over http, but does seem to work over https. Consider changing the origin URL to https://example.com:1234/",
-		ValidateHTTPService("http://example.com:1234/", client.Transport).Error())
+		"127.0.0.1:1234 doesn't seem to work over http, but does seem to work over https. Consider changing the origin URL to https://127.0.0.1:1234/",
+		ValidateHTTPService(originURL, hostname, client.Transport).Error())
 }
 
 func createMockServerAndClient(handler http.Handler) (*httptest.Server, *http.Client, error) {

websocket/notice_page.go

@@ -0,0 +1,312 @@
+package websocket
+
+func nonWebSocketRequestPage() []byte {
+	return []byte(`<!DOCTYPE html>
+		<html lang="en">
+		
+		<head>
+			<meta charset="UTF-8">
+			<meta name="robots" content="noindex">
+			<meta name="viewport" content="initial-scale=1, maximum-scale=1, user-scalable=no, width=device-width">
+			<title>Cloudflare Access</title>
+		
+			<style>
+				body {
+					margin: 0;
+					padding: 0;
+					border: 0;
+					font-size: 16px;
+					line-height: 1.7em;
+					font-family: "Open Sans", sans-serif;
+					color: #424242;
+					background: #f3f1fe;
+				}
+		
+				.main-title {
+					font-size: 30px;
+					color: #fff;
+					padding: 40px;
+					background: rgb(129, 118, 181);
+					background: linear-gradient(72deg, rgba(129, 118, 181, 1) 0%, rgba(127, 120, 183, 1) 35%, rgba(119, 195, 224, 1) 100%);
+				}
+		
+				.main-content {
+					padding: 0;
+					margin: 0
+				}
+		
+				.section-1 {
+					display: flex;
+					flex-direction: row;
+					padding: 40px;
+					box-shadow: 0 4px 9px 1px rgba(129, 118, 181, 0.25);
+					background: #fff;
+					flex-wrap: wrap;
+		
+					justify-content: space-between;
+				}
+		
+				.logo-section {
+					padding: 20px;
+					display: flex;
+					justify-content: center;
+					padding-right: 40px;
+				}
+		
+				.logo-section > div {
+					margin: auto;
+				}
+		
+				.logo {
+					width: auto;
+					height: auto;
+					min-width: 250px;
+				}
+		
+				.section-1-content {
+					display: flex;
+					flex-direction: row;
+					flex-wrap: wrap;
+					margin: 40px;
+					justify-content: space-between;
+					flex-basis: calc(70% - 100px);
+					max-width: calc(70% - 100px);
+					border-right: 1px solid #d8d0ff;
+				}
+		
+				.section-1-content .main-message {
+					flex-basis: calc(49% - 40px);
+					max-width: calc(49% - 40px);
+					padding: 20px;
+					display: flex;
+					flex-direction: column;
+					justify-content: center;
+				}
+		
+				.section-1-content .debug-details {
+					padding: 20px;
+					display: flex;
+					flex-direction: column;
+					justify-content: center;
+					color: #7e7e7e;
+					flex-basis: calc(49% - 40px);
+					max-width: calc(49% - 40px);
+					flex-basis: calc(49% - 40px);
+					max-width: calc(49% - 40px);
+					align-items: center;
+					justify-content: center;
+					position: relative;
+				}
+		
+				.section-1-content .main-message .title {
+					font-size: 50px;
+					margin-bottom: 20px;
+				}
+		
+				.section-1-content .main-message .sub-title {
+					color: #8f8f8f;
+					font-size: 20px;
+				}
+		
+				.section-2 {
+					padding: 50px 100px;
+					color: rgb(71, 64, 106);
+					font-size: 18px;
+				}
+		
+				.section-2 .zd-link-message {
+					margin-top: 10px;
+				}
+		
+				.section-2 .cf-link {
+					margin-top: 30px;
+				}
+		
+				.section-2 .cf-link-message {
+					display: inline;
+				}
+		
+				.section-3 {
+					color: rgb(71, 64, 106);
+		
+					padding: 40px 80px 20px;
+				}
+		
+				footer {
+					border-top: 1px solid #d8d0ff;
+					padding-top: 20px;
+					text-align: center;
+				}
+		
+				a {
+					text-decoration: none;
+					color: #2400cf;
+				}
+		
+				a:visited {
+					color: #3f2ba2;
+				}
+		
+				.Message-is-warning {
+					color: #cc8400;
+				}
+		
+				.Message-is-success {
+					color: #028402;
+				}
+		
+				.appName {
+					text-transform: capitalize;
+				}
+		
+				.org-logo {
+					width: auto;
+					height: auto;
+					max-width: 155px;
+				}
+		
+				.watermark-logo::after {
+					opacity: 0.05;
+					height: 100%;
+					width: 70%;
+					background-repeat: no-repeat;
+					background-position: center;
+					content: "#";
+					z-index: 1;
+					position: absolute;
+				}
+		
+				.watermark-logo .debug-text {
+					display: flex;
+					flex-direction: column;
+					justify-items: center;
+					z-index: 10;
+				}
+		
+				.watermark-logo p, .watermark-logo div {
+					opacity: 1;
+				}
+		
+				.app-title {
+					color: #5f5f5f;
+				}
+		
+				@media only screen and (max-width: 1046px) {
+					.section-1-content .main-message {
+						flex-basis: 100%;
+						max-width: 100%;
+						padding: 20px;
+					}
+		
+					.section-1-content .debug-details {
+						flex-basis: 100%;
+						max-width: 100%;
+						padding: 20px;
+						margin-top: 50px;
+					}
+		
+					.logo {
+						min-width: 200px;
+					}
+				}
+		
+				@media only screen and (max-width: 890px) {
+					.section-1 {
+						flex-wrap: wrap-reverse;
+						padding: 5px;
+					}
+		
+					.section-1-content {
+						padding: 5px;
+					}
+		
+					.section-2 {
+						padding: 50px 30px;
+					}
+		
+					.logo {
+						min-width: 150px;
+					}
+		
+					.logo-section {
+						flex-basis: 100%;
+						max-width: 100%;
+					}
+		
+					.section-1-content {
+						border-right: 0;
+						flex-basis: 100%;
+						max-width: 100%;
+					}
+				}
+			</style>
+		</head>
+		
+		
+		<body>
+		<header class="main-title">
+				Cloudflare Access
+		</header>
+		<div class="main-content">
+			<div class="section-1">
+				<div class="section-1-content">
+					<div class="main-message">
+						<div class="title">  Success </div>
+						<div class="sub-title">
+						You are now logged in and can reach this application over SSH from your command line. 
+							You can close this browser window.
+						</div>
+					</div>
+				</div>
+		
+				<div class="logo-section">
+					<div>
+		
+						<svg class="logo" width="250" viewBox="0 0 122 53" xmlns="http://www.w3.org/2000/svg">
+							<rect class="" width="100%" height="100%" fill="none" style=""/>
+							<defs>
+								<style>.cls-5 {
+									fill: #fff
+								}
+		
+								.st0 {
+									fill: #8176b5
+								}</style>
+							</defs>
+							<g class="currentLayer" style="">
+								<path class="" d="m113.65 12.721l-6.72-1.56-1.2-0.48-30.843 0.24v14.882l38.763 0.12z"
+									  fill="#fff"/>
+								<path class=""
+									  d="m101.05 24.482c0.36-1.2 0.24-2.4-0.36-3.12s-1.44-1.2-2.52-1.32l-20.882-0.24c-0.12 0-0.24-0.12-0.36-0.12-0.12-0.12-0.12-0.24 0-0.36 0.12-0.24 0.24-0.36 0.48-0.36l21.002-0.24c2.52-0.12 5.16-2.16 6.12-4.56l1.2-3.12c0-0.12 0.12-0.24 0-0.36-1.32-6.121-6.84-10.682-13.32-10.682-6.001 0-11.162 3.84-12.962 9.241-1.2-0.84-2.64-1.32-4.32-1.2-2.88 0.24-5.16 2.64-5.52 5.52-0.12 0.72 0 1.44 0.12 2.16-4.681 0.12-8.521 3.961-8.521 8.761 0 0.48 0 0.84 0.12 1.32 0 0.24 0.24 0.36 0.36 0.36h38.523c0.24 0 0.48-0.12 0.48-0.36l0.36-1.32z"
+									  fill="#f48120"/>
+								<path class=""
+									  d="m107.65 11.041h-0.6c-0.12 0-0.24 0.12-0.36 0.24l-0.84 2.88c-0.36 1.2-0.24 2.4 0.36 3.12s1.44 1.2 2.52 1.32l4.44 0.24c0.12 0 0.24 0.12 0.36 0.12 0.12 0.12 0.12 0.241 0 0.361-0.12 0.24-0.24 0.36-0.48 0.36l-4.56 0.24c-2.52 0.12-5.16 2.16-6.12 4.56l-0.24 1.08c-0.12 0.12 0 0.36 0.24 0.36h15.84c0.24 0 0.36-0.12 0.36-0.36 0.24-0.96 0.48-2.04 0.48-3.12 0-6.24-5.16-11.4-11.4-11.4"
+									  fill="#faad3f"/>
+								<path class="st0"
+									  d="m120.61 32.643c-0.6 0-1.08-0.48-1.08-1.08s0.48-1.08 1.08-1.08 1.08 0.48 1.08 1.08-0.48 1.08-1.08 1.08m0-1.92c-0.48 0-0.84 0.36-0.84 0.84s0.36 0.84 0.84 0.84 0.84-0.36 0.84-0.84-0.36-0.84-0.84-0.84m0.48 1.44h-0.24l-0.24-0.36h-0.24v0.36h-0.24v-1.08h0.6c0.24 0 0.36 0.12 0.36 0.36 0 0.12-0.12 0.24-0.24 0.36l0.24 0.36zm-0.36-0.6c0.12 0 0.12 0 0.12-0.12s-0.12-0.12-0.12-0.12h-0.36v0.36h0.36zm-107.65-1.08h2.64v7.2h4.562v2.28h-7.2zm9.962 4.68c0-2.76 2.16-4.92 5.16-4.92s5.04 2.16 5.04 4.92-2.16 4.92-5.16 4.92c-2.88 0-5.04-2.16-5.04-4.92m7.56 0c0-1.44-0.96-2.64-2.4-2.64s-2.4 1.2-2.4 2.52 0.96 2.52 2.4 2.52c1.44 0.24 2.4-0.96 2.4-2.4m5.88 0.6v-5.28h2.64v5.28c0 1.32 0.72 2.04 1.801 2.04s1.8-0.6 1.8-1.92v-5.4h2.64v5.28c0 3.12-1.8 4.44-4.44 4.44-2.76-0.12-4.44-1.44-4.44-4.44m12.84-5.28h3.721c3.36 0 5.4 1.92 5.4 4.68s-2.04 4.8-5.4 4.8h-3.6v-9.48zm3.721 7.08c1.56 0 2.64-0.84 2.64-2.4s-1.08-2.4-2.64-2.4h-1.08v4.8h1.08zm9.12-7.08h7.561v2.28h-4.92v1.56h4.44v2.16h-4.44v3.48h-2.64zm11.282 0h2.64v7.2h4.56v2.28h-7.2zm14.04-0.12h2.641l4.08 9.6h-2.88l-0.72-1.68h-3.72l-0.72 1.68h-2.76l4.08-9.6zm2.401 5.88l-1.08-2.64-1.08 2.64h2.16zm7.68-5.76h4.441c1.44 0 2.4 0.36 3.12 1.08 0.6 0.6 0.84 1.32 0.84 2.16 0 1.44-0.72 2.4-1.92 2.88l2.28 3.36h-3l-1.92-2.88h-1.2v2.88h-2.64v-9.48zm4.321 4.56c0.84 0 1.44-0.48 1.44-1.08 0-0.72-0.6-1.08-1.44-1.08h-1.68v2.28h1.68zm7.8-4.56h7.681v2.16h-5.04v1.44h4.56v2.16h-4.56v1.44h5.16v2.28h-7.8zm-102.37 5.88a2.37 2.37 0 0 1 -2.16 1.44c-1.44 0-2.4-1.2-2.4-2.52s0.96-2.52 2.4-2.52c1.08 0 1.92 0.72 2.28 1.56h2.76c-0.48-2.28-2.4-3.96-5.04-3.96-2.88 0-5.16 2.16-5.16 4.92s2.16 4.92 5.04 4.92c2.52 0 4.44-1.68 5.04-3.84h-2.76z"
+									  fill="#8176b5"/>
+								<path class="st0"
+									  d="m53.092 43.286h2.614l4.04 9.68h-2.852l-0.713-1.695h-3.683l-0.713 1.694h-2.733l4.04-9.68zm2.376 5.928h-2.138l1.07-2.662 1.069 2.662zm58.031-5.928c1.71 0 3.8 0.783 3.8 2.966h-2.93c0-1.281-2.371-0.985-2.123 0 0.223 0.886 1.533 0.783 2.22 0.914 2.422 0.46 3.12 1.804 3.12 2.765 0 1.353-0.64 3.216-4.087 3.216-1.854 0-4.393-0.546-4.387-3.086h2.99c0 1.223 2.494 1.332 2.494 0.13 0-0.664-1.166-1.085-2.35-1.245-0.875-0.119-2.912-0.73-2.912-2.634 0-1.048 0.457-3.026 4.165-3.026zm-11.504 0c1.71 0 3.8 0.783 3.8 2.966h-2.931c0-1.281-2.37-0.985-2.123 0 0.223 0.886 1.534 0.783 2.22 0.914 2.422 0.46 3.12 1.804 3.12 2.765 0 1.353-0.639 3.216-4.086 3.216-1.854 0-4.394-0.546-4.387-3.086h2.99c0 1.223 2.494 1.332 2.494 0.13 0-0.664-1.167-1.085-2.35-1.245-0.876-0.119-2.913-0.73-2.913-2.634 0-1.048 0.458-3.026 4.166-3.026zm-15.05 0.202h7.406v2.225h-4.877v1.433h4.417v2.073h-4.417v1.5h4.942v2.226h-7.47v-9.457zm-18.326 5.702l2.692 0.016c-0.492 2.158-2.397 3.776-4.827 3.776-2.84 0-4.942-2.174-4.942-4.888v-0.034c0-2.714 2.135-4.922 4.975-4.922 2.496 0 4.433 1.669 4.86 3.928h-2.693c-0.328-0.91-1.133-1.568-2.183-1.568-1.396 0-2.332 1.163-2.332 2.529v0.033c0 1.349 0.952 2.546 2.348 2.546 0.985 0 1.74-0.59 2.102-1.416zm12.326 0l2.693 0.016c-0.493 2.158-2.397 3.776-4.828 3.776-2.84 0-4.942-2.174-4.942-4.888v-0.034c0-2.714 2.135-4.922 4.975-4.922 2.496 0 4.433 1.669 4.86 3.928h-2.691c-0.329-0.91-1.133-1.568-2.184-1.568-1.396 0-2.332 1.163-2.332 2.529v0.033c0 1.349 0.953 2.546 2.348 2.546 0.985 0 1.74-0.59 2.102-1.416z"
+									  fill="#8176b5" fill-rule="evenodd"/>
+							</g>
+						</svg>
+		
+					</div>
+				</div>
+			</div>
+			<div class="section-2">
+			</div>
+			<div class="section-3 ">
+				<footer>
+					<a href="https://support.cloudflare.com/hc/en-us" target="_blank"> Help </a>
+					•
+					<span>Performance &amp; Security by
+				  <a href="https://www.cloudflare.com/products/cloudflare-access/" target="_blank">Cloudflare Access</a>
+				</span>
+				</footer>
+			</div>
+		</div>
+		</body>
+		
+		</html>`)
+}

websocket/websocket.go

@@ -9,11 +9,24 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"time"
 
 	"github.com/gorilla/websocket"
+	"github.com/sirupsen/logrus"
 )
 
-var stripWebsocketHeaders = []string {
+const (
+	// Time allowed to write a message to the peer.
+	writeWait = 10 * time.Second
+
+	// Time allowed to read the next pong message from the peer.
+	pongWait = 60 * time.Second
+
+	// Send pings to peer with this period. Must be less than pongWait.
+	pingPeriod = (pongWait * 9) / 10
+)
+
+var stripWebsocketHeaders = []string{
 	"Upgrade",
 	"Connection",
 	"Sec-Websocket-Key",
@@ -21,6 +34,32 @@ var stripWebsocketHeaders = []string {
 	"Sec-Websocket-Extensions",
 }
 
+// Conn is a wrapper around the standard gorilla websocket
+// but implements a ReadWriter
+type Conn struct {
+	*websocket.Conn
+}
+
+// Read will read messages from the websocket connection
+func (c *Conn) Read(p []byte) (int, error) {
+	_, message, err := c.Conn.ReadMessage()
+	if err != nil {
+		return 0, err
+	}
+
+	return copy(p, message), nil
+
+}
+
+// Write will write messages to the websocket connection
+func (c *Conn) Write(p []byte) (int, error) {
+	if err := c.Conn.WriteMessage(websocket.BinaryMessage, p); err != nil {
+		return 0, err
+	}
+
+	return len(p), nil
+}
+
 // IsWebSocketUpgrade checks to see if the request is a WebSocket connection.
 func IsWebSocketUpgrade(req *http.Request) bool {
 	return websocket.IsWebSocketUpgrade(req)
@@ -36,7 +75,7 @@ func ClientConnect(req *http.Request, tlsClientConfig *tls.Config) (*websocket.C
 	d := &websocket.Dialer{TLSClientConfig: tlsClientConfig}
 	conn, response, err := d.Dial(req.URL.String(), wsHeaders)
 	if err != nil {
-		return nil, nil, err
+		return nil, response, err
 	}
 	response.Header.Set("Sec-WebSocket-Accept", generateAcceptKey(req))
 	return conn, response, err
@@ -74,18 +113,65 @@ func Stream(conn, backendConn io.ReadWriter) {
 	<-proxyDone
 }
 
+// StartProxyServer will start a websocket server that will decode
+// the websocket data and write the resulting data to the provided
+// address
+func StartProxyServer(logger *logrus.Logger, listener net.Listener, remote string, shutdownC <-chan struct{}) error {
+	upgrader := websocket.Upgrader{
+		ReadBufferSize:  1024,
+		WriteBufferSize: 1024,
+	}
+
+	httpServer := &http.Server{Addr: listener.Addr().String(), Handler: nil}
+	go func() {
+		<-shutdownC
+		httpServer.Close()
+	}()
+
+	http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		stream, err := net.Dial("tcp", remote)
+		if err != nil {
+			logger.WithError(err).Error("Cannot connect to remote.")
+			return
+		}
+		defer stream.Close()
+
+		if !websocket.IsWebSocketUpgrade(r) {
+			w.Write(nonWebSocketRequestPage())
+			return
+		}
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			logger.WithError(err).Error("failed to upgrade")
+			return
+		}
+		conn.SetReadDeadline(time.Now().Add(pongWait))
+		conn.SetPongHandler(func(string) error { conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
+		done := make(chan struct{})
+		go pinger(logger, conn, done)
+		defer func() {
+			<-done
+			conn.Close()
+		}()
+		Stream(&Conn{conn}, stream)
+	})
+
+	return httpServer.Serve(listener)
+}
+
 // 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
+		wsHeaders[key] = val
 	}
 	// Assume the header keys are in canonical format.
-	for _,  header := range stripWebsocketHeaders {
+	for _, header := range stripWebsocketHeaders {
 		wsHeaders.Del(header)
 	}
+	wsHeaders.Set("Host", req.Host) // See TUN-1097
 	return wsHeaders
 }
 
@@ -115,3 +201,19 @@ func changeRequestScheme(req *http.Request) string {
 		return req.URL.Scheme
 	}
 }
+
+// pinger simulates the websocket connection to keep it alive
+func pinger(logger *logrus.Logger, ws *websocket.Conn, done chan struct{}) {
+	ticker := time.NewTicker(pingPeriod)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			if err := ws.WriteControl(websocket.PingMessage, []byte{}, time.Now().Add(writeWait)); err != nil {
+				logger.WithError(err).Debug("failed to send ping message")
+			}
+		case <-done:
+			return
+		}
+	}
+}

websocket/websocket_test.go

@@ -1,100 +1,138 @@
 package websocket
 
 import (
-  "crypto/tls"
-  "io"
-  "math/rand"
-  "net/http"
-  "testing"
+	"crypto/tls"
+	"io"
+	"math/rand"
+	"net/http"
+	"testing"
 
-  "github.com/sirupsen/logrus"
-  "github.com/stretchr/testify/assert"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
 
-  "golang.org/x/net/websocket"
+	"golang.org/x/net/websocket"
 
-  "github.com/cloudflare/cloudflared/hello"
-  "github.com/cloudflare/cloudflared/tlsconfig"
+	"github.com/cloudflare/cloudflared/hello"
+	"github.com/cloudflare/cloudflared/tlsconfig"
 )
 
 const (
-  // example in Sec-Websocket-Key in rfc6455
-  testSecWebsocketKey    = "dGhlIHNhbXBsZSBub25jZQ=="
-  // example Sec-Websocket-Accept in rfc6455
-  testSecWebsocketAccept = "s3pPLMBiTxaQ9kYGzzhZRbK+xOo="
+	// example in Sec-Websocket-Key in rfc6455
+	testSecWebsocketKey = "dGhlIHNhbXBsZSBub25jZQ=="
+	// example Sec-Websocket-Accept in rfc6455
+	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, 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")
+	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
+	return req
 }
 
 func websocketClientTLSConfig(t *testing.T) *tls.Config {
-  certPool, err := tlsconfig.LoadOriginCertPool(nil)
-  assert.NoError(t, err)
-  assert.NotNil(t, certPool)
-  return &tls.Config{RootCAs: certPool}
+	certPool, err := tlsconfig.LoadOriginCertPool(nil)
+	assert.NoError(t, err)
+	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"))
+	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))
+	req := testRequest(t, "http://example.com", nil)
+	assert.Equal(t, testSecWebsocketAccept, generateAcceptKey(req))
 }
 
 func TestServe(t *testing.T) {
-  logger := logrus.New()
-  shutdownC := make(chan struct{})
-  errC := make(chan error)
-  listener, err := hello.CreateTLSListener("localhost:1111")
-  assert.NoError(t, err)
-  defer listener.Close()
+	logger := logrus.New()
+	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(logger, listener, shutdownC)
-  }()
+	go func() {
+		errC <- hello.StartHelloWorldServer(logger, listener, shutdownC)
+	}()
 
-  req := testRequest(t, "https://localhost:1111/ws", nil)
+	req := testRequest(t, "https://localhost:1111/ws", nil)
 
-  tlsConfig := websocketClientTLSConfig(t)
-  assert.NotNil(t, tlsConfig)
-  conn, resp, err := ClientConnect(req, tlsConfig)
-  assert.NoError(t, err)
-  assert.Equal(t, testSecWebsocketAccept, resp.Header.Get("Sec-WebSocket-Accept"))
+	tlsConfig := websocketClientTLSConfig(t)
+	assert.NotNil(t, tlsConfig)
+	conn, resp, err := ClientConnect(req, tlsConfig)
+	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)
+	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)
-  }
+		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
+	conn.Close()
+	close(shutdownC)
+	<-errC
 }
+
+// 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 := logrus.New()
+// 	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()
+// }