Release 2019.9.1

This reverts commit 817c3be9da5465043c2a2fda6c48f7ada760682e.

.gitignore

@@ -9,5 +9,8 @@ guide/public
 \#*\#
 cscope.*
 cloudflared
+cloudflared.exe
 !cmd/cloudflared/
-.DS_Store
+.DS_Store
+*-session.log
+ssh_server_tests/.env

Makefile

@@ -39,6 +39,10 @@ container:
 test: vet
 	go test -v -race $(VERSION_FLAGS) ./...
 
+.PHONY: test-ssh-server
+test-ssh-server:
+	docker-compose -f ssh_server_tests/docker-compose.yml up
+
 .PHONY: cloudflared-deb
 cloudflared-deb: cloudflared
 	mkdir -p $(PACKAGE_DIR)

RELEASE_NOTES

@@ -1,3 +1,22 @@
+2019.9.1
+- 2019-09-23 TUN-2334: remove tlsConfig.ServerName special case
+- 2019-09-23 AUTH-2077: Quotes open browser command in windows
+- 2019-09-11 AUTH-2050: Adds time.sleep to temporarily avoid hitting tunnel muxer dealock issue
+- 2019-09-10 AUTH-2056: Writes stderr to its own stream for non-pty connections
+- 2019-09-16 TUN-2307: Capnp is the only serialization format used in tunnelpogs
+- 2019-09-18 TUN-2315: Replace Scope with IntentLabel
+- 2019-09-17 TUN-2309: Split ConnectResult into ConnectError and ConnectSuccess, each implementing its own capnp serialization logic
+- 2019-09-18 AUTH-2052: Adds tests for SSH server
+- 2019-09-18 AUTH-2067: Log commands correctly
+- 2019-09-19 AUTH-2055: Verifies token at edge on access login
+- 2019-09-04 TUN-2201: change SRV records used by cloudflared
+- 2019-09-06 TUN-2280: Revert "TUN-2260: add name/group to CapnpConnectParameters, remove Scope"
+- 2019-09-03 AUTH-1943 hooked up uploader to logger, added timestamp to session logs, add tests
+- 2019-09-04 AUTH-2036: Refactor user retrieval, shutdown after ssh server stops, add custom version string
+- 2019-09-06 AUTH-1942 added event log to ssh server
+- 2019-09-04 AUTH-2037: Adds support for ssh port forwarding
+- 2019-09-05 TUN-2276: Path encoding broken
+
 2019.9.0
 - 2019-09-05 TUN-2279: Revert path encoding fix
 - 2019-08-30 AUTH-2021 - check error for failing tests

awsuploader/directory_upload_manager.go

@@ -77,7 +77,7 @@ func (m *DirectoryUploadManager) sweep() {
 		checkTime := time.Now().Add(-time.Duration(retentionTime))
 
 		//delete the file it is stale
-		if info.ModTime().After(checkTime) {
+		if info.ModTime().Before(checkTime) {
 			os.Remove(path)
 			return nil
 		}

carrier/carrier.go

@@ -114,7 +114,7 @@ func createWebsocketStream(options *StartOptions) (*cloudflaredWebsocket.Conn, e
 
 	wsConn, resp, err := cloudflaredWebsocket.ClientConnect(req, nil)
 	defer closeRespBody(resp)
-	if err != nil && isAccessResponse(resp) {
+	if err != nil && IsAccessResponse(resp) {
 		wsConn, err = createAccessAuthenticatedStream(options)
 		if err != nil {
 			return nil, err
@@ -126,10 +126,10 @@ func createWebsocketStream(options *StartOptions) (*cloudflaredWebsocket.Conn, e
 	return &cloudflaredWebsocket.Conn{Conn: wsConn}, nil
 }
 
-// isAccessResponse checks the http Response to see if the url location
+// IsAccessResponse checks the http Response to see if the url location
 // contains the Access structure.
-func isAccessResponse(resp *http.Response) bool {
-	if resp == nil || resp.StatusCode <= 300 {
+func IsAccessResponse(resp *http.Response) bool {
+	if resp == nil || resp.StatusCode != http.StatusFound {
 		return false
 	}
 
@@ -156,7 +156,7 @@ func createAccessAuthenticatedStream(options *StartOptions) (*websocket.Conn, er
 		return wsConn, nil
 	}
 
-	if !isAccessResponse(resp) {
+	if !IsAccessResponse(resp) {
 		return nil, err
 	}
 
@@ -179,7 +179,7 @@ func createAccessAuthenticatedStream(options *StartOptions) (*websocket.Conn, er
 
 // createAccessWebSocketStream builds an Access request and makes a connection
 func createAccessWebSocketStream(options *StartOptions) (*websocket.Conn, *http.Response, error) {
-	req, err := buildAccessRequest(options)
+	req, err := BuildAccessRequest(options)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -187,7 +187,7 @@ func createAccessWebSocketStream(options *StartOptions) (*websocket.Conn, *http.
 }
 
 // buildAccessRequest builds an HTTP request with the Access token set
-func buildAccessRequest(options *StartOptions) (*http.Request, error) {
+func BuildAccessRequest(options *StartOptions) (*http.Request, error) {
 	req, err := http.NewRequest(http.MethodGet, options.OriginURL, nil)
 	if err != nil {
 		return nil, err

carrier/carrier_test.go

@@ -102,14 +102,14 @@ func TestIsAccessResponse(t *testing.T) {
 		ExpectedOut bool
 	}{
 		{"nil response", nil, false},
-		{"redirect with no location", &http.Response{StatusCode: http.StatusPermanentRedirect}, false},
+		{"redirect with no location", &http.Response{StatusCode: http.StatusFound}, false},
 		{"200 ok", &http.Response{StatusCode: http.StatusOK}, false},
-		{"redirect with location", &http.Response{StatusCode: http.StatusPermanentRedirect, Header: validLocationHeader}, true},
-		{"redirect with invalid location", &http.Response{StatusCode: http.StatusPermanentRedirect, Header: invalidLocationHeader}, false},
+		{"redirect with location", &http.Response{StatusCode: http.StatusFound, Header: validLocationHeader}, true},
+		{"redirect with invalid location", &http.Response{StatusCode: http.StatusFound, Header: invalidLocationHeader}, false},
 	}
 
 	for i, tc := range testCases {
-		if isAccessResponse(tc.In) != tc.ExpectedOut {
+		if IsAccessResponse(tc.In) != tc.ExpectedOut {
 			t.Fatalf("Failed case %d -- %s", i, tc.Description)
 		}
 	}

cmd/cloudflared/access/cmd.go

@@ -1,17 +1,20 @@
 package access
 
 import (
-	"errors"
 	"fmt"
+	"net/http"
 	"net/url"
 	"os"
 	"strings"
 	"text/template"
+	"time"
 
+	"github.com/cloudflare/cloudflared/carrier"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/shell"
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/token"
 	"github.com/cloudflare/cloudflared/sshgen"
 	"github.com/cloudflare/cloudflared/validation"
+	"github.com/pkg/errors"
 	"golang.org/x/net/idna"
 
 	"github.com/cloudflare/cloudflared/log"
@@ -188,9 +191,14 @@ func login(c *cli.Context) error {
 		logger.Errorf("Please provide the url of the Access application\n")
 		return err
 	}
-	token, err := token.FetchToken(appURL)
-	if err != nil {
-		logger.Errorf("Failed to fetch token: %s\n", err)
+	if err := verifyTokenAtEdge(appURL, c); err != nil {
+		logger.WithError(err).Error("Could not verify token")
+		return err
+	}
+
+	token, err := token.GetTokenIfExists(appURL)
+	if err != nil || token == "" {
+		fmt.Fprintln(os.Stderr, "Unable to find token for provided application.")
 		return err
 	}
 	fmt.Fprintf(os.Stdout, "Successfully fetched your token:\n\n%s\n\n", string(token))
@@ -372,3 +380,59 @@ func isFileThere(candidate string) bool {
 	}
 	return true
 }
+
+// verifyTokenAtEdge checks for a token on disk, or generates a new one.
+// Then makes a request to to the origin with the token to ensure it is valid.
+// Returns nil if token is valid.
+func verifyTokenAtEdge(appUrl *url.URL, c *cli.Context) error {
+	headers := buildRequestHeaders(c.StringSlice(sshHeaderFlag))
+	if c.IsSet(sshTokenIDFlag) {
+		headers.Add("CF-Access-Client-Id", c.String(sshTokenIDFlag))
+	}
+	if c.IsSet(sshTokenSecretFlag) {
+		headers.Add("CF-Access-Client-Secret", c.String(sshTokenSecretFlag))
+	}
+	options := &carrier.StartOptions{OriginURL: appUrl.String(), Headers: headers}
+
+	if valid, err := isTokenValid(options); err != nil {
+		return err
+	} else if valid {
+		return nil
+	}
+
+	if err := token.RemoveTokenIfExists(appUrl); err != nil {
+		return err
+	}
+
+	if valid, err := isTokenValid(options); err != nil {
+		return err
+	} else if !valid {
+		return errors.New("failed to verify token")
+	}
+
+	return nil
+}
+
+// isTokenValid makes a request to the origin and returns true if the response was not a 302.
+func isTokenValid(options *carrier.StartOptions) (bool, error) {
+	req, err := carrier.BuildAccessRequest(options)
+	if err != nil {
+		return false, errors.Wrap(err, "Could not create access request")
+	}
+
+	// Do not follow redirects
+	client := &http.Client{
+		CheckRedirect: func(req *http.Request, via []*http.Request) error {
+			return http.ErrUseLastResponse
+		},
+		Timeout: time.Second * 5,
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return false, err
+	}
+	defer resp.Body.Close()
+
+	// A redirect to login means the token was invalid.
+	return !carrier.IsAccessResponse(resp), nil
+}

cmd/cloudflared/shell/launch_browser_darwin.go

@@ -0,0 +1,11 @@
+//+build darwin
+
+package shell
+
+import (
+	"os/exec"
+)
+
+func getBrowserCmd(url string) *exec.Cmd {
+	return exec.Command("open", url)
+}

cmd/cloudflared/shell/launch_browser_other.go

@@ -0,0 +1,11 @@
+//+build !windows,!darwin,!linux,!netbsd,!freebsd,!openbsd
+
+package shell
+
+import (
+	"os/exec"
+)
+
+func getBrowserCmd(url string) *exec.Cmd {
+	return nil
+}

cmd/cloudflared/shell/launch_browser_unix.go

@@ -0,0 +1,11 @@
+//+build linux freebsd openbsd netbsd
+
+package shell
+
+import (
+	"os/exec"
+)
+
+func getBrowserCmd(url string) *exec.Cmd {
+	return exec.Command("xdg-open", url)
+}

cmd/cloudflared/shell/launch_browser_windows.go

@@ -0,0 +1,18 @@
+//+build windows
+
+package shell
+
+import (
+	"fmt"
+	"os/exec"
+	"syscall"
+)
+
+func getBrowserCmd(url string) *exec.Cmd {
+	cmd := exec.Command("cmd")
+	// CmdLine is only defined when compiling for windows.
+	// Empty string is the cmd proc "Title". Needs to be included because the start command will interpret the first
+	// quoted string as that field and we want to quote the URL.
+	cmd.SysProcAttr = &syscall.SysProcAttr{CmdLine: fmt.Sprintf(`/c start "" "%s"`, url)}
+	return cmd
+}

cmd/cloudflared/shell/shell.go

@@ -4,25 +4,11 @@ import (
 	"io"
 	"os"
 	"os/exec"
-	"runtime"
 )
 
 // OpenBrowser opens the specified URL in the default browser of the user
 func OpenBrowser(url string) error {
-	var cmd string
-	var args []string
-
-	switch runtime.GOOS {
-	case "windows":
-		cmd = "cmd"
-		args = []string{"/c", "start"}
-	case "darwin":
-		cmd = "open"
-	default: // "linux", "freebsd", "openbsd", "netbsd"
-		cmd = "xdg-open"
-	}
-	args = append(args, url)
-	return exec.Command(cmd, args...).Start()
+	return getBrowserCmd(url).Start()
 }
 
 // Run will kick off a shell task and pipe the results to the respective std pipes

cmd/cloudflared/transfer/transfer.go

@@ -18,7 +18,7 @@ import (
 )
 
 const (
-	baseStoreURL  = "https://login.cloudflarewarp.com/"
+	baseStoreURL  = "https://login.argotunnel.com/"
 	clientTimeout = time.Second * 60
 )
 

cmd/cloudflared/tunnel/cmd.go

@@ -73,6 +73,11 @@ const (
 
 	// s3URLFlag is the S3 URL of SSH log uploader (e.g. don't use AWS s3 and use google storage bucket instead)
 	s3URLFlag = "s3-url-host"
+
+	// disablePortForwarding disables both remote and local ssh port forwarding
+	enablePortForwardingFlag = "enable-port-forwarding"
+
+	noIntentMsg = "The --intent argument is required. Cloudflared looks up an Intent to determine what configuration to use (i.e. which tunnels to start). If you don't have any Intents yet, you can use a placeholder Intent Label for now. Then, when you make an Intent with that label, cloudflared will get notified and open the tunnels you specified in that Intent."
 )
 
 var (
@@ -363,33 +368,42 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
 
 	if c.IsSet("ssh-server") {
 		if runtime.GOOS != "darwin" && runtime.GOOS != "linux" {
-			logger.Errorf("--ssh-server is not supported on %s", runtime.GOOS)
-			return errors.New(fmt.Sprintf("--ssh-server is not supported on %s", runtime.GOOS))
+			msg := fmt.Sprintf("--ssh-server is not supported on %s", runtime.GOOS)
+			logger.Error(msg)
+			return errors.New(msg)
 
 		}
 
 		logger.Infof("ssh-server set")
 
+		logManager := sshlog.NewEmptyManager()
 		if c.IsSet(bucketNameFlag) && c.IsSet(regionNameFlag) && c.IsSet(accessKeyIDFlag) && c.IsSet(secretIDFlag) {
 			uploader, err := awsuploader.NewFileUploader(c.String(bucketNameFlag), c.String(regionNameFlag),
 				c.String(accessKeyIDFlag), c.String(secretIDFlag), c.String(sessionTokenIDFlag), c.String(s3URLFlag))
 			if err != nil {
-				logger.WithError(err).Error("Cannot create uploader for SSH Server")
-				return errors.Wrap(err, "Cannot create uploader for SSH Server")
+				msg := "Cannot create uploader for SSH Server"
+				logger.WithError(err).Error(msg)
+				return errors.Wrap(err, msg)
 			}
 
-			os.Mkdir(sshLogFileDirectory, 0600)
+			if err := os.MkdirAll(sshLogFileDirectory, 0600); err != nil {
+				msg := fmt.Sprintf("Cannot create SSH log file directory %s", sshLogFileDirectory)
+				logger.WithError(err).Errorf(msg)
+				return errors.Wrap(err, msg)
+			}
+
+			logManager = sshlog.New(sshLogFileDirectory)
 
 			uploadManager := awsuploader.NewDirectoryUploadManager(logger, uploader, sshLogFileDirectory, 30*time.Minute, shutdownC)
 			uploadManager.Start()
 		}
 
-		logManager := sshlog.New()
 		sshServerAddress := "127.0.0.1:" + c.String(sshPortFlag)
-		server, err := sshserver.New(logManager, logger, sshServerAddress, shutdownC, c.Duration(sshIdleTimeoutFlag), c.Duration(sshMaxTimeoutFlag))
+		server, err := sshserver.New(logManager, logger, version, sshServerAddress, shutdownC, c.Duration(sshIdleTimeoutFlag), c.Duration(sshMaxTimeoutFlag), c.Bool(enablePortForwardingFlag))
 		if err != nil {
-			logger.WithError(err).Error("Cannot create new SSH Server")
-			return errors.Wrap(err, "Cannot create new SSH Server")
+			msg := "Cannot create new SSH Server"
+			logger.WithError(err).Error(msg)
+			return errors.Wrap(err, msg)
 		}
 		wg.Add(1)
 		go func() {
@@ -397,6 +411,8 @@ func StartServer(c *cli.Context, version string, shutdownC, graceShutdownC chan
 			if err = server.Start(); err != nil && err != ssh.ErrServerClosed {
 				logger.WithError(err).Error("SSH server error")
 			}
+			// TODO: remove when declarative tunnels are implemented.
+			close(shutdownC)
 		}()
 		c.Set("url", "ssh://"+sshServerAddress)
 	}
@@ -506,19 +522,16 @@ func startDeclarativeTunnel(ctx context.Context,
 		return err
 	}
 
-	name := c.String("name")
-	group := c.String("group")
-	if group == "" {
-		err := fmt.Errorf("--group must be specified")
-		logger.WithError(err).Error("unable to parse group name")
-		return err
+	intentLabel := c.String("intent")
+	if intentLabel == "" {
+		logger.Error("--intent was empty")
+		return fmt.Errorf(noIntentMsg)
 	}
 
 	cloudflaredConfig := &connection.CloudflaredConfig{
 		BuildInfo:     buildInfo,
 		CloudflaredID: cloudflaredID,
-		Name:          name,
-		Group:         group,
+		IntentLabel:   intentLabel,
 		Tags:          tags,
 	}
 
@@ -942,15 +955,9 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 			Hidden:  true,
 		}),
 		altsrc.NewStringFlag(&cli.StringFlag{
-			Name:    "name",
-			Usage:   "Friendly name for this cloudflared instance.",
-			EnvVars: []string{"TUNNEL_DECLARATIVE_NAME"},
-			Hidden:  true,
-		}),
-		altsrc.NewStringFlag(&cli.StringFlag{
-			Name:    "group",
-			Usage:   "The group whose behavior this cloudflared instance will adopt. This behavior can be configured by editing the group's 'intent' in the Declarative Tunnel UI.",
-			EnvVars: []string{"TUNNEL_DECLARATIVE_GROUP"},
+			Name:    "intent",
+			Usage:   "The label of an Intent from which `cloudflared` should gets its tunnels from. Intents can be created in the Origin Registry UI.",
+			EnvVars: []string{"TUNNEL_INTENT"},
 			Hidden:  true,
 		}),
 		altsrc.NewDurationFlag(&cli.DurationFlag{
@@ -963,7 +970,7 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 		altsrc.NewStringFlag(&cli.StringFlag{
 			Name:    sshPortFlag,
 			Usage:   "Localhost port that cloudflared SSH server will run on",
-			Value:   "22",
+			Value:   "2222",
 			EnvVars: []string{"LOCAL_SSH_PORT"},
 			Hidden:  true,
 		}),
@@ -1015,5 +1022,11 @@ func tunnelFlags(shouldHide bool) []cli.Flag {
 			EnvVars: []string{"S3_URL"},
 			Hidden:  true,
 		}),
+		altsrc.NewBoolFlag(&cli.BoolFlag{
+			Name:    enablePortForwardingFlag,
+			Usage:   "Enables remote and local SSH port forwarding",
+			EnvVars: []string{"ENABLE_PORT_FORWARDING"},
+			Hidden:  true,
+		}),
 	}
 }

cmd/cloudflared/tunnel/login.go

@@ -15,7 +15,7 @@ import (
 
 const (
 	baseLoginURL     = "https://dash.cloudflare.com/argotunnel"
-	callbackStoreURL = "https://login.cloudflarewarp.com/"
+	callbackStoreURL = "https://login.argotunnel.com/"
 )
 
 func login(c *cli.Context) error {

connection/connection.go

@@ -50,7 +50,7 @@ func (c *Connection) Serve(ctx context.Context) error {
 }
 
 // Connect is used to establish connections with cloudflare's edge network
-func (c *Connection) Connect(ctx context.Context, parameters *tunnelpogs.ConnectParameters, logger *logrus.Entry) (*pogs.ConnectResult, error) {
+func (c *Connection) Connect(ctx context.Context, parameters *tunnelpogs.ConnectParameters, logger *logrus.Entry) (pogs.ConnectResult, error) {
 	openStreamCtx, cancel := context.WithTimeout(ctx, openStreamTimeout)
 	defer cancel()
 

connection/discovery.go

@@ -13,13 +13,13 @@ import (
 )
 
 const (
-	// Used to discover HA Warp servers
-	srvService = "warp"
+	// Used to discover HA origintunneld servers
+	srvService = "origintunneld"
 	srvProto   = "tcp"
-	srvName    = "cloudflarewarp.com"
+	srvName    = "argotunnel.com"
 
 	// Used to fallback to DoT when we can't use the default resolver to
-	// discover HA Warp servers (GitHub issue #75).
+	// discover HA origintunneld servers (GitHub issue #75).
 	dotServerName = "cloudflare-dns.com"
 	dotServerAddr = "1.1.1.1:853"
 	dotTimeout    = time.Duration(15 * time.Second)
@@ -30,8 +30,8 @@ const (
 
 var friendlyDNSErrorLines = []string{
 	`Please try the following things to diagnose this issue:`,
-	`  1. ensure that cloudflarewarp.com is returning "warp" service records.`,
-	`     Run your system's equivalent of: dig srv _warp._tcp.cloudflarewarp.com`,
+	`  1. ensure that argotunnel.com is returning "origintunneld" service records.`,
+	`     Run your system's equivalent of: dig srv _origintunneld._tcp.argotunnel.com`,
 	`  2. ensure that your DNS resolver is not returning compressed SRV records.`,
 	`     See GitHub issue https://github.com/golang/go/issues/27546`,
 	`     For example, you could use Cloudflare's 1.1.1.1 as your resolver:`,
@@ -102,7 +102,7 @@ func EdgeDiscovery(logger *logrus.Entry) ([]*net.TCPAddr, error) {
 		// Try to fall back to DoT from Cloudflare directly.
 		//
 		// Note: Instead of DoT, we could also have used DoH. Either of these:
-		//     - directly via the JSON API (https://1.1.1.1/dns-query?ct=application/dns-json&name=_warp._tcp.cloudflarewarp.com&type=srv)
+		//     - directly via the JSON API (https://1.1.1.1/dns-query?ct=application/dns-json&name=_origintunneld._tcp.argotunnel.com&type=srv)
 		//     - indirectly via `tunneldns.NewUpstreamHTTPS()`
 		// But both of these cases miss out on a key feature from the stdlib:
 		//     "The returned records are sorted by priority and randomized by weight within a priority."
@@ -119,7 +119,7 @@ func EdgeDiscovery(logger *logrus.Entry) ([]*net.TCPAddr, error) {
 			for _, s := range friendlyDNSErrorLines {
 				logger.Errorln(s)
 			}
-			return nil, errors.Wrap(err, "Could not lookup srv records on _warp._tcp.cloudflarewarp.com")
+			return nil, errors.Wrap(err, "Could not lookup srv records on _origintunneld._tcp.argotunnel.com")
 		}
 		// Accept the fallback results and keep going
 		addrs = fallbackAddrs

connection/manager.go

@@ -17,8 +17,9 @@ import (
 )
 
 const (
-	quickStartLink = "https://developers.cloudflare.com/argo-tunnel/quickstart/"
-	faqLink        = "https://developers.cloudflare.com/argo-tunnel/faq/"
+	quickStartLink    = "https://developers.cloudflare.com/argo-tunnel/quickstart/"
+	faqLink           = "https://developers.cloudflare.com/argo-tunnel/faq/"
+	defaultRetryAfter = time.Second * 5
 )
 
 // EdgeManager manages connections with the edge
@@ -47,8 +48,7 @@ type CloudflaredConfig struct {
 	CloudflaredID uuid.UUID
 	Tags          []pogs.Tag
 	BuildInfo     *buildinfo.BuildInfo
-	Name          string
-	Group         string
+	IntentLabel   string
 }
 
 func NewEdgeManager(
@@ -88,8 +88,12 @@ func (em *EdgeManager) Run(ctx context.Context) error {
 		// Create/delete connection one at a time, so we don't need to adjust for connections that are being created/deleted
 		// in shouldCreateConnection or shouldReduceConnection calculation
 		if em.state.shouldCreateConnection(em.serviceDiscoverer.AvailableAddrs()) {
-			if err := em.newConnection(ctx); err != nil {
-				em.logger.WithError(err).Error("cannot create new connection")
+			if connErr := em.newConnection(ctx); connErr != nil {
+				if !connErr.ShouldRetry {
+					em.logger.WithError(connErr).Error(em.noRetryMessage())
+					return connErr
+				}
+				em.logger.WithError(connErr).Error("cannot create new connection")
 			}
 		} else if em.state.shouldReduceConnection() {
 			if err := em.closeConnection(ctx); err != nil {
@@ -104,11 +108,11 @@ func (em *EdgeManager) UpdateConfigurable(newConfigurable *EdgeManagerConfigurab
 	em.state.updateConfigurable(newConfigurable)
 }
 
-func (em *EdgeManager) newConnection(ctx context.Context) error {
+func (em *EdgeManager) newConnection(ctx context.Context) *pogs.ConnectError {
 	edgeIP := em.serviceDiscoverer.Addr()
 	edgeConn, err := em.dialEdge(ctx, edgeIP)
 	if err != nil {
-		return errors.Wrap(err, "dial edge error")
+		return retryConnection(fmt.Sprintf("dial edge error: %v", err))
 	}
 	configurable := em.state.getConfigurable()
 	// Establish a muxed connection with the edge
@@ -122,12 +126,12 @@ func (em *EdgeManager) newConnection(ctx context.Context) error {
 		Logger:            em.logger.WithField("subsystem", "muxer"),
 	})
 	if err != nil {
-		return errors.Wrap(err, "couldn't perform handshake with edge")
+		retryConnection(fmt.Sprintf("couldn't perform handshake with edge: %v", err))
 	}
 
 	h2muxConn, err := newConnection(muxer, edgeIP)
 	if err != nil {
-		return errors.Wrap(err, "couldn't create h2mux connection")
+		return retryConnection(fmt.Sprintf("couldn't create h2mux connection: %v", err))
 	}
 
 	go em.serveConn(ctx, h2muxConn)
@@ -137,24 +141,20 @@ func (em *EdgeManager) newConnection(ctx context.Context) error {
 		CloudflaredVersion:  em.cloudflaredConfig.BuildInfo.CloudflaredVersion,
 		NumPreviousAttempts: 0,
 		OriginCert:          em.state.getUserCredential(),
-		Name:                em.cloudflaredConfig.Name,
-		Group:               em.cloudflaredConfig.Group,
+		IntentLabel:         em.cloudflaredConfig.IntentLabel,
 		Tags:                em.cloudflaredConfig.Tags,
 	}, em.logger)
 	if err != nil {
 		h2muxConn.Shutdown()
-		return errors.Wrap(err, "couldn't connect to edge")
+		return retryConnection(fmt.Sprintf("couldn't connect to edge: %v", err))
 	}
 
-	if connErr := connResult.Err; connErr != nil {
-		if !connErr.ShouldRetry {
-			return errors.Wrap(connErr, em.noRetryMessage())
-		}
-		return errors.Wrapf(connErr, "edge responded with RetryAfter=%v", connErr.RetryAfter)
+	if connErr := connResult.ConnectError(); connErr != nil {
+		return connErr
 	}
 
 	em.state.newConnection(h2muxConn)
-	em.logger.Infof("connected to %s", connResult.ServerInfo.LocationName)
+	em.logger.Infof("connected to %s", connResult.ConnectedTo())
 	return nil
 }
 
@@ -284,3 +284,11 @@ func (ems *edgeManagerState) getUserCredential() []byte {
 	defer ems.RUnlock()
 	return ems.userCredential
 }
+
+func retryConnection(cause string) *pogs.ConnectError {
+	return &pogs.ConnectError{
+		Cause:       cause,
+		RetryAfter:  defaultRetryAfter,
+		ShouldRetry: true,
+	}
+}

ssh_server_tests/Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3-buster
+
+RUN wget https://bin.equinox.io/c/VdrWdbjqyF/cloudflared-stable-linux-amd64.deb \
+  && dpkg -i cloudflared-stable-linux-amd64.deb
+
+RUN pip install pexpect
+
+COPY tests.py .
+COPY ssh /root/.ssh
+RUN chmod 600 /root/.ssh/id_rsa
+
+ARG SSH_HOSTNAME
+RUN bash -c 'sed -i "s/{{hostname}}/${SSH_HOSTNAME}/g" /root/.ssh/authorized_keys_config'
+RUN bash -c 'sed -i "s/{{hostname}}/${SSH_HOSTNAME}/g" /root/.ssh/short_lived_cert_config'

ssh_server_tests/README.md

@@ -0,0 +1,23 @@
+# Cloudflared SSH server smoke tests
+
+Runs several tests in a docker container against a server that is started out of band of these tests.
+Cloudflared token also needs to be retrieved out of band.
+SSH server hostname and user need to be configured in a docker environment file
+
+
+## Running tests
+
+* Build cloudflared:
+make cloudflared
+
+* Start server:
+sudo ./cloudflared tunnel --hostname HOSTNAME --ssh-server
+
+* Fetch token: 
+./cloudflared access login HOSTNAME
+
+* Create docker env file:
+echo "SSH_HOSTNAME=HOSTNAME\nSSH_USER=USERNAME\n" > ssh_server_tests/.env
+
+* Run tests:
+make test-ssh-server

ssh_server_tests/docker-compose.yml

@@ -0,0 +1,19 @@
+version: "3.1"
+
+services:
+  ssh_test:
+    build:
+      context: .
+      args:
+        - SSH_HOSTNAME=${SSH_HOSTNAME}
+    volumes:
+      - "~/.cloudflared/:/root/.cloudflared"
+    env_file:
+     - .env
+    environment:
+      - AUTHORIZED_KEYS_SSH_CONFIG=/root/.ssh/authorized_keys_config
+      - SHORT_LIVED_CERT_SSH_CONFIG=/root/.ssh/short_lived_cert_config
+      - REMOTE_SCP_FILENAME=scp_test.txt
+      - ROOT_ONLY_TEST_FILE_PATH=~/permission_test.txt
+
+    entrypoint: "python tests.py"

ssh_server_tests/ssh/authorized_keys_config

@@ -0,0 +1,5 @@
+Host *
+    AddressFamily inet
+
+Host {{hostname}}
+  ProxyCommand /usr/local/bin/cloudflared access ssh --hostname %h

ssh_server_tests/ssh/id_rsa

@@ -0,0 +1,49 @@
+-----BEGIN OPENSSH PRIVATE KEY-----
+b3BlbnNzaC1rZXktdjEAAAAABG5vbmUAAAAEbm9uZQAAAAAAAAABAAACFwAAAAdzc2gtcn
+NhAAAAAwEAAQAAAgEAvi26NDQ8cYTTztqPe9ZgF5HR/rIo5FoDgL5NbbZKW6h0txP9Fd8s
+id9Bgmo+aGkeM327tPVVMQ6UFmdRksOCIDWQDjNLF8b6S+Fu95tvMKSbGreRoR32OvgZKV
+I6KmOsF4z4GIv9naPplZswtKEUhSSI+/gPdAs9wfwalqZ77e82QJ727bYMeC3lzuoT+KBI
+dYufJ4OQhLtpHrqhB5sn7s6+oCv/u85GSln5SIC18Hi2t9lW4tgb5tH8P0kEDDWGfPS5ok
+qGi4kFTvwBXOCS2r4dhi5hRkpP7PqG4np0OCfvK5IRRJ27fCnj0loc+puZJAxnPMbuJr64
+vwxRx78PM/V0PDUsl0P6aR/vbe0XmF9FGqbWf2Tar1p4r6C9/bMzcDz8seYT8hzLIHP3+R
+l1hdlsTLm+1EzhaExKId+tjXegKGG4nU24h6qHEnRxLQDMwEsdkfj4E1pVypZJXVyNj99D
+o84vi0EUnu7R4HmQb/C+Pu7qMDtLT3Zk7O5Mg4LQ+cTz9V0noYEAyG46nAB4U/nJzBnV1J
++aAdpioHmUAYhLYlQ9Kiy7LCJi92g9Wqa4wxMKxBbO5ZeH++p2p2lUi/oQNqx/2dLYFmy0
+wxvJHbZIhAaFbOeCvHg1ucIAQznli2jOr2qoB+yKRRPAp/3NXnZg1v7ce2CkwiAD52wjtC
+kAAAdILMJUeyzCVHsAAAAHc3NoLXJzYQAAAgEAvi26NDQ8cYTTztqPe9ZgF5HR/rIo5FoD
+gL5NbbZKW6h0txP9Fd8sid9Bgmo+aGkeM327tPVVMQ6UFmdRksOCIDWQDjNLF8b6S+Fu95
+tvMKSbGreRoR32OvgZKVI6KmOsF4z4GIv9naPplZswtKEUhSSI+/gPdAs9wfwalqZ77e82
+QJ727bYMeC3lzuoT+KBIdYufJ4OQhLtpHrqhB5sn7s6+oCv/u85GSln5SIC18Hi2t9lW4t
+gb5tH8P0kEDDWGfPS5okqGi4kFTvwBXOCS2r4dhi5hRkpP7PqG4np0OCfvK5IRRJ27fCnj
+0loc+puZJAxnPMbuJr64vwxRx78PM/V0PDUsl0P6aR/vbe0XmF9FGqbWf2Tar1p4r6C9/b
+MzcDz8seYT8hzLIHP3+Rl1hdlsTLm+1EzhaExKId+tjXegKGG4nU24h6qHEnRxLQDMwEsd
+kfj4E1pVypZJXVyNj99Do84vi0EUnu7R4HmQb/C+Pu7qMDtLT3Zk7O5Mg4LQ+cTz9V0noY
+EAyG46nAB4U/nJzBnV1J+aAdpioHmUAYhLYlQ9Kiy7LCJi92g9Wqa4wxMKxBbO5ZeH++p2
+p2lUi/oQNqx/2dLYFmy0wxvJHbZIhAaFbOeCvHg1ucIAQznli2jOr2qoB+yKRRPAp/3NXn
+Zg1v7ce2CkwiAD52wjtCkAAAADAQABAAACAQCbnVsyAFQ9J00Rg/HIiUATyTQlzq57O9SF
+8jH1RiZOHedzLx32WaleH5rBFiJ+2RTnWUjQ57aP77fpJR2wk93UcT+w/vPBPwXsNUjRvx
+Qan3ZzRCYbyiKDWiNslmYV7X0RwD36CAK8jTVDP7t48h2SXLTiSLaMY+5i3uD6yLu7k/O2
+qNyw4jgN1rCmwQ8acD0aQec3NAZ7NcbsaBX/3Uutsup0scwOZtlJWZoLY5Z8cKpCgcsAz4
+j1NHnNZvey7dFgSffj/ktdvf7kBH0w/GnuJ4aNF0Jte70u0kiw5TZYBQVFh74tgUu6a6SJ
+qUbxIYUL5EJNjxGsDn+phHEemw3aMv0CwZG6Tqaionlna7bLsl9Bg1HTGclczVWx8uqC+M
+6agLmkhYCHG0rVj8h5smjXAQXtmvIDVYDOlJZZoF9VAOCj6QfmJUH1NAGpCs1HDHbeOxGA
+OLCh4d3F4rScPqhGdtSt4W13VFIvXn2Qqoz9ufepZsee1SZqpcerxywx2wN9ZAzu+X8lTN
+i+TA2B3vWpqqucOEsp4JwDN+VMKZqKUGUDWcm/eHSaG6wq0q734LUlgM85TjaIg8QsNtWV
+giB1nWwsYIuH4rsFNFGEwURYdGBcw6idH0GZ7I4RaIB5F9oOza1d601E0APHYrtnx9yOiK
+nOtJ+5ZmVZovaDRfu1aQAAAQBU/EFaNUzoVhO04pS2L6BlByt963bOIsSJhdlEzek5AAli
+eaf1S/PD6xWCc0IGY+GZE0HPbhsKYanjqOpWldcA2T7fzf4oz4vFBfUkPYo/MLSlLCYsDd
+IH3wBkCssnfR5EkzNgxnOvq646Nl64BMvxwSIXGPktdq9ZALxViwricSRzCFURnh5vLHWU
+wBzSgAA0UlZ9E64GtAv066+AoZCp83GhTLRC4o0naE2e/K4op4BCFHLrZ8eXmDRK3NJj80
+Vkn+uhrk+SHmbjIhmS57Pv9p8TWyRvemph/nMUuZGKBUu2X+JQxggck0KigIrXjsmciCsM
+BIM3mYDDfjYbyVhTAAABAQDkV8O1bWUsAIqk7RU+iDZojN5kaO+zUvj1TafX8QX1sY6pu4
+Z2cfSEka1532BaehM95bQm7BCPw4cYg56XidmCQTZ9WaWqxVrOo48EKXUtZMZx6nKFOKlq
+MT2XTMnGT9n7kFCfEjSVkAjuJ9ZTFLOaoXAaVRnxeHQwOKaup5KKP9GSzNIw328U+96s3V
+WKHeT4pMjHBccgW/qX/tRRidZw5in5uBC9Ew5y3UACFTkNOnhUwVfyUNbBZJ2W36msQ3KD
+AN7nOrQHqhd3NFyCEy2ovIAKVBacr/VEX6EsRUshIehJzz8EY9f3kXL7WT2QDoz2giPeBJ
+HJdEpXt43UpszjAAABAQDVNpqNdHUlCs9XnbIvc6ZRrNh79wt65YFfvh/QEuA33KnA6Ri6
+EgnV5IdUWXS/UFaYcm2udydrBpVIVifSYl3sioHBylpri23BEy38PKwVXvghUtfpN6dWGn
+NZUG25fQPtIzqi+lo953ZjIj+Adi17AeVv4P4NiLrZeM9lXfWf2pEPOecxXs1IwAf9IiDQ
+WepAwRLsu42eEnHA+DSJPZUkSbISfM5X345k0g6EVATX/yLL3CsqClPzPtsqjh6rbEfFg3
+2OfIMcWV77gOlGWGQ+bUHc8kV6xJqV9QVacLWzfLvIqHF0wQMf8WLOVHEzkfiq4VjwhVqr
+/+FFvljm5nSDAAAAEW1pa2VAQzAyWTUwVEdKR0g4AQ==
+-----END OPENSSH PRIVATE KEY-----

ssh_server_tests/ssh/id_rsa.pub

@@ -0,0 +1 @@
+ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAACAQC+Lbo0NDxxhNPO2o971mAXkdH+sijkWgOAvk1ttkpbqHS3E/0V3yyJ30GCaj5oaR4zfbu09VUxDpQWZ1GSw4IgNZAOM0sXxvpL4W73m28wpJsat5GhHfY6+BkpUjoqY6wXjPgYi/2do+mVmzC0oRSFJIj7+A90Cz3B/BqWpnvt7zZAnvbttgx4LeXO6hP4oEh1i58ng5CEu2keuqEHmyfuzr6gK/+7zkZKWflIgLXweLa32Vbi2Bvm0fw/SQQMNYZ89LmiSoaLiQVO/AFc4JLavh2GLmFGSk/s+obienQ4J+8rkhFEnbt8KePSWhz6m5kkDGc8xu4mvri/DFHHvw8z9XQ8NSyXQ/ppH+9t7ReYX0UaptZ/ZNqvWnivoL39szNwPPyx5hPyHMsgc/f5GXWF2WxMub7UTOFoTEoh362Nd6AoYbidTbiHqocSdHEtAMzASx2R+PgTWlXKlkldXI2P30Ojzi+LQRSe7tHgeZBv8L4+7uowO0tPdmTs7kyDgtD5xPP1XSehgQDIbjqcAHhT+cnMGdXUn5oB2mKgeZQBiEtiVD0qLLssImL3aD1aprjDEwrEFs7ll4f76nanaVSL+hA2rH/Z0tgWbLTDG8kdtkiEBoVs54K8eDW5wgBDOeWLaM6vaqgH7IpFE8Cn/c1edmDW/tx7YKTCIAPnbCO0KQ== mike@C02Y50TGJGH8

ssh_server_tests/ssh/short_lived_cert_config

@@ -0,0 +1,11 @@
+Host *
+    AddressFamily inet
+
+Host {{hostname}}
+  ProxyCommand bash -c '/usr/local/bin/cloudflared access ssh-gen --hostname %h; ssh -F /root/.ssh/short_lived_cert_config -tt %r@cfpipe-{{hostname}} >&2 <&1'
+
+Host cfpipe-{{hostname}}
+  HostName {{hostname}}
+  ProxyCommand /usr/local/bin/cloudflared access ssh --hostname %h
+  IdentityFile ~/.cloudflared/{{hostname}}-cf_key
+  CertificateFile ~/.cloudflared/{{hostname}}-cf_key-cert.pub

ssh_server_tests/tests.py

@@ -0,0 +1,195 @@
+"""
+Cloudflared Integration tests
+"""
+
+import unittest
+import subprocess
+import os
+import tempfile
+from contextlib import contextmanager
+
+from pexpect import pxssh
+
+
+class TestSSHBase(unittest.TestCase):
+    """
+    SSH test base class containing constants and helper funcs
+    """
+
+    HOSTNAME = os.environ["SSH_HOSTNAME"]
+    SSH_USER = os.environ["SSH_USER"]
+    SSH_TARGET = f"{SSH_USER}@{HOSTNAME}"
+    AUTHORIZED_KEYS_SSH_CONFIG = os.environ["AUTHORIZED_KEYS_SSH_CONFIG"]
+    SHORT_LIVED_CERT_SSH_CONFIG = os.environ["SHORT_LIVED_CERT_SSH_CONFIG"]
+    SSH_OPTIONS = {"StrictHostKeyChecking": "no"}
+
+    @classmethod
+    def get_ssh_command(cls, pty=True):
+        """
+        Return ssh command arg list. If pty is true, a PTY is forced for the session.
+        """
+        cmd = [
+            "ssh",
+            "-o",
+            "StrictHostKeyChecking=no",
+            "-F",
+            cls.AUTHORIZED_KEYS_SSH_CONFIG,
+            cls.SSH_TARGET,
+        ]
+        if not pty:
+            cmd += ["-T"]
+        else:
+            cmd += ["-tt"]
+
+        return cmd
+
+    @classmethod
+    @contextmanager
+    def ssh_session_manager(cls, *args, **kwargs):
+        """
+        Context manager for interacting with a pxssh session.
+        Disables pty echo on the remote server and ensures session is terminated afterward.
+        """
+        session = pxssh.pxssh(options=cls.SSH_OPTIONS)
+
+        session.login(
+            cls.HOSTNAME,
+            username=cls.SSH_USER,
+            original_prompt=r"[#@$]",
+            ssh_config=kwargs.get("ssh_config", cls.AUTHORIZED_KEYS_SSH_CONFIG),
+            ssh_tunnels=kwargs.get("ssh_tunnels", {}),
+        )
+        try:
+            session.sendline("stty -echo")
+            session.prompt()
+            yield session
+        finally:
+            session.logout()
+
+    @staticmethod
+    def get_command_output(session, cmd):
+        """
+        Executes command on remote ssh server and waits for prompt.
+        Returns command output
+        """
+        session.sendline(cmd)
+        session.prompt()
+        return session.before.decode().strip()
+
+    def exec_command(self, cmd, shell=False):
+        """
+        Executes command locally. Raises Assertion error for non-zero return code.
+        Returns stdout and stderr
+        """
+        proc = subprocess.Popen(
+            cmd, stderr=subprocess.PIPE, stdout=subprocess.PIPE, shell=shell
+        )
+        raw_out, raw_err = proc.communicate()
+
+        out = raw_out.decode()
+        err = raw_err.decode()
+        self.assertEqual(proc.returncode, 0, msg=f"stdout: {out} stderr: {err}")
+        return out.strip(), err.strip()
+
+
+class TestSSHCommandExec(TestSSHBase):
+    """
+    Tests inline ssh command exec
+    """
+
+    # Name of file to be downloaded over SCP on remote server.
+    REMOTE_SCP_FILENAME = os.environ["REMOTE_SCP_FILENAME"]
+
+    @classmethod
+    def get_scp_base_command(cls):
+        return [
+            "scp",
+            "-o",
+            "StrictHostKeyChecking=no",
+            "-v",
+            "-F",
+            cls.AUTHORIZED_KEYS_SSH_CONFIG,
+        ]
+
+    @unittest.skip(
+        "This creates files on the remote. Should be skipped until server is dockerized."
+    )
+    def test_verbose_scp_sink_mode(self):
+        with tempfile.NamedTemporaryFile() as fl:
+            self.exec_command(
+                self.get_scp_base_command() + [fl.name, f"{self.SSH_TARGET}:"]
+            )
+
+    def test_verbose_scp_source_mode(self):
+        with tempfile.TemporaryDirectory() as tmpdirname:
+            self.exec_command(
+                self.get_scp_base_command()
+                + [f"{self.SSH_TARGET}:{self.REMOTE_SCP_FILENAME}", tmpdirname]
+            )
+            local_filename = os.path.join(tmpdirname, self.REMOTE_SCP_FILENAME)
+
+            self.assertTrue(os.path.exists(local_filename))
+            self.assertTrue(os.path.getsize(local_filename) > 0)
+
+    def test_pty_command(self):
+        base_cmd = self.get_ssh_command()
+
+        out, _ = self.exec_command(base_cmd + ["whoami"])
+        self.assertEqual(out.strip().lower(), self.SSH_USER.lower())
+
+        out, _ = self.exec_command(base_cmd + ["tty"])
+        self.assertNotEqual(out, "not a tty")
+
+    def test_non_pty_command(self):
+        base_cmd = self.get_ssh_command(pty=False)
+
+        out, _ = self.exec_command(base_cmd + ["whoami"])
+        self.assertEqual(out.strip().lower(), self.SSH_USER.lower())
+
+        out, _ = self.exec_command(base_cmd + ["tty"])
+        self.assertEqual(out, "not a tty")
+
+
+class TestSSHShell(TestSSHBase):
+    """
+    Tests interactive SSH shell
+    """
+
+    # File path to a file on the remote server with root only read privileges.
+    ROOT_ONLY_TEST_FILE_PATH = os.environ["ROOT_ONLY_TEST_FILE_PATH"]
+
+    def test_ssh_pty(self):
+        with self.ssh_session_manager() as session:
+
+            # Test shell launched as correct user
+            username = self.get_command_output(session, "whoami")
+            self.assertEqual(username.lower(), self.SSH_USER.lower())
+
+            # Test USER env variable set
+            user_var = self.get_command_output(session, "echo $USER")
+            self.assertEqual(user_var.lower(), self.SSH_USER.lower())
+
+            # Test HOME env variable set to true user home.
+            home_env = self.get_command_output(session, "echo $HOME")
+            pwd = self.get_command_output(session, "pwd")
+            self.assertEqual(pwd, home_env)
+
+            # Test shell launched in correct user home dir.
+            self.assertIn(username, pwd)
+
+            # Ensure shell launched with correct user's permissions and privs.
+            # Cant read root owned 0700 files.
+            output = self.get_command_output(
+                session, f"cat {self.ROOT_ONLY_TEST_FILE_PATH}"
+            )
+            self.assertIn("Permission denied", output)
+
+    def test_short_lived_cert_auth(self):
+        with self.ssh_session_manager(
+            ssh_config=self.SHORT_LIVED_CERT_SSH_CONFIG
+        ) as session:
+            username = self.get_command_output(session, "whoami")
+            self.assertEqual(username.lower(), self.SSH_USER.lower())
+
+
+unittest.main()

sshlog/empty_manager.go

@@ -0,0 +1,37 @@
+package sshlog
+
+import (
+	"io"
+
+	"github.com/sirupsen/logrus"
+)
+
+//empty manager implements the Manager but does nothing (for testing and to disable logging unless the logs are set)
+type emptyManager struct {
+}
+
+type emptyWriteCloser struct {
+}
+
+// NewEmptyManager creates a new instance of a log empty log manager that does nothing
+func NewEmptyManager() Manager {
+	return &emptyManager{}
+}
+
+func (m *emptyManager) NewLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
+	return &emptyWriteCloser{}, nil
+}
+
+func (m *emptyManager) NewSessionLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
+	return &emptyWriteCloser{}, nil
+}
+
+// emptyWriteCloser
+
+func (w *emptyWriteCloser) Write(p []byte) (n int, err error) {
+	return len(p), nil
+}
+
+func (w *emptyWriteCloser) Close() error {
+	return nil
+}

sshlog/go.capnp

@@ -0,0 +1,15 @@
+# Generate go.capnp.out with:
+# capnp compile -o- go.capnp > go.capnp.out
+# Must run inside this directory to preserve paths.
+
+@0xd12a1c51fedd6c88;
+
+annotation package(file) :Text;
+annotation import(file) :Text;
+annotation doc(struct, field, enum) :Text;
+annotation tag(enumerant) :Text;
+annotation notag(enumerant) :Void;
+annotation customtype(field) :Text;
+annotation name(struct, field, union, enum, enumerant, interface, method, param, annotation, const, group) :Text;
+
+$package("capnp");

sshlog/logger.go

@@ -2,6 +2,7 @@ package sshlog
 
 import (
 	"bufio"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
@@ -12,19 +13,22 @@ import (
 )
 
 const (
-	logTimeFormat = "2006-01-02T15-04-05.000"
-	megabyte      = 1024 * 1024
+	logTimeFormat        = "2006-01-02T15-04-05.000"
+	megabyte             = 1024 * 1024
+	defaultFileSizeLimit = 100 * megabyte
 )
 
 // Logger will buffer and write events to disk
 type Logger struct {
 	sync.Mutex
-	filename    string
-	file        *os.File
-	writeBuffer *bufio.Writer
-	logger      *logrus.Logger
-	done        chan struct{}
-	once        sync.Once
+	filename      string
+	file          *os.File
+	writeBuffer   *bufio.Writer
+	logger        *logrus.Logger
+	flushInterval time.Duration
+	maxFileSize   int64
+	done          chan struct{}
+	once          sync.Once
 }
 
 // NewLogger creates a Logger instance. A buffer is created that needs to be
@@ -34,16 +38,23 @@ type Logger struct {
 // logger variable is a logrus that will log all i/o, filesystem error etc, that
 // that shouldn't end execution of the logger, but are useful to report to the
 // caller.
-func NewLogger(filename string, logger *logrus.Logger) (*Logger, error) {
+func NewLogger(filename string, logger *logrus.Logger, flushInterval time.Duration, maxFileSize int64) (*Logger, error) {
+	if logger == nil {
+		return nil, errors.New("logger can't be nil")
+	}
 	f, err := os.OpenFile(filename, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, os.FileMode(0600))
 	if err != nil {
 		return nil, err
 	}
 	l := &Logger{filename: filename,
-		file:        f,
-		writeBuffer: bufio.NewWriter(f),
-		logger:      logger,
-		done:        make(chan struct{})}
+		file:          f,
+		writeBuffer:   bufio.NewWriter(f),
+		logger:        logger,
+		flushInterval: flushInterval,
+		maxFileSize:   maxFileSize,
+		done:          make(chan struct{}),
+	}
+
 	go l.writer()
 	return l, nil
 }
@@ -70,7 +81,7 @@ func (l *Logger) Close() error {
 // writer is the run loop that handles draining the write buffer and syncing
 // data to disk.
 func (l *Logger) writer() {
-	ticker := time.NewTicker(time.Second)
+	ticker := time.NewTicker(l.flushInterval)
 	defer ticker.Stop()
 	for {
 		select {
@@ -118,7 +129,7 @@ func (l *Logger) shouldRotate() bool {
 		return false
 	}
 
-	return info.Size() >= 100*megabyte
+	return info.Size() >= l.maxFileSize
 }
 
 // rotate creates a new logfile with the existing filename and renames the

sshlog/logger_test.go

@@ -0,0 +1,90 @@
+package sshlog
+
+import (
+	"io/ioutil"
+	"log"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+)
+
+const logFileName = "test-logger.log"
+
+func createLogger(t *testing.T) *Logger {
+	os.Remove(logFileName)
+	l := logrus.New()
+	logger, err := NewLogger(logFileName, l, time.Millisecond, 1024)
+	if err != nil {
+		t.Fatal("couldn't create the logger!", err)
+	}
+	return logger
+}
+
+func TestWrite(t *testing.T) {
+	testStr := "hi"
+	logger := createLogger(t)
+	defer func() {
+		logger.Close()
+		os.Remove(logFileName)
+	}()
+
+	logger.Write([]byte(testStr))
+	time.Sleep(2 * time.Millisecond)
+	data, err := ioutil.ReadFile(logFileName)
+	if err != nil {
+		t.Fatal("couldn't read the log file!", err)
+	}
+	checkStr := string(data)
+	if checkStr != testStr {
+		t.Fatal("file data doesn't match!")
+	}
+}
+
+func TestFilenameRotation(t *testing.T) {
+	newName := rotationName("dir/bob/acoolloggername.log")
+
+	dir := filepath.Dir(newName)
+	if dir != "dir/bob" {
+		t.Fatal("rotation name doesn't respect the directory filepath:", newName)
+	}
+
+	filename := filepath.Base(newName)
+	if !strings.HasPrefix(filename, "acoolloggername") {
+		t.Fatal("rotation filename is wrong:", filename)
+	}
+
+	ext := filepath.Ext(newName)
+	if ext != ".log" {
+		t.Fatal("rotation file extension is wrong:", ext)
+	}
+}
+
+func TestRotation(t *testing.T) {
+	logger := createLogger(t)
+
+	for i := 0; i < 2000; i++ {
+		logger.Write([]byte("a string for testing rotation\n"))
+	}
+	logger.Close()
+
+	count := 0
+	filepath.Walk(".", func(path string, info os.FileInfo, err error) error {
+		if err != nil || info.IsDir() {
+			return nil
+		}
+		if strings.HasPrefix(info.Name(), "test-logger") {
+			log.Println("deleting: ", path)
+			os.Remove(path)
+			count++
+		}
+		return nil
+	})
+	if count < 2 {
+		t.Fatal("rotation didn't roll files:", count)
+	}
+
+}

sshlog/manager.go

@@ -2,6 +2,8 @@ package sshlog
 
 import (
 	"io"
+	"path/filepath"
+	"time"
 
 	"github.com/sirupsen/logrus"
 )
@@ -9,15 +11,24 @@ import (
 // Manager be managing logs bruh
 type Manager interface {
 	NewLogger(string, *logrus.Logger) (io.WriteCloser, error)
+	NewSessionLogger(string, *logrus.Logger) (io.WriteCloser, error)
 }
 
-type manager struct{}
+type manager struct {
+	baseDirectory string
+}
 
 // New creates a new instance of a log manager
-func New() Manager {
-	return &manager{}
+func New(baseDirectory string) Manager {
+	return &manager{
+		baseDirectory: baseDirectory,
+	}
 }
 
 func (m *manager) NewLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
-	return NewLogger(name, logger)
+	return NewLogger(filepath.Join(m.baseDirectory, name), logger, time.Second, defaultFileSizeLimit)
+}
+
+func (m *manager) NewSessionLogger(name string, logger *logrus.Logger) (io.WriteCloser, error) {
+	return NewSessionLogger(filepath.Join(m.baseDirectory, name), logger, time.Second, defaultFileSizeLimit)
 }

sshlog/session_log.capnp

@@ -0,0 +1,9 @@
+using Go = import "go.capnp";
+@0x8f43375162194466;
+$Go.package("sshlog");
+$Go.import("github.com/cloudflare/cloudflared/sshlog");
+
+struct SessionLog {
+    timestamp @0 :Text;
+    content @1 :Data;
+}

sshlog/session_log.capnp.go

@@ -0,0 +1,110 @@
+// Code generated by capnpc-go. DO NOT EDIT.
+
+package sshlog
+
+import (
+	capnp "zombiezen.com/go/capnproto2"
+	text "zombiezen.com/go/capnproto2/encoding/text"
+	schemas "zombiezen.com/go/capnproto2/schemas"
+)
+
+type SessionLog struct{ capnp.Struct }
+
+// SessionLog_TypeID is the unique identifier for the type SessionLog.
+const SessionLog_TypeID = 0xa13a07c504a5ab64
+
+func NewSessionLog(s *capnp.Segment) (SessionLog, error) {
+	st, err := capnp.NewStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2})
+	return SessionLog{st}, err
+}
+
+func NewRootSessionLog(s *capnp.Segment) (SessionLog, error) {
+	st, err := capnp.NewRootStruct(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2})
+	return SessionLog{st}, err
+}
+
+func ReadRootSessionLog(msg *capnp.Message) (SessionLog, error) {
+	root, err := msg.RootPtr()
+	return SessionLog{root.Struct()}, err
+}
+
+func (s SessionLog) String() string {
+	str, _ := text.Marshal(0xa13a07c504a5ab64, s.Struct)
+	return str
+}
+
+func (s SessionLog) Timestamp() (string, error) {
+	p, err := s.Struct.Ptr(0)
+	return p.Text(), err
+}
+
+func (s SessionLog) HasTimestamp() bool {
+	p, err := s.Struct.Ptr(0)
+	return p.IsValid() || err != nil
+}
+
+func (s SessionLog) TimestampBytes() ([]byte, error) {
+	p, err := s.Struct.Ptr(0)
+	return p.TextBytes(), err
+}
+
+func (s SessionLog) SetTimestamp(v string) error {
+	return s.Struct.SetText(0, v)
+}
+
+func (s SessionLog) Content() ([]byte, error) {
+	p, err := s.Struct.Ptr(1)
+	return []byte(p.Data()), err
+}
+
+func (s SessionLog) HasContent() bool {
+	p, err := s.Struct.Ptr(1)
+	return p.IsValid() || err != nil
+}
+
+func (s SessionLog) SetContent(v []byte) error {
+	return s.Struct.SetData(1, v)
+}
+
+// SessionLog_List is a list of SessionLog.
+type SessionLog_List struct{ capnp.List }
+
+// NewSessionLog creates a new list of SessionLog.
+func NewSessionLog_List(s *capnp.Segment, sz int32) (SessionLog_List, error) {
+	l, err := capnp.NewCompositeList(s, capnp.ObjectSize{DataSize: 0, PointerCount: 2}, sz)
+	return SessionLog_List{l}, err
+}
+
+func (s SessionLog_List) At(i int) SessionLog { return SessionLog{s.List.Struct(i)} }
+
+func (s SessionLog_List) Set(i int, v SessionLog) error { return s.List.SetStruct(i, v.Struct) }
+
+func (s SessionLog_List) String() string {
+	str, _ := text.MarshalList(0xa13a07c504a5ab64, s.List)
+	return str
+}
+
+// SessionLog_Promise is a wrapper for a SessionLog promised by a client call.
+type SessionLog_Promise struct{ *capnp.Pipeline }
+
+func (p SessionLog_Promise) Struct() (SessionLog, error) {
+	s, err := p.Pipeline.Struct()
+	return SessionLog{s}, err
+}
+
+const schema_8f43375162194466 = "x\xda\x120q`\x12d\x8dg`\x08dae\xfb" +
+	"\x9f\xb2z)\xcbQv\xab\x85\x0c\x82B\x8c\xff\xd3\\" +
+	"$\x93\x02\xcd\x9d\xfb\x19X\x99\xd8\x19\x18\x04E_\x09" +
+	"*\x82h\xd9r\x06\xc6\xff\xc5\xa9\xc5\xc5\x99\xf9y\xf1" +
+	"L9\xf9\xe9z\xc9\x89\x05y\x05V\xc1`!\xfe<" +
+	"\x9f\xfc\xf4\x00F\xc6@\x0ef\x16\x06\x06\x16F\x06\x06" +
+	"A\xcd \x06\x86@\x0df\xc6@\x13&FAFF" +
+	"\x11F\x90\xa0\xa1\x13\x03C\xa0\x0e3c\xa0\x05\x13\xe3" +
+	"\xff\x92\xcc\xdc\xd4\xe2\x92\xc4\\\x06\xc6\x02F\x1e\x06&" +
+	"F\x1e\x06\xc6\xfa\xe4\xfc\xbc\x92\xd4\xbc\x12F^\x06&" +
+	"F^\x06F@\x00\x00\x00\xff\xff\xdaK$\x1a"
+
+func init() {
+	schemas.Register(schema_8f43375162194466,
+		0xa13a07c504a5ab64)
+}

sshlog/session_logger.go

@@ -0,0 +1,71 @@
+package sshlog
+
+import (
+	"time"
+
+	"github.com/sirupsen/logrus"
+	capnp "zombiezen.com/go/capnproto2"
+	"zombiezen.com/go/capnproto2/pogs"
+)
+
+// SessionLogger will buffer and write events to disk using capnp proto for session replay
+type SessionLogger struct {
+	logger  *Logger
+	encoder *capnp.Encoder
+}
+
+type sessionLogData struct {
+	Timestamp string // The UTC timestamp of when the log occurred
+	Content   []byte // The shell output
+}
+
+// NewSessionLogger creates a new session logger by encapsulating a Logger object and writing capnp encoded messages to it
+func NewSessionLogger(filename string, logger *logrus.Logger, flushInterval time.Duration, maxFileSize int64) (*SessionLogger, error) {
+	l, err := NewLogger(filename, logger, flushInterval, maxFileSize)
+	if err != nil {
+		return nil, err
+	}
+	sessionLogger := &SessionLogger{
+		logger:  l,
+		encoder: capnp.NewEncoder(l),
+	}
+	return sessionLogger, nil
+}
+
+// Writes to a log buffer. Implements the io.Writer interface.
+func (l *SessionLogger) Write(p []byte) (n int, err error) {
+	return l.writeSessionLog(&sessionLogData{
+		Timestamp: time.Now().UTC().Format(time.RFC3339),
+		Content:   p,
+	})
+}
+
+// Close drains anything left in the buffer and cleans up any resources still
+// in use.
+func (l *SessionLogger) Close() error {
+	return l.logger.Close()
+}
+
+func (l *SessionLogger) writeSessionLog(p *sessionLogData) (int, error) {
+	msg, seg, err := capnp.NewMessage(capnp.SingleSegment(nil))
+	if err != nil {
+		return 0, err
+	}
+	log, err := NewRootSessionLog(seg)
+	if err != nil {
+		return 0, err
+	}
+	log.SetTimestamp(p.Timestamp)
+	log.SetContent(p.Content)
+
+	if err := l.encoder.Encode(msg); err != nil {
+		return 0, err
+	}
+	return len(p.Content), nil
+}
+
+func unmarshalSessionLog(s SessionLog) (*sessionLogData, error) {
+	p := new(sessionLogData)
+	err := pogs.Extract(p, SessionLog_TypeID, s.Struct)
+	return p, err
+}

sshlog/session_logger_test.go

@@ -0,0 +1,69 @@
+package sshlog
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	capnp "zombiezen.com/go/capnproto2"
+)
+
+const sessionLogFileName = "test-session-logger.log"
+
+func createSessionLogger(t *testing.T) *SessionLogger {
+	os.Remove(sessionLogFileName)
+	l := logrus.New()
+	logger, err := NewSessionLogger(sessionLogFileName, l, time.Millisecond, 1024)
+	if err != nil {
+		t.Fatal("couldn't create the logger!", err)
+	}
+	return logger
+}
+
+func TestSessionLogWrite(t *testing.T) {
+	testStr := "hi"
+	logger := createSessionLogger(t)
+	defer func() {
+		logger.Close()
+		os.Remove(sessionLogFileName)
+	}()
+
+	logger.Write([]byte(testStr))
+	time.Sleep(2 * time.Millisecond)
+	f, err := os.Open(sessionLogFileName)
+	if err != nil {
+		t.Fatal("couldn't read the log file!", err)
+	}
+	defer f.Close()
+
+	msg, err := capnp.NewDecoder(f).Decode()
+	if err != nil {
+		t.Fatal("couldn't read the capnp msg file!", err)
+	}
+
+	sessionLog, err := ReadRootSessionLog(msg)
+	if err != nil {
+		t.Fatal("couldn't read the session log from the msg!", err)
+	}
+
+	timeStr, err := sessionLog.Timestamp()
+	if err != nil {
+		t.Fatal("couldn't read the Timestamp field!", err)
+	}
+
+	_, terr := time.Parse(time.RFC3339, timeStr)
+	if terr != nil {
+		t.Fatal("couldn't parse the Timestamp into the expected RFC3339 format", terr)
+	}
+
+	data, err := sessionLog.Content()
+	if err != nil {
+		t.Fatal("couldn't read the Content field!", err)
+	}
+
+	checkStr := string(data)
+	if checkStr != testStr {
+		t.Fatal("file data doesn't match!")
+	}
+}

sshserver/authentication.go

@@ -27,7 +27,15 @@ func (s *SSHServer) configureAuthentication() {
 	s.PublicKeyHandler = s.authenticationHandler
 }
 
+// authenticationHandler is a callback that returns true if the user attempting to connect is authenticated.
 func (s *SSHServer) authenticationHandler(ctx ssh.Context, key ssh.PublicKey) bool {
+	sshUser, err := lookupUser(ctx.User())
+	if err != nil {
+		s.logger.Debugf("Invalid user: %s", ctx.User())
+		return false
+	}
+	ctx.SetValue("sshUser", sshUser)
+
 	cert, ok := key.(*gossh.Certificate)
 	if !ok {
 		return s.authorizedKeyHandler(ctx, key)
@@ -36,9 +44,9 @@ func (s *SSHServer) authenticationHandler(ctx ssh.Context, key ssh.PublicKey) bo
 }
 
 func (s *SSHServer) authorizedKeyHandler(ctx ssh.Context, key ssh.PublicKey) bool {
-	sshUser, err := s.getUserFunc(ctx.User())
-	if err != nil {
-		s.logger.Debugf("Invalid user: %s", ctx.User())
+	sshUser, ok := ctx.Value("sshUser").(*User)
+	if !ok {
+		s.logger.Error("Failed to retrieve user from context")
 		return false
 	}
 
@@ -55,7 +63,6 @@ func (s *SSHServer) authorizedKeyHandler(ctx ssh.Context, key ssh.PublicKey) boo
 	}
 
 	for len(authorizedKeysBytes) > 0 {
-
 		// Skips invalid keys. Returns error if no valid keys remain.
 		pubKey, _, _, rest, err := ssh.ParseAuthorizedKey(authorizedKeysBytes)
 		authorizedKeysBytes = rest
@@ -65,7 +72,6 @@ func (s *SSHServer) authorizedKeyHandler(ctx ssh.Context, key ssh.PublicKey) boo
 		}
 
 		if ssh.KeysEqual(pubKey, key) {
-			ctx.SetValue("sshUser", sshUser)
 			return true
 		}
 	}
@@ -83,13 +89,6 @@ func (s *SSHServer) shortLivedCertHandler(ctx ssh.Context, cert *gossh.Certifica
 	if err := checker.CheckCert(ctx.User(), cert); err != nil {
 		s.logger.Debug(err)
 		return false
-	} else {
-		sshUser, err := s.getUserFunc(ctx.User())
-		if err != nil {
-			s.logger.Debugf("Invalid user: %s", ctx.User())
-			return false
-		}
-		ctx.SetValue("sshUser", sshUser)
 	}
 	return true
 }
@@ -98,7 +97,7 @@ func getCACert() (ssh.PublicKey, error) {
 	caCertPath := path.Join(systemConfigPath, "ca.pub")
 	caCertBytes, err := ioutil.ReadFile(caCertPath)
 	if err != nil {
-		return nil, errors.Wrap(err, fmt.Sprintf("Failed to load CA certertificate %s", caCertPath))
+		return nil, errors.Wrap(err, fmt.Sprintf("Failed to load CA certificate %s", caCertPath))
 	}
 	caCert, _, _, _, err := ssh.ParseAuthorizedKey(caCertBytes)
 	if err != nil {

sshserver/authentication_test.go

@@ -22,7 +22,6 @@ import (
 )
 
 const (
-	validPrincipal       = "testUser"
 	testDir              = "testdata"
 	testUserKeyFilename  = "id_rsa.pub"
 	testCAFilename       = "ca.pub"
@@ -30,7 +29,10 @@ const (
 	testUserCertFilename = "id_rsa-cert.pub"
 )
 
-var logger, hook = test.NewNullLogger()
+var (
+	logger, hook = test.NewNullLogger()
+	mockUser     = &User{Username: "testUser", HomeDir: testDir}
+)
 
 func TestMain(m *testing.M) {
 	authorizedKeysDir = testUserKeyFilename
@@ -40,20 +42,20 @@ func TestMain(m *testing.M) {
 }
 
 func TestPublicKeyAuth_Success(t *testing.T) {
-	context, cancel := newMockContext(validPrincipal)
+	context, cancel := newMockContext(mockUser)
 	defer cancel()
 
-	sshServer := SSHServer{getUserFunc: getMockUser}
+	sshServer := SSHServer{logger: logger}
 
 	pubKey := getKey(t, testUserKeyFilename)
 	assert.True(t, sshServer.authorizedKeyHandler(context, pubKey))
 }
 
 func TestPublicKeyAuth_MissingKey(t *testing.T) {
-	context, cancel := newMockContext(validPrincipal)
+	context, cancel := newMockContext(mockUser)
 	defer cancel()
 
-	sshServer := SSHServer{logger: logger, getUserFunc: getMockUser}
+	sshServer := SSHServer{logger: logger}
 
 	pubKey := getKey(t, testOtherCAFilename)
 	assert.False(t, sshServer.authorizedKeyHandler(context, pubKey))
@@ -61,23 +63,27 @@ func TestPublicKeyAuth_MissingKey(t *testing.T) {
 }
 
 func TestPublicKeyAuth_InvalidUser(t *testing.T) {
-	context, cancel := newMockContext("notAUser")
+	context, cancel := newMockContext(&User{Username: "notAUser"})
 	defer cancel()
 
-	sshServer := SSHServer{logger: logger, getUserFunc: lookupUser}
+	sshServer := SSHServer{logger: logger}
 
 	pubKey := getKey(t, testUserKeyFilename)
-	assert.False(t, sshServer.authorizedKeyHandler(context, pubKey))
+	assert.False(t, sshServer.authenticationHandler(context, pubKey))
 	assert.Contains(t, hook.LastEntry().Message, "Invalid user")
 }
 
 func TestPublicKeyAuth_MissingFile(t *testing.T) {
-	currentUser, err := user.Current()
+	tempUser, err := user.Current()
 	require.Nil(t, err)
-	context, cancel := newMockContext(currentUser.Username)
+	currentUser, err := lookupUser(tempUser.Username)
+	require.Nil(t, err)
+
+	require.Nil(t, err)
+	context, cancel := newMockContext(currentUser)
 	defer cancel()
 
-	sshServer := SSHServer{Server: ssh.Server{}, logger: logger, getUserFunc: lookupUser}
+	sshServer := SSHServer{Server: ssh.Server{}, logger: logger}
 
 	pubKey := getKey(t, testUserKeyFilename)
 	assert.False(t, sshServer.authorizedKeyHandler(context, pubKey))
@@ -85,11 +91,11 @@ func TestPublicKeyAuth_MissingFile(t *testing.T) {
 }
 
 func TestShortLivedCerts_Success(t *testing.T) {
-	context, cancel := newMockContext(validPrincipal)
+	context, cancel := newMockContext(mockUser)
 	defer cancel()
 
 	caCert := getKey(t, testCAFilename)
-	sshServer := SSHServer{logger: log.CreateLogger(), caCert: caCert, getUserFunc: getMockUser}
+	sshServer := SSHServer{logger: log.CreateLogger(), caCert: caCert}
 
 	userCert, ok := getKey(t, testUserCertFilename).(*gossh.Certificate)
 	require.True(t, ok)
@@ -97,11 +103,11 @@ func TestShortLivedCerts_Success(t *testing.T) {
 }
 
 func TestShortLivedCerts_CAsDontMatch(t *testing.T) {
-	context, cancel := newMockContext(validPrincipal)
+	context, cancel := newMockContext(mockUser)
 	defer cancel()
 
 	caCert := getKey(t, testOtherCAFilename)
-	sshServer := SSHServer{logger: logger, caCert: caCert, getUserFunc: getMockUser}
+	sshServer := SSHServer{logger: logger, caCert: caCert}
 
 	userCert, ok := getKey(t, testUserCertFilename).(*gossh.Certificate)
 	require.True(t, ok)
@@ -109,25 +115,12 @@ func TestShortLivedCerts_CAsDontMatch(t *testing.T) {
 	assert.Equal(t, "CA certificate does not match user certificate signer", hook.LastEntry().Message)
 }
 
-func TestShortLivedCerts_UserDoesNotExist(t *testing.T) {
-	context, cancel := newMockContext(validPrincipal)
-	defer cancel()
-
-	caCert := getKey(t, testCAFilename)
-	sshServer := SSHServer{logger: logger, caCert: caCert, getUserFunc: lookupUser}
-
-	userCert, ok := getKey(t, testUserCertFilename).(*gossh.Certificate)
-	require.True(t, ok)
-	assert.False(t, sshServer.shortLivedCertHandler(context, userCert))
-	assert.Contains(t, hook.LastEntry().Message, "Invalid user")
-}
-
 func TestShortLivedCerts_InvalidPrincipal(t *testing.T) {
-	context, cancel := newMockContext("notAUser")
+	context, cancel := newMockContext(&User{Username: "NotAUser"})
 	defer cancel()
 
 	caCert := getKey(t, testCAFilename)
-	sshServer := SSHServer{logger: logger, caCert: caCert, getUserFunc: lookupUser}
+	sshServer := SSHServer{logger: logger, caCert: caCert}
 
 	userCert, ok := getKey(t, testUserCertFilename).(*gossh.Certificate)
 	require.True(t, ok)
@@ -135,14 +128,6 @@ func TestShortLivedCerts_InvalidPrincipal(t *testing.T) {
 	assert.Contains(t, hook.LastEntry().Message, "not in the set of valid principals for given certificate")
 }
 
-func getMockUser(_ string) (*User, error) {
-	return &User{
-		Username: validPrincipal,
-		HomeDir:  testDir,
-	}, nil
-
-}
-
 func getKey(t *testing.T, filename string) ssh.PublicKey {
 	path := path.Join(testDir, filename)
 	bytes, err := ioutil.ReadFile(path)
@@ -157,10 +142,13 @@ type mockSSHContext struct {
 	*sync.Mutex
 }
 
-func newMockContext(user string) (*mockSSHContext, context.CancelFunc) {
+func newMockContext(user *User) (*mockSSHContext, context.CancelFunc) {
 	innerCtx, cancel := context.WithCancel(context.Background())
 	mockCtx := &mockSSHContext{innerCtx, &sync.Mutex{}}
-	mockCtx.SetValue("user", user)
+	mockCtx.SetValue("sshUser", user)
+
+	// This naming is confusing but we cant change it because this mocks the SSHContext struct in gliderlabs/ssh
+	mockCtx.SetValue("user", user.Username)
 	return mockCtx, cancel
 }
 

sshserver/sshserver_unix.go

@@ -3,13 +3,17 @@
 package sshserver
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"os"
 	"os/exec"
 	"os/user"
+	"runtime"
 	"strconv"
+	"strings"
 	"syscall"
 	"time"
 	"unsafe"
@@ -22,16 +26,38 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
-type SSHServer struct {
-	ssh.Server
-	logger      *logrus.Logger
-	shutdownC   chan struct{}
-	caCert      ssh.PublicKey
-	getUserFunc func(string) (*User, error)
-	logManager  sshlog.Manager
+const (
+	auditEventAuth   = "auth"
+	auditEventStart  = "session_start"
+	auditEventStop   = "session_stop"
+	auditEventExec   = "exec"
+	auditEventScp    = "scp"
+	auditEventResize = "resize"
+	sshContextSessionID = "sessionID"
+	sshContextEventLogger = "eventLogger"
+)
+
+type auditEvent struct {
+	Event     string `json:"event,omitempty"`
+	EventType string `json:"event_type,omitempty"`
+	SessionID string `json:"session_id,omitempty"`
+	User      string `json:"user,omitempty"`
+	Login     string `json:"login,omitempty"`
+	Datetime  string `json:"datetime,omitempty"`
+	IPAddress string `json:"ip_address,omitempty"`
 }
 
-func New(logManager sshlog.Manager, logger *logrus.Logger, address string, shutdownC chan struct{}, idleTimeout, maxTimeout time.Duration) (*SSHServer, error) {
+// SSHServer adds on to the ssh.Server of the gliderlabs package
+type SSHServer struct {
+	ssh.Server
+	logger     *logrus.Logger
+	shutdownC  chan struct{}
+	caCert     ssh.PublicKey
+	logManager sshlog.Manager
+}
+
+// New creates a new SSHServer and configures its host keys and authenication by the data provided
+func New(logManager sshlog.Manager, logger *logrus.Logger, version, address string, shutdownC chan struct{}, idleTimeout, maxTimeout time.Duration, enablePortForwarding bool) (*SSHServer, error) {
 	currentUser, err := user.Current()
 	if err != nil {
 		return nil, err
@@ -40,12 +66,39 @@ func New(logManager sshlog.Manager, logger *logrus.Logger, address string, shutd
 		return nil, errors.New("cloudflared SSH server needs to run as root")
 	}
 
+	forwardHandler := &ssh.ForwardedTCPHandler{}
 	sshServer := SSHServer{
-		Server:      ssh.Server{Addr: address, MaxTimeout: maxTimeout, IdleTimeout: idleTimeout},
-		logger:      logger,
-		shutdownC:   shutdownC,
-		getUserFunc: lookupUser,
-		logManager:  logManager,
+		Server: ssh.Server{
+			Addr:        address,
+			MaxTimeout:  maxTimeout,
+			IdleTimeout: idleTimeout,
+			Version:     fmt.Sprintf("SSH-2.0-Cloudflare-Access_%s_%s", version, runtime.GOOS),
+			// Register SSH global Request handlers to respond to tcpip forwarding
+			RequestHandlers: map[string]ssh.RequestHandler{
+				"tcpip-forward":        forwardHandler.HandleSSHRequest,
+				"cancel-tcpip-forward": forwardHandler.HandleSSHRequest,
+			},
+			// Register SSH channel types
+			ChannelHandlers: map[string]ssh.ChannelHandler{
+				"session":      ssh.DefaultSessionHandler,
+				"direct-tcpip": ssh.DirectTCPIPHandler,
+			},
+		},
+		logger:     logger,
+		shutdownC:  shutdownC,
+		logManager: logManager,
+	}
+
+	// AUTH-2050: This is a temporary workaround of a timing issue in the tunnel muxer to allow further testing.
+	// TODO: Remove this
+	sshServer.ConnCallback = func(conn net.Conn) net.Conn {
+		time.Sleep(10 * time.Millisecond)
+		return conn
+	}
+
+	if enablePortForwarding {
+		sshServer.LocalPortForwardingCallback = allowForward
+		sshServer.ReversePortForwardingCallback = allowForward
 	}
 
 	if err := sshServer.configureHostKeys(); err != nil {
@@ -53,9 +106,11 @@ func New(logManager sshlog.Manager, logger *logrus.Logger, address string, shutd
 	}
 
 	sshServer.configureAuthentication()
+
 	return &sshServer, nil
 }
 
+// Start the SSH server listener to start handling SSH connections from clients
 func (s *SSHServer) Start() error {
 	s.logger.Infof("Starting SSH server at %s", s.Addr)
 
@@ -71,32 +126,38 @@ func (s *SSHServer) Start() error {
 }
 
 func (s *SSHServer) connectionHandler(session ssh.Session) {
-	sessionID, err := uuid.NewRandom()
+	sessionUUID, err := uuid.NewRandom()
+
 	if err != nil {
 		if _, err := io.WriteString(session, "Failed to generate session ID\n"); err != nil {
 			s.logger.WithError(err).Error("Failed to generate session ID: Failed to write to SSH session")
 		}
-		s.CloseSession(session)
+		s.errorAndExit(session, "", nil)
+		return
 	}
+	sessionID := sessionUUID.String()
+
+	eventLogger, err := s.logManager.NewLogger(fmt.Sprintf("%s-event.log", sessionID), s.logger)
+	if err != nil {
+		if _, err := io.WriteString(session, "Failed to create event log\n"); err != nil {
+			s.logger.WithError(err).Error("Failed to create event log: Failed to write to create event logger")
+		}
+		s.errorAndExit(session, "", nil)
+		return
+	}
+
+	sshContext, ok := session.Context().(ssh.Context)
+	if !ok {
+		s.logger.Error("Could not retrieve session context")
+		s.errorAndExit(session, "", nil)
+	}
+
+	sshContext.SetValue(sshContextSessionID, sessionID)
+	sshContext.SetValue(sshContextEventLogger, eventLogger)
 
 	// Get uid and gid of user attempting to login
-	sshUser, ok := session.Context().Value("sshUser").(*User)
-	if !ok || sshUser == nil {
-		s.logger.Error("Error retrieving credentials from session")
-		s.CloseSession(session)
-		return
-	}
-
-	uidInt, err := stringToUint32(sshUser.Uid)
-	if err != nil {
-		s.logger.WithError(err).Error("Invalid user")
-		s.CloseSession(session)
-		return
-	}
-	gidInt, err := stringToUint32(sshUser.Gid)
-	if err != nil {
-		s.logger.WithError(err).Error("Invalid user group")
-		s.CloseSession(session)
+	sshUser, uidInt, gidInt, success := s.getSSHUser(session, eventLogger)
+	if !success {
 		return
 	}
 
@@ -104,8 +165,16 @@ func (s *SSHServer) connectionHandler(session ssh.Session) {
 	var cmd *exec.Cmd
 	if session.RawCommand() != "" {
 		cmd = exec.Command(sshUser.Shell, "-c", session.RawCommand())
+
+		event := auditEventExec
+		if strings.HasPrefix(session.RawCommand(), "scp") {
+			event = auditEventScp
+		}
+		s.logAuditEvent(session, event)
 	} else {
 		cmd = exec.Command(sshUser.Shell)
+		s.logAuditEvent(session, auditEventStart)
+		defer s.logAuditEvent(session, auditEventStop)
 	}
 	// Supplementary groups are not explicitly specified. They seem to be inherited by default.
 	cmd.SysProcAttr = &syscall.SysProcAttr{Credential: &syscall.Credential{Uid: uidInt, Gid: gidInt}, Setsid: true}
@@ -114,54 +183,81 @@ func (s *SSHServer) connectionHandler(session ssh.Session) {
 	cmd.Env = append(cmd.Env, fmt.Sprintf("HOME=%s", sshUser.HomeDir))
 	cmd.Dir = sshUser.HomeDir
 
-	ptyReq, winCh, isPty := session.Pty()
 	var shellInput io.WriteCloser
 	var shellOutput io.ReadCloser
+	pr, pw := io.Pipe()
+	defer pw.Close()
+
+	ptyReq, winCh, isPty := session.Pty()
 
 	if isPty {
 		cmd.Env = append(cmd.Env, fmt.Sprintf("TERM=%s", ptyReq.Term))
-		shellInput, shellOutput, err = s.startPtySession(cmd, winCh)
+		tty, err := s.startPtySession(cmd, winCh, func() {
+			s.logAuditEvent(session, auditEventResize)
+		})
+		shellInput = tty
+		shellOutput = tty
 		if err != nil {
 			s.logger.WithError(err).Error("Failed to start pty session")
 			close(s.shutdownC)
 			return
 		}
 	} else {
-		shellInput, shellOutput, err = s.startNonPtySession(cmd)
+		var shellError io.ReadCloser
+		shellInput, shellOutput, shellError, err = s.startNonPtySession(cmd)
 		if err != nil {
 			s.logger.WithError(err).Error("Failed to start non-pty session")
 			close(s.shutdownC)
 			return
 		}
+
+		// Write stderr to both the command recorder, and remote user
+		go func() {
+			mw := io.MultiWriter(pw, session.Stderr())
+			if _, err := io.Copy(mw, shellError); err != nil {
+				s.logger.WithError(err).Error("Failed to write stderr to user")
+			}
+		}()
 	}
 
-	// Write incoming commands to shell
+	sessionLogger, err := s.logManager.NewSessionLogger(fmt.Sprintf("%s-session.log", sessionID), s.logger)
+	if err != nil {
+		if _, err := io.WriteString(session, "Failed to create log\n"); err != nil {
+			s.logger.WithError(err).Error("Failed to create log: Failed to write to SSH session")
+		}
+		s.errorAndExit(session, "", nil)
+		return
+	}
 	go func() {
+		defer sessionLogger.Close()
+		defer pr.Close()
+		_, err := io.Copy(sessionLogger, pr)
+		if err != nil {
+			s.logger.WithError(err).Error("Failed to write session log")
+		}
+	}()
+
+	// Write stdin to shell
+	go func() {
+
+		/*
+			Only close shell stdin for non-pty sessions because they have distinct stdin, stdout, and stderr.
+			This is done to prevent commands like SCP from hanging after all data has been sent.
+			PTY sessions share one file for all three streams and the shell process closes it.
+			Closing it here also closes shellOutput and causes an error on copy().
+		*/
+		if !isPty {
+			defer shellInput.Close()
+		}
 		if _, err := io.Copy(shellInput, session); err != nil {
 			s.logger.WithError(err).Error("Failed to write incoming command to pty")
 		}
 	}()
 
-	pr, pw := io.Pipe()
-	defer pr.Close()
-	defer pw.Close()
-
-	logger, err := s.logManager.NewLogger(fmt.Sprintf("%s-session.log", sessionID), s.logger)
-	if err != nil {
-		if _, err := io.WriteString(session, "Failed to create log\n"); err != nil {
-			s.logger.WithError(err).Error("Failed to create log: Failed to write to SSH session")
-		}
-		s.CloseSession(session)
-	}
-	defer logger.Close()
-	go func() {
-		io.Copy(logger, pr)
-	}()
-
-	// Write outgoing command output to both the command recorder, and remote user
+	// Write stdout to both the command recorder, and remote user
 	mw := io.MultiWriter(pw, session)
 	if _, err := io.Copy(mw, shellOutput); err != nil {
-		s.logger.WithError(err).Error("Failed to write command output to user")
+		s.logger.WithError(err).Error("Failed to write stdout to user")
 	}
 
 	// Wait for all resources associated with cmd to be released
@@ -171,32 +267,65 @@ func (s *SSHServer) connectionHandler(session ssh.Session) {
 	}
 }
 
-func (s *SSHServer) CloseSession(session ssh.Session) {
-	if err := session.Exit(1); err != nil {
-		s.logger.WithError(err).Error("Failed to close SSH session")
+// getSSHUser gets the ssh user, uid, and gid of the user attempting to login
+func (s *SSHServer) getSSHUser(session ssh.Session, eventLogger io.WriteCloser) (*User, uint32, uint32, bool) {
+	// Get uid and gid of user attempting to login
+	sshUser, ok := session.Context().Value("sshUser").(*User)
+	if !ok || sshUser == nil {
+		s.errorAndExit(session, "Error retrieving credentials from session", nil)
+		return nil, 0, 0, false
+	}
+	s.logAuditEvent(session, auditEventAuth)
+
+	uidInt, err := stringToUint32(sshUser.Uid)
+	if err != nil {
+		s.errorAndExit(session, "Invalid user", err)
+		return sshUser, 0, 0, false
+	}
+	gidInt, err := stringToUint32(sshUser.Gid)
+	if err != nil {
+		s.errorAndExit(session, "Invalid user group", err)
+		return sshUser, 0, 0, false
+	}
+	return sshUser, uidInt, gidInt, true
+}
+
+// errorAndExit reports an error with the session and exits
+func (s *SSHServer) errorAndExit(session ssh.Session, errText string, err error) {
+	if exitError := session.Exit(1); exitError != nil {
+		s.logger.WithError(exitError).Error("Failed to close SSH session")
+	} else if err != nil {
+		s.logger.WithError(err).Error(errText)
+	} else if errText != "" {
+		s.logger.Error(errText)
 	}
 }
 
-func (s *SSHServer) startNonPtySession(cmd *exec.Cmd) (io.WriteCloser, io.ReadCloser, error) {
-	in, err := cmd.StdinPipe()
+func (s *SSHServer) startNonPtySession(cmd *exec.Cmd) (stdin io.WriteCloser, stdout io.ReadCloser, stderr io.ReadCloser, err error) {
+	stdin, err = cmd.StdinPipe()
 	if err != nil {
-		return nil, nil, err
+		return
 	}
-	out, err := cmd.StdoutPipe()
+	stdout, err = cmd.StdoutPipe()
 	if err != nil {
-		return nil, nil, err
+		return
 	}
-	cmd.Stderr = cmd.Stdout
+
+	stderr, err = cmd.StderrPipe()
+	if err != nil {
+		return
+	}
+
 	if err = cmd.Start(); err != nil {
-		return nil, nil, err
+		return
 	}
-	return in, out, nil
+	return
 }
 
-func (s *SSHServer) startPtySession(cmd *exec.Cmd, winCh <-chan ssh.Window) (io.WriteCloser, io.ReadCloser, error) {
+func (s *SSHServer) startPtySession(cmd *exec.Cmd, winCh <-chan ssh.Window, logCallback func()) (io.ReadWriteCloser, error) {
 	tty, err := pty.Start(cmd)
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 
 	// Handle terminal window size changes
@@ -207,10 +336,50 @@ func (s *SSHServer) startPtySession(cmd *exec.Cmd, winCh <-chan ssh.Window) (io.
 				close(s.shutdownC)
 				return
 			}
+			logCallback()
 		}
 	}()
 
-	return tty, tty, nil
+	return tty, nil
+}
+
+func (s *SSHServer) logAuditEvent(session ssh.Session, eventType string) {
+	username := "unknown"
+	sshUser, ok := session.Context().Value("sshUser").(*User)
+	if ok && sshUser != nil {
+		username = sshUser.Username
+	}
+
+	sessionID, ok := session.Context().Value(sshContextSessionID).(string)
+	if !ok {
+		s.logger.Error("Failed to retrieve sessionID from context")
+		return
+	}
+	writer, ok := session.Context().Value(sshContextEventLogger).(io.WriteCloser)
+	if !ok {
+		s.logger.Error("Failed to retrieve eventLogger from context")
+		return
+	}
+
+	event := auditEvent{
+		Event:     session.RawCommand(),
+		EventType: eventType,
+		SessionID: sessionID,
+		User:      username,
+		Login:     username,
+		Datetime:  time.Now().UTC().Format(time.RFC3339),
+		IPAddress: session.RemoteAddr().String(),
+	}
+	data, err := json.Marshal(&event)
+	if err != nil {
+		s.logger.WithError(err).Error("Failed to marshal audit event. malformed audit object")
+		return
+	}
+	line := string(data) + "\n"
+	if _, err := writer.Write([]byte(line)); err != nil {
+		s.logger.WithError(err).Error("Failed to write audit event.")
+	}
+
 }
 
 // Sets PTY window size for terminal
@@ -223,5 +392,8 @@ func setWinsize(f *os.File, w, h int) syscall.Errno {
 func stringToUint32(str string) (uint32, error) {
 	uid, err := strconv.ParseUint(str, 10, 32)
 	return uint32(uid), err
-
+}
+
+func allowForward(_ ssh.Context, _ string, _ uint32) bool {
+	return true
 }

sshserver/sshserver_windows.go

@@ -13,7 +13,7 @@ import (
 
 type SSHServer struct{}
 
-func New(_ sshlog.Manager, _ *logrus.Logger, _ string, _ chan struct{}, _, _ time.Duration) (*SSHServer, error) {
+func New(_ sshlog.Manager, _ *logrus.Logger, _, _ string, _ chan struct{}, _, _ time.Duration, _ bool) (*SSHServer, error) {
 	return nil, errors.New("cloudflared ssh server is not supported on windows")
 }
 

streamhandler/request.go

@@ -45,16 +45,11 @@ func H2RequestHeadersToH1Request(h2 []h2mux.Header, h1 *http.Request) error {
 			// Otherwise the host header will be based on the origin URL
 			h1.Host = header.Value
 		case ":path":
-			u, err := url.Parse(header.Value)
-			if err != nil {
-				return fmt.Errorf("unparseable path")
-			}
-			resolved := h1.URL.ResolveReference(u)
-			// prevent escaping base URL
-			if !strings.HasPrefix(resolved.String(), h1.URL.String()) {
-				return fmt.Errorf("invalid path")
-			}
-			h1.URL = resolved
+			// We can't just set `URL.Path`, because there's no way to ask the library to *not* escape it,
+			// causing https://github.com/cloudflare/cloudflared/issues/124.
+			// The only way to bypass the URL escape seems to currently be `URL.Opaque`.
+			// See https://github.com/golang/go/issues/5777
+			h1.URL.Opaque = fmt.Sprintf("//%v%v", h1.URL.Host, header.Value)
 		case "content-length":
 			contentLength, err := strconv.ParseInt(header.Value, 10, 64)
 			if err != nil {

streamhandler/request_test.go

@@ -0,0 +1,192 @@
+package streamhandler
+
+import (
+	"fmt"
+	"net/http"
+	"testing"
+
+	"github.com/cloudflare/cloudflared/h2mux"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestH2RequestHeadersToH1Request_RegularHeaders(t *testing.T) {
+	request, err := http.NewRequest(http.MethodGet, "http://example.com", nil)
+	assert.NoError(t, err)
+
+	headersConversionErr := H2RequestHeadersToH1Request(
+		[]h2mux.Header{
+			h2mux.Header{
+				Name:  "Mock header 1",
+				Value: "Mock value 1",
+			},
+			h2mux.Header{
+				Name:  "Mock header 2",
+				Value: "Mock value 2",
+			},
+		},
+		request,
+	)
+
+	assert.Equal(t, http.Header{
+		"Mock header 1": []string{"Mock value 1"},
+		"Mock header 2": []string{"Mock value 2"},
+	}, request.Header)
+
+	assert.NoError(t, headersConversionErr)
+}
+
+func TestH2RequestHeadersToH1Request_NoHeaders(t *testing.T) {
+	request, err := http.NewRequest(http.MethodGet, "http://example.com", nil)
+	assert.NoError(t, err)
+
+	headersConversionErr := H2RequestHeadersToH1Request(
+		[]h2mux.Header{},
+		request,
+	)
+
+	assert.Equal(t, http.Header{}, request.Header)
+
+	assert.NoError(t, headersConversionErr)
+}
+
+func TestH2RequestHeadersToH1Request_InvalidHostPath(t *testing.T) {
+	request, err := http.NewRequest(http.MethodGet, "http://example.com", nil)
+	assert.NoError(t, err)
+
+	headersConversionErr := H2RequestHeadersToH1Request(
+		[]h2mux.Header{
+			h2mux.Header{
+				Name:  ":path",
+				Value: "//bad_path/",
+			},
+			h2mux.Header{
+				Name:  "Mock header",
+				Value: "Mock value",
+			},
+		},
+		request,
+	)
+
+	assert.Equal(t, http.Header{
+		"Mock header": []string{"Mock value"},
+	}, request.Header)
+
+	assert.Equal(t, "http://example.com//bad_path/", request.URL.String())
+
+	assert.NoError(t, headersConversionErr)
+}
+
+func TestH2RequestHeadersToH1Request_HostPathWithQuery(t *testing.T) {
+	request, err := http.NewRequest(http.MethodGet, "http://example.com/", nil)
+	assert.NoError(t, err)
+
+	headersConversionErr := H2RequestHeadersToH1Request(
+		[]h2mux.Header{
+			h2mux.Header{
+				Name:  ":path",
+				Value: "/?query=mock%20value",
+			},
+			h2mux.Header{
+				Name:  "Mock header",
+				Value: "Mock value",
+			},
+		},
+		request,
+	)
+
+	assert.Equal(t, http.Header{
+		"Mock header": []string{"Mock value"},
+	}, request.Header)
+
+	assert.Equal(t, "http://example.com/?query=mock%20value", request.URL.String())
+
+	assert.NoError(t, headersConversionErr)
+}
+
+func TestH2RequestHeadersToH1Request_HostPathWithURLEncoding(t *testing.T) {
+	request, err := http.NewRequest(http.MethodGet, "http://example.com/", nil)
+	assert.NoError(t, err)
+
+	headersConversionErr := H2RequestHeadersToH1Request(
+		[]h2mux.Header{
+			h2mux.Header{
+				Name:  ":path",
+				Value: "/mock%20path",
+			},
+			h2mux.Header{
+				Name:  "Mock header",
+				Value: "Mock value",
+			},
+		},
+		request,
+	)
+
+	assert.Equal(t, http.Header{
+		"Mock header": []string{"Mock value"},
+	}, request.Header)
+
+	assert.Equal(t, "http://example.com/mock%20path", request.URL.String())
+
+	assert.NoError(t, headersConversionErr)
+}
+
+func TestH2RequestHeadersToH1Request_WeirdURLs(t *testing.T) {
+	expectedPaths := []string{
+		"",
+		"/",
+		"//",
+		"/%20",
+		"/ ",
+		"/ a ",
+		"/a%20b",
+		"/foo/bar;param?query#frag",
+		"/a␠b",
+		"/a-umlaut-ä",
+		"/a-umlaut-%C3%A4",
+		"/a-umlaut-%c3%a4",
+		"/a#b#c",
+		"/a#b␠c",
+		"/a#b%20c",
+		"/a#b c",
+		"/\\",
+		"/a\\",
+		"/a\\b",
+		"/a,b.c.",
+		"/.",
+		"/a`",
+		"/a[0]",
+		"/?a[0]=5 &b[]=",
+		"/?a=%22b%20%22",
+	}
+
+	for index, expectedPath := range expectedPaths {
+		requestURL := "https://example.com"
+		expectedURL := fmt.Sprintf("https://example.com%v", expectedPath)
+
+		request, err := http.NewRequest(http.MethodGet, requestURL, nil)
+		assert.NoError(t, err)
+
+		headersConversionErr := H2RequestHeadersToH1Request(
+			[]h2mux.Header{
+				h2mux.Header{
+					Name:  ":path",
+					Value: expectedPath,
+				},
+				h2mux.Header{
+					Name:  "Mock header",
+					Value: "Mock value",
+				},
+			},
+			request,
+		)
+
+		assert.Equal(t, http.Header{
+			"Mock header": []string{"Mock value"},
+		}, request.Header)
+
+		assert.Equal(t, expectedURL, request.URL.String(), fmt.Sprintf("Failed URL index: %v", index))
+
+		assert.NoError(t, headersConversionErr)
+	}
+}

streamhandler/stream_handler.go

@@ -93,7 +93,7 @@ func (s *StreamHandler) UpdateConfig(newConfig []*pogs.ReverseProxyConfig) (fail
 	toAdd := s.tunnelHostnameMapper.ToAdd(newConfig)
 	for _, tunnelConfig := range toAdd {
 		tunnelHostname := tunnelConfig.TunnelHostname
-		originSerice, err := tunnelConfig.OriginConfigJSONHandler.OriginConfig.Service()
+		originSerice, err := tunnelConfig.OriginConfig.Service()
 		if err != nil {
 			s.logger.WithField("tunnelHostname", tunnelHostname).WithError(err).Error("Invalid origin service config")
 			failedConfigs = append(failedConfigs, &pogs.FailedConfig{

streamhandler/stream_handler_test.go

@@ -49,10 +49,8 @@ func TestServeRequest(t *testing.T) {
 	reverseProxyConfigs := []*pogs.ReverseProxyConfig{
 		{
 			TunnelHostname: testTunnelHostname,
-			OriginConfigJSONHandler: &pogs.OriginConfigJSONHandler{
-				OriginConfig: &pogs.HTTPOriginConfig{
-					URLString: httpServer.URL,
-				},
+			OriginConfig: &pogs.HTTPOriginConfig{
+				URLString: httpServer.URL,
 			},
 		},
 	}
@@ -99,10 +97,8 @@ func TestServeBadRequest(t *testing.T) {
 	reverseProxyConfigs := []*pogs.ReverseProxyConfig{
 		{
 			TunnelHostname: testTunnelHostname,
-			OriginConfigJSONHandler: &pogs.OriginConfigJSONHandler{
-				OriginConfig: &pogs.HTTPOriginConfig{
-					URLString: "",
-				},
+			OriginConfig: &pogs.HTTPOriginConfig{
+				URLString: "",
 			},
 		},
 	}

tlsconfig/certreloader.go

@@ -5,7 +5,6 @@ import (
 	"crypto/x509"
 	"fmt"
 	"io/ioutil"
-	"net"
 	"runtime"
 	"sync"
 
@@ -18,6 +17,8 @@ import (
 const (
 	OriginCAPoolFlag = "origin-ca-pool"
 	CaCertFlag       = "cacert"
+
+	edgeTLSServerName = "cftunnel.com"
 )
 
 // CertReloader can load and reload a TLS certificate from a particular filepath.
@@ -126,7 +127,7 @@ func CreateTunnelConfig(c *cli.Context) (*tls.Config, error) {
 		rootCAs = append(rootCAs, c.String(CaCertFlag))
 	}
 
-	userConfig := &TLSParameters{RootCAs: rootCAs}
+	userConfig := &TLSParameters{RootCAs: rootCAs, ServerName: edgeTLSServerName}
 	tlsConfig, err := GetConfig(userConfig)
 	if err != nil {
 		return nil, err
@@ -142,10 +143,6 @@ func CreateTunnelConfig(c *cli.Context) (*tls.Config, error) {
 			rootCAPool.AddCert(cert)
 		}
 		tlsConfig.RootCAs = rootCAPool
-		tlsConfig.ServerName = "cftunnel.com"
-	} else if edgeAddrs := c.StringSlice("edge"); len(edgeAddrs) > 0 {
-		// Set for development environments and for testing specific origintunneld instances
-		tlsConfig.ServerName, _, _ = net.SplitHostPort(edgeAddrs[0])
 	}
 
 	if tlsConfig.ServerName == "" && !tlsConfig.InsecureSkipVerify {

tlsconfig/tlsconfig_test.go

@@ -1,5 +1,3 @@
-// TODO: Remove the above build tag and include this test when we start compiling with Golang 1.10.0+
-
 package tlsconfig
 
 import (

tunnelhostnamemapper/tunnelhostnamemapper_test.go

@@ -193,20 +193,20 @@ func TestTunnelHostnameMapper_ToRemove(t *testing.T) {
 
 func sampleConfig1() *pogs.ReverseProxyConfig {
 	return &pogs.ReverseProxyConfig{
-		TunnelHostname:          "mock.example.com",
-		OriginConfigJSONHandler: &pogs.OriginConfigJSONHandler{OriginConfig: &pogs.HelloWorldOriginConfig{}},
-		Retries:                 18,
-		ConnectionTimeout:       5 * time.Second,
-		CompressionQuality:      3,
+		TunnelHostname:     "mock.example.com",
+		OriginConfig:       &pogs.HelloWorldOriginConfig{},
+		Retries:            18,
+		ConnectionTimeout:  5 * time.Second,
+		CompressionQuality: 3,
 	}
 }
 
 func sampleConfig2() *pogs.ReverseProxyConfig {
 	return &pogs.ReverseProxyConfig{
-		TunnelHostname:          "mock2.example.com",
-		OriginConfigJSONHandler: &pogs.OriginConfigJSONHandler{OriginConfig: &pogs.HelloWorldOriginConfig{}},
-		Retries:                 18,
-		ConnectionTimeout:       5 * time.Second,
-		CompressionQuality:      3,
+		TunnelHostname:     "mock2.example.com",
+		OriginConfig:       &pogs.HelloWorldOriginConfig{},
+		Retries:            18,
+		ConnectionTimeout:  5 * time.Second,
+		CompressionQuality: 3,
 	}
 }

tunnelrpc/pogs/config.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/json"
 	"fmt"
 	"net"
 	"net/http"
@@ -29,11 +28,43 @@ import (
 
 // ClientConfig is a collection of FallibleConfig that determines how cloudflared should function
 type ClientConfig struct {
-	Version              Version               `json:"version"`
-	SupervisorConfig     *SupervisorConfig     `json:"supervisor_config"`
-	EdgeConnectionConfig *EdgeConnectionConfig `json:"edge_connection_config"`
-	DoHProxyConfigs      []*DoHProxyConfig     `json:"doh_proxy_configs" capnp:"dohProxyConfigs"`
-	ReverseProxyConfigs  []*ReverseProxyConfig `json:"reverse_proxy_configs"`
+	Version              Version
+	SupervisorConfig     *SupervisorConfig
+	EdgeConnectionConfig *EdgeConnectionConfig
+	DoHProxyConfigs      []*DoHProxyConfig `capnp:"dohProxyConfigs"`
+	ReverseProxyConfigs  []*ReverseProxyConfig
+}
+
+func (c *ClientConfig) MarshalBytes() ([]byte, error) {
+	msg, firstSeg, err := capnp.NewMessage(capnp.SingleSegment(nil))
+	if err != nil {
+		return nil, err
+	}
+	capnpEntity, err := tunnelrpc.NewRootClientConfig(firstSeg)
+	if err != nil {
+		return nil, err
+	}
+	err = MarshalClientConfig(capnpEntity, c)
+	if err != nil {
+		return nil, err
+	}
+	return msg.Marshal()
+}
+
+func UnmarshalClientConfigFromBytes(clientConfigBytes []byte) (*ClientConfig, error) {
+	msg, err := capnp.Unmarshal(clientConfigBytes)
+	if err != nil {
+		return nil, err
+	}
+	capnpClientConfig, err := tunnelrpc.ReadRootClientConfig(msg)
+	if err != nil {
+		return nil, err
+	}
+	pogsClientConfig, err := UnmarshalClientConfig(capnpClientConfig)
+	if err != nil {
+		return nil, err
+	}
+	return pogsClientConfig, nil
 }
 
 // Version type models the version of a ClientConfig
@@ -52,16 +83,17 @@ func (v Version) String() string {
 }
 
 // FallibleConfig is an interface implemented by configs that cloudflared might not be able to apply
+//go-sumtype:decl FallibleConfig
 type FallibleConfig interface {
 	FailReason(err error) string
-	jsonType() string
+	isFallibleConfig()
 }
 
 // SupervisorConfig specifies config of components managed by Supervisor other than ConnectionManager
 type SupervisorConfig struct {
-	AutoUpdateFrequency    time.Duration `json:"auto_update_frequency"`
-	MetricsUpdateFrequency time.Duration `json:"metrics_update_frequency"`
-	GracePeriod            time.Duration `json:"grace_period"`
+	AutoUpdateFrequency    time.Duration
+	MetricsUpdateFrequency time.Duration
+	GracePeriod            time.Duration
 }
 
 // FailReason impelents FallibleConfig interface for SupervisorConfig
@@ -69,23 +101,15 @@ func (sc *SupervisorConfig) FailReason(err error) string {
 	return fmt.Sprintf("Cannot apply SupervisorConfig, err: %v", err)
 }
 
-func (sc *SupervisorConfig) MarshalJSON() ([]byte, error) {
-	marshaler := make(map[string]SupervisorConfig, 1)
-	marshaler[sc.jsonType()] = *sc
-	return json.Marshal(marshaler)
-}
-
-func (sc *SupervisorConfig) jsonType() string {
-	return "supervisor_config"
-}
+func (_ *SupervisorConfig) isFallibleConfig() {}
 
 // EdgeConnectionConfig specifies what parameters and how may connections should ConnectionManager establish with edge
 type EdgeConnectionConfig struct {
-	NumHAConnections    uint8         `json:"num_ha_connections"`
-	HeartbeatInterval   time.Duration `json:"heartbeat_interval"`
-	Timeout             time.Duration `json:"timeout"`
-	MaxFailedHeartbeats uint64        `json:"max_failed_heartbeats"`
-	UserCredentialPath  string        `json:"user_credential_path"`
+	NumHAConnections    uint8
+	HeartbeatInterval   time.Duration
+	Timeout             time.Duration
+	MaxFailedHeartbeats uint64
+	UserCredentialPath  string
 }
 
 // FailReason impelents FallibleConfig interface for EdgeConnectionConfig
@@ -93,21 +117,13 @@ func (cmc *EdgeConnectionConfig) FailReason(err error) string {
 	return fmt.Sprintf("Cannot apply EdgeConnectionConfig, err: %v", err)
 }
 
-func (cmc *EdgeConnectionConfig) MarshalJSON() ([]byte, error) {
-	marshaler := make(map[string]EdgeConnectionConfig, 1)
-	marshaler[cmc.jsonType()] = *cmc
-	return json.Marshal(marshaler)
-}
-
-func (cmc *EdgeConnectionConfig) jsonType() string {
-	return "edge_connection_config"
-}
+func (_ *EdgeConnectionConfig) isFallibleConfig() {}
 
 // DoHProxyConfig is configuration for DNS over HTTPS service
 type DoHProxyConfig struct {
-	ListenHost string   `json:"listen_host"`
-	ListenPort uint16   `json:"listen_port"`
-	Upstreams  []string `json:"upstreams"`
+	ListenHost string
+	ListenPort uint16
+	Upstreams  []string
 }
 
 // FailReason impelents FallibleConfig interface for DoHProxyConfig
@@ -115,23 +131,15 @@ func (dpc *DoHProxyConfig) FailReason(err error) string {
 	return fmt.Sprintf("Cannot apply DoHProxyConfig, err: %v", err)
 }
 
-func (dpc *DoHProxyConfig) MarshalJSON() ([]byte, error) {
-	marshaler := make(map[string]DoHProxyConfig, 1)
-	marshaler[dpc.jsonType()] = *dpc
-	return json.Marshal(marshaler)
-}
-
-func (dpc *DoHProxyConfig) jsonType() string {
-	return "doh_proxy_config"
-}
+func (_ *DoHProxyConfig) isFallibleConfig() {}
 
 // ReverseProxyConfig how and for what hostnames can this cloudflared proxy
 type ReverseProxyConfig struct {
-	TunnelHostname          h2mux.TunnelHostname     `json:"tunnel_hostname"`
-	OriginConfigJSONHandler *OriginConfigJSONHandler `json:"origin_config"`
-	Retries                 uint64                   `json:"retries"`
-	ConnectionTimeout       time.Duration            `json:"connection_timeout"`
-	CompressionQuality      uint64                   `json:"compression_quality"`
+	TunnelHostname     h2mux.TunnelHostname
+	OriginConfig       OriginConfig
+	Retries            uint64
+	ConnectionTimeout  time.Duration
+	CompressionQuality uint64
 }
 
 func NewReverseProxyConfig(
@@ -145,11 +153,11 @@ func NewReverseProxyConfig(
 		return nil, fmt.Errorf("NewReverseProxyConfig: originConfigUnmarshaler was null")
 	}
 	return &ReverseProxyConfig{
-		TunnelHostname:          h2mux.TunnelHostname(tunnelHostname),
-		OriginConfigJSONHandler: &OriginConfigJSONHandler{originConfig},
-		Retries:                 retries,
-		ConnectionTimeout:       connectionTimeout,
-		CompressionQuality:      compressionQuality,
+		TunnelHostname:     h2mux.TunnelHostname(tunnelHostname),
+		OriginConfig:       originConfig,
+		Retries:            retries,
+		ConnectionTimeout:  connectionTimeout,
+		CompressionQuality: compressionQuality,
 	}, nil
 }
 
@@ -158,58 +166,29 @@ func (rpc *ReverseProxyConfig) FailReason(err error) string {
 	return fmt.Sprintf("Cannot apply ReverseProxyConfig, err: %v", err)
 }
 
-func (rpc *ReverseProxyConfig) MarshalJSON() ([]byte, error) {
-	marshaler := make(map[string]ReverseProxyConfig, 1)
-	marshaler[rpc.jsonType()] = *rpc
-	return json.Marshal(marshaler)
-}
-
-func (rpc *ReverseProxyConfig) jsonType() string {
-	return "reverse_proxy_config"
-}
+func (_ *ReverseProxyConfig) isFallibleConfig() {}
 
 //go-sumtype:decl OriginConfig
 type OriginConfig interface {
 	// Service returns a OriginService used to proxy to the origin
 	Service() (originservice.OriginService, error)
 	// go-sumtype requires at least one unexported method, otherwise it will complain that interface is not sealed
-	jsonType() string
-}
-
-type originType int
-
-const (
-	httpType originType = iota
-	wsType
-	helloWorldType
-)
-
-func (ot originType) String() string {
-	switch ot {
-	case httpType:
-		return "Http"
-	case wsType:
-		return "WebSocket"
-	case helloWorldType:
-		return "HelloWorld"
-	default:
-		return "unknown"
-	}
+	isOriginConfig()
 }
 
 type HTTPOriginConfig struct {
-	URLString              string        `capnp:"urlString" json:"url_string" mapstructure:"url_string"`
-	TCPKeepAlive           time.Duration `capnp:"tcpKeepAlive" json:"tcp_keep_alive" mapstructure:"tcp_keep_alive"`
-	DialDualStack          bool          `json:"dial_dual_stack" mapstructure:"dial_dual_stack"`
-	TLSHandshakeTimeout    time.Duration `capnp:"tlsHandshakeTimeout" json:"tls_handshake_timeout" mapstructure:"tls_handshake_timeout"`
-	TLSVerify              bool          `capnp:"tlsVerify" json:"tls_verify" mapstructure:"tls_verify"`
-	OriginCAPool           string        `json:"origin_ca_pool" mapstructure:"origin_ca_pool"`
-	OriginServerName       string        `json:"origin_server_name" mapstructure:"origin_server_name"`
-	MaxIdleConnections     uint64        `json:"max_idle_connections" mapstructure:"max_idle_connections"`
-	IdleConnectionTimeout  time.Duration `json:"idle_connection_timeout" mapstructure:"idle_connection_timeout"`
-	ProxyConnectionTimeout time.Duration `json:"proxy_connection_timeout" mapstructure:"proxy_connection_timeout"`
-	ExpectContinueTimeout  time.Duration `json:"expect_continue_timeout" mapstructure:"expect_continue_timeout"`
-	ChunkedEncoding        bool          `json:"chunked_encoding" mapstructure:"chunked_encoding"`
+	URLString              string        `capnp:"urlString"`
+	TCPKeepAlive           time.Duration `capnp:"tcpKeepAlive"`
+	DialDualStack          bool
+	TLSHandshakeTimeout    time.Duration `capnp:"tlsHandshakeTimeout"`
+	TLSVerify              bool          `capnp:"tlsVerify"`
+	OriginCAPool           string
+	OriginServerName       string
+	MaxIdleConnections     uint64
+	IdleConnectionTimeout  time.Duration
+	ProxyConnectionTimeout time.Duration
+	ExpectContinueTimeout  time.Duration
+	ChunkedEncoding        bool
 }
 
 func (hc *HTTPOriginConfig) Service() (originservice.OriginService, error) {
@@ -248,15 +227,13 @@ func (hc *HTTPOriginConfig) Service() (originservice.OriginService, error) {
 	return originservice.NewHTTPService(transport, url, hc.ChunkedEncoding), nil
 }
 
-func (_ *HTTPOriginConfig) jsonType() string {
-	return httpType.String()
-}
+func (*HTTPOriginConfig) isOriginConfig() {}
 
 type WebSocketOriginConfig struct {
-	URLString        string `capnp:"urlString" json:"url_string" mapstructure:"url_string"`
-	TLSVerify        bool   `capnp:"tlsVerify" json:"tls_verify" mapstructure:"tls_verify"`
-	OriginCAPool     string `json:"origin_ca_pool" mapstructure:"origin_ca_pool"`
-	OriginServerName string `json:"origin_server_name" mapstructure:"origin_server_name"`
+	URLString        string `capnp:"urlString"`
+	TLSVerify        bool   `capnp:"tlsVerify"`
+	OriginCAPool     string
+	OriginServerName string
 }
 
 func (wsc *WebSocketOriginConfig) Service() (originservice.OriginService, error) {
@@ -277,13 +254,11 @@ func (wsc *WebSocketOriginConfig) Service() (originservice.OriginService, error)
 	return originservice.NewWebSocketService(tlsConfig, url)
 }
 
-func (_ *WebSocketOriginConfig) jsonType() string {
-	return wsType.String()
-}
+func (*WebSocketOriginConfig) isOriginConfig() {}
 
 type HelloWorldOriginConfig struct{}
 
-func (_ *HelloWorldOriginConfig) Service() (originservice.OriginService, error) {
+func (*HelloWorldOriginConfig) Service() (originservice.OriginService, error) {
 	helloCert, err := tlsconfig.GetHelloCertificateX509()
 	if err != nil {
 		return nil, errors.Wrap(err, "Cannot get Hello World server certificate")
@@ -308,9 +283,7 @@ func (_ *HelloWorldOriginConfig) Service() (originservice.OriginService, error)
 	return originservice.NewHelloWorldService(transport)
 }
 
-func (_ *HelloWorldOriginConfig) jsonType() string {
-	return helloWorldType.String()
-}
+func (*HelloWorldOriginConfig) isOriginConfig() {}
 
 /*
  * Boilerplate to convert between these structs and the primitive structs
@@ -519,9 +492,9 @@ func UnmarshalDoHProxyConfig(s tunnelrpc.DoHProxyConfig) (*DoHProxyConfig, error
 
 func MarshalReverseProxyConfig(s tunnelrpc.ReverseProxyConfig, p *ReverseProxyConfig) error {
 	s.SetTunnelHostname(p.TunnelHostname.String())
-	switch config := p.OriginConfigJSONHandler.OriginConfig.(type) {
+	switch config := p.OriginConfig.(type) {
 	case *HTTPOriginConfig:
-		ss, err := s.Origin().NewHttp()
+		ss, err := s.OriginConfig().NewHttp()
 		if err != nil {
 			return err
 		}
@@ -529,7 +502,7 @@ func MarshalReverseProxyConfig(s tunnelrpc.ReverseProxyConfig, p *ReverseProxyCo
 			return err
 		}
 	case *WebSocketOriginConfig:
-		ss, err := s.Origin().NewWebsocket()
+		ss, err := s.OriginConfig().NewWebsocket()
 		if err != nil {
 			return err
 		}
@@ -537,7 +510,7 @@ func MarshalReverseProxyConfig(s tunnelrpc.ReverseProxyConfig, p *ReverseProxyCo
 			return err
 		}
 	case *HelloWorldOriginConfig:
-		ss, err := s.Origin().NewHelloWorld()
+		ss, err := s.OriginConfig().NewHelloWorld()
 		if err != nil {
 			return err
 		}
@@ -560,9 +533,9 @@ func UnmarshalReverseProxyConfig(s tunnelrpc.ReverseProxyConfig) (*ReverseProxyC
 		return nil, err
 	}
 	p.TunnelHostname = h2mux.TunnelHostname(tunnelHostname)
-	switch s.Origin().Which() {
-	case tunnelrpc.ReverseProxyConfig_origin_Which_http:
-		ss, err := s.Origin().Http()
+	switch s.OriginConfig().Which() {
+	case tunnelrpc.ReverseProxyConfig_originConfig_Which_http:
+		ss, err := s.OriginConfig().Http()
 		if err != nil {
 			return nil, err
 		}
@@ -570,9 +543,9 @@ func UnmarshalReverseProxyConfig(s tunnelrpc.ReverseProxyConfig) (*ReverseProxyC
 		if err != nil {
 			return nil, err
 		}
-		p.OriginConfigJSONHandler = &OriginConfigJSONHandler{config}
-	case tunnelrpc.ReverseProxyConfig_origin_Which_websocket:
-		ss, err := s.Origin().Websocket()
+		p.OriginConfig = config
+	case tunnelrpc.ReverseProxyConfig_originConfig_Which_websocket:
+		ss, err := s.OriginConfig().Websocket()
 		if err != nil {
 			return nil, err
 		}
@@ -580,9 +553,9 @@ func UnmarshalReverseProxyConfig(s tunnelrpc.ReverseProxyConfig) (*ReverseProxyC
 		if err != nil {
 			return nil, err
 		}
-		p.OriginConfigJSONHandler = &OriginConfigJSONHandler{config}
-	case tunnelrpc.ReverseProxyConfig_origin_Which_helloWorld:
-		ss, err := s.Origin().HelloWorld()
+		p.OriginConfig = config
+	case tunnelrpc.ReverseProxyConfig_originConfig_Which_helloWorld:
+		ss, err := s.OriginConfig().HelloWorld()
 		if err != nil {
 			return nil, err
 		}
@@ -590,7 +563,7 @@ func UnmarshalReverseProxyConfig(s tunnelrpc.ReverseProxyConfig) (*ReverseProxyC
 		if err != nil {
 			return nil, err
 		}
-		p.OriginConfigJSONHandler = &OriginConfigJSONHandler{config}
+		p.OriginConfig = config
 	}
 	p.Retries = s.Retries()
 	p.ConnectionTimeout = time.Duration(s.ConnectionTimeout())
@@ -690,13 +663,13 @@ func (i ClientService_PogsImpl) UseConfiguration(p tunnelrpc.ClientService_useCo
 }
 
 type UseConfigurationResult struct {
-	Success       bool            `json:"success"`
-	FailedConfigs []*FailedConfig `json:"failed_configs"`
+	Success       bool
+	FailedConfigs []*FailedConfig
 }
 
 type FailedConfig struct {
-	Config FallibleConfig `json:"config"`
-	Reason string         `json:"reason"`
+	Config FallibleConfig
+	Reason string
 }
 
 func MarshalFailedConfig(s tunnelrpc.FailedConfig, p *FailedConfig) error {

tunnelrpc/pogs/config_test.go

@@ -61,13 +61,13 @@ func ClientConfigTestCases() []*ClientConfig {
 			sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
 			}),
 			sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-				c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleHTTPOriginConfig()}
+				c.OriginConfig = sampleHTTPOriginConfig()
 			}),
 			sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-				c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleHTTPOriginConfigUnixPath()}
+				c.OriginConfig = sampleHTTPOriginConfigUnixPath()
 			}),
 			sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-				c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleWebSocketOriginConfig()}
+				c.OriginConfig = sampleWebSocketOriginConfig()
 			}),
 		}
 	}
@@ -83,21 +83,14 @@ func ClientConfigTestCases() []*ClientConfig {
 }
 
 func TestClientConfig(t *testing.T) {
-	for i, testCase := range ClientConfigTestCases() {
-		_, seg, err := capnp.NewMessage(capnp.SingleSegment(nil))
-		capnpEntity, err := tunnelrpc.NewClientConfig(seg)
-		if !assert.NoError(t, err) {
-			t.Fatal("Couldn't initialize a new message")
-		}
-		err = MarshalClientConfig(capnpEntity, testCase)
-		if !assert.NoError(t, err, "testCase index %v failed to marshal", i) {
-			continue
-		}
-		result, err := UnmarshalClientConfig(capnpEntity)
-		if !assert.NoError(t, err, "testCase index %v failed to unmarshal", i) {
-			continue
-		}
-		assert.Equal(t, testCase, result, "testCase index %v didn't preserve struct through marshalling and unmarshalling", i)
+	for _, testCase := range ClientConfigTestCases() {
+		b, err := testCase.MarshalBytes()
+		assert.NoError(t, err)
+
+		clientConfig, err := UnmarshalClientConfigFromBytes(b)
+		assert.NoError(t, err)
+
+		assert.Equal(t, testCase, clientConfig)
 	}
 }
 
@@ -167,13 +160,13 @@ func TestReverseProxyConfig(t *testing.T) {
 	testCases := []*ReverseProxyConfig{
 		sampleReverseProxyConfig(),
 		sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-			c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleHTTPOriginConfig()}
+			c.OriginConfig = sampleHTTPOriginConfig()
 		}),
 		sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-			c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleHTTPOriginConfigUnixPath()}
+			c.OriginConfig = sampleHTTPOriginConfigUnixPath()
 		}),
 		sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-			c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleWebSocketOriginConfig()}
+			c.OriginConfig = sampleWebSocketOriginConfig()
 		}),
 	}
 	for i, testCase := range testCases {
@@ -323,11 +316,11 @@ func sampleDoHProxyConfig(overrides ...func(*DoHProxyConfig)) *DoHProxyConfig {
 // applies any number of overrides to it, and returns it.
 func sampleReverseProxyConfig(overrides ...func(*ReverseProxyConfig)) *ReverseProxyConfig {
 	sample := &ReverseProxyConfig{
-		TunnelHostname:          "mock-non-lb-tunnel.example.com",
-		OriginConfigJSONHandler: &OriginConfigJSONHandler{&HelloWorldOriginConfig{}},
-		Retries:                 18,
-		ConnectionTimeout:       5 * time.Second,
-		CompressionQuality:      3,
+		TunnelHostname:     "mock-non-lb-tunnel.example.com",
+		OriginConfig:       &HelloWorldOriginConfig{},
+		Retries:            18,
+		ConnectionTimeout:  5 * time.Second,
+		CompressionQuality: 3,
 	}
 	sample.ensureNoZeroFields()
 	for _, f := range overrides {

tunnelrpc/pogs/json.go

@@ -1,68 +0,0 @@
-package pogs
-
-import (
-	"encoding/json"
-	"fmt"
-
-	"github.com/mitchellh/mapstructure"
-	"github.com/pkg/errors"
-)
-
-// OriginConfigJSONHandler is a wrapper to serialize OriginConfig with type information, and deserialize JSON
-// into an OriginConfig.
-type OriginConfigJSONHandler struct {
-	OriginConfig OriginConfig
-}
-
-func (ocjh *OriginConfigJSONHandler) MarshalJSON() ([]byte, error) {
-	marshaler := make(map[string]OriginConfig, 1)
-	marshaler[ocjh.OriginConfig.jsonType()] = ocjh.OriginConfig
-	return json.Marshal(marshaler)
-}
-
-func (ocjh *OriginConfigJSONHandler) UnmarshalJSON(b []byte) error {
-	var originJSON map[string]interface{}
-	if err := json.Unmarshal(b, &originJSON); err != nil {
-		return errors.Wrapf(err, "cannot unmarshal %s into originJSON", string(b))
-	}
-
-	if originConfig, ok := originJSON[httpType.String()]; ok {
-		httpOriginConfig := &HTTPOriginConfig{}
-		if err := mapstructure.Decode(originConfig, httpOriginConfig); err != nil {
-			return errors.Wrapf(err, "cannot decode %+v into HTTPOriginConfig", originConfig)
-		}
-		ocjh.OriginConfig = httpOriginConfig
-		return nil
-	}
-
-	if originConfig, ok := originJSON[wsType.String()]; ok {
-		wsOriginConfig := &WebSocketOriginConfig{}
-		if err := mapstructure.Decode(originConfig, wsOriginConfig); err != nil {
-			return errors.Wrapf(err, "cannot decode %+v into WebSocketOriginConfig", originConfig)
-		}
-		ocjh.OriginConfig = wsOriginConfig
-		return nil
-	}
-
-	if originConfig, ok := originJSON[helloWorldType.String()]; ok {
-		helloWorldOriginConfig := &HelloWorldOriginConfig{}
-		if err := mapstructure.Decode(originConfig, helloWorldOriginConfig); err != nil {
-			return errors.Wrapf(err, "cannot decode %+v into HelloWorldOriginConfig", originConfig)
-		}
-		ocjh.OriginConfig = helloWorldOriginConfig
-		return nil
-	}
-
-	return fmt.Errorf("cannot unmarshal %s into OriginConfig", string(b))
-}
-
-// FallibleConfigMarshaler is a wrapper for FallibleConfig to implement custom marshal logic
-type FallibleConfigMarshaler struct {
-	FallibleConfig FallibleConfig
-}
-
-func (fcm *FallibleConfigMarshaler) MarshalJSON() ([]byte, error) {
-	marshaler := make(map[string]FallibleConfig, 1)
-	marshaler[fcm.FallibleConfig.jsonType()] = fcm.FallibleConfig
-	return json.Marshal(marshaler)
-}

tunnelrpc/pogs/json_test.go

@@ -1,285 +0,0 @@
-package pogs
-
-import (
-	"encoding/json"
-	"strings"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestUnmarshalOrigin(t *testing.T) {
-	tests := []struct {
-		jsonLiteral          string
-		exceptedOriginConfig OriginConfig
-	}{
-		{
-			jsonLiteral: `{
-				"Http":{
-					"url_string":"https.example.com",
-					"tcp_keep_alive":7000000000,
-					"dial_dual_stack":true,
-					"tls_handshake_timeout":11000000000,
-					"tls_verify":true,
-					"origin_ca_pool":"/etc/cert.pem",
-					"origin_server_name":"secure.example.com",
-					"max_idle_connections":19,
-					"idle_connection_timeout":17000000000,
-					"proxy_connection_timeout":15000000000,
-					"expect_continue_timeout":21000000000,
-					"chunked_encoding":true
-				}
-			}`,
-			exceptedOriginConfig: sampleHTTPOriginConfig(),
-		},
-		{
-			jsonLiteral: `{
-				"WebSocket":{
-					"url_string":"ssh://example.com",
-					"tls_verify":true,
-					"origin_ca_pool":"/etc/cert.pem",
-					"origin_server_name":"secure.example.com"
-				}
-			}`,
-			exceptedOriginConfig: sampleWebSocketOriginConfig(),
-		},
-		{
-			jsonLiteral: `{
-				"HelloWorld": {}
-			}`,
-			exceptedOriginConfig: &HelloWorldOriginConfig{},
-		},
-	}
-
-	for _, test := range tests {
-		originConfigJSON := prettyToValidJSON(test.jsonLiteral)
-		var OriginConfigJSONHandler OriginConfigJSONHandler
-		err := json.Unmarshal([]byte(originConfigJSON), &OriginConfigJSONHandler)
-		assert.NoError(t, err)
-		assert.Equal(t, test.exceptedOriginConfig, OriginConfigJSONHandler.OriginConfig)
-	}
-}
-
-func TestUnmarshalClientConfig(t *testing.T) {
-	prettyClientConfigJSON := `{
-		"version":10,
-		"supervisor_config":{
-			"auto_update_frequency":86400000000000,
-			"metrics_update_frequency":300000000000,
-			"grace_period":30000000000
-		},
-		"edge_connection_config":{
-			"num_ha_connections":4,
-			"heartbeat_interval":5000000000,
-			"timeout":30000000000,
-			"max_failed_heartbeats":5,
-			"user_credential_path":"~/.cloudflared/cert.pem"
-		},
-		"doh_proxy_configs":[{
-			"listen_host": "localhost",
-			"listen_port": 53,
-			"upstreams": ["https://1.1.1.1/dns-query", "https://1.0.0.1/dns-query"]
-		}],
-		"reverse_proxy_configs":[{
-			"tunnel_hostname":"sdfjadk33.cftunnel.com",
-			"origin_config":{
-				"Http":{
-					"url_string":"https://127.0.0.1:8080",
-					"tcp_keep_alive":30000000000,
-					"dial_dual_stack":true,
-					"tls_handshake_timeout":10000000000,
-					"tls_verify":true,
-					"origin_ca_pool":"",
-					"origin_server_name":"",
-					"max_idle_connections":100,
-					"idle_connection_timeout":90000000000,
-					"proxy_connection_timeout":90000000000,
-					"expect_continue_timeout":90000000000,
-					"chunked_encoding":true
-				}
-			},
-			"retries":5,
-			"connection_timeout":30,
-			"compression_quality":0
-		}]
-	}`
-	// replace new line and tab
-	clientConfigJSON := prettyToValidJSON(prettyClientConfigJSON)
-
-	var clientConfig ClientConfig
-	err := json.Unmarshal([]byte(clientConfigJSON), &clientConfig)
-	assert.NoError(t, err)
-
-	assert.Equal(t, Version(10), clientConfig.Version)
-
-	supervisorConfig := SupervisorConfig{
-		AutoUpdateFrequency:    time.Hour * 24,
-		MetricsUpdateFrequency: time.Second * 300,
-		GracePeriod:            time.Second * 30,
-	}
-	assert.Equal(t, supervisorConfig, *clientConfig.SupervisorConfig)
-
-	edgeConnectionConfig := EdgeConnectionConfig{
-		NumHAConnections:    4,
-		HeartbeatInterval:   time.Second * 5,
-		Timeout:             time.Second * 30,
-		MaxFailedHeartbeats: 5,
-		UserCredentialPath:  "~/.cloudflared/cert.pem",
-	}
-	assert.Equal(t, edgeConnectionConfig, *clientConfig.EdgeConnectionConfig)
-
-	dohProxyConfig := DoHProxyConfig{
-		ListenHost: "localhost",
-		ListenPort: 53,
-		Upstreams:  []string{"https://1.1.1.1/dns-query", "https://1.0.0.1/dns-query"},
-	}
-
-	assert.Len(t, clientConfig.DoHProxyConfigs, 1)
-	assert.Equal(t, dohProxyConfig, *clientConfig.DoHProxyConfigs[0])
-
-	reverseProxyConfig := ReverseProxyConfig{
-		TunnelHostname: "sdfjadk33.cftunnel.com",
-		OriginConfigJSONHandler: &OriginConfigJSONHandler{
-			OriginConfig: &HTTPOriginConfig{
-				URLString:              "https://127.0.0.1:8080",
-				TCPKeepAlive:           time.Second * 30,
-				DialDualStack:          true,
-				TLSHandshakeTimeout:    time.Second * 10,
-				TLSVerify:              true,
-				OriginCAPool:           "",
-				OriginServerName:       "",
-				MaxIdleConnections:     100,
-				IdleConnectionTimeout:  time.Second * 90,
-				ProxyConnectionTimeout: time.Second * 90,
-				ExpectContinueTimeout:  time.Second * 90,
-				ChunkedEncoding:        true,
-			},
-		},
-		Retries:            5,
-		ConnectionTimeout:  30,
-		CompressionQuality: 0,
-	}
-
-	assert.Len(t, clientConfig.ReverseProxyConfigs, 1)
-	assert.Equal(t, reverseProxyConfig, *clientConfig.ReverseProxyConfigs[0])
-}
-
-func TestMarshalFallibleConfig(t *testing.T) {
-	tests := []struct {
-		fallibleConfig     FallibleConfig
-		expctedJSONLiteral string
-	}{
-		{
-			fallibleConfig: sampleSupervisorConfig(),
-			expctedJSONLiteral: `{
-				"supervisor_config":{
-					"auto_update_frequency":75600000000000,
-					"metrics_update_frequency":660000000000,
-					"grace_period":31000000000
-				}
-			}`,
-		},
-		{
-			fallibleConfig: sampleEdgeConnectionConfig(),
-			expctedJSONLiteral: `{
-				"edge_connection_config":{
-					"num_ha_connections":49,
-					"heartbeat_interval":5000000000,
-					"timeout":9000000000,
-					"max_failed_heartbeats":9001,
-					"user_credential_path":"/Users/example/.cloudflared/cert.pem"
-				}
-			}`,
-		},
-		{
-			fallibleConfig: sampleDoHProxyConfig(),
-			expctedJSONLiteral: `{
-				"doh_proxy_config":{
-					"listen_host":"127.0.0.1",
-					"listen_port":53,
-					"upstreams":["1.1.1.1","1.0.0.1"]
-				}
-			}`,
-		},
-		{
-			fallibleConfig: sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-				c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleHTTPOriginConfig()}
-			}),
-			expctedJSONLiteral: `{
-				"reverse_proxy_config":{
-					"tunnel_hostname":"mock-non-lb-tunnel.example.com",
-					"origin_config":{
-						"Http":{
-							"url_string":"https.example.com",
-							"tcp_keep_alive":7000000000,
-							"dial_dual_stack":true,
-							"tls_handshake_timeout":11000000000,
-							"tls_verify":true,
-							"origin_ca_pool":"/etc/cert.pem",
-							"origin_server_name":"secure.example.com",
-							"max_idle_connections":19,
-							"idle_connection_timeout":17000000000,
-							"proxy_connection_timeout":15000000000,
-							"expect_continue_timeout":21000000000,
-							"chunked_encoding":true
-						}
-					},
-					"retries":18,
-					"connection_timeout":5000000000,
-					"compression_quality":3
-				}
-			}`,
-		},
-		{
-			fallibleConfig: sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-				c.OriginConfigJSONHandler = &OriginConfigJSONHandler{sampleWebSocketOriginConfig()}
-			}),
-			expctedJSONLiteral: `{
-				"reverse_proxy_config":{
-					"tunnel_hostname":"mock-non-lb-tunnel.example.com",
-					"origin_config":{
-						"WebSocket":{
-							"url_string":"ssh://example.com",
-							"tls_verify":true,
-							"origin_ca_pool":"/etc/cert.pem",
-							"origin_server_name":"secure.example.com"
-						}
-					},
-					"retries":18,
-					"connection_timeout":5000000000,
-					"compression_quality":3
-				}
-			}`,
-		},
-		{
-			fallibleConfig: sampleReverseProxyConfig(func(c *ReverseProxyConfig) {
-				c.OriginConfigJSONHandler = &OriginConfigJSONHandler{&HelloWorldOriginConfig{}}
-			}),
-			expctedJSONLiteral: `{
-				"reverse_proxy_config":{
-					"tunnel_hostname":"mock-non-lb-tunnel.example.com",
-					"origin_config":{
-						"HelloWorld":{}
-					},
-					"retries":18,
-					"connection_timeout":5000000000,
-					"compression_quality":3
-				}
-			}`,
-		},
-	}
-
-	for _, test := range tests {
-		b, err := json.Marshal(test.fallibleConfig)
-		assert.NoError(t, err)
-		assert.Equal(t, prettyToValidJSON(test.expctedJSONLiteral), string(b))
-	}
-
-}
-
-type prettyJSON string
-
-func prettyToValidJSON(prettyJSON string) string {
-	return strings.ReplaceAll(strings.ReplaceAll(prettyJSON, "\n", ""), "\t", "")
-}

tunnelrpc/pogs/tunnelrpc.go

@@ -2,10 +2,12 @@ package pogs
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/cloudflare/cloudflared/tunnelrpc"
 	"github.com/google/uuid"
+	"github.com/pkg/errors"
 
 	log "github.com/sirupsen/logrus"
 	capnp "zombiezen.com/go/capnproto2"
@@ -73,28 +75,129 @@ func UnmarshalRegistrationOptions(s tunnelrpc.RegistrationOptions) (*Registratio
 	return p, err
 }
 
-type ConnectResult struct {
-	Err          *ConnectError
-	ServerInfo   ServerInfo
-	ClientConfig ClientConfig
+// ConnectResult models the result of Connect RPC, implemented by ConnectError and ConnectSuccess.
+type ConnectResult interface {
+	ConnectError() *ConnectError
+	ConnectedTo() string
+	ClientConfig() *ClientConfig
+	Marshal(s tunnelrpc.ConnectResult) error
 }
 
-func MarshalConnectResult(s tunnelrpc.ConnectResult, p *ConnectResult) error {
-	return pogs.Insert(tunnelrpc.ConnectResult_TypeID, s.Struct, p)
+func MarshalConnectResult(s tunnelrpc.ConnectResult, p ConnectResult) error {
+	return p.Marshal(s)
 }
 
-func UnmarshalConnectResult(s tunnelrpc.ConnectResult) (*ConnectResult, error) {
-	p := new(ConnectResult)
-	err := pogs.Extract(p, tunnelrpc.ConnectResult_TypeID, s.Struct)
-	return p, err
+func UnmarshalConnectResult(s tunnelrpc.ConnectResult) (ConnectResult, error) {
+	switch s.Result().Which() {
+	case tunnelrpc.ConnectResult_result_Which_err:
+		capnpConnectError, err := s.Result().Err()
+		if err != nil {
+			return nil, err
+		}
+		return UnmarshalConnectError(capnpConnectError)
+	case tunnelrpc.ConnectResult_result_Which_success:
+		capnpConnectSuccess, err := s.Result().Success()
+		if err != nil {
+			return nil, err
+		}
+		return UnmarshalConnectSuccess(capnpConnectSuccess)
+	default:
+		return nil, fmt.Errorf("Unmarshal %v not implemented yet", s.Result().Which().String())
+	}
 }
 
+// ConnectSuccess is the concrete returned type when Connect RPC succeed
+type ConnectSuccess struct {
+	ServerLocationName string
+	Config             *ClientConfig
+}
+
+func (*ConnectSuccess) ConnectError() *ConnectError {
+	return nil
+}
+
+func (cs *ConnectSuccess) ConnectedTo() string {
+	return cs.ServerLocationName
+}
+
+func (cs *ConnectSuccess) ClientConfig() *ClientConfig {
+	return cs.Config
+}
+
+func (cs *ConnectSuccess) Marshal(s tunnelrpc.ConnectResult) error {
+	capnpConnectSuccess, err := s.Result().NewSuccess()
+	if err != nil {
+		return err
+	}
+
+	err = capnpConnectSuccess.SetServerLocationName(cs.ServerLocationName)
+	if err != nil {
+		return errors.Wrap(err, "failed to set ConnectSuccess.ServerLocationName")
+	}
+
+	if cs.Config != nil {
+		capnpClientConfig, err := capnpConnectSuccess.NewClientConfig()
+		if err != nil {
+			return errors.Wrap(err, "failed to initialize ConnectSuccess.ClientConfig")
+		}
+		if err := MarshalClientConfig(capnpClientConfig, cs.Config); err != nil {
+			return errors.Wrap(err, "failed to marshal ClientConfig")
+		}
+	}
+
+	return nil
+}
+
+func UnmarshalConnectSuccess(s tunnelrpc.ConnectSuccess) (*ConnectSuccess, error) {
+	p := new(ConnectSuccess)
+
+	serverLocationName, err := s.ServerLocationName()
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get tunnelrpc.ConnectSuccess.ServerLocationName")
+	}
+	p.ServerLocationName = serverLocationName
+
+	if s.HasClientConfig() {
+		capnpClientConfig, err := s.ClientConfig()
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to get tunnelrpc.ConnectSuccess.ClientConfig")
+		}
+		p.Config, err = UnmarshalClientConfig(capnpClientConfig)
+		if err != nil {
+			return nil, errors.Wrap(err, "failed to get unmarshal ClientConfig")
+		}
+	}
+
+	return p, nil
+}
+
+// ConnectError is the concrete returned type when Connect RPC encounters some error
 type ConnectError struct {
 	Cause       string
 	RetryAfter  time.Duration
 	ShouldRetry bool
 }
 
+func (ce *ConnectError) ConnectError() *ConnectError {
+	return ce
+}
+
+func (*ConnectError) ConnectedTo() string {
+	return ""
+}
+
+func (*ConnectError) ClientConfig() *ClientConfig {
+	return nil
+}
+
+func (ce *ConnectError) Marshal(s tunnelrpc.ConnectResult) error {
+	capnpConnectError, err := s.Result().NewErr()
+	if err != nil {
+		return err
+	}
+	return MarshalConnectError(capnpConnectError, ce)
+}
+
 func MarshalConnectError(s tunnelrpc.ConnectError, p *ConnectError) error {
 	return pogs.Insert(tunnelrpc.ConnectError_TypeID, s.Struct, p)
 }
@@ -134,8 +237,7 @@ type ConnectParameters struct {
 	NumPreviousAttempts uint8
 	Tags                []Tag
 	CloudflaredVersion  string
-	Name                string
-	Group               string
+	IntentLabel         string
 }
 
 func MarshalConnectParameters(s tunnelrpc.CapnpConnectParameters, p *ConnectParameters) error {
@@ -168,13 +270,7 @@ func MarshalConnectParameters(s tunnelrpc.CapnpConnectParameters, p *ConnectPara
 	if err := s.SetCloudflaredVersion(p.CloudflaredVersion); err != nil {
 		return err
 	}
-	if err := s.SetName(p.Name); err != nil {
-		return err
-	}
-	if err := s.SetGroup(p.Group); err != nil {
-		return err
-	}
-	return nil
+	return s.SetIntentLabel(p.IntentLabel)
 }
 
 func UnmarshalConnectParameters(s tunnelrpc.CapnpConnectParameters) (*ConnectParameters, error) {
@@ -215,24 +311,14 @@ func UnmarshalConnectParameters(s tunnelrpc.CapnpConnectParameters) (*ConnectPar
 		return nil, err
 	}
 
-	name, err := s.Name()
-	if err != nil {
-		return nil, err
-	}
-
-	group, err := s.Group()
-	if err != nil {
-		return nil, err
-	}
-
+	intentLabel, err := s.IntentLabel()
 	return &ConnectParameters{
 		OriginCert:          originCert,
 		CloudflaredID:       cloudflaredID,
 		NumPreviousAttempts: s.NumPreviousAttempts(),
 		Tags:                tags,
 		CloudflaredVersion:  cloudflaredVersion,
-		Name:                name,
-		Group:               group,
+		IntentLabel:         intentLabel,
 	}, nil
 }
 
@@ -240,7 +326,7 @@ type TunnelServer interface {
 	RegisterTunnel(ctx context.Context, originCert []byte, hostname string, options *RegistrationOptions) (*TunnelRegistration, error)
 	GetServerInfo(ctx context.Context) (*ServerInfo, error)
 	UnregisterTunnel(ctx context.Context, gracePeriodNanoSec int64) error
-	Connect(ctx context.Context, paramaters *ConnectParameters) (*ConnectResult, error)
+	Connect(ctx context.Context, parameters *ConnectParameters) (ConnectResult, error)
 }
 
 func TunnelServer_ServerToClient(s TunnelServer) tunnelrpc.TunnelServer {
@@ -301,11 +387,11 @@ func (i TunnelServer_PogsImpl) UnregisterTunnel(p tunnelrpc.TunnelServer_unregis
 }
 
 func (i TunnelServer_PogsImpl) Connect(p tunnelrpc.TunnelServer_connect) error {
-	paramaters, err := p.Params.Parameters()
+	parameters, err := p.Params.Parameters()
 	if err != nil {
 		return err
 	}
-	pogsParameters, err := UnmarshalConnectParameters(paramaters)
+	pogsParameters, err := UnmarshalConnectParameters(parameters)
 	if err != nil {
 		return err
 	}
@@ -318,7 +404,7 @@ func (i TunnelServer_PogsImpl) Connect(p tunnelrpc.TunnelServer_connect) error {
 	if err != nil {
 		return err
 	}
-	return MarshalConnectResult(result, connectResult)
+	return connectResult.Marshal(result)
 }
 
 type TunnelServer_PogsClient struct {
@@ -382,7 +468,7 @@ func (c TunnelServer_PogsClient) UnregisterTunnel(ctx context.Context, gracePeri
 
 func (c TunnelServer_PogsClient) Connect(ctx context.Context,
 	parameters *ConnectParameters,
-) (*ConnectResult, error) {
+) (ConnectResult, error) {
 	client := tunnelrpc.TunnelServer{Client: c.Client}
 	promise := client.Connect(ctx, func(p tunnelrpc.TunnelServer_connect_Params) error {
 		connectParameters, err := p.NewParameters()

tunnelrpc/pogs/tunnelrpc_test.go

@@ -11,21 +11,21 @@ import (
 	capnp "zombiezen.com/go/capnproto2"
 )
 
-func sampleTestConnectResult() *ConnectResult {
-	return &ConnectResult{
-		Err: &ConnectError{
+func TestConnectResult(t *testing.T) {
+	testCases := []ConnectResult{
+		&ConnectError{
 			Cause:       "it broke",
 			ShouldRetry: false,
 			RetryAfter:  2 * time.Second,
 		},
-		ServerInfo:   ServerInfo{LocationName: "computer"},
-		ClientConfig: *sampleClientConfig(),
-	}
-}
-
-func TestConnectResult(t *testing.T) {
-	testCases := []*ConnectResult{
-		sampleTestConnectResult(),
+		&ConnectSuccess{
+			ServerLocationName: "SFO",
+			Config:             sampleClientConfig(),
+		},
+		&ConnectSuccess{
+			ServerLocationName: "",
+			Config:             nil,
+		},
 	}
 	for i, testCase := range testCases {
 		_, seg, err := capnp.NewMessage(capnp.SingleSegment(nil))
@@ -49,7 +49,7 @@ func TestConnectParameters(t *testing.T) {
 	testCases := []*ConnectParameters{
 		sampleConnectParameters(),
 		sampleConnectParameters(func(c *ConnectParameters) {
-			c.Name = ""
+			c.IntentLabel = "my_intent"
 		}),
 		sampleConnectParameters(func(c *ConnectParameters) {
 			c.Tags = nil
@@ -89,8 +89,7 @@ func sampleConnectParameters(overrides ...func(*ConnectParameters)) *ConnectPara
 			},
 		},
 		CloudflaredVersion: "7.0",
-		Name:               "My Computer",
-		Group:              "www",
+		IntentLabel:        "my_intent",
 	}
 	sample.ensureNoZeroFields()
 	for _, f := range overrides {

tunnelrpc/tunnelrpc.capnp

@@ -57,18 +57,15 @@ struct CapnpConnectParameters {
     tags @3 :List(Tag);
     # release version of cloudflared
     cloudflaredVersion @4 :Text;
-    # friendly name for this cloudflared instance
-    name @5 :Text;
-    # group whose behavior this cloudflared instance will adopt
-    group @6 :Text;
+    # which intent this cloudflared instance should get its behaviour from
+    intentLabel @5 :Text;
 }
 
 struct ConnectResult {
-    err @0 :ConnectError;
-    # Information about the server this connection is established with
-    serverInfo @1 :ServerInfo;
-    # How this cloudflared instance should be configured
-    clientConfig @2 :ClientConfig;
+    result :union {
+        err @0 :ConnectError;
+        success @1 :ConnectSuccess;
+    }
 }
 
 struct ConnectError {
@@ -78,6 +75,13 @@ struct ConnectError {
     shouldRetry @2 :Bool;
 }
 
+struct ConnectSuccess {
+    # Information about the server this connection is established with
+    serverLocationName @0 :Text;
+    # How this cloudflared instance should be configured. This can be null if there isn't an intent for this origin yet
+    clientConfig @1 :ClientConfig;
+}
+
 struct ClientConfig {
     # Version of this configuration. This value is opaque, but is guaranteed
     # to monotonically increase in value. Any configuration supplied to
@@ -129,7 +133,7 @@ struct EdgeConnectionConfig {
 
 struct ReverseProxyConfig {
     tunnelHostname @0 :Text;
-    origin :union {
+    originConfig :union {
         http @1 :HTTPOriginConfig;
         websocket @2 :WebSocketOriginConfig;
         helloWorld @3 :HelloWorldOriginConfig;