Release 2023.6.0

This code was leaking streams because it wasn't closing the quic stream
after unregistering from the edge.

Makefile

@@ -140,6 +140,7 @@ container:
 generate-docker-version:
 	echo latest $(VERSION) > versions
 
+
 .PHONY: test
 test: vet
 ifndef CI
@@ -147,9 +148,18 @@ ifndef CI
 else
 	@mkdir -p .cover
 	go test -v -mod=vendor -race $(LDFLAGS) -coverprofile=".cover/c.out" ./...
-	go tool cover -html ".cover/c.out" -o .cover/all.html
 endif
 
+.PHONY: cover
+cover:
+	@echo ""
+	@echo "=====> Total test coverage: <====="
+	@echo ""
+	# Print the overall coverage here for quick access.
+	$Q go tool cover -func ".cover/c.out" | grep "total:" | awk '{print $$3}'
+	# Generate the HTML report that can be viewed from the browser in CI.
+	$Q go tool cover -html ".cover/c.out" -o .cover/all.html
+
 .PHONY: test-ssh-server
 test-ssh-server:
 	docker-compose -f ssh_server_tests/docker-compose.yml up

RELEASE_NOTES

@@ -1,3 +1,8 @@
+2023.6.0
+- 2023-06-15 TUN-7471: Fixes cloudflared not closing the quic stream on unregister UDP session
+- 2023-06-09 TUN-7463: Add default ingress rule if no ingress rules are provided when updating the configuration
+- 2023-05-31 TUN-7447: Add a cover build to report code coverage
+
 2023.5.1
 - 2023-05-16 TUN-7424: Add CORS headers to host_details responses
 - 2023-05-11 TUN-7421: Add *.cloudflare.com to permitted Origins for management WebSocket requests

cfsetup.yaml

@@ -29,6 +29,12 @@ buster: &buster
       - export GOARCH=amd64
       - export FIPS=true
       - make cloudflared
+  cover: 
+    build_dir: *build_dir
+    builddeps: *build_deps
+    pre-cache: *build_pre_cache
+    post-cache: 
+     - make cover
   # except FIPS (handled in github-fips-release-pkgs) and macos (handled in github-release-macos-amd64)
   github-release-pkgs:
     build_dir: *build_dir

cmd/cloudflared/tunnel/cmd.go

@@ -399,7 +399,7 @@ func StartServer(
 		}
 	}
 
-	localRules := []ingress.Rule{}
+	internalRules := []ingress.Rule{}
 	if features.Contains(features.FeatureManagementLogs) {
 		serviceIP := c.String("service-op-ip")
 		if edgeAddrs, err := edgediscovery.ResolveEdge(log, tunnelConfig.Region, tunnelConfig.EdgeIPVersion); err == nil {
@@ -416,9 +416,9 @@ func StartServer(
 			logger.ManagementLogger.Log,
 			logger.ManagementLogger,
 		)
-		localRules = []ingress.Rule{ingress.NewManagementRule(mgmt)}
+		internalRules = []ingress.Rule{ingress.NewManagementRule(mgmt)}
 	}
-	orchestrator, err := orchestration.NewOrchestrator(ctx, orchestratorConfig, tunnelConfig.Tags, localRules, tunnelConfig.Log)
+	orchestrator, err := orchestration.NewOrchestrator(ctx, orchestratorConfig, tunnelConfig.Tags, internalRules, tunnelConfig.Log)
 	if err != nil {
 		return err
 	}

connection/quic.go

@@ -357,7 +357,7 @@ func (q *QUICConnection) serveUDPSession(session *datagramsession.Session, close
 // closeUDPSession first unregisters the session from session manager, then it tries to unregister from edge
 func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUID, message string) {
 	q.sessionManager.UnregisterSession(ctx, sessionID, message, false)
-	stream, err := q.session.OpenStream()
+	quicStream, err := q.session.OpenStream()
 	if err != nil {
 		// Log this at debug because this is not an error if session was closed due to lost connection
 		// with edge
@@ -367,6 +367,9 @@ func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUI
 			Msgf("Failed to open quic stream to unregister udp session with edge")
 		return
 	}
+
+	stream := quicpogs.NewSafeStreamCloser(quicStream)
+	defer stream.Close()
 	rpcClientStream, err := quicpogs.NewRPCClientStream(ctx, stream, q.logger)
 	if err != nil {
 		// Log this at debug because this is not an error if session was closed due to lost connection
@@ -375,6 +378,8 @@ func (q *QUICConnection) closeUDPSession(ctx context.Context, sessionID uuid.UUI
 			Msgf("Failed to open rpc stream to unregister udp session with edge")
 		return
 	}
+	defer rpcClientStream.Close()
+
 	if err := rpcClientStream.UnregisterUdpSession(ctx, sessionID, message); err != nil {
 		q.logger.Err(err).Str("sessionID", sessionID.String()).
 			Msgf("Failed to unregister udp session with edge")

ingress/ingress.go

@@ -44,7 +44,7 @@ func (ing Ingress) FindMatchingRule(hostname, path string) (*Rule, int) {
 	if err == nil {
 		hostname = host
 	}
-	for i, rule := range ing.LocalRules {
+	for i, rule := range ing.InternalRules {
 		if rule.Matches(hostname, path) {
 			// Local rule matches return a negative rule index to distiguish local rules from user-defined rules in logs
 			// Full range would be [-1 .. )
@@ -77,7 +77,7 @@ func matchHost(ruleHost, reqHost string) bool {
 // Ingress maps eyeball requests to origins.
 type Ingress struct {
 	// Set of ingress rules that are not added to remote config, e.g. management
-	LocalRules []Rule
+	InternalRules []Rule
 	// Rules that are provided by the user from remote or local configuration
 	Rules    []Rule              `json:"ingress"`
 	Defaults OriginRequestConfig `json:"originRequest"`
@@ -110,16 +110,18 @@ func ParseIngressFromConfigAndCLI(conf *config.Configuration, c *cli.Context, lo
 	//   --bastion for ssh bastion service
 	ingressRules, err = parseCLIIngress(c, false)
 	if errors.Is(err, ErrNoIngressRulesCLI) {
-		// Only log a warning if the tunnel is not a remotely managed tunnel and the config
-		// will be loaded after connecting.
+		// If no token is provided, the probability of NOT being a remotely managed tunnel is higher.
+		// So, we should warn the user that no ingress rules were found, because remote configuration will most likely not exist.
 		if !c.IsSet("token") {
 			log.Warn().Msgf(ErrNoIngressRulesCLI.Error())
 		}
 		return newDefaultOrigin(c, log), nil
 	}
+
 	if err != nil {
 		return Ingress{}, err
 	}
+
 	return ingressRules, nil
 }
 
@@ -148,14 +150,10 @@ func parseCLIIngress(c *cli.Context, allowURLFromArgs bool) (Ingress, error) {
 // 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)
+	defaultRule := GetDefaultIngressRules(log)
 	defaults := originRequestFromSingleRule(c)
 	ingress := Ingress{
-		Rules: []Rule{
-			{
-				Service: &noRulesService,
-			},
-		},
+		Rules:    defaultRule,
 		Defaults: defaults,
 	}
 	return ingress
@@ -219,6 +217,17 @@ func (ing Ingress) CatchAll() *Rule {
 	return &ing.Rules[len(ing.Rules)-1]
 }
 
+// Gets the default ingress rule that will be return 503 status
+// code for all incoming requests.
+func GetDefaultIngressRules(log *zerolog.Logger) []Rule {
+	noRulesService := newDefaultStatusCode(log)
+	return []Rule{
+		{
+			Service: &noRulesService,
+		},
+	}
+}
+
 func validateAccessConfiguration(cfg *config.AccessConfig) error {
 	if !cfg.Required {
 		return nil

orchestration/orchestrator.go

@@ -28,8 +28,8 @@ type Orchestrator struct {
 	lock sync.RWMutex
 	// Underlying value is proxy.Proxy, can be read without the lock, but still needs the lock to update
 	proxy atomic.Value
-	// Set of local ingress rules defined at cloudflared startup (separate from user-defined ingress rules)
-	localRules         []ingress.Rule
+	// Set of internal ingress rules defined at cloudflared startup (separate from user-defined ingress rules)
+	internalRules      []ingress.Rule
 	warpRoutingEnabled atomic.Bool
 	config             *Config
 	tags               []tunnelpogs.Tag
@@ -41,13 +41,13 @@ type Orchestrator struct {
 	proxyShutdownC chan<- struct{}
 }
 
-func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag, localRules []ingress.Rule, log *zerolog.Logger) (*Orchestrator, error) {
+func NewOrchestrator(ctx context.Context, config *Config, tags []tunnelpogs.Tag, internalRules []ingress.Rule, log *zerolog.Logger) (*Orchestrator, error) {
 	o := &Orchestrator{
 		// Lowest possible version, any remote configuration will have version higher than this
 		// Starting at -1 allows a configuration migration (local to remote) to override the current configuration as it
 		// will start at version 0.
 		currentVersion: -1,
-		localRules:     localRules,
+		internalRules:  internalRules,
 		config:         config,
 		tags:           tags,
 		log:            log,
@@ -116,8 +116,13 @@ func (o *Orchestrator) updateIngress(ingressRules ingress.Ingress, warpRouting i
 	default:
 	}
 
-	// Assign the local ingress rules to the parsed ingress
-	ingressRules.LocalRules = o.localRules
+	// Assign the internal ingress rules to the parsed ingress
+	ingressRules.InternalRules = o.internalRules
+
+	// Check if ingress rules are empty, and add the default route if so.
+	if ingressRules.IsEmpty() {
+		ingressRules.Rules = ingress.GetDefaultIngressRules(o.log)
+	}
 
 	// Start new proxy before closing the ones from last version.
 	// The upside is we don't need to restart proxy from last version, which can fail

orchestration/orchestrator_test.go

@@ -91,15 +91,15 @@ func TestUpdateConfiguration(t *testing.T) {
         "enabled": true,
         "connectTimeout": 10
     }
-}	
+}
 `)
 
 	updateWithValidation(t, orchestrator, 2, configJSONV2)
 	configV2 := orchestrator.config
-	// Validate local ingress rules
-	require.Equal(t, "management.argotunnel.com", configV2.Ingress.LocalRules[0].Hostname)
-	require.True(t, configV2.Ingress.LocalRules[0].Matches("management.argotunnel.com", "/ping"))
-	require.Equal(t, "management", configV2.Ingress.LocalRules[0].Service.String())
+	// Validate internal ingress rules
+	require.Equal(t, "management.argotunnel.com", configV2.Ingress.InternalRules[0].Hostname)
+	require.True(t, configV2.Ingress.InternalRules[0].Matches("management.argotunnel.com", "/ping"))
+	require.Equal(t, "management", configV2.Ingress.InternalRules[0].Service.String())
 	// Validate ingress rule 0
 	require.Equal(t, "jira.tunnel.org", configV2.Ingress.Rules[0].Hostname)
 	require.True(t, configV2.Ingress.Rules[0].Matches("jira.tunnel.org", "/login"))
@@ -145,7 +145,7 @@ func TestUpdateConfiguration(t *testing.T) {
 {
 	"originRequest":
 }
-	
+
 `)
 
 	resp = orchestrator.UpdateConfig(3, invalidJSON)
@@ -165,7 +165,7 @@ func TestUpdateConfiguration(t *testing.T) {
     "warp-routing": {
         "enabled": false
     }
-}	
+}
 `)
 	updateWithValidation(t, orchestrator, 10, configJSONV10)
 	configV10 := orchestrator.config
@@ -204,12 +204,34 @@ func TestUpdateConfiguration_FromMigration(t *testing.T) {
     "warp-routing": {
         "enabled": true
     }
-}	
+}
 `)
 	updateWithValidation(t, orchestrator, 0, configJSONV2)
 	require.Len(t, orchestrator.config.Ingress.Rules, 1)
 }
 
+// Validates that the default ingress rule will be set if there is no rule provided from the remote.
+func TestUpdateConfiguration_WithoutIngressRule(t *testing.T) {
+	initConfig := &Config{
+		Ingress: &ingress.Ingress{},
+	}
+	orchestrator, err := NewOrchestrator(context.Background(), initConfig, testTags, []ingress.Rule{}, &testLogger)
+	require.NoError(t, err)
+	initOriginProxy, err := orchestrator.GetOriginProxy()
+	require.NoError(t, err)
+	require.Implements(t, (*connection.OriginProxy)(nil), initOriginProxy)
+
+	// We need to create an empty RemoteConfigJSON because that will get unmarshalled to a RemoteConfig
+	emptyConfig := &ingress.RemoteConfigJSON{}
+	configBytes, err := json.Marshal(emptyConfig)
+	if err != nil {
+		require.FailNow(t, "The RemoteConfigJSON shouldn't fail while being marshalled")
+	}
+
+	updateWithValidation(t, orchestrator, 0, configBytes)
+	require.Len(t, orchestrator.config.Ingress.Rules, 1)
+}
+
 // TestConcurrentUpdateAndRead makes sure orchestrator can receive updates and return origin proxy concurrently
 func TestConcurrentUpdateAndRead(t *testing.T) {
 	const (