config: add support for TCP proxy chaining (#5053)

Add a distinction between TCP routes depending on whether the To URL(s)
have the scheme tcp://. For routes with a TCP upstream, configure Envoy
to terminate CONNECT requests and open a TCP tunnel to the upstream
service (this is the current behavior). For routes without a TCP
upstream, configure Envoy to proxy CONNECT requests to the upstream.

This new mode can allow an upstream proxy server to terminate a CONNECT
request and open its own TCP tunnel to the final destination server.
(Note that this will typically require setting the preserve_host_header
option as well.)

Note that this requires Envoy 1.30 or later.

config/envoyconfig/routes.go

@@ -415,11 +415,14 @@ func (b *Builder) buildPolicyRouteRouteAction(options *config.Options, policy *c
 	}
 
 	if policy.IsTCP() {
-		upgradeConfigs = append(upgradeConfigs, &envoy_config_route_v3.RouteAction_UpgradeConfig{
+		uc := &envoy_config_route_v3.RouteAction_UpgradeConfig{
-			UpgradeType:   "CONNECT",
+			UpgradeType: "CONNECT",
-			Enabled:       &wrapperspb.BoolValue{Value: true},
+			Enabled:     &wrapperspb.BoolValue{Value: true},
-			ConnectConfig: &envoy_config_route_v3.RouteAction_UpgradeConfig_ConnectConfig{},
+		}
-		})
+		if policy.IsTCPUpstream() {
+			uc.ConnectConfig = &envoy_config_route_v3.RouteAction_UpgradeConfig_ConnectConfig{}
+		}
+		upgradeConfigs = append(upgradeConfigs, uc)
 	}
 	action := &envoy_config_route_v3.RouteAction{
 		ClusterSpecifier: &envoy_config_route_v3.RouteAction_Cluster{

config/envoyconfig/routes_test.go

@@ -1042,7 +1042,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 			Policies: []config.Policy{
 				{
 					From:                "tcp+https://example.com:22",
-					To:                  mustParseWeightedURLs(t, "https://to.example.com"),
+					To:                  mustParseWeightedURLs(t, "tcp://to.example.com"),
 					PassIdentityHeaders: ptr(true),
 				},
 				{
@@ -1123,7 +1123,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 						"checkSettings": {
 							"contextExtensions": {
 								"internal": "false",
-								"route_id": "15730681265277585877"
+								"route_id": "10474912405080199536"
 							}
 						}
 					}
@@ -1166,7 +1166,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 					"upgradeConfigs": [
 						{ "enabled": false, "upgradeType": "websocket"},
 						{ "enabled": false, "upgradeType": "spdy/3.1"},
-						{ "enabled": true, "upgradeType": "CONNECT", "connectConfig": {} }
+						{ "enabled": true, "upgradeType": "CONNECT" }
 					]
 				},
 				"requestHeadersToRemove": [

config/policy.go

@@ -466,10 +466,17 @@ func (p *Policy) Validate() error {
 		return errEitherToOrRedirectOrResponseRequired
 	}
 
+	toSchemes := make(map[string]struct{})
 	for _, u := range p.To {
 		if err = u.Validate(); err != nil {
 			return fmt.Errorf("config: %s: %w", u.URL.String(), err)
 		}
+		toSchemes[u.URL.Scheme] = struct{}{}
+	}
+
+	// It is an error to mix TCP and non-TCP To URLs.
+	if _, hasTCP := toSchemes["tcp"]; hasTCP && len(toSchemes) > 1 {
+		return fmt.Errorf("config: cannot mix tcp and non-tcp To URLs")
 	}
 
 	// Only allow public access if no other whitelists are in place
@@ -648,6 +655,11 @@ func (p *Policy) IsTCP() bool {
 	return strings.HasPrefix(p.From, "tcp")
 }
 
+// IsTCPUpstream returns true if the route has a TCP upstream (To) URL
+func (p *Policy) IsTCPUpstream() bool {
+	return len(p.To) > 0 && p.To[0].URL.Scheme == "tcp"
+}
+
 // AllAllowedDomains returns all the allowed domains.
 func (p *Policy) AllAllowedDomains() []string {
 	var ads []string

config/policy_test.go

@@ -50,6 +50,8 @@ func Test_PolicyValidate(t *testing.T) {
 		{"bad kube service account token file", Policy{From: "https://httpbin.corp.example", To: mustParseWeightedURLs(t, "https://internal-host-name"), KubernetesServiceAccountTokenFile: "testdata/missing.token"}, true},
 		{"good kube service account token", Policy{From: "https://httpbin.corp.example", To: mustParseWeightedURLs(t, "https://internal-host-name"), KubernetesServiceAccountToken: "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIiLCJpYXQiOjE1OTY1MDk4MjIsImV4cCI6MTYyODA0NTgyMiwiYXVkIjoid3d3LmV4YW1wbGUuY29tIiwic3ViIjoianJvY2tldEBleGFtcGxlLmNvbSIsIkdpdmVuTmFtZSI6IkpvaG5ueSIsIlN1cm5hbWUiOiJSb2NrZXQiLCJFbWFpbCI6Impyb2NrZXRAZXhhbXBsZS5jb20iLCJSb2xlIjpbIk1hbmFnZXIiLCJQcm9qZWN0IEFkbWluaXN0cmF0b3IiXX0.H0I6ccQrL6sKobsKQj9dqNcLw_INhU9_xJsVyCkgkiY"}, false},
 		{"bad kube service account token and file", Policy{From: "https://httpbin.corp.example", To: mustParseWeightedURLs(t, "https://internal-host-name"), KubernetesServiceAccountToken: "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJPbmxpbmUgSldUIEJ1aWxkZXIiLCJpYXQiOjE1OTY1MDk4MjIsImV4cCI6MTYyODA0NTgyMiwiYXVkIjoid3d3LmV4YW1wbGUuY29tIiwic3ViIjoianJvY2tldEBleGFtcGxlLmNvbSIsIkdpdmVuTmFtZSI6IkpvaG5ueSIsIlN1cm5hbWUiOiJSb2NrZXQiLCJFbWFpbCI6Impyb2NrZXRAZXhhbXBsZS5jb20iLCJSb2xlIjpbIk1hbmFnZXIiLCJQcm9qZWN0IEFkbWluaXN0cmF0b3IiXX0.H0I6ccQrL6sKobsKQj9dqNcLw_INhU9_xJsVyCkgkiY", KubernetesServiceAccountTokenFile: "testdata/kubeserviceaccount.token"}, true},
+		{"TCP To URLs", Policy{From: "tcp+https://httpbin.corp.example:4000", To: mustParseWeightedURLs(t, "tcp://one.example.com:5000", "tcp://two.example.com:5000")}, false},
+		{"mix of TCP and non-TCP To URLs", Policy{From: "tcp+https://httpbin.corp.example:4000", To: mustParseWeightedURLs(t, "https://example.com", "tcp://example.com:5000")}, true},
 	}
 
 	for _, tt := range tests {
@@ -340,3 +342,30 @@ func TestPolicy_SortOrder(t *testing.T) {
 		})
 	}
 }
+
+func TestPolicy_IsTCP(t *testing.T) {
+	p1 := Policy{From: "https://example.com"}
+	assert.False(t, p1.IsTCP())
+
+	p2 := Policy{From: "tcp+https://example.com"}
+	assert.True(t, p2.IsTCP())
+}
+
+func TestPolicy_IsTCPUpstream(t *testing.T) {
+	p1 := Policy{
+		From: "tcp+https://example.com:1234",
+		To:   mustParseWeightedURLs(t, "https://one.example.com", "https://two.example.com"),
+	}
+	assert.False(t, p1.IsTCPUpstream())
+
+	p2 := Policy{
+		From: "tcp+https://example.com:1234",
+		To:   mustParseWeightedURLs(t, "tcp://one.example.com:4000", "tcp://two.example.com:4000"),
+	}
+	assert.True(t, p2.IsTCPUpstream())
+
+	p3 := Policy{
+		From: "tcp+https://example.com:1234",
+	}
+	assert.False(t, p3.IsTCPUpstream())
+}