Release 2023.4.1

cloudflared tail will now fetch the management token from by making
a request to the Cloudflare API using the cert.pem (acquired from
cloudflared login).

Refactored some of the credentials code into it's own package as
to allow for easier use between subcommands outside of
`cloudflared tunnel`.

CHANGES.md

@@ -1,6 +1,10 @@
+## 2023.4.1
+### New Features
+- You can now stream your logs from your remote cloudflared to your local terminal with `cloudflared tail <TUNNEL-ID>`. This new feature requires the remote cloudflared to be version 2023.4.1 or higher.
+
 ## 2023.3.2
 ### Notices
-- Due to the nature of QuickTunnels (https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/do-more-with-tunnels/trycloudflare/) and its intended usage for testing and experiment of Cloudflare Tunnels, starting from 2023.3.2, QuickTunnels only make a single connection to the edge. If users want to use Tunnels in a production enverionment, they should move to Named Tunnels instead. (https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/install-and-setup/tunnel-guide/remote/#set-up-a-tunnel-remotely-dashboard-setup)
+- Due to the nature of QuickTunnels (https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/do-more-with-tunnels/trycloudflare/) and its intended usage for testing and experiment of Cloudflare Tunnels, starting from 2023.3.2, QuickTunnels only make a single connection to the edge. If users want to use Tunnels in a production environment, they should move to Named Tunnels instead. (https://developers.cloudflare.com/cloudflare-one/connections/connect-apps/install-and-setup/tunnel-guide/remote/#set-up-a-tunnel-remotely-dashboard-setup)
 
 ## 2023.3.1
 ### Breaking Change
@@ -25,7 +29,7 @@
 
 ## 2022.12.0
 ### Improvements
-- cloudflared now attempts to try other edge addresses before falling back to a lower protoocol.
+- cloudflared now attempts to try other edge addresses before falling back to a lower protocol.
 - cloudflared tunnel no longer spins up a quick tunnel. The call has to be explicit and provide a --url flag.
 - cloudflared will now randomly pick the first or second region to connect to instead of always connecting to region2 first.
 

RELEASE_NOTES

@@ -1,3 +1,13 @@
+2023.4.1
+- 2023-04-13 TUN-7368: Report destination address for TCP requests in logs
+- 2023-04-12 TUN-7134: Acquire token for cloudflared tail
+- 2023-04-12 TUN-7131: Add cloudflared log event to connection messages and enable streaming logs
+- 2023-04-11 TUN-7132 TUN-7136: Add filter support for streaming logs
+- 2023-04-06 TUN-7354: Don't warn for empty ingress rules when using --token
+- 2023-04-06 TUN-7128: Categorize logs from public hostname locations
+- 2023-04-06 TUN-7351: Add streaming logs session ping and timeout
+- 2023-04-06 TUN-7335: Fix cloudflared update not working in windows
+
 2023.4.0
 - 2023-04-07 TUN-7356: Bump golang.org/x/net package to 0.7.0
 - 2023-04-07 TUN-7357: Bump to go 1.19.6

certutil/certutil.go

@@ -1,58 +0,0 @@
-package certutil
-
-import (
-	"encoding/json"
-	"encoding/pem"
-	"fmt"
-)
-
-type namedTunnelToken struct {
-	ZoneID    string `json:"zoneID"`
-	AccountID string `json:"accountID"`
-	APIToken  string `json:"apiToken"`
-}
-
-type OriginCert struct {
-	ZoneID    string
-	APIToken  string
-	AccountID string
-}
-
-func DecodeOriginCert(blocks []byte) (*OriginCert, error) {
-	if len(blocks) == 0 {
-		return nil, fmt.Errorf("Cannot decode empty certificate")
-	}
-	originCert := OriginCert{}
-	block, rest := pem.Decode(blocks)
-	for {
-		if block == nil {
-			break
-		}
-		switch block.Type {
-		case "PRIVATE KEY", "CERTIFICATE":
-			// this is for legacy purposes.
-			break
-		case "ARGO TUNNEL TOKEN":
-			if originCert.ZoneID != "" || originCert.APIToken != "" {
-				return nil, fmt.Errorf("Found multiple tokens in the certificate")
-			}
-			// The token is a string,
-			// Try the newer JSON format
-			ntt := namedTunnelToken{}
-			if err := json.Unmarshal(block.Bytes, &ntt); err == nil {
-				originCert.ZoneID = ntt.ZoneID
-				originCert.APIToken = ntt.APIToken
-				originCert.AccountID = ntt.AccountID
-			}
-		default:
-			return nil, fmt.Errorf("Unknown block %s in the certificate", block.Type)
-		}
-		block, rest = pem.Decode(rest)
-	}
-
-	if originCert.ZoneID == "" || originCert.APIToken == "" {
-		return nil, fmt.Errorf("Missing token in the certificate")
-	}
-
-	return &originCert, nil
-}

certutil/certutil_test.go

@@ -1,51 +0,0 @@
-package certutil
-
-import (
-	"fmt"
-	"io/ioutil"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestLoadOriginCert(t *testing.T) {
-	cert, err := DecodeOriginCert([]byte{})
-	assert.Equal(t, fmt.Errorf("Cannot decode empty certificate"), err)
-	assert.Nil(t, cert)
-
-	blocks, err := ioutil.ReadFile("test-cert-unknown-block.pem")
-	assert.Nil(t, err)
-	cert, err = DecodeOriginCert(blocks)
-	assert.Equal(t, fmt.Errorf("Unknown block RSA PRIVATE KEY in the certificate"), err)
-	assert.Nil(t, cert)
-}
-
-func TestJSONArgoTunnelTokenEmpty(t *testing.T) {
-	cert, err := DecodeOriginCert([]byte{})
-	blocks, err := ioutil.ReadFile("test-cert-no-token.pem")
-	assert.Nil(t, err)
-	cert, err = DecodeOriginCert(blocks)
-	assert.Equal(t, fmt.Errorf("Missing token in the certificate"), err)
-	assert.Nil(t, cert)
-}
-
-func TestJSONArgoTunnelToken(t *testing.T) {
-	// The given cert's Argo Tunnel Token was generated by base64 encoding this JSON:
-	// {
-	// "zoneID": "7b0a4d77dfb881c1a3b7d61ea9443e19",
-	// "apiToken": "test-service-key",
-	// "accountID": "abcdabcdabcdabcd1234567890abcdef"
-	// }
-	CloudflareTunnelTokenTest(t, "test-cloudflare-tunnel-cert-json.pem")
-}
-
-func CloudflareTunnelTokenTest(t *testing.T, path string) {
-	blocks, err := ioutil.ReadFile(path)
-	assert.Nil(t, err)
-	cert, err := DecodeOriginCert(blocks)
-	assert.Nil(t, err)
-	assert.NotNil(t, cert)
-	assert.Equal(t, "7b0a4d77dfb881c1a3b7d61ea9443e19", cert.ZoneID)
-	key := "test-service-key"
-	assert.Equal(t, key, cert.APIToken)
-}

cfapi/client.go

@@ -8,6 +8,7 @@ type TunnelClient interface {
 	CreateTunnel(name string, tunnelSecret []byte) (*TunnelWithToken, error)
 	GetTunnel(tunnelID uuid.UUID) (*Tunnel, error)
 	GetTunnelToken(tunnelID uuid.UUID) (string, error)
+	GetManagementToken(tunnelID uuid.UUID) (string, error)
 	DeleteTunnel(tunnelID uuid.UUID) error
 	ListTunnels(filter *TunnelFilter) ([]*Tunnel, error)
 	ListActiveClients(tunnelID uuid.UUID) ([]*ActiveClient, error)

cfapi/tunnel.go

@@ -50,6 +50,10 @@ type newTunnel struct {
 	TunnelSecret []byte `json:"tunnel_secret"`
 }
 
+type managementRequest struct {
+	Resources []string `json:"resources"`
+}
+
 type CleanupParams struct {
 	queryParams url.Values
 }
@@ -133,6 +137,28 @@ func (r *RESTClient) GetTunnelToken(tunnelID uuid.UUID) (token string, err error
 	return "", r.statusCodeToError("get tunnel token", resp)
 }
 
+func (r *RESTClient) GetManagementToken(tunnelID uuid.UUID) (token string, err error) {
+	endpoint := r.baseEndpoints.accountLevel
+	endpoint.Path = path.Join(endpoint.Path, fmt.Sprintf("%v/management", tunnelID))
+
+	body := &managementRequest{
+		Resources: []string{"logs"},
+	}
+
+	resp, err := r.sendRequest("POST", endpoint, body)
+	if err != nil {
+		return "", errors.Wrap(err, "REST request failed")
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode == http.StatusOK {
+		err = parseResponse(resp.Body, &token)
+		return token, err
+	}
+
+	return "", r.statusCodeToError("get tunnel token", resp)
+}
+
 func (r *RESTClient) DeleteTunnel(tunnelID uuid.UUID) error {
 	endpoint := r.baseEndpoints.accountLevel
 	endpoint.Path = path.Join(endpoint.Path, fmt.Sprintf("%v", tunnelID))

cmd/cloudflared/cliutil/build_info.go

@@ -47,3 +47,7 @@ func (bi *BuildInfo) GetBuildTypeMsg() string {
 	}
 	return fmt.Sprintf(" with %s", bi.BuildType)
 }
+
+func (bi *BuildInfo) UserAgent() string {
+	return fmt.Sprintf("cloudflared/%s", bi.CloudflaredVersion)
+}

cmd/cloudflared/main.go

@@ -90,7 +90,7 @@ func main() {
 	updater.Init(Version)
 	tracing.Init(Version)
 	token.Init(Version)
-	tail.Init(Version)
+	tail.Init(bInfo)
 	runApp(app, graceShutdownC)
 }
 

cmd/cloudflared/tail/cmd.go

@@ -2,6 +2,7 @@ package tail
 
 import (
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -10,28 +11,32 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/google/uuid"
 	"github.com/mattn/go-colorable"
 	"github.com/rs/zerolog"
 	"github.com/urfave/cli/v2"
 	"nhooyr.io/websocket"
 
+	"github.com/cloudflare/cloudflared/cmd/cloudflared/cliutil"
+	"github.com/cloudflare/cloudflared/credentials"
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/management"
 )
 
 var (
-	version string
+	buildInfo *cliutil.BuildInfo
 )
 
-func Init(v string) {
-	version = v
+func Init(bi *cliutil.BuildInfo) {
+	buildInfo = bi
 }
 
 func Command() *cli.Command {
 	return &cli.Command{
-		Name:   "tail",
-		Action: Run,
-		Usage:  "Stream logs from a remote cloudflared",
+		Name:      "tail",
+		Action:    Run,
+		Usage:     "Stream logs from a remote cloudflared",
+		UsageText: "cloudflared tail [tail command options] [TUNNEL-ID]",
 		Flags: []cli.Flag{
 			&cli.StringFlag{
 				Name:    "connector-id",
@@ -39,6 +44,17 @@ func Command() *cli.Command {
 				Value:   "",
 				EnvVars: []string{"TUNNEL_MANAGEMENT_CONNECTOR"},
 			},
+			&cli.StringSliceFlag{
+				Name:    "event",
+				Usage:   "Filter by specific Events (cloudflared, http, tcp, udp) otherwise, defaults to send all events",
+				EnvVars: []string{"TUNNEL_MANAGEMENT_FILTER_EVENTS"},
+			},
+			&cli.StringFlag{
+				Name:    "level",
+				Usage:   "Filter by specific log levels (debug, info, warn, error)",
+				EnvVars: []string{"TUNNEL_MANAGEMENT_FILTER_LEVEL"},
+				Value:   "debug",
+			},
 			&cli.StringFlag{
 				Name:    "token",
 				Usage:   "Access token for a specific tunnel",
@@ -61,9 +77,15 @@ func Command() *cli.Command {
 			&cli.StringFlag{
 				Name:    logger.LogLevelFlag,
 				Value:   "info",
-				Usage:   "Application logging level {debug, info, warn, error, fatal}. ",
+				Usage:   "Application logging level {debug, info, warn, error, fatal}",
 				EnvVars: []string{"TUNNEL_LOGLEVEL"},
 			},
+			&cli.StringFlag{
+				Name:    credentials.OriginCertFlag,
+				Usage:   "Path to the certificate generated for your origin when you run cloudflared login.",
+				EnvVars: []string{"TUNNEL_ORIGIN_CERT"},
+				Value:   credentials.FindDefaultOriginCertPath(),
+			},
 		},
 	}
 }
@@ -113,6 +135,94 @@ func createLogger(c *cli.Context) *zerolog.Logger {
 	return &log
 }
 
+// parseFilters will attempt to parse provided filters to send to with the EventStartStreaming
+func parseFilters(c *cli.Context) (*management.StreamingFilters, error) {
+	var level *management.LogLevel
+	var events []management.LogEventType
+
+	argLevel := c.String("level")
+	argEvents := c.StringSlice("event")
+
+	if argLevel != "" {
+		l, ok := management.ParseLogLevel(argLevel)
+		if !ok {
+			return nil, fmt.Errorf("invalid --level filter provided, please use one of the following Log Levels: debug, info, warn, error")
+		}
+		level = &l
+	}
+
+	for _, v := range argEvents {
+		t, ok := management.ParseLogEventType(v)
+		if !ok {
+			return nil, fmt.Errorf("invalid --event filter provided, please use one of the following EventTypes: cloudflared, http, tcp, udp")
+		}
+		events = append(events, t)
+	}
+
+	if level == nil && len(events) == 0 {
+		// When no filters are provided, do not return a StreamingFilters struct
+		return nil, nil
+	}
+
+	return &management.StreamingFilters{
+		Level:  level,
+		Events: events,
+	}, nil
+}
+
+// getManagementToken will make a call to the Cloudflare API to acquire a management token for the requested tunnel.
+func getManagementToken(c *cli.Context, log *zerolog.Logger) (string, error) {
+	userCreds, err := credentials.Read(c.String(credentials.OriginCertFlag), log)
+	if err != nil {
+		return "", err
+	}
+
+	client, err := userCreds.Client(c.String("api-url"), buildInfo.UserAgent(), log)
+	if err != nil {
+		return "", err
+	}
+
+	tunnelIDString := c.Args().First()
+	if tunnelIDString == "" {
+		return "", errors.New("no tunnel ID provided")
+	}
+	tunnelID, err := uuid.Parse(tunnelIDString)
+	if err != nil {
+		return "", errors.New("unable to parse provided tunnel id as a valid UUID")
+	}
+
+	token, err := client.GetManagementToken(tunnelID)
+	if err != nil {
+		return "", err
+	}
+
+	return token, nil
+}
+
+// buildURL will build the management url to contain the required query parameters to authenticate the request.
+func buildURL(c *cli.Context, log *zerolog.Logger) (url.URL, error) {
+	var err error
+	managementHostname := c.String("management-hostname")
+	token := c.String("token")
+	if token == "" {
+		token, err = getManagementToken(c, log)
+		if err != nil {
+			return url.URL{}, fmt.Errorf("unable to acquire management token for requested tunnel id: %w", err)
+		}
+	}
+	query := url.Values{}
+	query.Add("access_token", token)
+	connector := c.String("connector-id")
+	if connector != "" {
+		connectorID, err := uuid.Parse(connector)
+		if err != nil {
+			return url.URL{}, fmt.Errorf("unabled to parse 'connector-id' flag into a valid UUID: %w", err)
+		}
+		query.Add("connector_id", connectorID.String())
+	}
+	return url.URL{Scheme: "wss", Host: managementHostname, Path: "/logs", RawQuery: query.Encode()}, nil
+}
+
 // Run implements a foreground runner
 func Run(c *cli.Context) error {
 	log := createLogger(c)
@@ -121,12 +231,20 @@ func Run(c *cli.Context) error {
 	signal.Notify(signals, syscall.SIGTERM, syscall.SIGINT)
 	defer signal.Stop(signals)
 
-	managementHostname := c.String("management-hostname")
-	token := c.String("token")
-	u := url.URL{Scheme: "wss", Host: managementHostname, Path: "/logs", RawQuery: "access_token=" + token}
+	filters, err := parseFilters(c)
+	if err != nil {
+		log.Error().Err(err).Msgf("invalid filters provided")
+		return nil
+	}
+
+	u, err := buildURL(c, log)
+	if err != nil {
+		log.Err(err).Msg("unable to construct management request URL")
+		return nil
+	}
 
 	header := make(http.Header)
-	header.Add("User-Agent", "cloudflared/"+version)
+	header.Add("User-Agent", buildInfo.UserAgent())
 	trace := c.String("trace")
 	if trace != "" {
 		header["cf-trace-id"] = []string{trace}
@@ -148,11 +266,17 @@ func Run(c *cli.Context) error {
 	// Once connection is established, send start_streaming event to begin receiving logs
 	err = management.WriteEvent(conn, ctx, &management.EventStartStreaming{
 		ClientEvent: management.ClientEvent{Type: management.StartStreaming},
+		Filters:     filters,
 	})
 	if err != nil {
 		log.Error().Err(err).Msg("unable to request logs from management tunnel")
 		return nil
 	}
+	log.Debug().
+		Str("tunnel-id", c.Args().First()).
+		Str("connector-id", c.String("connector-id")).
+		Interface("filters", filters).
+		Msg("connected")
 
 	readerDone := make(chan struct{})
 
@@ -187,7 +311,12 @@ func Run(c *cli.Context) error {
 					}
 					// Output all the logs received to stdout
 					for _, l := range logs.Logs {
-						fmt.Printf("%s %s %s %s\n", l.Timestamp, l.Level, l.Event, l.Message)
+						fields, err := json.Marshal(l.Fields)
+						if err != nil {
+							fields = []byte("unable to parse fields")
+							log.Debug().Msgf("unable to parse fields from event %+v", l)
+						}
+						fmt.Printf("%s %s %s %s %s\n", l.Time, l.Level, l.Event, l.Message, fields)
 					}
 				case management.UnknownServerEventType:
 					fallthrough

cmd/cloudflared/tunnel/cmd.go

@@ -28,6 +28,7 @@ import (
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/updater"
 	"github.com/cloudflare/cloudflared/config"
 	"github.com/cloudflare/cloudflared/connection"
+	"github.com/cloudflare/cloudflared/credentials"
 	"github.com/cloudflare/cloudflared/features"
 	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/logger"
@@ -751,10 +752,10 @@ func configureCloudflaredFlags(shouldHide bool) []cli.Flag {
 			Hidden: shouldHide,
 		},
 		altsrc.NewStringFlag(&cli.StringFlag{
-			Name:    "origincert",
+			Name:    credentials.OriginCertFlag,
 			Usage:   "Path to the certificate generated for your origin when you run cloudflared login.",
 			EnvVars: []string{"TUNNEL_ORIGIN_CERT"},
-			Value:   findDefaultOriginCertPath(),
+			Value:   credentials.FindDefaultOriginCertPath(),
 			Hidden:  shouldHide,
 		}),
 		altsrc.NewDurationFlag(&cli.DurationFlag{

cmd/cloudflared/tunnel/configuration.go

@@ -3,17 +3,14 @@ package tunnel
 import (
 	"crypto/tls"
 	"fmt"
-	"io/ioutil"
 	mathRand "math/rand"
 	"net"
 	"net/netip"
 	"os"
-	"path/filepath"
 	"strings"
 	"time"
 
 	"github.com/google/uuid"
-	homedir "github.com/mitchellh/go-homedir"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/urfave/cli/v2"
@@ -33,7 +30,6 @@ import (
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
-const LogFieldOriginCertPath = "originCertPath"
 const secretValue = "*****"
 
 var (
@@ -46,18 +42,6 @@ var (
 	configFlags = []string{"autoupdate-freq", "no-autoupdate", "retries", "protocol", "loglevel", "transport-loglevel", "origincert", "metrics", "metrics-update-freq", "edge-ip-version", "edge-bind-address"}
 )
 
-// returns the first path that contains a cert.pem file. If none of the DefaultConfigSearchDirectories
-// contains a cert.pem file, return empty string
-func findDefaultOriginCertPath() string {
-	for _, defaultConfigDir := range config.DefaultConfigSearchDirectories() {
-		originCertPath, _ := homedir.Expand(filepath.Join(defaultConfigDir, config.DefaultCredentialFile))
-		if ok, _ := config.FileExists(originCertPath); ok {
-			return originCertPath
-		}
-	}
-	return ""
-}
-
 func generateRandomClientID(log *zerolog.Logger) (string, error) {
 	u, err := uuid.NewRandom()
 	if err != nil {
@@ -128,62 +112,6 @@ func dnsProxyStandAlone(c *cli.Context, namedTunnel *connection.NamedTunnelPrope
 			namedTunnel != nil) // named tunnel
 }
 
-func findOriginCert(originCertPath string, log *zerolog.Logger) (string, error) {
-	if originCertPath == "" {
-		log.Info().Msgf("Cannot determine default origin certificate path. No file %s in %v", config.DefaultCredentialFile, config.DefaultConfigSearchDirectories())
-		if isRunningFromTerminal() {
-			log.Error().Msgf("You need to specify the origin certificate path with --origincert option, or set TUNNEL_ORIGIN_CERT environment variable. See %s for more information.", argumentsUrl)
-			return "", fmt.Errorf("client didn't specify origincert path when running from terminal")
-		} else {
-			log.Error().Msgf("You need to specify the origin certificate path by specifying the origincert option in the configuration file, or set TUNNEL_ORIGIN_CERT environment variable. See %s for more information.", serviceUrl)
-			return "", fmt.Errorf("client didn't specify origincert path")
-		}
-	}
-	var err error
-	originCertPath, err = homedir.Expand(originCertPath)
-	if err != nil {
-		log.Err(err).Msgf("Cannot resolve origin certificate path")
-		return "", fmt.Errorf("cannot resolve path %s", originCertPath)
-	}
-	// Check that the user has acquired a certificate using the login command
-	ok, err := config.FileExists(originCertPath)
-	if err != nil {
-		log.Error().Err(err).Msgf("Cannot check if origin cert exists at path %s", originCertPath)
-		return "", fmt.Errorf("cannot check if origin cert exists at path %s", originCertPath)
-	}
-	if !ok {
-		log.Error().Msgf(`Cannot find a valid certificate for your origin at the path:
-
-    %s
-
-If the path above is wrong, specify the path with the -origincert option.
-If you don't have a certificate signed by Cloudflare, run the command:
-
-	%s login
-`, originCertPath, os.Args[0])
-		return "", fmt.Errorf("cannot find a valid certificate at the path %s", originCertPath)
-	}
-
-	return originCertPath, nil
-}
-
-func readOriginCert(originCertPath string) ([]byte, error) {
-	// Easier to send the certificate as []byte via RPC than decoding it at this point
-	originCert, err := ioutil.ReadFile(originCertPath)
-	if err != nil {
-		return nil, fmt.Errorf("cannot read %s to load origin certificate", originCertPath)
-	}
-	return originCert, nil
-}
-
-func getOriginCert(originCertPath string, log *zerolog.Logger) ([]byte, error) {
-	if originCertPath, err := findOriginCert(originCertPath, log); err != nil {
-		return nil, err
-	} else {
-		return readOriginCert(originCertPath)
-	}
-}
-
 func prepareTunnelConfig(
 	c *cli.Context,
 	info *cliutil.BuildInfo,

cmd/cloudflared/tunnel/credential_finder.go

@@ -5,6 +5,7 @@ import (
 	"path/filepath"
 
 	"github.com/cloudflare/cloudflared/config"
+	"github.com/cloudflare/cloudflared/credentials"
 
 	"github.com/google/uuid"
 	"github.com/rs/zerolog"
@@ -56,13 +57,13 @@ func newSearchByID(id uuid.UUID, c *cli.Context, log *zerolog.Logger, fs fileSys
 }
 
 func (s searchByID) Path() (string, error) {
-	originCertPath := s.c.String("origincert")
+	originCertPath := s.c.String(credentials.OriginCertFlag)
 	originCertLog := s.log.With().
-		Str(LogFieldOriginCertPath, originCertPath).
+		Str("originCertPath", originCertPath).
 		Logger()
 
 	// Fallback to look for tunnel credentials in the origin cert directory
-	if originCertPath, err := findOriginCert(originCertPath, &originCertLog); err == nil {
+	if originCertPath, err := credentials.FindOriginCert(originCertPath, &originCertLog); err == nil {
 		originCertDir := filepath.Dir(originCertPath)
 		if filePath, err := tunnelFilePath(s.id, originCertDir); err == nil {
 			if s.fs.validFilePath(filePath) {

cmd/cloudflared/tunnel/login.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/cloudflare/cloudflared/cmd/cloudflared/cliutil"
 	"github.com/cloudflare/cloudflared/config"
+	"github.com/cloudflare/cloudflared/credentials"
 	"github.com/cloudflare/cloudflared/logger"
 	"github.com/cloudflare/cloudflared/token"
 )
@@ -85,7 +86,7 @@ func checkForExistingCert() (string, bool, error) {
 	if err != nil {
 		return "", false, err
 	}
-	path := filepath.Join(configPath, config.DefaultCredentialFile)
+	path := filepath.Join(configPath, credentials.DefaultCredentialFile)
 	fileInfo, err := os.Stat(path)
 	if err == nil && fileInfo.Size() > 0 {
 		return path, true, nil

cmd/cloudflared/tunnel/subcommand_context.go

@@ -13,9 +13,9 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/urfave/cli/v2"
 
-	"github.com/cloudflare/cloudflared/certutil"
 	"github.com/cloudflare/cloudflared/cfapi"
 	"github.com/cloudflare/cloudflared/connection"
+	"github.com/cloudflare/cloudflared/credentials"
 	"github.com/cloudflare/cloudflared/logger"
 )
 
@@ -37,7 +37,7 @@ type subcommandContext struct {
 
 	// These fields should be accessed using their respective Getter
 	tunnelstoreClient cfapi.Client
-	userCredential    *userCredential
+	userCredential    *credentials.User
 }
 
 func newSubcommandContext(c *cli.Context) (*subcommandContext, error) {
@@ -56,65 +56,28 @@ func (sc *subcommandContext) credentialFinder(tunnelID uuid.UUID) CredFinder {
 	return newSearchByID(tunnelID, sc.c, sc.log, sc.fs)
 }
 
-type userCredential struct {
-	cert     *certutil.OriginCert
-	certPath string
-}
-
 func (sc *subcommandContext) client() (cfapi.Client, error) {
 	if sc.tunnelstoreClient != nil {
 		return sc.tunnelstoreClient, nil
 	}
-	credential, err := sc.credential()
+	cred, err := sc.credential()
 	if err != nil {
 		return nil, err
 	}
-	userAgent := fmt.Sprintf("cloudflared/%s", buildInfo.Version())
-	client, err := cfapi.NewRESTClient(
-		sc.c.String("api-url"),
-		credential.cert.AccountID,
-		credential.cert.ZoneID,
-		credential.cert.APIToken,
-		userAgent,
-		sc.log,
-	)
-
+	sc.tunnelstoreClient, err = cred.Client(sc.c.String("api-url"), buildInfo.UserAgent(), sc.log)
 	if err != nil {
 		return nil, err
 	}
-	sc.tunnelstoreClient = client
-	return client, nil
+	return sc.tunnelstoreClient, nil
 }
 
-func (sc *subcommandContext) credential() (*userCredential, error) {
+func (sc *subcommandContext) credential() (*credentials.User, error) {
 	if sc.userCredential == nil {
-		originCertPath := sc.c.String("origincert")
-		originCertLog := sc.log.With().
-			Str(LogFieldOriginCertPath, originCertPath).
-			Logger()
-
-		originCertPath, err := findOriginCert(originCertPath, &originCertLog)
+		uc, err := credentials.Read(sc.c.String(credentials.OriginCertFlag), sc.log)
 		if err != nil {
-			return nil, errors.Wrap(err, "Error locating origin cert")
-		}
-		blocks, err := readOriginCert(originCertPath)
-		if err != nil {
-			return nil, errors.Wrapf(err, "Can't read origin cert from %s", originCertPath)
-		}
-
-		cert, err := certutil.DecodeOriginCert(blocks)
-		if err != nil {
-			return nil, errors.Wrap(err, "Error decoding origin cert")
-		}
-
-		if cert.AccountID == "" {
-			return nil, errors.Errorf(`Origin certificate needs to be refreshed before creating new tunnels.\nDelete %s and run "cloudflared login" to obtain a new cert.`, originCertPath)
-		}
-
-		sc.userCredential = &userCredential{
-			cert:     cert,
-			certPath: originCertPath,
+			return nil, err
 		}
+		sc.userCredential = uc
 	}
 	return sc.userCredential, nil
 }
@@ -175,13 +138,13 @@ func (sc *subcommandContext) create(name string, credentialsFilePath string, sec
 		return nil, err
 	}
 	tunnelCredentials := connection.Credentials{
-		AccountTag:   credential.cert.AccountID,
+		AccountTag:   credential.AccountID(),
 		TunnelSecret: tunnelSecret,
 		TunnelID:     tunnel.ID,
 	}
 	usedCertPath := false
 	if credentialsFilePath == "" {
-		originCertDir := filepath.Dir(credential.certPath)
+		originCertDir := filepath.Dir(credential.CertPath())
 		credentialsFilePath, err = tunnelFilePath(tunnelCredentials.TunnelID, originCertDir)
 		if err != nil {
 			return nil, err

cmd/cloudflared/tunnel/subcommand_context_test.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/cloudflare/cloudflared/cfapi"
 	"github.com/cloudflare/cloudflared/connection"
+	"github.com/cloudflare/cloudflared/credentials"
 )
 
 type mockFileSystem struct {
@@ -37,7 +38,7 @@ func Test_subcommandContext_findCredentials(t *testing.T) {
 		log               *zerolog.Logger
 		fs                fileSystem
 		tunnelstoreClient cfapi.Client
-		userCredential    *userCredential
+		userCredential    *credentials.User
 	}
 	type args struct {
 		tunnelID uuid.UUID
@@ -249,7 +250,7 @@ func Test_subcommandContext_Delete(t *testing.T) {
 		isUIEnabled       bool
 		fs                fileSystem
 		tunnelstoreClient *deleteMockTunnelStore
-		userCredential    *userCredential
+		userCredential    *credentials.User
 	}
 	type args struct {
 		tunnelIDs []uuid.UUID

cmd/cloudflared/updater/update.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"runtime"
+	"strings"
 	"time"
 
 	"github.com/facebookgo/grace/gracenet"
@@ -95,11 +96,24 @@ func CheckForUpdate(options updateOptions) (CheckResult, error) {
 		url = StagingUpdateURL
 	}
 
+	if runtime.GOOS == "windows" {
+		cfdPath = encodeWindowsPath(cfdPath)
+	}
+
 	s := NewWorkersService(version, url, cfdPath, Options{IsBeta: options.isBeta,
 		IsForced: options.isForced, RequestedVersion: options.intendedVersion})
 
 	return s.Check()
 }
+func encodeWindowsPath(path string) string {
+	// We do this because Windows allows spaces in directories such as
+	// Program Files but does not allow these directories to be spaced in batch files.
+	targetPath := strings.Replace(path, "Program Files (x86)", "PROGRA~2", -1)
+	// This is to do the same in 32 bit systems. We do this second so that the first
+	// replace is for x86 dirs.
+	targetPath = strings.Replace(targetPath, "Program Files", "PROGRA~1", -1)
+	return targetPath
+}
 
 func applyUpdate(options updateOptions, update CheckResult) UpdateOutcome {
 	if update.Version() == "" || options.updateDisabled {

cmd/cloudflared/updater/workers_update.go

@@ -25,14 +25,13 @@ const (
 	// rename cloudflared.exe.new to cloudflared.exe
 	// delete cloudflared.exe.old
 	// start the service
-	// delete the batch file
-	windowsUpdateCommandTemplate = `@echo off
-sc stop cloudflared >nul 2>&1
+	// exit with code 0 if we've reached this point indicating success.
+	windowsUpdateCommandTemplate = `sc stop cloudflared >nul 2>&1
 rename "{{.TargetPath}}" {{.OldName}}
 rename "{{.NewPath}}" {{.BinaryName}}
 del "{{.OldPath}}"
 sc start cloudflared >nul 2>&1
-del {{.BatchName}}`
+exit /b 0`
 	batchFileName = "cfd_update.bat"
 )
 
@@ -214,8 +213,9 @@ func isValidChecksum(checksum, filePath string) error {
 // writeBatchFile writes a batch file out to disk
 // see the dicussion on why it has to be done this way
 func writeBatchFile(targetPath string, newPath string, oldPath string) error {
-	os.Remove(batchFileName) //remove any failed updates before download
-	f, err := os.Create(batchFileName)
+	batchFilePath := filepath.Join(filepath.Dir(targetPath), batchFileName)
+	os.Remove(batchFilePath) //remove any failed updates before download
+	f, err := os.Create(batchFilePath)
 	if err != nil {
 		return err
 	}
@@ -241,6 +241,16 @@ func writeBatchFile(targetPath string, newPath string, oldPath string) error {
 
 // run each OS command for windows
 func runWindowsBatch(batchFile string) error {
-	cmd := exec.Command("cmd", "/c", batchFile)
-	return cmd.Start()
+	defer os.Remove(batchFile)
+	cmd := exec.Command("cmd", "/C", batchFile)
+	_, err := cmd.Output()
+	// Remove the batch file we created. Don't let this interfere with the error
+	// we report.
+	if err != nil {
+		if exitError, ok := err.(*exec.ExitError); ok {
+			return fmt.Errorf("Error during update : %s;", string(exitError.Stderr))
+		}
+
+	}
+	return err
 }

config/configuration.go

@@ -39,8 +39,6 @@ var (
 )
 
 const (
-	DefaultCredentialFile = "cert.pem"
-
 	// BastionFlag is to enable bastion, or jump host, operation
 	BastionFlag = "bastion"
 )

connection/control.go

@@ -2,13 +2,13 @@ package connection
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"net"
 	"time"
 
 	"github.com/rs/zerolog"
 
+	"github.com/cloudflare/cloudflared/management"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
 )
 
@@ -85,7 +85,7 @@ func (c *controlStream) ServeControlStream(
 		return err
 	}
 
-	c.observer.logServerInfo(c.connIndex, registrationDetails.Location, c.edgeAddress, fmt.Sprintf("Connection %s registered with protocol: %s", registrationDetails.UUID, c.protocol))
+	c.observer.logConnected(registrationDetails.UUID, c.connIndex, registrationDetails.Location, c.edgeAddress, c.protocol)
 	c.observer.sendConnectedEvent(c.connIndex, c.protocol, registrationDetails.Location)
 	c.connectedFuse.Connected()
 
@@ -116,7 +116,11 @@ func (c *controlStream) waitForUnregister(ctx context.Context, rpcClient NamedTu
 
 	c.observer.sendUnregisteringEvent(c.connIndex)
 	rpcClient.GracefulShutdown(ctx, c.gracePeriod)
-	c.observer.log.Info().Uint8(LogFieldConnIndex, c.connIndex).Msg("Unregistered tunnel connection")
+	c.observer.log.Info().
+		Int(management.EventTypeKey, int(management.Cloudflared)).
+		Uint8(LogFieldConnIndex, c.connIndex).
+		IPAddr(LogFieldIPAddress, c.edgeAddress).
+		Msg("Unregistered tunnel connection")
 }
 
 func (c *controlStream) IsStopped() bool {

connection/observer.go

@@ -4,12 +4,17 @@ import (
 	"net"
 	"strings"
 
+	"github.com/google/uuid"
 	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/management"
 )
 
 const (
+	LogFieldConnectionID      = "connection"
 	LogFieldLocation          = "location"
 	LogFieldIPAddress         = "ip"
+	LogFieldProtocol          = "protocol"
 	observerChannelBufferSize = 16
 )
 
@@ -41,13 +46,16 @@ func (o *Observer) RegisterSink(sink EventSink) {
 	o.addSinkChan <- sink
 }
 
-func (o *Observer) logServerInfo(connIndex uint8, location string, address net.IP, msg string) {
+func (o *Observer) logConnected(connectionID uuid.UUID, connIndex uint8, location string, address net.IP, protocol Protocol) {
 	o.sendEvent(Event{Index: connIndex, EventType: Connected, Location: location})
 	o.log.Info().
+		Int(management.EventTypeKey, int(management.Cloudflared)).
+		Str(LogFieldConnectionID, connectionID.String()).
 		Uint8(LogFieldConnIndex, connIndex).
 		Str(LogFieldLocation, location).
 		IPAddr(LogFieldIPAddress, address).
-		Msg(msg)
+		Str(LogFieldProtocol, protocol.String()).
+		Msg("Registered tunnel connection")
 	o.metrics.registerServerLocation(uint8ToString(connIndex), location)
 }
 

credentials/credentials.go

@@ -0,0 +1,83 @@
+package credentials
+
+import (
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/cfapi"
+)
+
+const (
+	logFieldOriginCertPath = "originCertPath"
+)
+
+type User struct {
+	cert     *OriginCert
+	certPath string
+}
+
+func (c User) AccountID() string {
+	return c.cert.AccountID
+}
+
+func (c User) ZoneID() string {
+	return c.cert.ZoneID
+}
+
+func (c User) APIToken() string {
+	return c.cert.APIToken
+}
+
+func (c User) CertPath() string {
+	return c.certPath
+}
+
+// Client uses the user credentials to create a Cloudflare API client
+func (c *User) Client(apiURL string, userAgent string, log *zerolog.Logger) (cfapi.Client, error) {
+	if apiURL == "" {
+		return nil, errors.New("An api-url was not provided for the Cloudflare API client")
+	}
+	client, err := cfapi.NewRESTClient(
+		apiURL,
+		c.cert.AccountID,
+		c.cert.ZoneID,
+		c.cert.APIToken,
+		userAgent,
+		log,
+	)
+
+	if err != nil {
+		return nil, err
+	}
+	return client, nil
+}
+
+// Read will load and read the origin cert.pem to load the user credentials
+func Read(originCertPath string, log *zerolog.Logger) (*User, error) {
+	originCertLog := log.With().
+		Str(logFieldOriginCertPath, originCertPath).
+		Logger()
+
+	originCertPath, err := FindOriginCert(originCertPath, &originCertLog)
+	if err != nil {
+		return nil, errors.Wrap(err, "Error locating origin cert")
+	}
+	blocks, err := readOriginCert(originCertPath)
+	if err != nil {
+		return nil, errors.Wrapf(err, "Can't read origin cert from %s", originCertPath)
+	}
+
+	cert, err := decodeOriginCert(blocks)
+	if err != nil {
+		return nil, errors.Wrap(err, "Error decoding origin cert")
+	}
+
+	if cert.AccountID == "" {
+		return nil, errors.Errorf(`Origin certificate needs to be refreshed before creating new tunnels.\nDelete %s and run "cloudflared login" to obtain a new cert.`, originCertPath)
+	}
+
+	return &User{
+		cert:     cert,
+		certPath: originCertPath,
+	}, nil
+}

credentials/credentials_test.go

@@ -0,0 +1,38 @@
+package credentials
+
+import (
+	"io/fs"
+	"os"
+	"path"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCredentialsRead(t *testing.T) {
+	file, err := os.ReadFile("test-cloudflare-tunnel-cert-json.pem")
+	require.NoError(t, err)
+	dir := t.TempDir()
+	certPath := path.Join(dir, originCertFile)
+	os.WriteFile(certPath, file, fs.ModePerm)
+	user, err := Read(certPath, &nopLog)
+	require.NoError(t, err)
+	require.Equal(t, certPath, user.CertPath())
+	require.Equal(t, "test-service-key", user.APIToken())
+	require.Equal(t, "7b0a4d77dfb881c1a3b7d61ea9443e19", user.ZoneID())
+	require.Equal(t, "abcdabcdabcdabcd1234567890abcdef", user.AccountID())
+}
+
+func TestCredentialsClient(t *testing.T) {
+	user := User{
+		certPath: "/tmp/cert.pem",
+		cert: &OriginCert{
+			ZoneID:    "7b0a4d77dfb881c1a3b7d61ea9443e19",
+			AccountID: "abcdabcdabcdabcd1234567890abcdef",
+			APIToken:  "test-service-key",
+		},
+	}
+	client, err := user.Client("example.com", "cloudflared/test", &nopLog)
+	require.NoError(t, err)
+	require.NotNil(t, client)
+}

credentials/origin_cert.go

@@ -0,0 +1,130 @@
+package credentials
+
+import (
+	"encoding/json"
+	"encoding/pem"
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/mitchellh/go-homedir"
+	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/config"
+)
+
+const (
+	DefaultCredentialFile = "cert.pem"
+	OriginCertFlag        = "origincert"
+)
+
+type namedTunnelToken struct {
+	ZoneID    string `json:"zoneID"`
+	AccountID string `json:"accountID"`
+	APIToken  string `json:"apiToken"`
+}
+
+type OriginCert struct {
+	ZoneID    string
+	APIToken  string
+	AccountID string
+}
+
+// FindDefaultOriginCertPath returns the first path that contains a cert.pem file. If none of the
+// DefaultConfigSearchDirectories contains a cert.pem file, return empty string
+func FindDefaultOriginCertPath() string {
+	for _, defaultConfigDir := range config.DefaultConfigSearchDirectories() {
+		originCertPath, _ := homedir.Expand(filepath.Join(defaultConfigDir, DefaultCredentialFile))
+		if ok := fileExists(originCertPath); ok {
+			return originCertPath
+		}
+	}
+	return ""
+}
+
+func decodeOriginCert(blocks []byte) (*OriginCert, error) {
+	if len(blocks) == 0 {
+		return nil, fmt.Errorf("Cannot decode empty certificate")
+	}
+	originCert := OriginCert{}
+	block, rest := pem.Decode(blocks)
+	for {
+		if block == nil {
+			break
+		}
+		switch block.Type {
+		case "PRIVATE KEY", "CERTIFICATE":
+			// this is for legacy purposes.
+			break
+		case "ARGO TUNNEL TOKEN":
+			if originCert.ZoneID != "" || originCert.APIToken != "" {
+				return nil, fmt.Errorf("Found multiple tokens in the certificate")
+			}
+			// The token is a string,
+			// Try the newer JSON format
+			ntt := namedTunnelToken{}
+			if err := json.Unmarshal(block.Bytes, &ntt); err == nil {
+				originCert.ZoneID = ntt.ZoneID
+				originCert.APIToken = ntt.APIToken
+				originCert.AccountID = ntt.AccountID
+			}
+		default:
+			return nil, fmt.Errorf("Unknown block %s in the certificate", block.Type)
+		}
+		block, rest = pem.Decode(rest)
+	}
+
+	if originCert.ZoneID == "" || originCert.APIToken == "" {
+		return nil, fmt.Errorf("Missing token in the certificate")
+	}
+
+	return &originCert, nil
+}
+
+func readOriginCert(originCertPath string) ([]byte, error) {
+	originCert, err := os.ReadFile(originCertPath)
+	if err != nil {
+		return nil, fmt.Errorf("cannot read %s to load origin certificate", originCertPath)
+	}
+
+	return originCert, nil
+}
+
+// FindOriginCert will check to make sure that the certificate exists at the specified file path.
+func FindOriginCert(originCertPath string, log *zerolog.Logger) (string, error) {
+	if originCertPath == "" {
+		log.Error().Msgf("Cannot determine default origin certificate path. No file %s in %v. You need to specify the origin certificate path by specifying the origincert option in the configuration file, or set TUNNEL_ORIGIN_CERT environment variable", DefaultCredentialFile, config.DefaultConfigSearchDirectories())
+		return "", fmt.Errorf("client didn't specify origincert path")
+	}
+	var err error
+	originCertPath, err = homedir.Expand(originCertPath)
+	if err != nil {
+		log.Err(err).Msgf("Cannot resolve origin certificate path")
+		return "", fmt.Errorf("cannot resolve path %s", originCertPath)
+	}
+	// Check that the user has acquired a certificate using the login command
+	ok := fileExists(originCertPath)
+	if !ok {
+		log.Error().Msgf(`Cannot find a valid certificate for your origin at the path:
+
+    %s
+
+If the path above is wrong, specify the path with the -origincert option.
+If you don't have a certificate signed by Cloudflare, run the command:
+
+	cloudflared login
+`, originCertPath)
+		return "", fmt.Errorf("cannot find a valid certificate at the path %s", originCertPath)
+	}
+
+	return originCertPath, nil
+}
+
+// FileExists checks to see if a file exist at the provided path.
+func fileExists(path string) bool {
+	fileStat, err := os.Stat(path)
+	if err != nil {
+		return false
+	}
+	return !fileStat.IsDir()
+}

credentials/origin_cert_test.go

@@ -0,0 +1,110 @@
+package credentials
+
+import (
+	"fmt"
+	"io/fs"
+	"os"
+	"path"
+	"testing"
+
+	"github.com/rs/zerolog"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	originCertFile = "cert.pem"
+)
+
+var (
+	nopLog = zerolog.Nop().With().Logger()
+)
+
+func TestLoadOriginCert(t *testing.T) {
+	cert, err := decodeOriginCert([]byte{})
+	assert.Equal(t, fmt.Errorf("Cannot decode empty certificate"), err)
+	assert.Nil(t, cert)
+
+	blocks, err := os.ReadFile("test-cert-unknown-block.pem")
+	assert.NoError(t, err)
+	cert, err = decodeOriginCert(blocks)
+	assert.Equal(t, fmt.Errorf("Unknown block RSA PRIVATE KEY in the certificate"), err)
+	assert.Nil(t, cert)
+}
+
+func TestJSONArgoTunnelTokenEmpty(t *testing.T) {
+	blocks, err := os.ReadFile("test-cert-no-token.pem")
+	assert.NoError(t, err)
+	cert, err := decodeOriginCert(blocks)
+	assert.Equal(t, fmt.Errorf("Missing token in the certificate"), err)
+	assert.Nil(t, cert)
+}
+
+func TestJSONArgoTunnelToken(t *testing.T) {
+	// The given cert's Argo Tunnel Token was generated by base64 encoding this JSON:
+	// {
+	// "zoneID": "7b0a4d77dfb881c1a3b7d61ea9443e19",
+	// "apiToken": "test-service-key",
+	// "accountID": "abcdabcdabcdabcd1234567890abcdef"
+	// }
+	CloudflareTunnelTokenTest(t, "test-cloudflare-tunnel-cert-json.pem")
+}
+
+func CloudflareTunnelTokenTest(t *testing.T, path string) {
+	blocks, err := os.ReadFile(path)
+	assert.NoError(t, err)
+	cert, err := decodeOriginCert(blocks)
+	assert.NoError(t, err)
+	assert.NotNil(t, cert)
+	assert.Equal(t, "7b0a4d77dfb881c1a3b7d61ea9443e19", cert.ZoneID)
+	key := "test-service-key"
+	assert.Equal(t, key, cert.APIToken)
+}
+
+type mockFile struct {
+	path string
+	data []byte
+	err  error
+}
+
+type mockFileSystem struct {
+	files map[string]mockFile
+}
+
+func newMockFileSystem(files ...mockFile) *mockFileSystem {
+	fs := mockFileSystem{map[string]mockFile{}}
+	for _, f := range files {
+		fs.files[f.path] = f
+	}
+	return &fs
+}
+
+func (fs *mockFileSystem) ReadFile(path string) ([]byte, error) {
+	if f, ok := fs.files[path]; ok {
+		return f.data, f.err
+	}
+	return nil, os.ErrNotExist
+}
+
+func (fs *mockFileSystem) ValidFilePath(path string) bool {
+	_, exists := fs.files[path]
+	return exists
+}
+
+func TestFindOriginCert_Valid(t *testing.T) {
+	file, err := os.ReadFile("test-cloudflare-tunnel-cert-json.pem")
+	require.NoError(t, err)
+	dir := t.TempDir()
+	certPath := path.Join(dir, originCertFile)
+	os.WriteFile(certPath, file, fs.ModePerm)
+	path, err := FindOriginCert(certPath, &nopLog)
+	require.NoError(t, err)
+	require.Equal(t, certPath, path)
+}
+
+func TestFindOriginCert_Missing(t *testing.T) {
+	dir := t.TempDir()
+	certPath := path.Join(dir, originCertFile)
+	_, err := FindOriginCert(certPath, &nopLog)
+	require.Error(t, err)
+}

edgediscovery/allregions/discovery.go

@@ -9,6 +9,8 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
+
+	"github.com/cloudflare/cloudflared/management"
 )
 
 const (
@@ -108,16 +110,20 @@ var friendlyDNSErrorLines = []string{
 
 // EdgeDiscovery implements HA service discovery lookup.
 func edgeDiscovery(log *zerolog.Logger, srvService string) ([][]*EdgeAddr, error) {
-	log.Debug().Str("domain", "_"+srvService+"._"+srvProto+"."+srvName).Msg("looking up edge SRV record")
+	logger := log.With().Int(management.EventTypeKey, int(management.Cloudflared)).Logger()
+	logger.Debug().
+		Int(management.EventTypeKey, int(management.Cloudflared)).
+		Str("domain", "_"+srvService+"._"+srvProto+"."+srvName).
+		Msg("edge discovery: looking up edge SRV record")
 
 	_, addrs, err := netLookupSRV(srvService, srvProto, srvName)
 	if err != nil {
 		_, fallbackAddrs, fallbackErr := fallbackLookupSRV(srvService, srvProto, srvName)
 		if fallbackErr != nil || len(fallbackAddrs) == 0 {
 			// use the original DNS error `err` in messages, not `fallbackErr`
-			log.Err(err).Msg("Error looking up Cloudflare edge IPs: the DNS query failed")
+			logger.Err(err).Msg("edge discovery: error looking up Cloudflare edge IPs: the DNS query failed")
 			for _, s := range friendlyDNSErrorLines {
-				log.Error().Msg(s)
+				logger.Error().Msg(s)
 			}
 			return nil, errors.Wrapf(err, "Could not lookup srv records on _%v._%v.%v", srvService, srvProto, srvName)
 		}
@@ -131,9 +137,13 @@ func edgeDiscovery(log *zerolog.Logger, srvService string) ([][]*EdgeAddr, error
 		if err != nil {
 			return nil, err
 		}
-		for _, e := range edgeAddrs {
-			log.Debug().Msgf("Edge Address: %+v", *e)
+		logAddrs := make([]string, len(edgeAddrs))
+		for i, e := range edgeAddrs {
+			logAddrs[i] = e.UDP.IP.String()
 		}
+		logger.Debug().
+			Strs("addresses", logAddrs).
+			Msg("edge discovery: resolved edge addresses")
 		resolvedAddrPerCNAME = append(resolvedAddrPerCNAME, edgeAddrs)
 	}
 
@@ -188,13 +198,15 @@ func ResolveAddrs(addrs []string, log *zerolog.Logger) (resolved []*EdgeAddr) {
 	for _, addr := range addrs {
 		tcpAddr, err := net.ResolveTCPAddr("tcp", addr)
 		if err != nil {
-			log.Error().Str(logFieldAddress, addr).Err(err).Msg("failed to resolve to TCP address")
+			log.Error().Int(management.EventTypeKey, int(management.Cloudflared)).
+				Str(logFieldAddress, addr).Err(err).Msg("edge discovery: failed to resolve to TCP address")
 			continue
 		}
 
 		udpAddr, err := net.ResolveUDPAddr("udp", addr)
 		if err != nil {
-			log.Error().Str(logFieldAddress, addr).Err(err).Msg("failed to resolve to UDP address")
+			log.Error().Int(management.EventTypeKey, int(management.Cloudflared)).
+				Str(logFieldAddress, addr).Err(err).Msg("edge discovery: failed to resolve to UDP address")
 			continue
 		}
 		version := V6

edgediscovery/edgediscovery.go

@@ -6,6 +6,7 @@ import (
 	"github.com/rs/zerolog"
 
 	"github.com/cloudflare/cloudflared/edgediscovery/allregions"
+	"github.com/cloudflare/cloudflared/management"
 )
 
 const (
@@ -74,33 +75,35 @@ func (ed *Edge) GetAddrForRPC() (*allregions.EdgeAddr, error) {
 
 // GetAddr gives this proxy connection an edge Addr. Prefer Addrs this connection has already used.
 func (ed *Edge) GetAddr(connIndex int) (*allregions.EdgeAddr, error) {
-	log := ed.log.With().Int(LogFieldConnIndex, connIndex).Logger()
+	log := ed.log.With().
+		Int(LogFieldConnIndex, connIndex).
+		Int(management.EventTypeKey, int(management.Cloudflared)).
+		Logger()
 	ed.Lock()
 	defer ed.Unlock()
 
 	// If this connection has already used an edge addr, return it.
 	if addr := ed.regions.AddrUsedBy(connIndex); addr != nil {
-		log.Debug().Msg("edgediscovery - GetAddr: Returning same address back to proxy connection")
+		log.Debug().IPAddr(LogFieldIPAddress, addr.UDP.IP).Msg("edge discovery: returning same edge address back to pool")
 		return addr, nil
 	}
 
 	// Otherwise, give it an unused one
 	addr := ed.regions.GetUnusedAddr(nil, connIndex)
 	if addr == nil {
-		log.Debug().Msg("edgediscovery - GetAddr: No addresses left to give proxy connection")
+		log.Debug().Msg("edge discovery: no addresses left in pool to give proxy connection")
 		return nil, errNoAddressesLeft
 	}
-	log = ed.log.With().
-		Int(LogFieldConnIndex, connIndex).
-		IPAddr(LogFieldIPAddress, addr.UDP.IP).Logger()
-	log.Debug().Msgf("edgediscovery - GetAddr: Giving connection its new address")
+	log.Debug().IPAddr(LogFieldIPAddress, addr.UDP.IP).Msg("edge discovery: giving new address to connection")
 	return addr, nil
 }
 
 // GetDifferentAddr gives back the proxy connection's edge Addr and uses a new one.
 func (ed *Edge) GetDifferentAddr(connIndex int, hasConnectivityError bool) (*allregions.EdgeAddr, error) {
-	log := ed.log.With().Int(LogFieldConnIndex, connIndex).Logger()
-
+	log := ed.log.With().
+		Int(LogFieldConnIndex, connIndex).
+		Int(management.EventTypeKey, int(management.Cloudflared)).
+		Logger()
 	ed.Lock()
 	defer ed.Unlock()
 
@@ -110,14 +113,14 @@ func (ed *Edge) GetDifferentAddr(connIndex int, hasConnectivityError bool) (*all
 	}
 	addr := ed.regions.GetUnusedAddr(oldAddr, connIndex)
 	if addr == nil {
-		log.Debug().Msg("edgediscovery - GetDifferentAddr: No addresses left to give proxy connection")
+		log.Debug().Msg("edge discovery: no addresses left in pool to give proxy connection")
 		// note: if oldAddr were not nil, it will become available on the next iteration
 		return nil, errNoAddressesLeft
 	}
-	log = ed.log.With().
-		Int(LogFieldConnIndex, connIndex).
-		IPAddr(LogFieldIPAddress, addr.UDP.IP).Logger()
-	log.Debug().Msgf("edgediscovery - GetDifferentAddr: Giving connection its new address from the address list: %v", ed.regions.AvailableAddrs())
+	log.Debug().
+		IPAddr(LogFieldIPAddress, addr.UDP.IP).
+		Int("available", ed.regions.AvailableAddrs()).
+		Msg("edge discovery: giving new address to connection")
 	return addr, nil
 }
 
@@ -133,8 +136,9 @@ func (ed *Edge) AvailableAddrs() int {
 func (ed *Edge) GiveBack(addr *allregions.EdgeAddr, hasConnectivityError bool) bool {
 	ed.Lock()
 	defer ed.Unlock()
-	log := ed.log.With().
-		IPAddr(LogFieldIPAddress, addr.UDP.IP).Logger()
-	log.Debug().Msgf("edgediscovery - GiveBack: Address now unused")
+	ed.log.Debug().
+		Int(management.EventTypeKey, int(management.Cloudflared)).
+		IPAddr(LogFieldIPAddress, addr.UDP.IP).
+		Msg("edge discovery: gave back address to the pool")
 	return ed.regions.GiveBack(addr, hasConnectivityError)
 }

features/features.go

@@ -16,6 +16,7 @@ var (
 		FeatureSerializedHeaders,
 		FeatureDatagramV2,
 		FeatureQUICSupportEOF,
+		FeatureManagementLogs,
 	}
 )
 

ingress/ingress.go

@@ -94,7 +94,7 @@ func ParseIngress(conf *config.Configuration) (Ingress, error) {
 // ParseIngressFromConfigAndCLI will parse the configuration rules from config files for ingress
 // rules and then attempt to parse CLI for ingress rules.
 // Will always return at least one valid ingress rule. If none are provided by the user, the default
-// will be to return 502 status code for all incoming requests.
+// will be to return 503 status code for all incoming requests.
 func ParseIngressFromConfigAndCLI(conf *config.Configuration, c *cli.Context, log *zerolog.Logger) (Ingress, error) {
 	// Attempt to parse ingress rules from configuration
 	ingressRules, err := ParseIngress(conf)
@@ -110,7 +110,11 @@ func ParseIngressFromConfigAndCLI(conf *config.Configuration, c *cli.Context, lo
 	//   --bastion for ssh bastion service
 	ingressRules, err = parseCLIIngress(c, false)
 	if errors.Is(err, ErrNoIngressRulesCLI) {
-		log.Warn().Msgf(ErrNoIngressRulesCLI.Error())
+		// Only log a warning if the tunnel is not a remotely managed tunnel and the config
+		// will be loaded after connecting.
+		if !c.IsSet("token") {
+			log.Warn().Msgf(ErrNoIngressRulesCLI.Error())
+		}
 		return newDefaultOrigin(c, log), nil
 	}
 	if err != nil {

management/events.go

@@ -48,7 +48,12 @@ type ClientEvent struct {
 // Additional filters can be provided to augment the log events requested.
 type EventStartStreaming struct {
 	ClientEvent
-	Filters []string `json:"filters"`
+	Filters *StreamingFilters `json:"filters,omitempty"`
+}
+
+type StreamingFilters struct {
+	Events []LogEventType `json:"events,omitempty"`
+	Level  *LogLevel      `json:"level,omitempty"`
 }
 
 // EventStopStreaming signifies that the client wishes to halt receiving log events.
@@ -59,21 +64,37 @@ type EventStopStreaming struct {
 // EventLog is the event that the server sends to the client with the log events.
 type EventLog struct {
 	ServerEvent
-	Logs []Log `json:"logs"`
+	Logs []*Log `json:"logs"`
 }
 
 // LogEventType is the way that logging messages are able to be filtered.
 // Example: assigning LogEventType.Cloudflared to a zerolog event will allow the client to filter for only
 // the Cloudflared-related events.
-type LogEventType int
+type LogEventType int8
 
 const (
-	Cloudflared LogEventType = 0
-	HTTP        LogEventType = 1
-	TCP         LogEventType = 2
-	UDP         LogEventType = 3
+	// Cloudflared events are signficant to cloudflared operations like connection state changes.
+	// Cloudflared is also the default event type for any events that haven't been separated into a proper event type.
+	Cloudflared LogEventType = iota
+	HTTP
+	TCP
+	UDP
 )
 
+func ParseLogEventType(s string) (LogEventType, bool) {
+	switch s {
+	case "cloudflared":
+		return Cloudflared, true
+	case "http":
+		return HTTP, true
+	case "tcp":
+		return TCP, true
+	case "udp":
+		return UDP, true
+	}
+	return -1, false
+}
+
 func (l LogEventType) String() string {
 	switch l {
 	case Cloudflared:
@@ -89,24 +110,99 @@ func (l LogEventType) String() string {
 	}
 }
 
+func (l LogEventType) MarshalJSON() ([]byte, error) {
+	return json.Marshal(l.String())
+}
+
+func (e *LogEventType) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return errors.New("unable to unmarshal LogEventType string")
+	}
+	if event, ok := ParseLogEventType(s); ok {
+		*e = event
+		return nil
+	}
+	return errors.New("unable to unmarshal LogEventType")
+}
+
 // LogLevel corresponds to the zerolog logging levels
 // "panic", "fatal", and "trace" are exempt from this list as they are rarely used and, at least
 // the the first two are limited to failure conditions that lead to cloudflared shutting down.
-type LogLevel string
+type LogLevel int8
 
 const (
-	Debug LogLevel = "debug"
-	Info  LogLevel = "info"
-	Warn  LogLevel = "warn"
-	Error LogLevel = "error"
+	Debug LogLevel = 0
+	Info  LogLevel = 1
+	Warn  LogLevel = 2
+	Error LogLevel = 3
+)
+
+func ParseLogLevel(l string) (LogLevel, bool) {
+	switch l {
+	case "debug":
+		return Debug, true
+	case "info":
+		return Info, true
+	case "warn":
+		return Warn, true
+	case "error":
+		return Error, true
+	}
+	return -1, false
+}
+
+func (l LogLevel) String() string {
+	switch l {
+	case Debug:
+		return "debug"
+	case Info:
+		return "info"
+	case Warn:
+		return "warn"
+	case Error:
+		return "error"
+	default:
+		return ""
+	}
+}
+
+func (l LogLevel) MarshalJSON() ([]byte, error) {
+	return json.Marshal(l.String())
+}
+
+func (l *LogLevel) UnmarshalJSON(data []byte) error {
+	var s string
+	if err := json.Unmarshal(data, &s); err != nil {
+		return errors.New("unable to unmarshal LogLevel string")
+	}
+	if level, ok := ParseLogLevel(s); ok {
+		*l = level
+		return nil
+	}
+	return fmt.Errorf("unable to unmarshal LogLevel")
+}
+
+const (
+	// TimeKey aligns with the zerolog.TimeFieldName
+	TimeKey = "time"
+	// LevelKey aligns with the zerolog.LevelFieldName
+	LevelKey = "level"
+	// LevelKey aligns with the zerolog.MessageFieldName
+	MessageKey = "message"
+	// EventTypeKey is the custom JSON key of the LogEventType in ZeroLogEvent
+	EventTypeKey = "event"
+	// FieldsKey is a custom JSON key to match and store every other key for a zerolog event
+	FieldsKey = "fields"
 )
 
 // Log is the basic structure of the events that are sent to the client.
 type Log struct {
-	Event     LogEventType `json:"event"`
-	Timestamp string       `json:"timestamp"`
-	Level     LogLevel     `json:"level"`
-	Message   string       `json:"message"`
+	Time    string                 `json:"time,omitempty"`
+	Level   LogLevel               `json:"level,omitempty"`
+	Message string                 `json:"message,omitempty"`
+	Event   LogEventType           `json:"event,omitempty"`
+	Fields  map[string]interface{} `json:"fields,omitempty"`
 }
 
 // IntoClientEvent unmarshals the provided ClientEvent into the proper type.

management/events_test.go

@@ -11,14 +11,83 @@ import (
 	"github.com/cloudflare/cloudflared/internal/test"
 )
 
+var (
+	debugLevel *LogLevel
+	infoLevel  *LogLevel
+	warnLevel  *LogLevel
+	errorLevel *LogLevel
+)
+
+func init() {
+	// created here because we can't do a reference to a const enum, i.e. &Info
+	debugLevel := new(LogLevel)
+	*debugLevel = Debug
+	infoLevel := new(LogLevel)
+	*infoLevel = Info
+	warnLevel := new(LogLevel)
+	*warnLevel = Warn
+	errorLevel := new(LogLevel)
+	*errorLevel = Error
+}
+
 func TestIntoClientEvent_StartStreaming(t *testing.T) {
-	event := ClientEvent{
-		Type:  StartStreaming,
-		event: []byte(`{"type": "start_streaming"}`),
+	for _, test := range []struct {
+		name     string
+		expected EventStartStreaming
+	}{
+		{
+			name:     "no filters",
+			expected: EventStartStreaming{ClientEvent: ClientEvent{Type: StartStreaming}},
+		},
+		{
+			name: "level filter",
+			expected: EventStartStreaming{
+				ClientEvent: ClientEvent{Type: StartStreaming},
+				Filters: &StreamingFilters{
+					Level: infoLevel,
+				},
+			},
+		},
+		{
+			name: "events filter",
+			expected: EventStartStreaming{
+				ClientEvent: ClientEvent{Type: StartStreaming},
+				Filters: &StreamingFilters{
+					Events: []LogEventType{Cloudflared, HTTP},
+				},
+			},
+		},
+		{
+			name: "level and events filters",
+			expected: EventStartStreaming{
+				ClientEvent: ClientEvent{Type: StartStreaming},
+				Filters: &StreamingFilters{
+					Level:  infoLevel,
+					Events: []LogEventType{Cloudflared},
+				},
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			data, err := json.Marshal(test.expected)
+			require.NoError(t, err)
+			event := ClientEvent{}
+			err = json.Unmarshal(data, &event)
+			require.NoError(t, err)
+			event.event = data
+			ce, ok := IntoClientEvent[EventStartStreaming](&event, StartStreaming)
+			require.True(t, ok)
+			require.Equal(t, test.expected.ClientEvent, ce.ClientEvent)
+			if test.expected.Filters != nil {
+				f := ce.Filters
+				ef := test.expected.Filters
+				if ef.Level != nil {
+					require.Equal(t, *ef.Level, *f.Level)
+				}
+				require.ElementsMatch(t, ef.Events, f.Events)
+			}
+		})
 	}
-	ce, ok := IntoClientEvent[EventStartStreaming](&event, StartStreaming)
-	require.True(t, ok)
-	require.Equal(t, EventStartStreaming{ClientEvent: ClientEvent{Type: StartStreaming}}, *ce)
 }
 
 func TestIntoClientEvent_StopStreaming(t *testing.T) {
@@ -62,12 +131,12 @@ func TestIntoServerEvent_Invalid(t *testing.T) {
 func TestReadServerEvent(t *testing.T) {
 	sentEvent := EventLog{
 		ServerEvent: ServerEvent{Type: Logs},
-		Logs: []Log{
+		Logs: []*Log{
 			{
-				Timestamp: time.Now().UTC().Format(time.RFC3339),
-				Event:     HTTP,
-				Level:     Info,
-				Message:   "test",
+				Time:    time.Now().UTC().Format(time.RFC3339),
+				Event:   HTTP,
+				Level:   Info,
+				Message: "test",
 			},
 		},
 	}

management/logger.go

@@ -18,14 +18,6 @@ const (
 	logWindow = 30
 )
 
-// ZeroLogEvent is the json structure that zerolog stores it's events as
-type ZeroLogEvent struct {
-	Time    string       `json:"time,omitempty"`
-	Level   LogLevel     `json:"level,omitempty"`
-	Type    LogEventType `json:"type,omitempty"`
-	Message string       `json:"message,omitempty"`
-}
-
 // Logger manages the number of management streaming log sessions
 type Logger struct {
 	sessions []*Session
@@ -48,29 +40,63 @@ func NewLogger() *Logger {
 }
 
 type LoggerListener interface {
-	Listen() *Session
+	Listen(*StreamingFilters) *Session
 	Close(*Session)
 }
 
 type Session struct {
 	// Buffered channel that holds the recent log events
-	listener chan *ZeroLogEvent
+	listener chan *Log
 	// Types of log events that this session will provide through the listener
-	filters []LogEventType
+	filters *StreamingFilters
 }
 
-func newListener(size int) *Session {
-	return &Session{
-		listener: make(chan *ZeroLogEvent, size),
-		filters:  []LogEventType{},
+func newSession(size int, filters *StreamingFilters) *Session {
+	s := &Session{
+		listener: make(chan *Log, size),
+	}
+	if filters != nil {
+		s.filters = filters
+	} else {
+		s.filters = &StreamingFilters{}
+	}
+	return s
+}
+
+// Insert attempts to insert the log to the session. If the log event matches the provided session filters, it
+// will be applied to the listener.
+func (s *Session) Insert(log *Log) {
+	// Level filters are optional
+	if s.filters.Level != nil {
+		if *s.filters.Level > log.Level {
+			return
+		}
+	}
+	// Event filters are optional
+	if len(s.filters.Events) != 0 && !contains(s.filters.Events, log.Event) {
+		return
+	}
+	select {
+	case s.listener <- log:
+	default:
+		// buffer is full, discard
 	}
 }
 
+func contains(array []LogEventType, t LogEventType) bool {
+	for _, v := range array {
+		if v == t {
+			return true
+		}
+	}
+	return false
+}
+
 // Listen creates a new Session that will append filtered log events as they are created.
-func (l *Logger) Listen() *Session {
+func (l *Logger) Listen(filters *StreamingFilters) *Session {
 	l.mu.Lock()
 	defer l.mu.Unlock()
-	listener := newListener(logWindow)
+	listener := newSession(logWindow, filters)
 	l.sessions = append(l.sessions, listener)
 	return listener
 }
@@ -104,35 +130,14 @@ func (l *Logger) Write(p []byte) (int, error) {
 	if len(l.sessions) == 0 {
 		return len(p), nil
 	}
-	var event ZeroLogEvent
-	iter := json.BorrowIterator(p)
-	defer json.ReturnIterator(iter)
-	iter.ReadVal(&event)
-	if iter.Error != nil {
-		l.Log.Debug().Msg("unable to unmarshal log event")
+	event, err := parseZerologEvent(p)
+	// drop event if unable to parse properly
+	if err != nil {
+		l.Log.Debug().Msg("unable to parse log event")
 		return len(p), nil
 	}
-	for _, listener := range l.sessions {
-		// no filters means all types are allowed
-		if len(listener.filters) != 0 {
-			valid := false
-			// make sure listener is subscribed to this event type
-			for _, t := range listener.filters {
-				if t == event.Type {
-					valid = true
-					break
-				}
-			}
-			if !valid {
-				continue
-			}
-		}
-
-		select {
-		case listener.listener <- &event:
-		default:
-			// buffer is full, discard
-		}
+	for _, session := range l.sessions {
+		session.Insert(event)
 	}
 	return len(p), nil
 }
@@ -140,3 +145,62 @@ func (l *Logger) Write(p []byte) (int, error) {
 func (l *Logger) WriteLevel(level zerolog.Level, p []byte) (n int, err error) {
 	return l.Write(p)
 }
+
+func parseZerologEvent(p []byte) (*Log, error) {
+	var fields map[string]interface{}
+	iter := json.BorrowIterator(p)
+	defer json.ReturnIterator(iter)
+	iter.ReadVal(&fields)
+	if iter.Error != nil {
+		return nil, iter.Error
+	}
+	logTime := time.Now().UTC().Format(zerolog.TimeFieldFormat)
+	if t, ok := fields[TimeKey]; ok {
+		if t, ok := t.(string); ok {
+			logTime = t
+		}
+	}
+	logLevel := Debug
+	// A zerolog Debug event can be created and then an error can be added
+	// via .Err(error), if so, we upgrade the level to error.
+	if _, hasError := fields["error"]; hasError {
+		logLevel = Error
+	} else {
+		if level, ok := fields[LevelKey]; ok {
+			if level, ok := level.(string); ok {
+				if logLevel, ok = ParseLogLevel(level); !ok {
+					logLevel = Debug
+				}
+			}
+		}
+	}
+	// Assume the event type is Cloudflared if unable to parse/find. This could be from log events that haven't
+	// yet been tagged with the appropriate EventType yet.
+	logEvent := Cloudflared
+	e := fields[EventTypeKey]
+	if e != nil {
+		if eventNumber, ok := e.(float64); ok {
+			logEvent = LogEventType(eventNumber)
+		}
+	}
+	logMessage := ""
+	if m, ok := fields[MessageKey]; ok {
+		if m, ok := m.(string); ok {
+			logMessage = m
+		}
+	}
+	event := Log{
+		Time:    logTime,
+		Level:   logLevel,
+		Event:   logEvent,
+		Message: logMessage,
+	}
+	// Remove the keys that have top level keys on Log
+	delete(fields, TimeKey)
+	delete(fields, LevelKey)
+	delete(fields, EventTypeKey)
+	delete(fields, MessageKey)
+	// The rest of the keys go into the Fields
+	event.Fields = fields
+	return &event, nil
+}

management/logger_test.go

@@ -2,9 +2,11 @@ package management
 
 import (
 	"testing"
+	"time"
 
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 // No listening sessions will not write to the channel
@@ -20,14 +22,15 @@ func TestLoggerWrite_OneSession(t *testing.T) {
 	logger := NewLogger()
 	zlog := zerolog.New(logger).With().Timestamp().Logger().Level(zerolog.InfoLevel)
 
-	session := logger.Listen()
+	session := logger.Listen(nil)
 	defer logger.Close(session)
-	zlog.Info().Int("type", int(HTTP)).Msg("hello")
+	zlog.Info().Int(EventTypeKey, int(HTTP)).Msg("hello")
 	select {
 	case event := <-session.listener:
+		assert.NotEmpty(t, event.Time)
 		assert.Equal(t, "hello", event.Message)
-		assert.Equal(t, LogLevel("info"), event.Level)
-		assert.Equal(t, HTTP, event.Type)
+		assert.Equal(t, Info, event.Level)
+		assert.Equal(t, HTTP, event.Event)
 	default:
 		assert.Fail(t, "expected an event to be in the listener")
 	}
@@ -38,16 +41,17 @@ func TestLoggerWrite_MultipleSessions(t *testing.T) {
 	logger := NewLogger()
 	zlog := zerolog.New(logger).With().Timestamp().Logger().Level(zerolog.InfoLevel)
 
-	session1 := logger.Listen()
+	session1 := logger.Listen(nil)
 	defer logger.Close(session1)
-	session2 := logger.Listen()
-	zlog.Info().Int("type", int(HTTP)).Msg("hello")
+	session2 := logger.Listen(nil)
+	zlog.Info().Int(EventTypeKey, int(HTTP)).Msg("hello")
 	for _, session := range []*Session{session1, session2} {
 		select {
 		case event := <-session.listener:
+			assert.NotEmpty(t, event.Time)
 			assert.Equal(t, "hello", event.Message)
-			assert.Equal(t, LogLevel("info"), event.Level)
-			assert.Equal(t, HTTP, event.Type)
+			assert.Equal(t, Info, event.Level)
+			assert.Equal(t, HTTP, event.Event)
 		default:
 			assert.Fail(t, "expected an event to be in the listener")
 		}
@@ -55,12 +59,13 @@ func TestLoggerWrite_MultipleSessions(t *testing.T) {
 
 	// Close session2 and make sure session1 still receives events
 	logger.Close(session2)
-	zlog.Info().Int("type", int(HTTP)).Msg("hello2")
+	zlog.Info().Int(EventTypeKey, int(HTTP)).Msg("hello2")
 	select {
 	case event := <-session1.listener:
+		assert.NotEmpty(t, event.Time)
 		assert.Equal(t, "hello2", event.Message)
-		assert.Equal(t, LogLevel("info"), event.Level)
-		assert.Equal(t, HTTP, event.Type)
+		assert.Equal(t, Info, event.Level)
+		assert.Equal(t, HTTP, event.Event)
 	default:
 		assert.Fail(t, "expected an event to be in the listener")
 	}
@@ -73,3 +78,160 @@ func TestLoggerWrite_MultipleSessions(t *testing.T) {
 		// pass
 	}
 }
+
+// Validate that the session filters events
+func TestSession_Insert(t *testing.T) {
+	infoLevel := new(LogLevel)
+	*infoLevel = Info
+	warnLevel := new(LogLevel)
+	*warnLevel = Warn
+	for _, test := range []struct {
+		name      string
+		filters   StreamingFilters
+		expectLog bool
+	}{
+		{
+			name:      "none",
+			expectLog: true,
+		},
+		{
+			name: "level",
+			filters: StreamingFilters{
+				Level: infoLevel,
+			},
+			expectLog: true,
+		},
+		{
+			name: "filtered out level",
+			filters: StreamingFilters{
+				Level: warnLevel,
+			},
+			expectLog: false,
+		},
+		{
+			name: "events",
+			filters: StreamingFilters{
+				Events: []LogEventType{HTTP},
+			},
+			expectLog: true,
+		},
+		{
+			name: "filtered out event",
+			filters: StreamingFilters{
+				Events: []LogEventType{Cloudflared},
+			},
+			expectLog: false,
+		},
+		{
+			name: "filter and event",
+			filters: StreamingFilters{
+				Level:  infoLevel,
+				Events: []LogEventType{HTTP},
+			},
+			expectLog: true,
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			session := newSession(4, &test.filters)
+			log := Log{
+				Time:    time.Now().UTC().Format(time.RFC3339),
+				Event:   HTTP,
+				Level:   Info,
+				Message: "test",
+			}
+			session.Insert(&log)
+			select {
+			case <-session.listener:
+				require.True(t, test.expectLog)
+			default:
+				require.False(t, test.expectLog)
+			}
+		})
+	}
+}
+
+// Validate that the session has a max amount of events to hold
+func TestSession_InsertOverflow(t *testing.T) {
+	session := newSession(1, nil)
+	log := Log{
+		Time:    time.Now().UTC().Format(time.RFC3339),
+		Event:   HTTP,
+		Level:   Info,
+		Message: "test",
+	}
+	// Insert 2 but only max channel size for 1
+	session.Insert(&log)
+	session.Insert(&log)
+	select {
+	case <-session.listener:
+		// pass
+	default:
+		require.Fail(t, "expected one log event")
+	}
+	// Second dequeue should fail
+	select {
+	case <-session.listener:
+		require.Fail(t, "expected no more remaining log events")
+	default:
+		// pass
+	}
+}
+
+type mockWriter struct {
+	event *Log
+	err   error
+}
+
+func (m *mockWriter) Write(p []byte) (int, error) {
+	m.event, m.err = parseZerologEvent(p)
+	return len(p), nil
+}
+
+// Validate all event types are set properly
+func TestParseZerologEvent_EventTypes(t *testing.T) {
+	writer := mockWriter{}
+	zlog := zerolog.New(&writer).With().Timestamp().Logger().Level(zerolog.InfoLevel)
+
+	for _, test := range []LogEventType{
+		Cloudflared,
+		HTTP,
+		TCP,
+		UDP,
+	} {
+		t.Run(test.String(), func(t *testing.T) {
+			defer func() { writer.err = nil }()
+			zlog.Info().Int(EventTypeKey, int(test)).Msg("test")
+			require.NoError(t, writer.err)
+			require.Equal(t, test, writer.event.Event)
+		})
+	}
+
+	// Invalid defaults to Cloudflared LogEventType
+	t.Run("invalid", func(t *testing.T) {
+		defer func() { writer.err = nil }()
+		zlog.Info().Str(EventTypeKey, "unknown").Msg("test")
+		require.NoError(t, writer.err)
+		require.Equal(t, Cloudflared, writer.event.Event)
+	})
+}
+
+// Validate top-level keys are removed from Fields
+func TestParseZerologEvent_Fields(t *testing.T) {
+	writer := mockWriter{}
+	zlog := zerolog.New(&writer).With().Timestamp().Logger().Level(zerolog.InfoLevel)
+	zlog.Info().Int(EventTypeKey, int(Cloudflared)).Str("test", "test").Msg("test message")
+	require.NoError(t, writer.err)
+	event := writer.event
+	require.NotEmpty(t, event.Time)
+	require.Equal(t, Cloudflared, event.Event)
+	require.Equal(t, Info, event.Level)
+	require.Equal(t, "test message", event.Message)
+
+	// Make sure Fields doesn't have other set keys used in the Log struct
+	require.NotEmpty(t, event.Fields)
+	require.Equal(t, "test", event.Fields["test"])
+	require.NotContains(t, event.Fields, EventTypeKey)
+	require.NotContains(t, event.Fields, LevelKey)
+	require.NotContains(t, event.Fields, MessageKey)
+	require.NotContains(t, event.Fields, TimeKey)
+}

management/service.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/rs/zerolog"
@@ -19,6 +20,9 @@ const (
 	// value will return this error to incoming requests.
 	StatusSessionLimitExceeded websocket.StatusCode = 4002
 	reasonSessionLimitExceeded                      = "limit exceeded for streaming sessions"
+
+	StatusIdleLimitExceeded websocket.StatusCode = 4003
+	reasonIdleLimitExceeded                      = "session was idle for too long"
 )
 
 type ManagementService struct {
@@ -96,12 +100,7 @@ func (m *ManagementService) streamLogs(c *websocket.Conn, ctx context.Context, s
 		case event := <-session.listener:
 			err := WriteEvent(c, ctx, &EventLog{
 				ServerEvent: ServerEvent{Type: Logs},
-				Logs: []Log{{
-					Event:     Cloudflared,
-					Timestamp: event.Time,
-					Level:     event.Level,
-					Message:   event.Message,
-				}},
+				Logs:        []*Log{event},
 			})
 			if err != nil {
 				// If the client (or the server) already closed the connection, don't attempt to close it again
@@ -132,13 +131,13 @@ func (m *ManagementService) startStreaming(c *websocket.Conn, ctx context.Contex
 		return
 	}
 	// Expect the first incoming request
-	_, ok := IntoClientEvent[EventStartStreaming](event, StartStreaming)
+	startEvent, ok := IntoClientEvent[EventStartStreaming](event, StartStreaming)
 	if !ok {
-		m.log.Err(c.Close(StatusInvalidCommand, reasonInvalidCommand)).Msgf("expected start_streaming as first recieved event")
+		m.log.Warn().Err(c.Close(StatusInvalidCommand, reasonInvalidCommand)).Msgf("expected start_streaming as first recieved event")
 		return
 	}
 	m.streaming.Store(true)
-	listener := m.logger.Listen()
+	listener := m.logger.Listen(startEvent.Filters)
 	m.log.Debug().Msgf("Streaming logs")
 	go m.streamLogs(c, ctx, listener)
 }
@@ -152,10 +151,20 @@ func (m *ManagementService) logs(w http.ResponseWriter, r *http.Request) {
 	}
 	// Make sure the connection is closed if other go routines fail to close the connection after completing.
 	defer c.Close(websocket.StatusInternalError, "")
-	ctx := r.Context()
+	ctx, cancel := context.WithCancel(r.Context())
+	defer cancel()
 	events := make(chan *ClientEvent)
 	go m.readEvents(c, ctx, events)
 
+	// Send a heartbeat ping to hold the connection open even if not streaming.
+	ping := time.NewTicker(15 * time.Second)
+	defer ping.Stop()
+
+	// Close the connection if no operation has occurred after the idle timeout.
+	idleTimeout := 5 * time.Minute
+	idle := time.NewTimer(idleTimeout)
+	defer idle.Stop()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -165,9 +174,11 @@ func (m *ManagementService) logs(w http.ResponseWriter, r *http.Request) {
 		case event := <-events:
 			switch event.Type {
 			case StartStreaming:
+				idle.Stop()
 				m.startStreaming(c, ctx, event)
 				continue
 			case StopStreaming:
+				idle.Reset(idleTimeout)
 				// TODO: limit StopStreaming to only halt streaming for clients that are already streaming
 				m.streaming.Store(false)
 			case UnknownClientEventType:
@@ -181,6 +192,11 @@ func (m *ManagementService) logs(w http.ResponseWriter, r *http.Request) {
 				}
 				return
 			}
+		case <-ping.C:
+			go c.Ping(ctx)
+		case <-idle.C:
+			c.Close(StatusIdleLimitExceeded, reasonIdleLimitExceeded)
+			return
 		}
 	}
 }

proxy/proxy.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cloudflare/cloudflared/cfio"
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/management"
 	"github.com/cloudflare/cloudflared/stream"
 	"github.com/cloudflare/cloudflared/tracing"
 	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
@@ -25,10 +26,12 @@ const (
 	// TagHeaderNamePrefix indicates a Cloudflared Warp Tag prefix that gets appended for warp traffic stream headers.
 	TagHeaderNamePrefix   = "Cf-Warp-Tag-"
 	LogFieldCFRay         = "cfRay"
+	LogFieldLBProbe       = "lbProbe"
 	LogFieldRule          = "ingressRule"
 	LogFieldOriginService = "originService"
 	LogFieldFlowID        = "flowID"
 	LogFieldConnIndex     = "connIndex"
+	LogFieldDestAddr      = "destAddr"
 
 	trailerHeaderName = "Trailer"
 )
@@ -169,14 +172,22 @@ func (p *Proxy) ProxyTCP(
 
 	tracedCtx := tracing.NewTracedContext(serveCtx, req.CfTraceID, p.log)
 
-	p.log.Debug().Str(LogFieldFlowID, req.FlowID).Uint8(LogFieldConnIndex, req.ConnIndex).Msg("tcp proxy stream started")
+	p.log.Debug().
+		Str(LogFieldFlowID, req.FlowID).
+		Str(LogFieldDestAddr, req.Dest).
+		Uint8(LogFieldConnIndex, req.ConnIndex).
+		Msg("tcp proxy stream started")
 
 	if err := p.proxyStream(tracedCtx, rwa, req.Dest, p.warpRouting.Proxy); err != nil {
 		p.logRequestError(err, req.CFRay, req.FlowID, "", ingress.ServiceWarpRouting)
 		return err
 	}
 
-	p.log.Debug().Str(LogFieldFlowID, req.FlowID).Uint8(LogFieldConnIndex, req.ConnIndex).Msg("tcp proxy stream finished successfully")
+	p.log.Debug().
+		Str(LogFieldFlowID, req.FlowID).
+		Str(LogFieldDestAddr, req.Dest).
+		Uint8(LogFieldConnIndex, req.ConnIndex).
+		Msg("tcp proxy stream finished successfully")
 
 	return nil
 }
@@ -339,7 +350,7 @@ func (p *Proxy) appendTagHeaders(r *http.Request) {
 type logFields struct {
 	cfRay     string
 	lbProbe   bool
-	rule      interface{}
+	rule      int
 	flowID    string
 	connIndex uint8
 }
@@ -353,45 +364,41 @@ func copyTrailers(w connection.ResponseWriter, response *http.Response) {
 }
 
 func (p *Proxy) logRequest(r *http.Request, fields logFields) {
+	log := p.log.With().Int(management.EventTypeKey, int(management.HTTP)).Logger()
+	event := log.Debug()
 	if fields.cfRay != "" {
-		p.log.Debug().Msgf("CF-RAY: %s %s %s %s", fields.cfRay, r.Method, r.URL, r.Proto)
-	} else if fields.lbProbe {
-		p.log.Debug().Msgf("CF-RAY: %s Load Balancer health check %s %s %s", fields.cfRay, r.Method, r.URL, r.Proto)
-	} else {
-		p.log.Debug().Msgf("All requests should have a CF-RAY header. Please open a support ticket with Cloudflare. %s %s %s ", r.Method, r.URL, r.Proto)
+		event = event.Str(LogFieldCFRay, fields.cfRay)
 	}
-	p.log.Debug().
-		Str("CF-RAY", fields.cfRay).
-		Str("Header", fmt.Sprintf("%+v", r.Header)).
+	if fields.lbProbe {
+		event = event.Bool(LogFieldLBProbe, fields.lbProbe)
+	}
+	if fields.cfRay == "" && !fields.lbProbe {
+		log.Debug().Msgf("All requests should have a CF-RAY header. Please open a support ticket with Cloudflare. %s %s %s ", r.Method, r.URL, r.Proto)
+	}
+	event.
+		Uint8(LogFieldConnIndex, fields.connIndex).
 		Str("host", r.Host).
 		Str("path", r.URL.Path).
-		Interface("rule", fields.rule).
-		Uint8(LogFieldConnIndex, fields.connIndex).
-		Msg("Inbound request")
-
-	if contentLen := r.ContentLength; contentLen == -1 {
-		p.log.Debug().Msgf("CF-RAY: %s Request Content length unknown", fields.cfRay)
-	} else {
-		p.log.Debug().Msgf("CF-RAY: %s Request content length %d", fields.cfRay, contentLen)
-	}
+		Interface(LogFieldRule, fields.rule).
+		Interface("headers", r.Header).
+		Int64("content-length", r.ContentLength).
+		Msgf("%s %s %s", r.Method, r.URL, r.Proto)
 }
 
 func (p *Proxy) logOriginResponse(resp *http.Response, fields logFields) {
 	responseByCode.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc()
+	event := p.log.Debug()
 	if fields.cfRay != "" {
-		p.log.Debug().Uint8(LogFieldConnIndex, fields.connIndex).Msgf("CF-RAY: %s Status: %s served by ingress %d", fields.cfRay, resp.Status, fields.rule)
-	} else if fields.lbProbe {
-		p.log.Debug().Uint8(LogFieldConnIndex, fields.connIndex).Msgf("Response to Load Balancer health check %s", resp.Status)
-	} else {
-		p.log.Debug().Uint8(LogFieldConnIndex, fields.connIndex).Msgf("Status: %s served by ingress %v", resp.Status, fields.rule)
+		event = event.Str(LogFieldCFRay, fields.cfRay)
 	}
-	p.log.Debug().Uint8(LogFieldConnIndex, fields.connIndex).Msgf("CF-RAY: %s Response Headers %+v", fields.cfRay, resp.Header)
-
-	if contentLen := resp.ContentLength; contentLen == -1 {
-		p.log.Debug().Uint8(LogFieldConnIndex, fields.connIndex).Msgf("CF-RAY: %s Response content length unknown", fields.cfRay)
-	} else {
-		p.log.Debug().Uint8(LogFieldConnIndex, fields.connIndex).Msgf("CF-RAY: %s Response content length %d", fields.cfRay, contentLen)
+	if fields.lbProbe {
+		event = event.Bool(LogFieldLBProbe, fields.lbProbe)
 	}
+	event.
+		Int(management.EventTypeKey, int(management.HTTP)).
+		Uint8(LogFieldConnIndex, fields.connIndex).
+		Int64("content-length", resp.ContentLength).
+		Msgf("%s", resp.Status)
 }
 
 func (p *Proxy) logRequestError(err error, cfRay string, flowID string, rule, service string) {
@@ -401,7 +408,9 @@ func (p *Proxy) logRequestError(err error, cfRay string, flowID string, rule, se
 		log = log.Str(LogFieldCFRay, cfRay)
 	}
 	if flowID != "" {
-		log = log.Str(LogFieldFlowID, flowID)
+		log = log.Str(LogFieldFlowID, flowID).Int(management.EventTypeKey, int(management.TCP))
+	} else {
+		log = log.Int(management.EventTypeKey, int(management.HTTP))
 	}
 	if rule != "" {
 		log = log.Str(LogFieldRule, rule)
@@ -409,7 +418,7 @@ func (p *Proxy) logRequestError(err error, cfRay string, flowID string, rule, se
 	if service != "" {
 		log = log.Str(LogFieldOriginService, service)
 	}
-	log.Msg("")
+	log.Send()
 }
 
 func getDestFromRule(rule *ingress.Rule, req *http.Request) (string, error) {

supervisor/tunnel.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cloudflare/cloudflared/features"
 	"github.com/cloudflare/cloudflared/h2mux"
 	"github.com/cloudflare/cloudflared/ingress"
+	"github.com/cloudflare/cloudflared/management"
 	"github.com/cloudflare/cloudflared/orchestration"
 	quicpogs "github.com/cloudflare/cloudflared/quic"
 	"github.com/cloudflare/cloudflared/retry"
@@ -238,6 +239,7 @@ func (e *EdgeTunnelServer) Serve(ctx context.Context, connIndex uint8, protocolF
 	}
 
 	logger := e.config.Log.With().
+		Int(management.EventTypeKey, int(management.Cloudflared)).
 		IPAddr(connection.LogFieldIPAddress, addr.UDP.IP).
 		Uint8(connection.LogFieldConnIndex, connIndex).
 		Logger()