Release 2022.5.1

The way apt works is:

1. It looks at the release file based on the `deb` added to sources.list.
2. It uses this release file to find the relative location of Packages or Packages.gz
3. It uses the pool information from packages to find the relative location of where the .deb file is located and then downloads and installs it.

This PR seeks to take advantage of this information by simply arranging
the files in a way apt expects thereby eliminating the need for an
orchestrating endpoint.

.teamcity/update-homebrew-core.sh

@@ -19,8 +19,8 @@ eval "$(tmp/homebrew/bin/brew shellenv)"
 brew update --force --quiet
 chmod -R go-w "$(brew --prefix)/share/zsh"
 
-git config user.name "cloudflare-warp-bot"
-git config user.email "warp-bot@cloudflare.com"
+git config --global user.name "cloudflare-warp-bot"
+git config --global user.email "warp-bot@cloudflare.com"
 
 # bump formula pr
 brew bump-formula-pr cloudflared --version="$VERSION"

RELEASE_NOTES

@@ -1,3 +1,12 @@
+2022.5.1
+- 2022-05-06 TUN-6146: Release_pkgs is now a generic command line script
+- 2022-05-06 TUN-6185: Fix tcpOverWSOriginService not using original scheme for String representation
+- 2022-05-05 TUN-6175: Simply debian packaging by structural upload
+- 2022-05-05 TUN-5945: Added support for Ubuntu releases
+- 2022-05-04 TUN-6054: Create and upload deb packages to R2
+- 2022-05-03 TUN-6161: Set git user/email for brew core release
+- 2022-05-03 TUN-6166: Fix mocked QUIC transport for UDP proxy manager to return expected error
+- 2022-04-27 TUN-6016: Push local managed tunnels configuration to the edge
 2022.5.0
 - 2022-05-02 TUN-6158: Update golang.org/x/crypto
 - 2022-04-20 VULN-8383 Bump yaml.v2 to yaml.v3

cmd/cloudflared/tunnel/cmd.go

@@ -343,13 +343,13 @@ func StartServer(
 		observer.SendURL(quickTunnelURL)
 	}
 
-	tunnelConfig, dynamicConfig, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
+	tunnelConfig, orchestratorConfig, err := prepareTunnelConfig(c, info, log, logTransport, observer, namedTunnel)
 	if err != nil {
 		log.Err(err).Msg("Couldn't start tunnel")
 		return err
 	}
 
-	orchestrator, err := orchestration.NewOrchestrator(ctx, dynamicConfig, tunnelConfig.Tags, tunnelConfig.Log)
+	orchestrator, err := orchestration.NewOrchestrator(ctx, orchestratorConfig, tunnelConfig.Tags, tunnelConfig.Log)
 	if err != nil {
 		return err
 	}
@@ -388,7 +388,7 @@ func StartServer(
 			info.Version(),
 			hostname,
 			metricsListener.Addr().String(),
-			dynamicConfig.Ingress,
+			orchestratorConfig.Ingress,
 			tunnelConfig.HAConnections,
 		)
 		app := tunnelUI.Launch(ctx, log, logTransport)

cmd/cloudflared/tunnel/configuration.go

@@ -43,6 +43,8 @@ var (
 
 	secretFlags     = [2]*altsrc.StringFlag{credentialsContentsFlag, tunnelTokenFlag}
 	defaultFeatures = []string{supervisor.FeatureAllowRemoteConfig, supervisor.FeatureSerializedHeaders}
+
+	configFlags = []string{"autoupdate-freq", "no-autoupdate", "retries", "protocol", "loglevel", "transport-loglevel", "origincert", "metrics", "metrics-update-freq"}
 )
 
 // returns the first path that contains a cert.pem file. If none of the DefaultConfigSearchDirectories
@@ -348,11 +350,24 @@ func prepareTunnelConfig(
 		ProtocolSelector: protocolSelector,
 		EdgeTLSConfigs:   edgeTLSConfigs,
 	}
-	dynamicConfig := &orchestration.Config{
+	orchestratorConfig := &orchestration.Config{
 		Ingress:            &ingressRules,
 		WarpRoutingEnabled: warpRoutingEnabled,
+		ConfigurationFlags: parseConfigFlags(c),
 	}
-	return tunnelConfig, dynamicConfig, nil
+	return tunnelConfig, orchestratorConfig, nil
+}
+
+func parseConfigFlags(c *cli.Context) map[string]string {
+	result := make(map[string]string)
+
+	for _, flag := range configFlags {
+		if v := c.String(flag); c.IsSet(flag) && v != "" {
+			result[flag] = v
+		}
+	}
+
+	return result
 }
 
 func gracePeriod(c *cli.Context) (time.Duration, error) {

config/configuration.go

@@ -177,9 +177,9 @@ func ValidateUrl(c *cli.Context, allowURLFromArgs bool) (*url.URL, error) {
 }
 
 type UnvalidatedIngressRule struct {
-	Hostname      string              `json:"hostname"`
-	Path          string              `json:"path"`
-	Service       string              `json:"service"`
+	Hostname      string              `json:"hostname,omitempty"`
+	Path          string              `json:"path,omitempty"`
+	Service       string              `json:"service,omitempty"`
 	OriginRequest OriginRequestConfig `yaml:"originRequest" json:"originRequest"`
 }
 
@@ -192,41 +192,41 @@ type UnvalidatedIngressRule struct {
 // - To specify a time.Duration in json, use int64 of the nanoseconds
 type OriginRequestConfig struct {
 	// HTTP proxy timeout for establishing a new connection
-	ConnectTimeout *CustomDuration `yaml:"connectTimeout" json:"connectTimeout"`
+	ConnectTimeout *CustomDuration `yaml:"connectTimeout" json:"connectTimeout,omitempty"`
 	// HTTP proxy timeout for completing a TLS handshake
-	TLSTimeout *CustomDuration `yaml:"tlsTimeout" json:"tlsTimeout"`
+	TLSTimeout *CustomDuration `yaml:"tlsTimeout" json:"tlsTimeout,omitempty"`
 	// HTTP proxy TCP keepalive duration
-	TCPKeepAlive *CustomDuration `yaml:"tcpKeepAlive" json:"tcpKeepAlive"`
+	TCPKeepAlive *CustomDuration `yaml:"tcpKeepAlive" json:"tcpKeepAlive,omitempty"`
 	// HTTP proxy should disable "happy eyeballs" for IPv4/v6 fallback
-	NoHappyEyeballs *bool `yaml:"noHappyEyeballs" json:"noHappyEyeballs"`
+	NoHappyEyeballs *bool `yaml:"noHappyEyeballs" json:"noHappyEyeballs,omitempty"`
 	// HTTP proxy maximum keepalive connection pool size
-	KeepAliveConnections *int `yaml:"keepAliveConnections" json:"keepAliveConnections"`
+	KeepAliveConnections *int `yaml:"keepAliveConnections" json:"keepAliveConnections,omitempty"`
 	// HTTP proxy timeout for closing an idle connection
-	KeepAliveTimeout *CustomDuration `yaml:"keepAliveTimeout" json:"keepAliveTimeout"`
+	KeepAliveTimeout *CustomDuration `yaml:"keepAliveTimeout" json:"keepAliveTimeout,omitempty"`
 	// Sets the HTTP Host header for the local webserver.
-	HTTPHostHeader *string `yaml:"httpHostHeader" json:"httpHostHeader"`
+	HTTPHostHeader *string `yaml:"httpHostHeader" json:"httpHostHeader,omitempty"`
 	// Hostname on the origin server certificate.
-	OriginServerName *string `yaml:"originServerName" json:"originServerName"`
+	OriginServerName *string `yaml:"originServerName" json:"originServerName,omitempty"`
 	// Path to the CA for the certificate of your origin.
 	// This option should be used only if your certificate is not signed by Cloudflare.
-	CAPool *string `yaml:"caPool" json:"caPool"`
+	CAPool *string `yaml:"caPool" json:"caPool,omitempty"`
 	// Disables TLS verification of the certificate presented by your origin.
 	// Will allow any certificate from the origin to be accepted.
 	// Note: The connection from your machine to Cloudflare's Edge is still encrypted.
-	NoTLSVerify *bool `yaml:"noTLSVerify" json:"noTLSVerify"`
+	NoTLSVerify *bool `yaml:"noTLSVerify" json:"noTLSVerify,omitempty"`
 	// Disables chunked transfer encoding.
 	// Useful if you are running a WSGI server.
-	DisableChunkedEncoding *bool `yaml:"disableChunkedEncoding" json:"disableChunkedEncoding"`
+	DisableChunkedEncoding *bool `yaml:"disableChunkedEncoding" json:"disableChunkedEncoding,omitempty"`
 	// Runs as jump host
-	BastionMode *bool `yaml:"bastionMode" json:"bastionMode"`
+	BastionMode *bool `yaml:"bastionMode" json:"bastionMode,omitempty"`
 	// Listen address for the proxy.
-	ProxyAddress *string `yaml:"proxyAddress" json:"proxyAddress"`
+	ProxyAddress *string `yaml:"proxyAddress" json:"proxyAddress,omitempty"`
 	// Listen port for the proxy.
-	ProxyPort *uint `yaml:"proxyPort" json:"proxyPort"`
+	ProxyPort *uint `yaml:"proxyPort" json:"proxyPort,omitempty"`
 	// Valid options are 'socks' or empty.
-	ProxyType *string `yaml:"proxyType" json:"proxyType"`
+	ProxyType *string `yaml:"proxyType" json:"proxyType,omitempty"`
 	// IP rules for the proxy service
-	IPRules []IngressIPRule `yaml:"ipRules" json:"ipRules"`
+	IPRules []IngressIPRule `yaml:"ipRules" json:"ipRules,omitempty"`
 }
 
 type IngressIPRule struct {

connection/connection.go

@@ -30,6 +30,7 @@ var switchingProtocolText = fmt.Sprintf("%d %s", http.StatusSwitchingProtocols,
 
 type Orchestrator interface {
 	UpdateConfig(version int32, config []byte) *pogs.UpdateConfigurationResponse
+	GetConfigJSON() ([]byte, error)
 	GetOriginProxy() (OriginProxy, error)
 }
 

connection/connection_test.go

@@ -42,6 +42,10 @@ type mockOrchestrator struct {
 	originProxy OriginProxy
 }
 
+func (mcr *mockOrchestrator) GetConfigJSON() ([]byte, error) {
+	return nil, fmt.Errorf("not implemented")
+}
+
 func (*mockOrchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
 	return &tunnelpogs.UpdateConfigurationResponse{
 		LastAppliedVersion: version,

connection/control.go

@@ -30,11 +30,15 @@ type controlStream struct {
 // ControlStreamHandler registers connections with origintunneld and initiates graceful shutdown.
 type ControlStreamHandler interface {
 	// ServeControlStream handles the control plane of the transport in the current goroutine calling this
-	ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions) error
+	ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error
 	// IsStopped tells whether the method above has finished
 	IsStopped() bool
 }
 
+type TunnelConfigJSONGetter interface {
+	GetConfigJSON() ([]byte, error)
+}
+
 // NewControlStream returns a new instance of ControlStreamHandler
 func NewControlStream(
 	observer *Observer,
@@ -63,15 +67,28 @@ func (c *controlStream) ServeControlStream(
 	ctx context.Context,
 	rw io.ReadWriteCloser,
 	connOptions *tunnelpogs.ConnectionOptions,
+	tunnelConfigGetter TunnelConfigJSONGetter,
 ) error {
 	rpcClient := c.newRPCClientFunc(ctx, rw, c.observer.log)
 
-	if err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.observer); err != nil {
+	registrationDetails, err := rpcClient.RegisterConnection(ctx, c.namedTunnelProperties, connOptions, c.connIndex, c.observer)
+	if err != nil {
 		rpcClient.Close()
 		return err
 	}
 	c.connectedFuse.Connected()
 
+	// if conn index is 0 and tunnel is not remotely managed, then send local ingress rules configuration
+	if c.connIndex == 0 && !registrationDetails.TunnelIsRemotelyManaged {
+		if tunnelConfig, err := tunnelConfigGetter.GetConfigJSON(); err == nil {
+			if err := rpcClient.SendLocalConfiguration(ctx, tunnelConfig, c.observer); err != nil {
+				c.observer.log.Err(err).Msg("unable to send local configuration")
+			}
+		} else {
+			c.observer.log.Err(err).Msg("failed to obtain current configuration")
+		}
+	}
+
 	c.waitForUnregister(ctx, rpcClient)
 	return nil
 }

connection/http2.go

@@ -117,7 +117,7 @@ func (c *HTTP2Connection) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 	switch connType {
 	case TypeControlStream:
-		if err := c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions); err != nil {
+		if err := c.controlStreamHandler.ServeControlStream(r.Context(), respWriter, c.connOptions, c.orchestrator); err != nil {
 			c.controlStreamErr = err
 			c.log.Error().Err(err)
 			respWriter.WriteErrorResponse()

connection/http2_test.go

@@ -15,6 +15,7 @@ import (
 	"time"
 
 	"github.com/gobwas/ws/wsutil"
+	"github.com/google/uuid"
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -166,18 +167,26 @@ type mockNamedTunnelRPCClient struct {
 	unregistered chan struct{}
 }
 
+func (mc mockNamedTunnelRPCClient) SendLocalConfiguration(c context.Context, config []byte, observer *Observer) error {
+	return nil
+}
+
 func (mc mockNamedTunnelRPCClient) RegisterConnection(
 	c context.Context,
 	properties *NamedTunnelProperties,
 	options *tunnelpogs.ConnectionOptions,
 	connIndex uint8,
 	observer *Observer,
-) error {
+) (*tunnelpogs.ConnectionDetails, error) {
 	if mc.shouldFail != nil {
-		return mc.shouldFail
+		return nil, mc.shouldFail
 	}
 	close(mc.registered)
-	return nil
+	return &tunnelpogs.ConnectionDetails{
+		Location:                "LIS",
+		UUID:                    uuid.New(),
+		TunnelIsRemotelyManaged: false,
+	}, nil
 }
 
 func (mc mockNamedTunnelRPCClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
@@ -477,7 +486,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 
 	select {
 	case <-rpcClientFactory.registered:
-		break //ok
+		break // ok
 	case <-time.Tick(time.Second):
 		t.Fatal("timeout out waiting for registration")
 	}
@@ -487,7 +496,7 @@ func TestGracefulShutdownHTTP2(t *testing.T) {
 
 	select {
 	case <-rpcClientFactory.unregistered:
-		break //ok
+		break // ok
 	case <-time.Tick(time.Second):
 		t.Fatal("timeout out waiting for unregistered signal")
 	}

connection/metrics.go

@@ -13,6 +13,7 @@ const (
 	MetricsNamespace = "cloudflared"
 	TunnelSubsystem  = "tunnel"
 	muxerSubsystem   = "muxer"
+	configSubsystem  = "config"
 )
 
 type muxerMetrics struct {
@@ -36,6 +37,11 @@ type muxerMetrics struct {
 	compRateAve      *prometheus.GaugeVec
 }
 
+type localConfigMetrics struct {
+	pushes       prometheus.Counter
+	pushesErrors prometheus.Counter
+}
+
 type tunnelMetrics struct {
 	timerRetries    prometheus.Gauge
 	serverLocations *prometheus.GaugeVec
@@ -51,6 +57,39 @@ type tunnelMetrics struct {
 	muxerMetrics        *muxerMetrics
 	tunnelsHA           tunnelsForHA
 	userHostnamesCounts *prometheus.CounterVec
+
+	localConfigMetrics *localConfigMetrics
+}
+
+func newLocalConfigMetrics() *localConfigMetrics {
+
+	pushesMetric := prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: MetricsNamespace,
+			Subsystem: configSubsystem,
+			Name:      "local_config_pushes",
+			Help:      "Number of local configuration pushes to the edge",
+		},
+	)
+
+	pushesErrorsMetric := prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: MetricsNamespace,
+			Subsystem: configSubsystem,
+			Name:      "local_config_pushes_errors",
+			Help:      "Number of errors occurred during local configuration pushes",
+		},
+	)
+
+	prometheus.MustRegister(
+		pushesMetric,
+		pushesErrorsMetric,
+	)
+
+	return &localConfigMetrics{
+		pushes:       pushesMetric,
+		pushesErrors: pushesErrorsMetric,
+	}
 }
 
 func newMuxerMetrics() *muxerMetrics {
@@ -386,6 +425,7 @@ func initTunnelMetrics() *tunnelMetrics {
 		regFail:             registerFail,
 		rpcFail:             rpcFail,
 		userHostnamesCounts: userHostnamesCounts,
+		localConfigMetrics:  newLocalConfigMetrics(),
 	}
 }
 

connection/quic.go

@@ -111,7 +111,7 @@ func (q *QUICConnection) Serve(ctx context.Context) error {
 
 func (q *QUICConnection) serveControlStream(ctx context.Context, controlStream quic.Stream) error {
 	// This blocks until the control plane is done.
-	err := q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions)
+	err := q.controlStreamHandler.ServeControlStream(ctx, controlStream, q.connOptions, q.orchestrator)
 	if err != nil {
 		// Not wrapping error here to be consistent with the http2 message.
 		return err

connection/quic_test.go

@@ -163,7 +163,7 @@ type fakeControlStream struct {
 	ControlStreamHandler
 }
 
-func (fakeControlStream) ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions) error {
+func (fakeControlStream) ServeControlStream(ctx context.Context, rw io.ReadWriteCloser, connOptions *tunnelpogs.ConnectionOptions, tunnelConfigGetter TunnelConfigJSONGetter) error {
 	<-ctx.Done()
 	return nil
 }

connection/rpc.go

@@ -58,6 +58,11 @@ type NamedTunnelRPCClient interface {
 		options *tunnelpogs.ConnectionOptions,
 		connIndex uint8,
 		observer *Observer,
+	) (*tunnelpogs.ConnectionDetails, error)
+	SendLocalConfiguration(
+		c context.Context,
+		config []byte,
+		observer *Observer,
 	) error
 	GracefulShutdown(ctx context.Context, gracePeriod time.Duration)
 	Close()
@@ -90,7 +95,7 @@ func (rsc *registrationServerClient) RegisterConnection(
 	options *tunnelpogs.ConnectionOptions,
 	connIndex uint8,
 	observer *Observer,
-) error {
+) (*tunnelpogs.ConnectionDetails, error) {
 	conn, err := rsc.client.RegisterConnection(
 		ctx,
 		properties.Credentials.Auth(),
@@ -101,10 +106,10 @@ func (rsc *registrationServerClient) RegisterConnection(
 	if err != nil {
 		if err.Error() == DuplicateConnectionError {
 			observer.metrics.regFail.WithLabelValues("dup_edge_conn", "registerConnection").Inc()
-			return errDuplicationConnection
+			return nil, errDuplicationConnection
 		}
 		observer.metrics.regFail.WithLabelValues("server_error", "registerConnection").Inc()
-		return serverRegistrationErrorFromRPC(err)
+		return nil, serverRegistrationErrorFromRPC(err)
 	}
 
 	observer.metrics.regSuccess.WithLabelValues("registerConnection").Inc()
@@ -112,7 +117,18 @@ func (rsc *registrationServerClient) RegisterConnection(
 	observer.logServerInfo(connIndex, conn.Location, fmt.Sprintf("Connection %s registered", conn.UUID))
 	observer.sendConnectedEvent(connIndex, conn.Location)
 
-	return nil
+	return conn, nil
+}
+
+func (rsc *registrationServerClient) SendLocalConfiguration(ctx context.Context, config []byte, observer *Observer) (err error) {
+	observer.metrics.localConfigMetrics.pushes.Inc()
+	defer func() {
+		if err != nil {
+			observer.metrics.localConfigMetrics.pushesErrors.Inc()
+		}
+	}()
+
+	return rsc.client.SendLocalConfiguration(ctx, config)
 }
 
 func (rsc *registrationServerClient) GracefulShutdown(ctx context.Context, gracePeriod time.Duration) {
@@ -274,7 +290,7 @@ func (h *h2muxConnection) registerNamedTunnel(
 	rpcClient := h.newRPCClientFunc(ctx, stream, h.observer.log)
 	defer rpcClient.Close()
 
-	if err = rpcClient.RegisterConnection(ctx, namedTunnel, connOptions, h.connIndex, h.observer); err != nil {
+	if _, err = rpcClient.RegisterConnection(ctx, namedTunnel, connOptions, h.connIndex, h.observer); err != nil {
 		return err
 	}
 	return nil

datagramsession/manager_test.go

@@ -255,7 +255,10 @@ func (rc *datagramChannel) Send(ctx context.Context, sessionID uuid.UUID, payloa
 	case <-ctx.Done():
 		return ctx.Err()
 	case <-rc.closedChan:
-		return fmt.Errorf("datagram channel closed")
+		return &errClosedSession{
+			message:  fmt.Errorf("datagram channel closed").Error(),
+			byRemote: true,
+		}
 	case rc.datagramChan <- &newDatagram{sessionID: sessionID, payload: payload}:
 		return nil
 	}
@@ -266,7 +269,11 @@ func (rc *datagramChannel) Receive(ctx context.Context) (uuid.UUID, []byte, erro
 	case <-ctx.Done():
 		return uuid.Nil, nil, ctx.Err()
 	case <-rc.closedChan:
-		return uuid.Nil, nil, fmt.Errorf("datagram channel closed")
+		err := &errClosedSession{
+			message:  fmt.Errorf("datagram channel closed").Error(),
+			byRemote: true,
+		}
+		return uuid.Nil, nil, err
 	case msg := <-rc.datagramChan:
 		return msg.sessionID, msg.payload, nil
 	}

ingress/config.go

@@ -47,18 +47,26 @@ type RemoteConfig struct {
 	WarpRouting config.WarpRoutingConfig
 }
 
-type remoteConfigJSON struct {
-	GlobalOriginRequest config.OriginRequestConfig      `json:"originRequest"`
+type RemoteConfigJSON struct {
+	GlobalOriginRequest *config.OriginRequestConfig     `json:"originRequest,omitempty"`
 	IngressRules        []config.UnvalidatedIngressRule `json:"ingress"`
 	WarpRouting         config.WarpRoutingConfig        `json:"warp-routing"`
 }
 
 func (rc *RemoteConfig) UnmarshalJSON(b []byte) error {
-	var rawConfig remoteConfigJSON
+	var rawConfig RemoteConfigJSON
+
 	if err := json.Unmarshal(b, &rawConfig); err != nil {
 		return err
 	}
-	ingress, err := validateIngress(rawConfig.IngressRules, originRequestFromConfig(rawConfig.GlobalOriginRequest))
+
+	// if nil, just assume the default values.
+	globalOriginRequestConfig := rawConfig.GlobalOriginRequest
+	if globalOriginRequestConfig == nil {
+		globalOriginRequestConfig = &config.OriginRequestConfig{}
+	}
+
+	ingress, err := validateIngress(rawConfig.IngressRules, originRequestFromConfig(*globalOriginRequestConfig))
 	if err != nil {
 		return err
 	}
@@ -387,3 +395,91 @@ func setConfig(defaults OriginRequestConfig, overrides config.OriginRequestConfi
 	cfg.setIPRules(overrides)
 	return cfg
 }
+
+func ConvertToRawOriginConfig(c OriginRequestConfig) config.OriginRequestConfig {
+	var connectTimeout *config.CustomDuration
+	var tlsTimeout *config.CustomDuration
+	var tcpKeepAlive *config.CustomDuration
+	var keepAliveConnections *int
+	var keepAliveTimeout *config.CustomDuration
+	var proxyAddress *string
+
+	if c.ConnectTimeout != defaultConnectTimeout {
+		connectTimeout = &c.ConnectTimeout
+	}
+	if c.TLSTimeout != defaultTLSTimeout {
+		tlsTimeout = &c.TLSTimeout
+	}
+	if c.TCPKeepAlive != defaultTCPKeepAlive {
+		tcpKeepAlive = &c.TCPKeepAlive
+	}
+	if c.KeepAliveConnections != defaultKeepAliveConnections {
+		keepAliveConnections = &c.KeepAliveConnections
+	}
+	if c.KeepAliveTimeout != defaultKeepAliveTimeout {
+		keepAliveTimeout = &c.KeepAliveTimeout
+	}
+	if c.ProxyAddress != defaultProxyAddress {
+		proxyAddress = &c.ProxyAddress
+	}
+
+	return config.OriginRequestConfig{
+		ConnectTimeout:         connectTimeout,
+		TLSTimeout:             tlsTimeout,
+		TCPKeepAlive:           tcpKeepAlive,
+		NoHappyEyeballs:        defaultBoolToNil(c.NoHappyEyeballs),
+		KeepAliveConnections:   keepAliveConnections,
+		KeepAliveTimeout:       keepAliveTimeout,
+		HTTPHostHeader:         emptyStringToNil(c.HTTPHostHeader),
+		OriginServerName:       emptyStringToNil(c.OriginServerName),
+		CAPool:                 emptyStringToNil(c.CAPool),
+		NoTLSVerify:            defaultBoolToNil(c.NoTLSVerify),
+		DisableChunkedEncoding: defaultBoolToNil(c.DisableChunkedEncoding),
+		BastionMode:            defaultBoolToNil(c.BastionMode),
+		ProxyAddress:           proxyAddress,
+		ProxyPort:              zeroUIntToNil(c.ProxyPort),
+		ProxyType:              emptyStringToNil(c.ProxyType),
+		IPRules:                convertToRawIPRules(c.IPRules),
+	}
+}
+
+func convertToRawIPRules(ipRules []ipaccess.Rule) []config.IngressIPRule {
+	result := make([]config.IngressIPRule, 0)
+	for _, r := range ipRules {
+		cidr := r.StringCIDR()
+
+		newRule := config.IngressIPRule{
+			Prefix: &cidr,
+			Ports:  r.Ports(),
+			Allow:  r.RulePolicy(),
+		}
+
+		result = append(result, newRule)
+	}
+
+	return result
+}
+
+func defaultBoolToNil(b bool) *bool {
+	if b == false {
+		return nil
+	}
+
+	return &b
+}
+
+func emptyStringToNil(s string) *string {
+	if s == "" {
+		return nil
+	}
+
+	return &s
+}
+
+func zeroUIntToNil(v uint) *uint {
+	if v == 0 {
+		return nil
+	}
+
+	return &v
+}

ingress/origin_service.go

@@ -109,6 +109,7 @@ func (o rawTCPService) MarshalJSON() ([]byte, error) {
 // tcpOverWSService models TCP origins serving eyeballs connecting over websocket, such as
 // cloudflared access commands.
 type tcpOverWSService struct {
+	scheme        string
 	dest          string
 	isBastion     bool
 	streamHandler streamHandlerFunc
@@ -130,7 +131,8 @@ func newTCPOverWSService(url *url.URL) *tcpOverWSService {
 		addPortIfMissing(url, 7864) // just a random port since there isn't a default in this case
 	}
 	return &tcpOverWSService{
-		dest: url.Host,
+		scheme: url.Scheme,
+		dest:   url.Host,
 	}
 }
 
@@ -160,7 +162,12 @@ func (o *tcpOverWSService) String() string {
 	if o.isBastion {
 		return ServiceBastion
 	}
-	return o.dest
+
+	if o.scheme != "" {
+		return fmt.Sprintf("%s://%s", o.scheme, o.dest)
+	} else {
+		return o.dest
+	}
 }
 
 func (o *tcpOverWSService) start(log *zerolog.Logger, _ <-chan struct{}, cfg OriginRequestConfig) error {

ipaccess/access.go

@@ -99,3 +99,15 @@ func (ipr *Rule) PortsString() string {
 	}
 	return "all"
 }
+
+func (ipr *Rule) Ports() []int {
+	return ipr.ports
+}
+
+func (ipr *Rule) RulePolicy() bool {
+	return ipr.allow
+}
+
+func (ipr *Rule) StringCIDR() string {
+	return ipr.ipNet.String()
+}

metrics/metrics.go

@@ -23,7 +23,7 @@ const (
 )
 
 type orchestrator interface {
-	GetConfigJSON() ([]byte, error)
+	GetVersionedConfigJSON() ([]byte, error)
 }
 
 func newMetricsHandler(
@@ -47,7 +47,7 @@ func newMetricsHandler(
 	})
 	if orchestrator != nil {
 		router.HandleFunc("/config", func(w http.ResponseWriter, r *http.Request) {
-			json, err := orchestrator.GetConfigJSON()
+			json, err := orchestrator.GetVersionedConfigJSON()
 			if err != nil {
 				w.WriteHeader(500)
 				_, _ = fmt.Fprintf(w, "ERR: %v", err)

orchestration/config.go

@@ -1,15 +1,66 @@
 package orchestration
 
 import (
+	"encoding/json"
+
+	"github.com/cloudflare/cloudflared/config"
 	"github.com/cloudflare/cloudflared/ingress"
 )
 
-type newConfig struct {
+type newRemoteConfig struct {
 	ingress.RemoteConfig
 	// Add more fields when we support other settings in tunnel orchestration
 }
 
+type newLocalConfig struct {
+	RemoteConfig       ingress.RemoteConfig
+	ConfigurationFlags map[string]string `json:"__configuration_flags,omitempty"`
+}
+
+// Config is the original config as read and parsed by cloudflared.
 type Config struct {
 	Ingress            *ingress.Ingress
 	WarpRoutingEnabled bool
+
+	// Extra settings used to configure this instance but that are not eligible for remotely management
+	// ie. (--protocol, --loglevel, ...)
+	ConfigurationFlags map[string]string
+}
+
+func (rc *newLocalConfig) MarshalJSON() ([]byte, error) {
+	var r = struct {
+		ConfigurationFlags map[string]string `json:"__configuration_flags,omitempty"`
+		ingress.RemoteConfigJSON
+	}{
+		ConfigurationFlags: rc.ConfigurationFlags,
+		RemoteConfigJSON: ingress.RemoteConfigJSON{
+			// UI doesn't support top level configs, so we reconcile to individual ingress configs.
+			GlobalOriginRequest: nil,
+			IngressRules:        convertToUnvalidatedIngressRules(rc.RemoteConfig.Ingress),
+			WarpRouting:         rc.RemoteConfig.WarpRouting,
+		},
+	}
+
+	return json.Marshal(r)
+}
+
+func convertToUnvalidatedIngressRules(i ingress.Ingress) []config.UnvalidatedIngressRule {
+	result := make([]config.UnvalidatedIngressRule, 0)
+	for _, rule := range i.Rules {
+		var path string
+		if rule.Path != nil {
+			path = rule.Path.String()
+		}
+
+		newRule := config.UnvalidatedIngressRule{
+			Hostname:      rule.Hostname,
+			Path:          path,
+			Service:       rule.Service.String(),
+			OriginRequest: ingress.ConvertToRawOriginConfig(rule.Config),
+		}
+
+		result = append(result, newRule)
+	}
+
+	return result
 }

orchestration/config_test.go

@@ -0,0 +1,82 @@
+package orchestration
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/cloudflare/cloudflared/ingress"
+)
+
+// TestNewLocalConfig_MarshalJSON tests that we are able to converte a compiled and validated config back
+// into an "unvalidated" format which is compatible with Remote Managed configurations.
+func TestNewLocalConfig_MarshalJSON(t *testing.T) {
+
+	rawConfig := []byte(`
+	{
+		"originRequest": {
+					"connectTimeout": 160,
+					"httpHostHeader": "default"
+		},
+		"ingress": [
+			{
+				"hostname": "tun.example.com",
+				"service": "https://localhost:8000"
+			},
+			{
+				"hostname": "*",
+				"service": "https://localhost:8001",
+				"originRequest": {
+					"connectTimeout": 121,
+					"tlsTimeout": 2,
+					"noHappyEyeballs": false,
+					"tcpKeepAlive": 2,
+					"keepAliveConnections": 2,
+					"keepAliveTimeout": 2,
+					"httpHostHeader": "def",
+					"originServerName": "b2",
+					"caPool": "/tmp/path1",
+					"noTLSVerify": false,
+					"disableChunkedEncoding": false,
+					"bastionMode": false,
+					"proxyAddress": "interface",
+					"proxyPort": 200,
+					"proxyType": "",
+					"ipRules": [
+						{
+							"prefix": "10.0.0.0/16",
+							"ports": [3000, 3030],
+							"allow": false
+						},
+						{
+							"prefix": "192.16.0.0/24",
+							"ports": [5000, 5050],
+							"allow": true
+						}
+					]
+				}
+			}
+		]
+	}
+	`)
+
+	var expectedConfig ingress.RemoteConfig
+	err := json.Unmarshal(rawConfig, &expectedConfig)
+	require.NoError(t, err)
+
+	c := &newLocalConfig{
+		RemoteConfig:       expectedConfig,
+		ConfigurationFlags: nil,
+	}
+
+	jsonSerde, err := json.Marshal(c)
+	require.NoError(t, err)
+
+	var config ingress.RemoteConfig
+	err = json.Unmarshal(jsonSerde, &config)
+	require.NoError(t, err)
+
+	require.Equal(t, config.WarpRouting.Enabled, false)
+	require.Equal(t, config.Ingress.Rules, expectedConfig.Ingress.Rules)
+}

orchestration/orchestrator.go

@@ -54,7 +54,7 @@ func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag,
 	return o, nil
 }
 
-// Update creates a new proxy with the new ingress rules
+// UpdateConfig creates a new proxy with the new ingress rules
 func (o *Orchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.UpdateConfigurationResponse {
 	o.lock.Lock()
 	defer o.lock.Unlock()
@@ -63,12 +63,12 @@ func (o *Orchestrator) UpdateConfig(version int32, config []byte) *tunnelpogs.Up
 		o.log.Debug().
 			Int32("current_version", o.currentVersion).
 			Int32("received_version", version).
-			Msg("Current version is equal or newer than receivied version")
+			Msg("Current version is equal or newer than received version")
 		return &tunnelpogs.UpdateConfigurationResponse{
 			LastAppliedVersion: o.currentVersion,
 		}
 	}
-	var newConf newConfig
+	var newConf newRemoteConfig
 	if err := json.Unmarshal(config, &newConf); err != nil {
 		o.log.Err(err).
 			Int32("version", version).
@@ -131,10 +131,26 @@ func (o *Orchestrator) updateIngress(ingressRules ingress.Ingress, warpRoutingEn
 	return nil
 }
 
-// GetConfigJSON returns the current version and configuration as JSON
+// GetConfigJSON returns the current json serialization of the config as the edge understands it
 func (o *Orchestrator) GetConfigJSON() ([]byte, error) {
 	o.lock.RLock()
 	defer o.lock.RUnlock()
+
+	c := &newLocalConfig{
+		RemoteConfig: ingress.RemoteConfig{
+			Ingress:     *o.config.Ingress,
+			WarpRouting: config.WarpRoutingConfig{Enabled: o.config.WarpRoutingEnabled},
+		},
+		ConfigurationFlags: o.config.ConfigurationFlags,
+	}
+
+	return json.Marshal(c)
+}
+
+// GetVersionedConfigJSON returns the current version and configuration as JSON
+func (o *Orchestrator) GetVersionedConfigJSON() ([]byte, error) {
+	o.lock.RLock()
+	defer o.lock.RUnlock()
 	var currentConfiguration = struct {
 		Version int32 `json:"version"`
 		Config  struct {

orchestration/orchestrator_test.go

@@ -2,6 +2,7 @@ package orchestration
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -641,6 +642,19 @@ func TestPersistentConnection(t *testing.T) {
 	wg.Wait()
 }
 
+func TestSerializeLocalConfig(t *testing.T) {
+	c := &newLocalConfig{
+		RemoteConfig: ingress.RemoteConfig{
+			Ingress:     ingress.Ingress{},
+			WarpRouting: config.WarpRoutingConfig{},
+		},
+		ConfigurationFlags: map[string]string{"a": "b"},
+	}
+
+	result, _ := json.Marshal(c)
+	fmt.Println(string(result))
+}
+
 func wsEcho(w http.ResponseWriter, r *http.Request) {
 	upgrader := gows.Upgrader{}
 

release_pkgs.py

@@ -0,0 +1,225 @@
+"""
+    This is a utility for creating deb and rpm packages, signing them 
+    and uploading them to a storage and adding metadata to workers KV.
+
+    It has two over-arching responsiblities:
+    1. Create deb and yum repositories from .deb and .rpm files. 
+       This is also responsible for signing the packages and generally preparing 
+       them to be in an uploadable state.
+    2. Upload these packages to a storage in a format that apt and yum expect.
+"""
+import subprocess
+import os
+import argparse
+import logging
+from hashlib import sha256
+
+import boto3
+from botocore.client import Config
+from botocore.exceptions import ClientError
+
+# The front facing R2 URL to access assets from.
+R2_ASSET_URL = 'https://demo-r2-worker.cloudflare-tunnel.workers.dev/'
+
+class PkgUploader:
+    def __init__(self, account_id, bucket_name, client_id, client_secret):
+        self.account_id = account_id
+        self.bucket_name = bucket_name
+        self.client_id = client_id
+        self.client_secret = client_secret
+
+    def upload_pkg_to_r2(self, filename, upload_file_path):
+        endpoint_url = f"https://{self.account_id}.r2.cloudflarestorage.com"
+        token_secret_hash = sha256(self.client_secret.encode()).hexdigest()
+         
+        config = Config(
+            region_name = 'auto',
+            s3={
+                "addressing_style": "path",
+            }
+        )
+         
+        r2 = boto3.client(
+            "s3",
+            endpoint_url=endpoint_url,
+            aws_access_key_id=self.client_id,
+            aws_secret_access_key=token_secret_hash,
+            config=config,
+        )
+
+        print(f"uploading asset: {filename} to {upload_file_path}...")
+        try:
+            r2.upload_file(filename, self.bucket_name, upload_file_path)
+        except ClientError as e:
+            raise e
+
+                  
+class PkgCreator:
+    """
+        The distribution conf is what dictates to reprepro, the debian packaging building
+        and signing tool we use, what distros to support, what GPG key to use for signing
+        and what to call the debian binary etc. This function creates it "./conf/distributions".
+
+        origin - name of your package (String)
+        label - label of your package (could be same as the name) (String)
+        release - release you want this to be distributed for (List of Strings)
+        components - could be a channel like main/stable/beta
+        archs - Architecture (List of Strings)
+        description - (String)
+        gpg_key_id - gpg key id of what you want to use to sign the packages.(String) 
+    """
+    def create_distribution_conf(self, 
+            file_path,
+            origin,
+            label,
+            releases,
+            archs,
+            components, 
+            description,
+            gpg_key_id ):
+        with open(file_path, "w") as distributions_file:
+            for release in releases:
+                distributions_file.write(f"Origin: {origin}\n")
+                distributions_file.write(f"Label: {label}\n")
+                distributions_file.write(f"Codename: {release}\n")
+                archs_list = " ".join(archs)
+                distributions_file.write(f"Architectures: {archs_list}\n")
+                distributions_file.write(f"Components: {components}\n")
+                distributions_file.write(f"Description: {description} - {release}\n")
+                distributions_file.write(f"SignWith: {gpg_key_id}\n")
+                distributions_file.write("\n")
+        return distributions_file
+
+    """
+        Uses the reprepro tool to generate packages, sign them and create the InRelease as specified
+        by the distribution_conf file. 
+
+        This function creates three folders db, pool and dist. 
+        db and pool contain information and metadata about builds. We can ignore these.
+        dist: contains all the pkgs and signed releases that are necessary for an apt download.
+    """
+    def create_deb_pkgs(self, release, deb_file):
+        self._clean_build_resources()
+        subprocess.call(("reprepro", "includedeb", release, deb_file))
+
+    """
+        This is mostly useful to clear previously built db, dist and pool resources.
+    """
+    def _clean_build_resources(self):
+        subprocess.call(("reprepro", "clearvanished"))
+
+"""
+    Walks through a directory and uploads it's assets to R2.
+    directory : root directory to walk through (String).
+    release: release string. If this value is none, a specific release path will not be created 
+              and the release will be uploaded to the default path. 
+    binary: name of the binary to upload
+"""
+def upload_from_directories(pkg_uploader, directory, release, binary):
+     for root, _ , files in os.walk(directory):
+        for file in files:
+            upload_file_name = os.path.join(binary, root, file)
+            if release:
+                upload_file_name = os.path.join(release, upload_file_name)
+            filename = os.path.join(root,file)
+            try: 
+                pkg_uploader.upload_pkg_to_r2(filename, upload_file_name)
+            except ClientError as e:
+                logging.error(e)
+                return 
+
+""" 
+    1. looks into a built_artifacts folder for cloudflared debs
+    2. creates Packages.gz, InRelease (signed) files
+    3. uploads them to Cloudflare R2 
+
+    pkg_creator, pkg_uploader: are instantiations of the two classes above.
+
+    gpg_key_id: is an id indicating the key the package should be signed with. The public key of this id will be 
+    uploaded to R2 so it can be presented to apt downloaders.
+
+    release_version: is the cloudflared release version.
+"""
+def create_deb_packaging(pkg_creator, pkg_uploader, releases, gpg_key_id, binary_name, archs, package_component, release_version):
+    # set configuration for package creation.
+    print(f"initialising configuration for {binary_name} , {archs}")
+    pkg_creator.create_distribution_conf(
+    "./conf/distributions",
+    binary_name,
+    binary_name,
+    releases,
+    archs,
+    package_component,
+    f"apt repository for {binary_name}",
+    gpg_key_id)
+
+    # create deb pkgs
+    for release in releases:
+        for arch in archs:
+            print(f"creating deb pkgs for {release} and {arch}...")
+            pkg_creator.create_deb_pkgs(release, f"./built_artifacts/cloudflared-linux-{arch}.deb")
+
+    print("uploading latest to r2...")
+    upload_from_directories(pkg_uploader, "dists", None, binary_name)
+    upload_from_directories(pkg_uploader, "pool", None, binary_name)
+
+    print(f"uploading versioned release {release_version} to r2...")
+    upload_from_directories(pkg_uploader, "dists", release_version, binary_name)
+    upload_from_directories(pkg_uploader, "pool", release_version, binary_name)
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description="Creates linux releases and uploads them in a packaged format"
+    )
+
+    parser.add_argument(
+            "--bucket", default=os.environ.get("R2_BUCKET_NAME"), help="R2 Bucket name"
+    )
+    parser.add_argument(
+            "--id", default=os.environ.get("R2_CLIENT_ID"), help="R2 Client ID"
+    )
+    parser.add_argument(
+            "--secret", default=os.environ.get("R2_CLIENT_SECRET"), help="R2 Client Secret"
+    )
+    parser.add_argument(
+            "--account", default=os.environ.get("R2_ACCOUNT_ID"), help="R2 Account Tag"
+    )
+    parser.add_argument(
+            "--release-tag", default=os.environ.get("RELEASE_VERSION"), help="Release version you want your pkgs to be\
+            prefixed with"
+    )
+
+    parser.add_argument(
+           "--binary", default=os.environ.get("BINARY_NAME"), help="The name of the binary the packages are for"
+    )
+
+    parser.add_argument(
+            "--gpg-key-id", default=os.environ.get("GPG_KEY_ID"), help="gpg key ID that's being used to sign release\
+            packages."
+    )
+
+    parser.add_argument(
+            "--deb-based-releases", default=["bookworm", "bullseye", "buster", "jammy", "impish", "focal", "bionic"],
+            help="list of debian based releases that need to be packaged for"
+    )
+
+    parser.add_argument(
+            "--archs", default=["amd64", "386", "arm64"], help="list of architectures we want to package for. Note that\
+            it is the caller's responsiblity to ensure that these debs are already present in a directory. This script\
+            will not build binaries or create their debs."
+            )
+    args = parser.parse_args()
+
+    return args
+
+if __name__ == "__main__":
+    try:
+        args = parse_args()
+    except Exception as e:
+        logging.exception(e)
+        exit(1)
+
+    pkg_creator = PkgCreator()
+    pkg_uploader = PkgUploader(args.account, args.bucket, args.id, args.secret)
+    create_deb_packaging(pkg_creator, pkg_uploader, args.deb_based_releases, args.gpg_key_id, args.binary, 
+            args.archs, "main", args.release_tag)

tunnelrpc/pogs/connectionrpc.go

@@ -175,6 +175,24 @@ func (c RegistrationServer_PogsClient) RegisterConnection(ctx context.Context, a
 	return nil, newRPCError("unknown result which %d", result.Which())
 }
 
+func (c RegistrationServer_PogsClient) SendLocalConfiguration(ctx context.Context, config []byte) error {
+	client := tunnelrpc.TunnelServer{Client: c.Client}
+	promise := client.UpdateLocalConfiguration(ctx, func(p tunnelrpc.RegistrationServer_updateLocalConfiguration_Params) error {
+		if err := p.SetConfig(config); err != nil {
+			return err
+		}
+
+		return nil
+	})
+
+	_, err := promise.Struct()
+	if err != nil {
+		return wrapRPCError(err)
+	}
+
+	return nil
+}
+
 func (c RegistrationServer_PogsClient) UnregisterConnection(ctx context.Context) error {
 	client := tunnelrpc.TunnelServer{Client: c.Client}
 	promise := client.UnregisterConnection(ctx, func(p tunnelrpc.RegistrationServer_unregisterConnection_Params) error {