Release 2023.3.1

The previous logic of var == x86 never fired for 386 arch windows
systems causing us to set ProgramFiles64Folder for the older windows
versions causing downloads to default to a different location. This
change fixes that.

.teamcity/update-homebrew.sh

@@ -14,17 +14,17 @@ if [[ ! -f "$FILENAME" ]] ; then
     exit 1
 fi
 
-if [[ "${GITHUB_PRIVATE_KEY:-}" == "" ]] ; then
-    echo "Missing GITHUB_PRIVATE_KEY"
+if [[ "${GITHUB_PRIVATE_KEY_B64:-}" == "" ]] ; then
+    echo "Missing GITHUB_PRIVATE_KEY_B64"
     exit 1
 fi
 
 # upload to s3 bucket for use by Homebrew formula
 s3cmd \
-    --acl-public --signature-v2 --access_key="$AWS_ACCESS_KEY_ID" --secret_key="$AWS_SECRET_ACCESS_KEY" --host-bucket="%(bucket)s.s3.cfdata.org" \
+    --acl-public --access_key="$AWS_ACCESS_KEY_ID" --secret_key="$AWS_SECRET_ACCESS_KEY" --host-bucket="%(bucket)s.s3.cfdata.org" \
     put "$FILENAME" "s3://cftunnel-docs/dl/cloudflared-$VERSION-darwin-amd64.tgz"
 s3cmd \
-    --acl-public --signature-v2 --access_key="$AWS_ACCESS_KEY_ID" --secret_key="$AWS_SECRET_ACCESS_KEY" --host-bucket="%(bucket)s.s3.cfdata.org" \
+    --acl-public --access_key="$AWS_ACCESS_KEY_ID" --secret_key="$AWS_SECRET_ACCESS_KEY" --host-bucket="%(bucket)s.s3.cfdata.org" \
     cp "s3://cftunnel-docs/dl/cloudflared-$VERSION-darwin-amd64.tgz" "s3://cftunnel-docs/dl/cloudflared-stable-darwin-amd64.tgz"
 SHA256=$(sha256sum "$FILENAME" | cut -b1-64)
 

CHANGES.md

@@ -1,3 +1,7 @@
+## 2023.3.1
+### Breaking Change
+- Running a tunnel without ingress rules defined in configuration file nor from the CLI flags will no longer provide a default ingress rule to localhost:8080 and instead will return HTTP response code 503 for all incoming HTTP requests.
+
 ## 2023.2.2
 ### Notices
 - Legacy tunnels were officially deprecated on December 1, 2022. Starting with this version, cloudflared no longer supports connecting legacy tunnels.

RELEASE_NOTES

@@ -1,3 +1,14 @@
+2023.3.1
+- 2023-03-13 TUN-7271: Return 503 status code when no ingress rules configured
+- 2023-03-10 TUN-7272: Fix cloudflared returning non supported status service which breaks configuration migration
+- 2023-03-09 TUN-7259: Add warning for missing ingress rules
+- 2023-03-09 TUN-7268: Default to Program Files as location for win32
+- 2023-03-07 TUN-7252: Remove h2mux connection
+- 2023-03-07 TUN-7253: Adopt http.ResponseWriter for connection.ResponseWriter
+- 2023-03-06 TUN-7245: Add bastion flag to origin service check
+- 2023-03-06 EDGESTORE-108: Remove deprecated s3v2 signature
+- 2023-03-02 TUN-7226: Fixed a missed rename
+
 2023.3.0
 - 2023-03-01 GH-352: Add Tunnel CLI option "edge-bind-address" (#870)
 - 2023-03-01 Fixed WIX template to allow MSI upgrades (#838)

cloudflared.wxs

@@ -1,9 +1,9 @@
 <?xml version="1.0"?>
 
-<?if $(var.Platform)="x86"?>
-    <?define Program_Files="ProgramFilesFolder"?>
-<?else ?>
+<?if $(var.Platform)="x64" ?>
     <?define Program_Files="ProgramFiles64Folder"?>
+<?else ?>
+    <?define Program_Files="ProgramFilesFolder"?>
 <?endif ?>
 <?ifndef var.Version?>
     <?error Undefined Version variable?>

cmd/cloudflared/tunnel/cmd.go

@@ -199,7 +199,7 @@ func TunnelCommand(c *cli.Context) error {
 	// Run a quick tunnel
 	// A unauthenticated named tunnel hosted on <random>.<quick-tunnels-service>.com
 	// We don't support running proxy-dns and a quick tunnel at the same time as the same process
-	shouldRunQuickTunnel := c.IsSet("url") || c.IsSet("hello-world")
+	shouldRunQuickTunnel := c.IsSet("url") || c.IsSet(ingress.HelloWorldFlag)
 	if !c.IsSet("proxy-dns") && c.String("quick-service") != "" && shouldRunQuickTunnel {
 		return RunQuickTunnel(sc)
 	}
@@ -215,6 +215,9 @@ func TunnelCommand(c *cli.Context) error {
 	}
 
 	if c.IsSet("proxy-dns") {
+		if shouldRunQuickTunnel {
+			return fmt.Errorf("running a quick tunnel with `proxy-dns` is not supported")
+		}
 		// NamedTunnelProperties are nil since proxy dns server does not need it.
 		// This is supported for legacy reasons: dns proxy server is not a tunnel and ideally should
 		// not run as part of cloudflared tunnel.
@@ -786,7 +789,7 @@ func configureProxyFlags(shouldHide bool) []cli.Flag {
 			Hidden:  shouldHide,
 		}),
 		altsrc.NewBoolFlag(&cli.BoolFlag{
-			Name:    "hello-world",
+			Name:    ingress.HelloWorldFlag,
 			Value:   false,
 			Usage:   "Run Hello World Server",
 			EnvVars: []string{"TUNNEL_HELLO_WORLD"},

cmd/cloudflared/tunnel/configuration.go

@@ -25,7 +25,6 @@ import (
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/edgediscovery"
 	"github.com/cloudflare/cloudflared/edgediscovery/allregions"
-	"github.com/cloudflare/cloudflared/h2mux"
 	"github.com/cloudflare/cloudflared/ingress"
 	"github.com/cloudflare/cloudflared/orchestration"
 	"github.com/cloudflare/cloudflared/supervisor"
@@ -125,7 +124,7 @@ func isSecretEnvVar(key string) bool {
 func dnsProxyStandAlone(c *cli.Context, namedTunnel *connection.NamedTunnelProperties) bool {
 	return c.IsSet("proxy-dns") &&
 		!(c.IsSet("name") || // adhoc-named tunnel
-			c.IsSet("hello-world") || // quick or named tunnel
+			c.IsSet(ingress.HelloWorldFlag) || // quick or named tunnel
 			namedTunnel != nil) // named tunnel
 }
 
@@ -228,23 +227,10 @@ func prepareTunnelConfig(
 		Arch:     info.OSArch(),
 	}
 	cfg := config.GetConfiguration()
-	ingressRules, err := ingress.ParseIngress(cfg)
-	if err != nil && err != ingress.ErrNoIngressRules {
+	ingressRules, err := ingress.ParseIngressFromConfigAndCLI(cfg, c, log)
+	if err != nil {
 		return nil, nil, err
 	}
-	if c.IsSet("url") {
-		// Ingress rules cannot be provided with --url flag
-		if !ingressRules.IsEmpty() {
-			return nil, nil, ingress.ErrURLIncompatibleWithIngress
-		} else {
-			// Only for quick or adhoc tunnels will we attempt to parse:
-			// --url, --hello-world, or --unix-socket flag for a tunnel ingress rule
-			ingressRules, err = ingress.NewSingleOrigin(c, false)
-			if err != nil {
-				return nil, nil, err
-			}
-		}
-	}
 
 	protocolSelector, err := connection.NewProtocolSelector(transportProtocol, namedTunnel.Credentials.AccountTag, c.IsSet(TunnelTokenFlag), c.Bool("post-quantum"), edgediscovery.ProtocolPercentage, connection.ResolveTTL, log)
 	if err != nil {
@@ -272,14 +258,6 @@ func prepareTunnelConfig(
 	if err != nil {
 		return nil, nil, err
 	}
-	muxerConfig := &connection.MuxerConfig{
-		HeartbeatInterval: c.Duration("heartbeat-interval"),
-		// Note TUN-3758 , we use Int because UInt is not supported with altsrc
-		MaxHeartbeats: uint64(c.Int("heartbeat-count")),
-		// Note TUN-3758 , we use Int because UInt is not supported with altsrc
-		CompressionSetting: h2mux.CompressionSetting(uint64(c.Int("compression-quality"))),
-		MetricsUpdateFreq:  c.Duration("metrics-update-freq"),
-	}
 	edgeIPVersion, err := parseConfigIPVersion(c.String("edge-ip-version"))
 	if err != nil {
 		return nil, nil, err
@@ -328,7 +306,6 @@ func prepareTunnelConfig(
 		Retries:            uint(c.Int("retries")),
 		RunFromTerminal:    isRunningFromTerminal(),
 		NamedTunnel:        namedTunnel,
-		MuxerConfig:        muxerConfig,
 		ProtocolSelector:   protocolSelector,
 		EdgeTLSConfigs:     edgeTLSConfigs,
 		NeedPQ:             needPQ,

component-tests/config.py

@@ -30,6 +30,7 @@ class NamedTunnelBaseConfig(BaseConfig):
     tunnel: str = None
     credentials_file: str = None
     ingress: list = None
+    hostname: str = None
 
     def __post_init__(self):
         if self.tunnel is None:
@@ -41,8 +42,10 @@ class NamedTunnelBaseConfig(BaseConfig):
 
     def merge_config(self, additional):
         config = super(NamedTunnelBaseConfig, self).merge_config(additional)
-        config['tunnel'] = self.tunnel
-        config['credentials-file'] = self.credentials_file
+        if 'tunnel' not in config:
+            config['tunnel'] = self.tunnel
+        if 'credentials-file' not in config:
+            config['credentials-file'] = self.credentials_file
         # In some cases we want to override default ingress, such as in config tests
         if 'ingress' not in config:
             config['ingress'] = self.ingress
@@ -61,7 +64,7 @@ class NamedTunnelConfig(NamedTunnelBaseConfig):
                            self.merge_config(additional_config))
 
     def get_url(self):
-        return "https://" + self.ingress[0]['hostname']
+        return "https://" + self.hostname
 
     def base_config(self):
         config = self.full_config.copy()
@@ -84,28 +87,9 @@ class NamedTunnelConfig(NamedTunnelBaseConfig):
     def get_credentials_json(self):
         with open(self.credentials_file) as json_file:
             return json.load(json_file)
-
-
+        
 @dataclass(frozen=True)
-class ClassicTunnelBaseConfig(BaseConfig):
-    hostname: str = None
-    origincert: str = None
-
-    def __post_init__(self):
-        if self.hostname is None:
-            raise TypeError("Field tunnel is not set")
-        if self.origincert is None:
-            raise TypeError("Field credentials_file is not set")
-
-    def merge_config(self, additional):
-        config = super(ClassicTunnelBaseConfig, self).merge_config(additional)
-        config['hostname'] = self.hostname
-        config['origincert'] = self.origincert
-        return config
-
-
-@dataclass(frozen=True)
-class ClassicTunnelConfig(ClassicTunnelBaseConfig):
+class QuickTunnelConfig(BaseConfig):
     full_config: dict = None
     additional_config: InitVar[dict] = {}
 
@@ -115,10 +99,6 @@ class ClassicTunnelConfig(ClassicTunnelBaseConfig):
         object.__setattr__(self, 'full_config',
                            self.merge_config(additional_config))
 
-    def get_url(self):
-        return "https://" + self.hostname
-
-
 @dataclass(frozen=True)
 class ProxyDnsConfig(BaseConfig):
     full_config = {

component-tests/conftest.py

@@ -5,14 +5,14 @@ from time import sleep
 import pytest
 import yaml
 
-from config import NamedTunnelConfig, ClassicTunnelConfig, ProxyDnsConfig
+from config import NamedTunnelConfig, ProxyDnsConfig, QuickTunnelConfig
 from constants import BACKOFF_SECS, PROXY_DNS_PORT
 from util import LOGGER
 
 
 class CfdModes(Enum):
     NAMED = auto()
-    CLASSIC = auto()
+    QUICK = auto()
     PROXY_DNS = auto()
 
 
@@ -26,7 +26,7 @@ def component_tests_config():
         config = yaml.safe_load(stream)
         LOGGER.info(f"component tests base config {config}")
 
-        def _component_tests_config(additional_config={}, cfd_mode=CfdModes.NAMED, run_proxy_dns=True):
+        def _component_tests_config(additional_config={}, cfd_mode=CfdModes.NAMED, run_proxy_dns=True, provide_ingress=True):
             if run_proxy_dns:
                 # Regression test for TUN-4177, running with proxy-dns should not prevent tunnels from running.
                 # So we run all tests with it.
@@ -36,18 +36,25 @@ def component_tests_config():
                 additional_config.pop("proxy-dns", None)
                 additional_config.pop("proxy-dns-port", None)
 
+            # Allows the ingress rules to be omitted from the provided config
+            ingress = []
+            if provide_ingress:
+                ingress = config['ingress']
+
+            # Provide the hostname to allow routing to the tunnel even if the ingress rule isn't defined in the config
+            hostname = config['ingress'][0]['hostname']
+
             if cfd_mode is CfdModes.NAMED:
                 return NamedTunnelConfig(additional_config=additional_config,
                                          cloudflared_binary=config['cloudflared_binary'],
                                          tunnel=config['tunnel'],
                                          credentials_file=config['credentials_file'],
-                                         ingress=config['ingress'])
-            elif cfd_mode is CfdModes.CLASSIC:
-                return ClassicTunnelConfig(
-                    additional_config=additional_config, cloudflared_binary=config['cloudflared_binary'],
-                    hostname=config['classic_hostname'], origincert=config['origincert'])
+                                         ingress=ingress,
+                                         hostname=hostname)
             elif cfd_mode is CfdModes.PROXY_DNS:
                 return ProxyDnsConfig(cloudflared_binary=config['cloudflared_binary'])
+            elif cfd_mode is CfdModes.QUICK:
+                return QuickTunnelConfig(additional_config=additional_config, cloudflared_binary=config['cloudflared_binary'])
             else:
                 raise Exception(f"Unknown cloudflared mode {cfd_mode}")
 

component-tests/constants.py

@@ -7,4 +7,4 @@ PROXY_DNS_PORT = 9053
 
 
 def protocols():
-    return ["h2mux", "http2", "quic"]
+    return ["http2", "quic"]

component-tests/test_quicktunnels.py

@@ -0,0 +1,33 @@
+#!/usr/bin/env python
+from conftest import CfdModes
+from constants import METRICS_PORT
+from util import LOGGER, start_cloudflared, wait_tunnel_ready, get_quicktunnel_url, send_requests
+
+class TestQuickTunnels:
+    def test_quick_tunnel(self, tmp_path, component_tests_config):
+        config = component_tests_config(cfd_mode=CfdModes.QUICK, run_proxy_dns=False)
+        LOGGER.debug(config)
+        with start_cloudflared(tmp_path, config, cfd_args=["--hello-world"], new_process=True):
+            wait_tunnel_ready(require_min_connections=4)
+            url = get_quicktunnel_url()
+            send_requests(url, 3, True)
+    
+    def test_quick_tunnel_url(self, tmp_path, component_tests_config):
+        config = component_tests_config(cfd_mode=CfdModes.QUICK, run_proxy_dns=False)
+        LOGGER.debug(config)
+        with start_cloudflared(tmp_path, config, cfd_args=["--url", f"http://localhost:{METRICS_PORT}/"], new_process=True):
+            wait_tunnel_ready()
+            url = get_quicktunnel_url()
+            send_requests(url+"/ready", 3, True)
+
+    def test_quick_tunnel_proxy_dns_url(self, tmp_path, component_tests_config):
+        config = component_tests_config(cfd_mode=CfdModes.QUICK, run_proxy_dns=True)
+        LOGGER.debug(config)
+        failed_start = start_cloudflared(tmp_path, config, cfd_args=["--url", f"http://localhost:{METRICS_PORT}/"], expect_success=False)
+        assert failed_start.returncode == 1, "Expected cloudflared to fail to run with `proxy-dns` and `hello-world`"
+    
+    def test_quick_tunnel_proxy_dns_hello_world(self, tmp_path, component_tests_config):
+        config = component_tests_config(cfd_mode=CfdModes.QUICK, run_proxy_dns=True)
+        LOGGER.debug(config)
+        failed_start = start_cloudflared(tmp_path, config, cfd_args=["--hello-world"], expect_success=False)
+        assert failed_start.returncode == 1, "Expected cloudflared to fail to run with `proxy-dns` and `url`"

component-tests/test_tunnel.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python
+import requests
+from conftest import CfdModes
+from constants import METRICS_PORT, MAX_RETRIES, BACKOFF_SECS
+from retrying import retry
+from util import LOGGER, start_cloudflared, wait_tunnel_ready, send_requests
+
+class TestTunnel:
+    '''Test tunnels with no ingress rules from config.yaml but ingress rules from CLI only'''
+
+    def test_tunnel_hello_world(self, tmp_path, component_tests_config):
+        config = component_tests_config(cfd_mode=CfdModes.NAMED, run_proxy_dns=False, provide_ingress=False)
+        LOGGER.debug(config)
+        with start_cloudflared(tmp_path, config, cfd_args=["run", "--hello-world"], new_process=True):
+            wait_tunnel_ready(tunnel_url=config.get_url(),
+                              require_min_connections=4)
+    
+    def test_tunnel_url(self, tmp_path, component_tests_config):
+        config = component_tests_config(cfd_mode=CfdModes.NAMED, run_proxy_dns=False, provide_ingress=False)
+        LOGGER.debug(config)
+        with start_cloudflared(tmp_path, config, cfd_args=["run", "--url", f"http://localhost:{METRICS_PORT}/"], new_process=True):
+            wait_tunnel_ready(require_min_connections=4)
+            send_requests(config.get_url()+"/ready", 3, True)
+
+    def test_tunnel_no_ingress(self, tmp_path, component_tests_config):
+        '''
+        Running a tunnel with no ingress rules provided from either config.yaml or CLI will still work but return 503
+        for all incoming requests.
+        '''
+        config = component_tests_config(cfd_mode=CfdModes.NAMED, run_proxy_dns=False, provide_ingress=False)
+        LOGGER.debug(config)
+        with start_cloudflared(tmp_path, config, cfd_args=["run"], new_process=True):
+            wait_tunnel_ready(require_min_connections=4)
+            resp = send_request(config.get_url()+"/")
+            assert resp.status_code == 503, "Expected cloudflared to return 503 for all requests with no ingress defined"
+            resp = send_request(config.get_url()+"/test")
+            assert resp.status_code == 503, "Expected cloudflared to return 503 for all requests with no ingress defined"
+
+
+@retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=BACKOFF_SECS * 1000)
+def send_request(url):
+    with requests.Session() as s:
+        return s.get(url, timeout=BACKOFF_SECS)

component-tests/util.py

@@ -35,7 +35,7 @@ def write_config(directory, config):
 
 
 def start_cloudflared(directory, config, cfd_args=["run"], cfd_pre_args=["tunnel"], new_process=False,
-                      allow_input=False, capture_output=True, root=False, skip_config_flag=False):
+                      allow_input=False, capture_output=True, root=False, skip_config_flag=False, expect_success=True):
 
     config_path = None
     if not skip_config_flag:
@@ -46,7 +46,7 @@ def start_cloudflared(directory, config, cfd_args=["run"], cfd_pre_args=["tunnel
     if new_process:
         return run_cloudflared_background(cmd, allow_input, capture_output)
     # By setting check=True, it will raise an exception if the process exits with non-zero exit code
-    return subprocess.run(cmd, check=True, capture_output=capture_output)
+    return subprocess.run(cmd, check=expect_success, capture_output=capture_output)
 
 
 def cloudflared_cmd(config, config_path, cfd_args, cfd_pre_args, root):
@@ -77,7 +77,18 @@ def run_cloudflared_background(cmd, allow_input, capture_output):
             cfd.terminate()
             if capture_output:
                 LOGGER.info(f"cloudflared log: {cfd.stderr.read()}")
+    
 
+def get_quicktunnel_url():
+    quicktunnel_url = f'http://localhost:{METRICS_PORT}/quicktunnel'
+    with requests.Session() as s:
+        resp = send_request(s, quicktunnel_url, True)
+
+        hostname = resp.json()["hostname"]
+        assert hostname, \
+            f"Quicktunnel endpoint returned {hostname} but we expected a url"
+
+        return f"https://{hostname}"
 
 def wait_tunnel_ready(tunnel_url=None, require_min_connections=1, cfd_logs=None):
     try:

connection/connection.go

@@ -200,6 +200,7 @@ func (h *HTTPResponseReadWriteAcker) AckConnection(tracePropagation string) erro
 type ResponseWriter interface {
 	WriteRespHeaders(status int, header http.Header) error
 	AddTrailer(trailerName, trailerValue string)
+	http.ResponseWriter
 	io.Writer
 }
 

connection/h2mux.go

@@ -1,46 +1,17 @@
 package connection
 
 import (
-	"context"
-	"io"
-	"net"
-	"net/http"
 	"time"
 
-	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
-	"golang.org/x/sync/errgroup"
 
 	"github.com/cloudflare/cloudflared/h2mux"
-	"github.com/cloudflare/cloudflared/tracing"
-	tunnelpogs "github.com/cloudflare/cloudflared/tunnelrpc/pogs"
-	"github.com/cloudflare/cloudflared/websocket"
 )
 
 const (
-	muxerTimeout      = 5 * time.Second
-	openStreamTimeout = 30 * time.Second
+	muxerTimeout = 5 * time.Second
 )
 
-type h2muxConnection struct {
-	orchestrator Orchestrator
-	gracePeriod  time.Duration
-	muxerConfig  *MuxerConfig
-	muxer        *h2mux.Muxer
-	// connectionID is only used by metrics, and prometheus requires labels to be string
-	connIndexStr string
-	connIndex    uint8
-
-	observer          *Observer
-	gracefulShutdownC <-chan struct{}
-	stoppedGracefully bool
-
-	log *zerolog.Logger
-
-	// newRPCClientFunc allows us to mock RPCs during testing
-	newRPCClientFunc func(context.Context, io.ReadWriteCloser, *zerolog.Logger) NamedTunnelRPCClient
-}
-
 type MuxerConfig struct {
 	HeartbeatInterval  time.Duration
 	MaxHeartbeats      uint64
@@ -59,180 +30,3 @@ func (mc *MuxerConfig) H2MuxerConfig(h h2mux.MuxedStreamHandler, log *zerolog.Lo
 		CompressionQuality: mc.CompressionSetting,
 	}
 }
-
-// NewTunnelHandler returns a TunnelHandler, origin LAN IP and error
-func NewH2muxConnection(
-	orchestrator Orchestrator,
-	gracePeriod time.Duration,
-	muxerConfig *MuxerConfig,
-	edgeConn net.Conn,
-	connIndex uint8,
-	observer *Observer,
-	gracefulShutdownC <-chan struct{},
-	log *zerolog.Logger,
-) (*h2muxConnection, error, bool) {
-	h := &h2muxConnection{
-		orchestrator:      orchestrator,
-		gracePeriod:       gracePeriod,
-		muxerConfig:       muxerConfig,
-		connIndexStr:      uint8ToString(connIndex),
-		connIndex:         connIndex,
-		observer:          observer,
-		gracefulShutdownC: gracefulShutdownC,
-		newRPCClientFunc:  newRegistrationRPCClient,
-		log:               log,
-	}
-
-	// Establish a muxed connection with the edge
-	// Client mux handshake with agent server
-	muxer, err := h2mux.Handshake(edgeConn, edgeConn, *muxerConfig.H2MuxerConfig(h, observer.logTransport), h2mux.ActiveStreams)
-	if err != nil {
-		recoverable := isHandshakeErrRecoverable(err, connIndex, observer)
-		return nil, err, recoverable
-	}
-	h.muxer = muxer
-	return h, nil, false
-}
-
-func (h *h2muxConnection) ServeNamedTunnel(ctx context.Context, namedTunnel *NamedTunnelProperties, connOptions *tunnelpogs.ConnectionOptions, connectedFuse ConnectedFuse) error {
-	errGroup, serveCtx := errgroup.WithContext(ctx)
-	errGroup.Go(func() error {
-		return h.serveMuxer(serveCtx)
-	})
-
-	errGroup.Go(func() error {
-		if err := h.registerNamedTunnel(serveCtx, namedTunnel, connOptions); err != nil {
-			return err
-		}
-		connectedFuse.Connected()
-		return nil
-	})
-
-	errGroup.Go(func() error {
-		h.controlLoop(serveCtx, connectedFuse, true)
-		return nil
-	})
-
-	err := errGroup.Wait()
-	if err == errMuxerStopped {
-		if h.stoppedGracefully {
-			return nil
-		}
-		h.observer.log.Info().Uint8(LogFieldConnIndex, h.connIndex).Msg("Unexpected muxer shutdown")
-	}
-	return err
-}
-
-func (h *h2muxConnection) serveMuxer(ctx context.Context) error {
-	// All routines should stop when muxer finish serving. When muxer is shutdown
-	// gracefully, it doesn't return an error, so we need to return errMuxerShutdown
-	// here to notify other routines to stop
-	err := h.muxer.Serve(ctx)
-	if err == nil {
-		return errMuxerStopped
-	}
-	return err
-}
-
-func (h *h2muxConnection) controlLoop(ctx context.Context, connectedFuse ConnectedFuse, isNamedTunnel bool) {
-	updateMetricsTicker := time.NewTicker(h.muxerConfig.MetricsUpdateFreq)
-	defer updateMetricsTicker.Stop()
-	var shutdownCompleted <-chan struct{}
-	for {
-		select {
-		case <-h.gracefulShutdownC:
-			if connectedFuse.IsConnected() {
-				h.unregister(isNamedTunnel)
-			}
-			h.stoppedGracefully = true
-			h.gracefulShutdownC = nil
-			shutdownCompleted = h.muxer.Shutdown()
-
-		case <-shutdownCompleted:
-			return
-
-		case <-ctx.Done():
-			// UnregisterTunnel blocks until the RPC call returns
-			if !h.stoppedGracefully && connectedFuse.IsConnected() {
-				h.unregister(isNamedTunnel)
-			}
-			h.muxer.Shutdown()
-			// don't wait for shutdown to finish when context is closed, this is the hard termination path
-			return
-
-		case <-updateMetricsTicker.C:
-			h.observer.metrics.updateMuxerMetrics(h.connIndexStr, h.muxer.Metrics())
-		}
-	}
-}
-
-func (h *h2muxConnection) newRPCStream(ctx context.Context, rpcName rpcName) (*h2mux.MuxedStream, error) {
-	openStreamCtx, openStreamCancel := context.WithTimeout(ctx, openStreamTimeout)
-	defer openStreamCancel()
-	stream, err := h.muxer.OpenRPCStream(openStreamCtx)
-	if err != nil {
-		return nil, err
-	}
-	return stream, nil
-}
-
-func (h *h2muxConnection) ServeStream(stream *h2mux.MuxedStream) error {
-	respWriter := &h2muxRespWriter{stream}
-
-	req, reqErr := h.newRequest(stream)
-	if reqErr != nil {
-		respWriter.WriteErrorResponse()
-		return reqErr
-	}
-
-	var sourceConnectionType = TypeHTTP
-	if websocket.IsWebSocketUpgrade(req) {
-		sourceConnectionType = TypeWebsocket
-	}
-
-	originProxy, err := h.orchestrator.GetOriginProxy()
-	if err != nil {
-		respWriter.WriteErrorResponse()
-		return err
-	}
-
-	err = originProxy.ProxyHTTP(respWriter, tracing.NewTracedHTTPRequest(req, h.connIndex, h.log), sourceConnectionType == TypeWebsocket)
-	if err != nil {
-		respWriter.WriteErrorResponse()
-	}
-	return err
-}
-
-func (h *h2muxConnection) newRequest(stream *h2mux.MuxedStream) (*http.Request, error) {
-	req, err := http.NewRequest("GET", "http://localhost:8080", h2mux.MuxedStreamReader{MuxedStream: stream})
-	if err != nil {
-		return nil, errors.Wrap(err, "Unexpected error from http.NewRequest")
-	}
-	err = H2RequestHeadersToH1Request(stream.Headers, req)
-	if err != nil {
-		return nil, errors.Wrap(err, "invalid request received")
-	}
-	return req, nil
-}
-
-type h2muxRespWriter struct {
-	*h2mux.MuxedStream
-}
-
-func (rp *h2muxRespWriter) AddTrailer(trailerName, trailerValue string) {
-	// do nothing. we don't support trailers over h2mux
-}
-
-func (rp *h2muxRespWriter) WriteRespHeaders(status int, header http.Header) error {
-	headers := H1ResponseToH2ResponseHeaders(status, header)
-	headers = append(headers, h2mux.Header{Name: ResponseMetaHeader, Value: responseMetaHeaderOrigin})
-	return rp.WriteHeaders(headers)
-}
-
-func (rp *h2muxRespWriter) WriteErrorResponse() {
-	_ = rp.WriteHeaders([]h2mux.Header{
-		{Name: ":status", Value: "502"},
-		{Name: ResponseMetaHeader, Value: responseMetaHeaderCfd},
-	})
-	_, _ = rp.Write([]byte("502 Bad Gateway"))
-}

connection/h2mux_test.go

@@ -1,301 +0,0 @@
-package connection
-
-import (
-	"context"
-	"fmt"
-	"io"
-	"net"
-	"net/http"
-	"strconv"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/gobwas/ws/wsutil"
-	"github.com/rs/zerolog"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/cloudflare/cloudflared/h2mux"
-)
-
-var (
-	testMuxerConfig = &MuxerConfig{
-		HeartbeatInterval:  time.Second * 5,
-		MaxHeartbeats:      5,
-		CompressionSetting: 0,
-		MetricsUpdateFreq:  time.Second * 5,
-	}
-)
-
-func newH2MuxConnection(t require.TestingT) (*h2muxConnection, *h2mux.Muxer) {
-	edgeConn, originConn := net.Pipe()
-	edgeMuxChan := make(chan *h2mux.Muxer)
-	go func() {
-		edgeMuxConfig := h2mux.MuxerConfig{
-			Log: &log,
-			Handler: h2mux.MuxedStreamFunc(func(stream *h2mux.MuxedStream) error {
-				// we only expect RPC traffic in client->edge direction, provide minimal support for mocking
-				require.True(t, stream.IsRPCStream())
-				return stream.WriteHeaders([]h2mux.Header{
-					{Name: ":status", Value: "200"},
-				})
-			}),
-		}
-		edgeMux, err := h2mux.Handshake(edgeConn, edgeConn, edgeMuxConfig, h2mux.ActiveStreams)
-		require.NoError(t, err)
-		edgeMuxChan <- edgeMux
-	}()
-	var connIndex = uint8(0)
-	testObserver := NewObserver(&log, &log)
-	h2muxConn, err, _ := NewH2muxConnection(testOrchestrator, testGracePeriod, testMuxerConfig, originConn, connIndex, testObserver, nil, &log)
-	require.NoError(t, err)
-	return h2muxConn, <-edgeMuxChan
-}
-
-func TestServeStreamHTTP(t *testing.T) {
-	tests := []testRequest{
-		{
-			name:           "ok",
-			endpoint:       "/ok",
-			expectedStatus: http.StatusOK,
-			expectedBody:   []byte(http.StatusText(http.StatusOK)),
-		},
-		{
-			name:           "large_file",
-			endpoint:       "/large_file",
-			expectedStatus: http.StatusOK,
-			expectedBody:   testLargeResp,
-		},
-		{
-			name:           "Bad request",
-			endpoint:       "/400",
-			expectedStatus: http.StatusBadRequest,
-			expectedBody:   []byte(http.StatusText(http.StatusBadRequest)),
-		},
-		{
-			name:           "Internal server error",
-			endpoint:       "/500",
-			expectedStatus: http.StatusInternalServerError,
-			expectedBody:   []byte(http.StatusText(http.StatusInternalServerError)),
-		},
-		{
-			name:           "Proxy error",
-			endpoint:       "/error",
-			expectedStatus: http.StatusBadGateway,
-			expectedBody:   nil,
-			isProxyError:   true,
-		},
-	}
-
-	ctx, cancel := context.WithCancel(context.Background())
-	h2muxConn, edgeMux := newH2MuxConnection(t)
-
-	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		_ = edgeMux.Serve(ctx)
-	}()
-	go func() {
-		defer wg.Done()
-		err := h2muxConn.serveMuxer(ctx)
-		require.Error(t, err)
-	}()
-
-	for _, test := range tests {
-		headers := []h2mux.Header{
-			{
-				Name:  ":path",
-				Value: test.endpoint,
-			},
-		}
-		stream, err := edgeMux.OpenStream(ctx, headers, nil)
-		require.NoError(t, err)
-		require.True(t, hasHeader(stream, ":status", strconv.Itoa(test.expectedStatus)))
-
-		if test.isProxyError {
-			assert.True(t, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderCfd))
-		} else {
-			assert.True(t, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderOrigin))
-			body := make([]byte, len(test.expectedBody))
-			_, err = stream.Read(body)
-			require.NoError(t, err)
-			require.Equal(t, test.expectedBody, body)
-		}
-	}
-	cancel()
-	wg.Wait()
-}
-
-func TestServeStreamWS(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	h2muxConn, edgeMux := newH2MuxConnection(t)
-
-	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		edgeMux.Serve(ctx)
-	}()
-	go func() {
-		defer wg.Done()
-		err := h2muxConn.serveMuxer(ctx)
-		require.Error(t, err)
-	}()
-
-	headers := []h2mux.Header{
-		{
-			Name:  ":path",
-			Value: "/ws/echo",
-		},
-		{
-			Name:  "connection",
-			Value: "upgrade",
-		},
-		{
-			Name:  "upgrade",
-			Value: "websocket",
-		},
-	}
-
-	readPipe, writePipe := io.Pipe()
-	stream, err := edgeMux.OpenStream(ctx, headers, readPipe)
-	require.NoError(t, err)
-
-	require.True(t, hasHeader(stream, ":status", strconv.Itoa(http.StatusSwitchingProtocols)))
-	assert.True(t, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderOrigin))
-
-	data := []byte("test websocket")
-	err = wsutil.WriteClientBinary(writePipe, data)
-	require.NoError(t, err)
-
-	respBody, err := wsutil.ReadServerBinary(stream)
-	require.NoError(t, err)
-	require.Equal(t, data, respBody, fmt.Sprintf("Expect %s, got %s", string(data), string(respBody)))
-
-	cancel()
-	wg.Wait()
-}
-
-func TestGracefulShutdownH2Mux(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	h2muxConn, edgeMux := newH2MuxConnection(t)
-
-	shutdownC := make(chan struct{})
-	unregisteredC := make(chan struct{})
-	h2muxConn.gracefulShutdownC = shutdownC
-	h2muxConn.newRPCClientFunc = func(_ context.Context, _ io.ReadWriteCloser, _ *zerolog.Logger) NamedTunnelRPCClient {
-		return &mockNamedTunnelRPCClient{
-			registered:   nil,
-			unregistered: unregisteredC,
-		}
-	}
-
-	var wg sync.WaitGroup
-	wg.Add(3)
-	go func() {
-		defer wg.Done()
-		_ = edgeMux.Serve(ctx)
-	}()
-	go func() {
-		defer wg.Done()
-		_ = h2muxConn.serveMuxer(ctx)
-	}()
-
-	go func() {
-		defer wg.Done()
-		h2muxConn.controlLoop(ctx, &mockConnectedFuse{}, true)
-	}()
-
-	time.Sleep(100 * time.Millisecond)
-	close(shutdownC)
-
-	select {
-	case <-unregisteredC:
-		break // ok
-	case <-time.Tick(time.Second):
-		assert.Fail(t, "timed out waiting for control loop to unregister")
-	}
-
-	cancel()
-	wg.Wait()
-
-	assert.True(t, h2muxConn.stoppedGracefully)
-	assert.Nil(t, h2muxConn.gracefulShutdownC)
-}
-
-func hasHeader(stream *h2mux.MuxedStream, name, val string) bool {
-	for _, header := range stream.Headers {
-		if header.Name == name && header.Value == val {
-			return true
-		}
-	}
-	return false
-}
-
-func benchmarkServeStreamHTTPSimple(b *testing.B, test testRequest) {
-	ctx, cancel := context.WithCancel(context.Background())
-	h2muxConn, edgeMux := newH2MuxConnection(b)
-
-	var wg sync.WaitGroup
-	wg.Add(2)
-	go func() {
-		defer wg.Done()
-		edgeMux.Serve(ctx)
-	}()
-	go func() {
-		defer wg.Done()
-		err := h2muxConn.serveMuxer(ctx)
-		require.Error(b, err)
-	}()
-
-	headers := []h2mux.Header{
-		{
-			Name:  ":path",
-			Value: test.endpoint,
-		},
-	}
-
-	body := make([]byte, len(test.expectedBody))
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		b.StartTimer()
-		stream, openstreamErr := edgeMux.OpenStream(ctx, headers, nil)
-		_, readBodyErr := stream.Read(body)
-		b.StopTimer()
-
-		require.NoError(b, openstreamErr)
-		assert.True(b, hasHeader(stream, ResponseMetaHeader, responseMetaHeaderOrigin))
-		require.True(b, hasHeader(stream, ":status", strconv.Itoa(http.StatusOK)))
-		require.NoError(b, readBodyErr)
-		require.Equal(b, test.expectedBody, body)
-	}
-
-	cancel()
-	wg.Wait()
-}
-
-func BenchmarkServeStreamHTTPSimple(b *testing.B) {
-	test := testRequest{
-		name:           "ok",
-		endpoint:       "/ok",
-		expectedStatus: http.StatusOK,
-		expectedBody:   []byte(http.StatusText(http.StatusOK)),
-	}
-
-	benchmarkServeStreamHTTPSimple(b, test)
-}
-
-func BenchmarkServeStreamHTTPLargeFile(b *testing.B) {
-	test := testRequest{
-		name:           "large_file",
-		endpoint:       "/large_file",
-		expectedStatus: http.StatusOK,
-		expectedBody:   testLargeResp,
-	}
-
-	benchmarkServeStreamHTTPSimple(b, test)
-}

connection/http2.go

@@ -197,6 +197,7 @@ type http2RespWriter struct {
 	flusher       http.Flusher
 	shouldFlush   bool
 	statusWritten bool
+	respHeaders   http.Header
 	log           *zerolog.Logger
 }
 
@@ -276,6 +277,14 @@ func (rp *http2RespWriter) WriteRespHeaders(status int, header http.Header) erro
 	return nil
 }
 
+func (rp *http2RespWriter) Header() http.Header {
+	return rp.respHeaders
+}
+
+func (rp *http2RespWriter) WriteHeader(status int) {
+	rp.WriteRespHeaders(status, rp.respHeaders)
+}
+
 func (rp *http2RespWriter) WriteErrorResponse() bool {
 	if rp.statusWritten {
 		return false

connection/quic.go

@@ -402,6 +402,7 @@ func (s *streamReadWriteAcker) AckConnection(tracePropagation string) error {
 // httpResponseAdapter translates responses written by the HTTP Proxy into ones that can be used in QUIC.
 type httpResponseAdapter struct {
 	*quicpogs.RequestServerStream
+	headers             http.Header
 	connectResponseSent bool
 }
 
@@ -426,6 +427,14 @@ func (hrw *httpResponseAdapter) WriteRespHeaders(status int, header http.Header)
 	return hrw.WriteConnectResponseData(nil, metadata...)
 }
 
+func (hrw *httpResponseAdapter) Header() http.Header {
+	return hrw.headers
+}
+
+func (hrw *httpResponseAdapter) WriteHeader(status int) {
+	hrw.WriteRespHeaders(status, hrw.headers)
+}
+
 func (hrw *httpResponseAdapter) WriteErrorResponse(err error) {
 	hrw.WriteConnectResponseData(err, quicpogs.Metadata{Key: "HttpStatus", Val: strconv.Itoa(http.StatusBadGateway)})
 }

connection/rpc.go

@@ -2,7 +2,6 @@ package connection
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"net"
 	"time"
@@ -152,178 +151,3 @@ const (
 	unregister   rpcName = "unregister"
 	authenticate rpcName = " authenticate"
 )
-
-func (h *h2muxConnection) registerTunnel(ctx context.Context, credentialSetter CredentialManager, classicTunnel *ClassicTunnelProperties, registrationOptions *tunnelpogs.RegistrationOptions) error {
-	h.observer.sendRegisteringEvent(registrationOptions.ConnectionID)
-
-	stream, err := h.newRPCStream(ctx, register)
-	if err != nil {
-		return err
-	}
-	rpcClient := NewTunnelServerClient(ctx, stream, h.observer.log)
-	defer rpcClient.Close()
-
-	_ = h.logServerInfo(ctx, rpcClient)
-	registration := rpcClient.client.RegisterTunnel(
-		ctx,
-		classicTunnel.OriginCert,
-		classicTunnel.Hostname,
-		registrationOptions,
-	)
-	if registrationErr := registration.DeserializeError(); registrationErr != nil {
-		// RegisterTunnel RPC failure
-		return h.processRegisterTunnelError(registrationErr, register)
-	}
-
-	credentialSetter.SetEventDigest(h.connIndex, registration.EventDigest)
-	return h.processRegistrationSuccess(registration, register, credentialSetter, classicTunnel)
-}
-
-type CredentialManager interface {
-	ReconnectToken() ([]byte, error)
-	EventDigest(connID uint8) ([]byte, error)
-	SetEventDigest(connID uint8, digest []byte)
-	ConnDigest(connID uint8) ([]byte, error)
-	SetConnDigest(connID uint8, digest []byte)
-}
-
-func (h *h2muxConnection) processRegistrationSuccess(
-	registration *tunnelpogs.TunnelRegistration,
-	name rpcName,
-	credentialManager CredentialManager, classicTunnel *ClassicTunnelProperties,
-) error {
-	for _, logLine := range registration.LogLines {
-		h.observer.log.Info().Msg(logLine)
-	}
-
-	if registration.TunnelID != "" {
-		h.observer.metrics.tunnelsHA.AddTunnelID(h.connIndex, registration.TunnelID)
-		h.observer.log.Info().Msgf("Each HA connection's tunnel IDs: %v", h.observer.metrics.tunnelsHA.String())
-	}
-
-	credentialManager.SetConnDigest(h.connIndex, registration.ConnDigest)
-	h.observer.metrics.userHostnamesCounts.WithLabelValues(registration.Url).Inc()
-
-	h.observer.log.Info().Msgf("Route propagating, it may take up to 1 minute for your new route to become functional")
-	h.observer.metrics.regSuccess.WithLabelValues(string(name)).Inc()
-	return nil
-}
-
-func (h *h2muxConnection) processRegisterTunnelError(err tunnelpogs.TunnelRegistrationError, name rpcName) error {
-	if err.Error() == DuplicateConnectionError {
-		h.observer.metrics.regFail.WithLabelValues("dup_edge_conn", string(name)).Inc()
-		return errDuplicationConnection
-	}
-	h.observer.metrics.regFail.WithLabelValues("server_error", string(name)).Inc()
-	return ServerRegisterTunnelError{
-		Cause:     err,
-		Permanent: err.IsPermanent(),
-	}
-}
-
-func (h *h2muxConnection) reconnectTunnel(ctx context.Context, credentialManager CredentialManager, classicTunnel *ClassicTunnelProperties, registrationOptions *tunnelpogs.RegistrationOptions) error {
-	token, err := credentialManager.ReconnectToken()
-	if err != nil {
-		return err
-	}
-	eventDigest, err := credentialManager.EventDigest(h.connIndex)
-	if err != nil {
-		return err
-	}
-	connDigest, err := credentialManager.ConnDigest(h.connIndex)
-	if err != nil {
-		return err
-	}
-
-	h.observer.log.Debug().Msg("initiating RPC stream to reconnect")
-	stream, err := h.newRPCStream(ctx, register)
-	if err != nil {
-		return err
-	}
-	rpcClient := NewTunnelServerClient(ctx, stream, h.observer.log)
-	defer rpcClient.Close()
-
-	_ = h.logServerInfo(ctx, rpcClient)
-	registration := rpcClient.client.ReconnectTunnel(
-		ctx,
-		token,
-		eventDigest,
-		connDigest,
-		classicTunnel.Hostname,
-		registrationOptions,
-	)
-	if registrationErr := registration.DeserializeError(); registrationErr != nil {
-		// ReconnectTunnel RPC failure
-		return h.processRegisterTunnelError(registrationErr, reconnect)
-	}
-	return h.processRegistrationSuccess(registration, reconnect, credentialManager, classicTunnel)
-}
-
-func (h *h2muxConnection) logServerInfo(ctx context.Context, rpcClient *tunnelServerClient) error {
-	// Request server info without blocking tunnel registration; must use capnp library directly.
-	serverInfoPromise := tunnelrpc.TunnelServer{Client: rpcClient.client.Client}.GetServerInfo(ctx, func(tunnelrpc.TunnelServer_getServerInfo_Params) error {
-		return nil
-	})
-	serverInfoMessage, err := serverInfoPromise.Result().Struct()
-	if err != nil {
-		h.observer.log.Err(err).Msg("Failed to retrieve server information")
-		return err
-	}
-	serverInfo, err := tunnelpogs.UnmarshalServerInfo(serverInfoMessage)
-	if err != nil {
-		h.observer.log.Err(err).Msg("Failed to retrieve server information")
-		return err
-	}
-	h.observer.logServerInfo(h.connIndex, serverInfo.LocationName, net.IP{}, "Connection established")
-	return nil
-}
-
-func (h *h2muxConnection) registerNamedTunnel(
-	ctx context.Context,
-	namedTunnel *NamedTunnelProperties,
-	connOptions *tunnelpogs.ConnectionOptions,
-) error {
-	stream, err := h.newRPCStream(ctx, register)
-	if err != nil {
-		return err
-	}
-	rpcClient := h.newRPCClientFunc(ctx, stream, h.observer.log)
-	defer rpcClient.Close()
-
-	registrationDetails, err := rpcClient.RegisterConnection(ctx, namedTunnel, connOptions, h.connIndex, nil, h.observer)
-	if err != nil {
-		return err
-	}
-	h.observer.logServerInfo(h.connIndex, registrationDetails.Location, nil, fmt.Sprintf("Connection %s registered", registrationDetails.UUID))
-	h.observer.sendConnectedEvent(h.connIndex, 0, registrationDetails.Location)
-
-	return nil
-}
-
-func (h *h2muxConnection) unregister(isNamedTunnel bool) {
-	h.observer.sendUnregisteringEvent(h.connIndex)
-
-	unregisterCtx, cancel := context.WithTimeout(context.Background(), h.gracePeriod)
-	defer cancel()
-
-	stream, err := h.newRPCStream(unregisterCtx, unregister)
-	if err != nil {
-		return
-	}
-	defer stream.Close()
-
-	if isNamedTunnel {
-		rpcClient := h.newRPCClientFunc(unregisterCtx, stream, h.observer.log)
-		defer rpcClient.Close()
-
-		rpcClient.GracefulShutdown(unregisterCtx, h.gracePeriod)
-	} else {
-		rpcClient := NewTunnelServerClient(unregisterCtx, stream, h.observer.log)
-		defer rpcClient.Close()
-
-		// gracePeriod is encoded in int64 using capnproto
-		_ = rpcClient.client.UnregisterTunnel(unregisterCtx, h.gracePeriod.Nanoseconds())
-	}
-
-	h.observer.log.Info().Uint8(LogFieldConnIndex, h.connIndex).Msg("Unregistered tunnel connection")
-}

ingress/config.go

@@ -113,7 +113,7 @@ func (rc *RemoteConfig) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
-func originRequestFromSingeRule(c *cli.Context) OriginRequestConfig {
+func originRequestFromSingleRule(c *cli.Context) OriginRequestConfig {
 	var connectTimeout = defaultHTTPConnectTimeout
 	var tlsTimeout = defaultTLSTimeout
 	var tcpKeepAlive = defaultTCPKeepAlive

ingress/config_test.go

@@ -411,7 +411,7 @@ func TestDefaultConfigFromCLI(t *testing.T) {
 		KeepAliveTimeout:     defaultKeepAliveTimeout,
 		ProxyAddress:         defaultProxyAddress,
 	}
-	actual := originRequestFromSingeRule(c)
+	actual := originRequestFromSingleRule(c)
 	require.Equal(t, expected, actual)
 }
 

ingress/ingress.go

@@ -20,6 +20,7 @@ import (
 
 var (
 	ErrNoIngressRules             = errors.New("The config file doesn't contain any ingress rules")
+	ErrNoIngressRulesCLI          = errors.New("No ingress rules were defined in provided config (if any) nor from the cli, cloudflared will return 503 for all incoming HTTP requests")
 	errLastRuleNotCatchAll        = errors.New("The last ingress rule must match all URLs (i.e. it should not have a hostname or path filter)")
 	errBadWildcard                = errors.New("Hostname patterns can have at most one wildcard character (\"*\") and it can only be used for subdomains, e.g. \"*.example.com\"")
 	errHostnameContainsPort       = errors.New("Hostname cannot contain a port")
@@ -70,16 +71,52 @@ type Ingress struct {
 	Defaults OriginRequestConfig `json:"originRequest"`
 }
 
-// NewSingleOrigin constructs an Ingress set with only one rule, constructed from
-// CLI parameters for quick tunnels like --url or --no-chunked-encoding.
-func NewSingleOrigin(c *cli.Context, allowURLFromArgs bool) (Ingress, error) {
+// ParseIngress parses ingress rules, but does not send HTTP requests to the origins.
+func ParseIngress(conf *config.Configuration) (Ingress, error) {
+	if len(conf.Ingress) == 0 {
+		return Ingress{}, ErrNoIngressRules
+	}
+	return validateIngress(conf.Ingress, originRequestFromConfig(conf.OriginRequest))
+}
+
+// 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.
+func ParseIngressFromConfigAndCLI(conf *config.Configuration, c *cli.Context, log *zerolog.Logger) (Ingress, error) {
+	// Attempt to parse ingress rules from configuration
+	ingressRules, err := ParseIngress(conf)
+	if err == nil && !ingressRules.IsEmpty() {
+		return ingressRules, nil
+	}
+	if err != ErrNoIngressRules {
+		return Ingress{}, err
+	}
+	// Attempt to parse ingress rules from CLI:
+	//   --url or --unix-socket flag for a tunnel HTTP ingress
+	//   --hello-world for a basic HTTP ingress self-served
+	//   --bastion for ssh bastion service
+	ingressRules, err = parseCLIIngress(c, false)
+	if errors.Is(err, ErrNoIngressRulesCLI) {
+		log.Warn().Msgf(ErrNoIngressRulesCLI.Error())
+		return newDefaultOrigin(c, log), nil
+	}
+	if err != nil {
+		return Ingress{}, err
+	}
+	return ingressRules, nil
+}
+
+// parseCLIIngress constructs an Ingress set with only one rule constructed from
+// CLI parameters: --url, --hello-world, --bastion, or --unix-socket
+func parseCLIIngress(c *cli.Context, allowURLFromArgs bool) (Ingress, error) {
 	service, err := parseSingleOriginService(c, allowURLFromArgs)
 	if err != nil {
 		return Ingress{}, err
 	}
 
 	// Construct an Ingress with the single rule.
-	defaults := originRequestFromSingeRule(c)
+	defaults := originRequestFromSingleRule(c)
 	ing := Ingress{
 		Rules: []Rule{
 			{
@@ -92,6 +129,22 @@ func NewSingleOrigin(c *cli.Context, allowURLFromArgs bool) (Ingress, error) {
 	return ing, err
 }
 
+// newDefaultOrigin always returns a 503 response code to help indicate that there are no ingress
+// rules setup, but the tunnel is reachable.
+func newDefaultOrigin(c *cli.Context, log *zerolog.Logger) Ingress {
+	noRulesService := newDefaultStatusCode(log)
+	defaults := originRequestFromSingleRule(c)
+	ingress := Ingress{
+		Rules: []Rule{
+			{
+				Service: &noRulesService,
+			},
+		},
+		Defaults: defaults,
+	}
+	return ingress
+}
+
 // WarpRoutingService starts a tcp stream between the origin and requests from
 // warp clients.
 type WarpRoutingService struct {
@@ -112,7 +165,7 @@ func NewWarpRoutingService(config WarpRoutingConfig) *WarpRoutingService {
 
 // Get a single origin service from the CLI/config.
 func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (OriginService, error) {
-	if c.IsSet("hello-world") {
+	if c.IsSet(HelloWorldFlag) {
 		return new(helloWorld), nil
 	}
 	if c.IsSet(config.BastionFlag) {
@@ -137,8 +190,7 @@ func parseSingleOriginService(c *cli.Context, allowURLFromArgs bool) (OriginServ
 		}
 		return &unixSocketPath{path: path, scheme: "http"}, nil
 	}
-	u, err := url.Parse("http://localhost:8080")
-	return &httpService{url: u}, err
+	return nil, ErrNoIngressRulesCLI
 }
 
 // IsEmpty checks if there are any ingress rules.
@@ -206,7 +258,7 @@ func validateIngress(ingress []config.UnvalidatedIngressRule, defaults OriginReq
 			}
 			srv := newStatusCode(statusCode)
 			service = &srv
-		} else if r.Service == HelloWorldService || r.Service == "hello-world" || r.Service == "helloworld" {
+		} else if r.Service == HelloWorldFlag || r.Service == HelloWorldService {
 			service = new(helloWorld)
 		} else if r.Service == ServiceSocksProxy {
 			rules := make([]ipaccess.Rule, len(r.OriginRequest.IPRules))
@@ -335,14 +387,6 @@ func (e errRuleShouldNotBeCatchAll) Error() string {
 		"will never be triggered.", e.index+1, e.hostname)
 }
 
-// ParseIngress parses ingress rules, but does not send HTTP requests to the origins.
-func ParseIngress(conf *config.Configuration) (Ingress, error) {
-	if len(conf.Ingress) == 0 {
-		return Ingress{}, ErrNoIngressRules
-	}
-	return validateIngress(conf.Ingress, originRequestFromConfig(conf.OriginRequest))
-}
-
 func isHTTPService(url *url.URL) bool {
 	return url.Scheme == "http" || url.Scheme == "https" || url.Scheme == "ws" || url.Scheme == "wss"
 }

ingress/ingress_test.go

@@ -43,7 +43,7 @@ ingress:
 	require.Equal(t, "https", s.scheme)
 }
 
-func Test_parseIngress(t *testing.T) {
+func TestParseIngress(t *testing.T) {
 	localhost8000 := MustParseURL(t, "https://localhost:8000")
 	localhost8001 := MustParseURL(t, "https://localhost:8001")
 	fourOhFour := newStatusCode(404)
@@ -517,7 +517,7 @@ func TestSingleOriginSetsConfig(t *testing.T) {
 
 	allowURLFromArgs := false
 	require.NoError(t, err)
-	ingress, err := NewSingleOrigin(cliCtx, allowURLFromArgs)
+	ingress, err := parseCLIIngress(cliCtx, allowURLFromArgs)
 	require.NoError(t, err)
 
 	assert.Equal(t, config.CustomDuration{Duration: time.Second}, ingress.Rules[0].Config.ConnectTimeout)
@@ -537,6 +537,119 @@ func TestSingleOriginSetsConfig(t *testing.T) {
 	assert.Equal(t, socksProxy, ingress.Rules[0].Config.ProxyType)
 }
 
+func TestSingleOriginServices(t *testing.T) {
+	host := "://localhost:8080"
+	httpURL := urlMustParse("http" + host)
+	tcpURL := urlMustParse("tcp" + host)
+	unix := "unix://service"
+	newCli := func(params ...string) *cli.Context {
+		flagSet := flag.NewFlagSet(t.Name(), flag.PanicOnError)
+		flagSet.Bool("hello-world", false, "")
+		flagSet.Bool("bastion", false, "")
+		flagSet.String("url", "", "")
+		flagSet.String("unix-socket", "", "")
+		cliCtx := cli.NewContext(cli.NewApp(), flagSet, nil)
+		for i := 0; i < len(params); i += 2 {
+			cliCtx.Set(params[i], params[i+1])
+		}
+
+		return cliCtx
+	}
+
+	tests := []struct {
+		name            string
+		cli             *cli.Context
+		expectedService OriginService
+		err             error
+	}{
+		{
+			name:            "Valid hello-world",
+			cli:             newCli("hello-world", "true"),
+			expectedService: &helloWorld{},
+		},
+		{
+			name:            "Valid bastion",
+			cli:             newCli("bastion", "true"),
+			expectedService: newBastionService(),
+		},
+		{
+			name:            "Valid http url",
+			cli:             newCli("url", httpURL.String()),
+			expectedService: &httpService{url: httpURL},
+		},
+		{
+			name:            "Valid tcp url",
+			cli:             newCli("url", tcpURL.String()),
+			expectedService: newTCPOverWSService(tcpURL),
+		},
+		{
+			name:            "Valid unix-socket",
+			cli:             newCli("unix-socket", unix),
+			expectedService: &unixSocketPath{path: unix, scheme: "http"},
+		},
+		{
+			name: "No origins defined",
+			cli:  newCli(),
+			err:  ErrNoIngressRulesCLI,
+		},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			ingress, err := parseCLIIngress(test.cli, false)
+			require.Equal(t, err, test.err)
+			if test.err != nil {
+				return
+			}
+			require.Equal(t, 1, len(ingress.Rules))
+			rule := ingress.Rules[0]
+			require.Equal(t, test.expectedService, rule.Service)
+		})
+	}
+}
+
+func urlMustParse(s string) *url.URL {
+	u, err := url.Parse(s)
+	if err != nil {
+		panic(err)
+	}
+	return u
+}
+
+func TestSingleOriginServices_URL(t *testing.T) {
+	host := "://localhost:8080"
+	newCli := func(param string, value string) *cli.Context {
+		flagSet := flag.NewFlagSet(t.Name(), flag.PanicOnError)
+		flagSet.String("url", "", "")
+		cliCtx := cli.NewContext(cli.NewApp(), flagSet, nil)
+		cliCtx.Set(param, value)
+		return cliCtx
+	}
+
+	httpTests := []string{"http", "https"}
+	for _, test := range httpTests {
+		t.Run(test, func(t *testing.T) {
+			url := urlMustParse(test + host)
+			ingress, err := parseCLIIngress(newCli("url", url.String()), false)
+			require.NoError(t, err)
+			require.Equal(t, 1, len(ingress.Rules))
+			rule := ingress.Rules[0]
+			require.Equal(t, &httpService{url: url}, rule.Service)
+		})
+	}
+
+	tcpTests := []string{"ssh", "rdp", "smb", "tcp"}
+	for _, test := range tcpTests {
+		t.Run(test, func(t *testing.T) {
+			url := urlMustParse(test + host)
+			ingress, err := parseCLIIngress(newCli("url", url.String()), false)
+			require.NoError(t, err)
+			require.Equal(t, 1, len(ingress.Rules))
+			rule := ingress.Rules[0]
+			require.Equal(t, newTCPOverWSService(url), rule.Service)
+		})
+	}
+}
+
 func TestFindMatchingRule(t *testing.T) {
 	ingress := Ingress{
 		Rules: []Rule{

ingress/origin_proxy.go

@@ -44,6 +44,9 @@ func (o *httpService) RoundTrip(req *http.Request) (*http.Response, error) {
 }
 
 func (o *statusCode) RoundTrip(_ *http.Request) (*http.Response, error) {
+	if o.defaultResp {
+		o.log.Warn().Msgf(ErrNoIngressRulesCLI.Error())
+	}
 	resp := &http.Response{
 		StatusCode: o.code,
 		Status:     fmt.Sprintf("%d %s", o.code, http.StatusText(o.code)),

ingress/origin_service.go

@@ -23,6 +23,7 @@ import (
 
 const (
 	HelloWorldService = "hello_world"
+	HelloWorldFlag    = "hello-world"
 	HttpStatusService = "http_status"
 )
 
@@ -246,12 +247,21 @@ func (o helloWorld) MarshalJSON() ([]byte, error) {
 // Typical use-case is "user wants the catch-all rule to just respond 404".
 type statusCode struct {
 	code int
+
+	// Set only when the user has not defined any ingress rules
+	defaultResp bool
+	log         *zerolog.Logger
 }
 
 func newStatusCode(status int) statusCode {
 	return statusCode{code: status}
 }
 
+// default status code (503) that is returned for requests to cloudflared that don't have any ingress rules setup
+func newDefaultStatusCode(log *zerolog.Logger) statusCode {
+	return statusCode{code: 503, defaultResp: true, log: log}
+}
+
 func (o *statusCode) String() string {
 	return fmt.Sprintf("http_status:%d", o.code)
 }

proxy/proxy_test.go

@@ -23,7 +23,6 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/cloudflare/cloudflared/cfio"
-
 	"github.com/cloudflare/cloudflared/config"
 	"github.com/cloudflare/cloudflared/connection"
 	"github.com/cloudflare/cloudflared/hello"
@@ -34,7 +33,7 @@ import (
 )
 
 var (
-	testTags        = []tunnelpogs.Tag{tunnelpogs.Tag{Name: "Name", Value: "value"}}
+	testTags        = []tunnelpogs.Tag{{Name: "Name", Value: "value"}}
 	noWarpRouting   = ingress.WarpRoutingConfig{}
 	testWarpRouting = ingress.WarpRoutingConfig{
 		Enabled:        true,
@@ -150,8 +149,7 @@ func TestProxySingleOrigin(t *testing.T) {
 	err := cliCtx.Set("hello-world", "true")
 	require.NoError(t, err)
 
-	allowURLFromArgs := false
-	ingressRule, err := ingress.NewSingleOrigin(cliCtx, allowURLFromArgs)
+	ingressRule, err := ingress.ParseIngressFromConfigAndCLI(&config.Configuration{}, cliCtx, &log)
 	require.NoError(t, err)
 
 	require.NoError(t, ingressRule.StartOrigins(&log, ctx.Done()))
@@ -701,8 +699,7 @@ func TestConnections(t *testing.T) {
 			cancel()
 			assert.Equal(t, test.want.err, err != nil)
 			assert.Equal(t, test.want.message, replayer.Bytes())
-			respPrinter := respWriter.(responsePrinter)
-			assert.Equal(t, test.want.headers, respPrinter.headers())
+			assert.Equal(t, test.want.headers, respWriter.Header())
 			replayer.rw.Reset()
 		})
 	}
@@ -795,10 +792,6 @@ func (p *pipedRequestBody) Close() error {
 	return nil
 }
 
-type responsePrinter interface {
-	headers() http.Header
-}
-
 type wsRespWriter struct {
 	w               io.Writer
 	responseHeaders http.Header
@@ -837,12 +830,16 @@ func (w *wsRespWriter) AddTrailer(trailerName, trailerValue string) {
 }
 
 // respHeaders is a test function to read respHeaders
-func (w *wsRespWriter) headers() http.Header {
+func (w *wsRespWriter) Header() http.Header {
 	// Removing indeterminstic header because it cannot be asserted.
 	w.responseHeaders.Del("Date")
 	return w.responseHeaders
 }
 
+func (w *wsRespWriter) WriteHeader(status int) {
+	// unused
+}
+
 type mockTCPRespWriter struct {
 	w               io.Writer
 	responseHeaders http.Header
@@ -863,7 +860,7 @@ func (m *mockTCPRespWriter) Write(p []byte) (n int, err error) {
 	return m.w.Write(p)
 }
 
-func (w *mockTCPRespWriter) AddTrailer(trailerName, trailerValue string) {
+func (m *mockTCPRespWriter) AddTrailer(trailerName, trailerValue string) {
 	// do nothing
 }
 
@@ -874,10 +871,14 @@ func (m *mockTCPRespWriter) WriteRespHeaders(status int, header http.Header) err
 }
 
 // respHeaders is a test function to read respHeaders
-func (m *mockTCPRespWriter) headers() http.Header {
+func (m *mockTCPRespWriter) Header() http.Header {
 	return m.responseHeaders
 }
 
+func (m *mockTCPRespWriter) WriteHeader(status int) {
+	// do nothing
+}
+
 func createSingleIngressConfig(t *testing.T, service string) ingress.Ingress {
 	ingressConfig := &config.Configuration{
 		Ingress: []config.UnvalidatedIngressRule{

supervisor/tunnel.go

@@ -69,7 +69,6 @@ type TunnelConfig struct {
 	PQKexIdx int
 
 	NamedTunnel      *connection.NamedTunnelProperties
-	MuxerConfig      *connection.MuxerConfig
 	ProtocolSelector connection.ProtocolSelector
 	EdgeTLSConfigs   map[connection.Protocol]*tls.Config
 	PacketConfig     *ingress.GlobalRouterConfig
@@ -91,7 +90,7 @@ func (c *TunnelConfig) registrationOptions(connectionID uint8, OriginLocalIP str
 		OriginLocalIP:        OriginLocalIP,
 		IsAutoupdated:        c.IsAutoupdated,
 		RunFromTerminal:      c.RunFromTerminal,
-		CompressionQuality:   uint64(c.MuxerConfig.CompressionSetting),
+		CompressionQuality:   0,
 		UUID:                 uuid.String(),
 		Features:             c.SupportedFeatures(),
 	}
@@ -106,7 +105,7 @@ func (c *TunnelConfig) connectionOptions(originLocalAddr string, numPreviousAtte
 		Client:              c.NamedTunnel.Client,
 		OriginLocalIP:       originIP,
 		ReplaceExisting:     c.ReplaceExisting,
-		CompressionQuality:  uint8(c.MuxerConfig.CompressionSetting),
+		CompressionQuality:  0,
 		NumPreviousAttempts: numPreviousAttempts,
 	}
 }
@@ -518,21 +517,7 @@ func (e *EdgeTunnelServer) serveConnection(
 		}
 
 	default:
-		edgeConn, err := edgediscovery.DialEdge(ctx, dialTimeout, e.config.EdgeTLSConfigs[protocol], addr.TCP, e.edgeBindAddr)
-		if err != nil {
-			connLog.ConnAwareLogger().Err(err).Msg("Unable to establish connection with Cloudflare edge")
-			return err, true
-		}
-
-		if err := e.serveH2mux(
-			ctx,
-			connLog,
-			edgeConn,
-			connIndex,
-			connectedFuse,
-		); err != nil {
-			return err, false
-		}
+		return fmt.Errorf("invalid protocol selected: %s", protocol), false
 	}
 	return
 }
@@ -545,55 +530,6 @@ func (r unrecoverableError) Error() string {
 	return r.err.Error()
 }
 
-func (e *EdgeTunnelServer) serveH2mux(
-	ctx context.Context,
-	connLog *ConnAwareLogger,
-	edgeConn net.Conn,
-	connIndex uint8,
-	connectedFuse *connectedFuse,
-) error {
-	if e.config.NeedPQ {
-		return unrecoverableError{errors.New("H2Mux transport does not support post-quantum")}
-	}
-	connLog.Logger().Debug().Msgf("Connecting via h2mux")
-	// Returns error from parsing the origin URL or handshake errors
-	handler, err, recoverable := connection.NewH2muxConnection(
-		e.orchestrator,
-		e.config.GracePeriod,
-		e.config.MuxerConfig,
-		edgeConn,
-		connIndex,
-		e.config.Observer,
-		e.gracefulShutdownC,
-		e.config.Log,
-	)
-	if err != nil {
-		if !recoverable {
-			return unrecoverableError{err}
-		}
-		return err
-	}
-
-	errGroup, serveCtx := errgroup.WithContext(ctx)
-
-	errGroup.Go(func() error {
-		connOptions := e.config.connectionOptions(edgeConn.LocalAddr().String(), uint8(connectedFuse.backoff.Retries()))
-		return handler.ServeNamedTunnel(serveCtx, e.config.NamedTunnel, connOptions, connectedFuse)
-	})
-
-	errGroup.Go(func() error {
-		err := listenReconnect(serveCtx, e.reconnectCh, e.gracefulShutdownC)
-		if err != nil {
-			// forcefully break the connection (this is only used for testing)
-			// errgroup will return context canceled for the handler.ServeClassicTunnel
-			connLog.Logger().Debug().Msg("Forcefully breaking h2mux connection")
-		}
-		return err
-	})
-
-	return errGroup.Wait()
-}
-
 func (e *EdgeTunnelServer) serveHTTP2(
 	ctx context.Context,
 	connLog *ConnAwareLogger,