UDP support (#5390)

config/envoyconfig/routes.go

@@ -423,6 +423,16 @@ func (b *Builder) buildPolicyRouteRouteAction(options *config.Options, policy *c
 		}
 		upgradeConfigs = append(upgradeConfigs, uc)
 	}
+	if policy.IsUDP() {
+		uc := &envoy_config_route_v3.RouteAction_UpgradeConfig{
+			UpgradeType: "CONNECT-UDP",
+			Enabled:     &wrapperspb.BoolValue{Value: true},
+		}
+		if policy.IsUDPUpstream() {
+			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{
 			Cluster: clusterName,
@@ -488,7 +498,7 @@ func toEnvoyHeaders(headers map[string]string) []*envoy_config_core_v3.HeaderVal
 func mkRouteMatch(policy *config.Policy) *envoy_config_route_v3.RouteMatch {
 	match := &envoy_config_route_v3.RouteMatch{}
 	switch {
-	case policy.IsTCP():
+	case policy.IsTCP(), policy.IsUDP():
 		match.PathSpecifier = &envoy_config_route_v3.RouteMatch_ConnectMatcher_{
 			ConnectMatcher: &envoy_config_route_v3.RouteMatch_ConnectMatcher{},
 		}
@@ -573,6 +583,7 @@ func getRouteIdleTimeout(policy *config.Policy) *durationpb.Duration {
 func shouldDisableStreamIdleTimeout(policy *config.Policy) bool {
 	return policy.AllowWebsockets ||
 		policy.IsTCP() ||
+		policy.IsUDP() ||
 		policy.IsForKubernetes() // disable for kubernetes so that tailing logs works (#2182)
 }
 

config/envoyconfig/routes_test.go

@@ -1229,6 +1229,99 @@ func Test_buildPolicyRoutes(t *testing.T) {
 	`, routes)
 	})
 
+	t.Run("udp", func(t *testing.T) {
+		routes, err := b.buildRoutesForPoliciesWithHost(&config.Config{Options: &config.Options{
+			CookieName:             "pomerium",
+			DefaultUpstreamTimeout: time.Second * 3,
+			SharedKey:              cryptutil.NewBase64Key(),
+			Policies: []config.Policy{
+				{
+					From:                "udp+https://example.com:22",
+					To:                  mustParseWeightedURLs(t, "udp://to.example.com"),
+					PassIdentityHeaders: ptr(true),
+				},
+			},
+		}}, "example.com:22")
+		require.NoError(t, err)
+
+		testutil.AssertProtoJSONEqual(t, `
+		[
+			{
+				"name": "policy-0",
+				"match": {
+					"connectMatcher": {}
+				},
+				"metadata": {
+					"filterMetadata": {
+						"envoy.filters.http.lua": {
+							"remove_impersonate_headers": false,
+							"remove_pomerium_authorization": true,
+							"remove_pomerium_cookie": "pomerium",
+							"rewrite_response_headers": []
+						}
+					}
+				},
+				"route": {
+					"autoHostRewrite": true,
+					"cluster": "policy-12",
+					"hashPolicy": [
+						{
+							"header": {
+								"headerName": "x-pomerium-routing-key"
+							},
+							"terminal": true
+						},
+						{
+							"connectionProperties": {
+								"sourceIp": true
+							},
+							"terminal": true
+						}
+					],
+					"idleTimeout": "0s",
+					"timeout": "0s",
+					"upgradeConfigs": [
+						{ "enabled": false, "upgradeType": "websocket"},
+						{ "enabled": false, "upgradeType": "spdy/3.1"},
+						{ "enabled": true, "upgradeType": "CONNECT-UDP", "connectConfig": {} }
+					]
+				},
+				"requestHeadersToRemove": [
+					"x-pomerium-reproxy-policy",
+					"x-pomerium-reproxy-policy-hmac"
+				],
+				"responseHeadersToAdd": [
+					{
+						"appendAction": "OVERWRITE_IF_EXISTS_OR_ADD",
+						"header": {
+						  "key": "X-Frame-Options",
+						  "value": "SAMEORIGIN"
+						}
+					},
+					{
+						"appendAction": "OVERWRITE_IF_EXISTS_OR_ADD",
+						"header": {
+						  "key": "X-XSS-Protection",
+						  "value": "1; mode=block"
+						}
+					}
+				],
+				"typedPerFilterConfig": {
+					"envoy.filters.http.ext_authz": {
+						"@type": "type.googleapis.com/envoy.extensions.filters.http.ext_authz.v3.ExtAuthzPerRoute",
+						"checkSettings": {
+							"contextExtensions": {
+								"internal": "false",
+								"route_id": "8231718688890004616"
+							}
+						}
+					}
+				}
+			}
+		]
+	`, routes)
+	})
+
 	t.Run("remove-pomerium-headers", func(t *testing.T) {
 		routes, err := b.buildRoutesForPoliciesWithHost(&config.Config{Options: &config.Options{
 			AuthenticateURLString:  "https://authenticate.example.com",
@@ -1267,7 +1360,7 @@ func Test_buildPolicyRoutes(t *testing.T) {
 					},
 					"route": {
 						"autoHostRewrite": true,
-						"cluster": "policy-12",
+						"cluster": "policy-13",
 						"hashPolicy": [
 							{
 								"header": {

config/policy.go

@@ -562,6 +562,9 @@ func (p *Policy) Validate() error {
 	if _, hasTCP := toSchemes["tcp"]; hasTCP && len(toSchemes) > 1 {
 		return fmt.Errorf("config: cannot mix tcp and non-tcp To URLs")
 	}
+	if _, hasUDP := toSchemes["udp"]; hasUDP && len(toSchemes) > 1 {
+		return fmt.Errorf("config: cannot mix udp and non-udp To URLs")
+	}
 
 	if err := p.Redirect.validate(); err != nil {
 		return fmt.Errorf("config: %w", err)
@@ -782,6 +785,16 @@ func (p *Policy) IsTCPUpstream() bool {
 	return len(p.To) > 0 && p.To[0].URL.Scheme == "tcp"
 }
 
+// IsUDP returns true if the route is for UDP.
+func (p *Policy) IsUDP() bool {
+	return strings.HasPrefix(p.From, "udp")
+}
+
+// IsUDPUpstream returns true if the route has a UDP upstream (To) URL
+func (p *Policy) IsUDPUpstream() bool {
+	return len(p.To) > 0 && p.To[0].URL.Scheme == "udp"
+}
+
 // AllAllowedDomains returns all the allowed domains.
 func (p *Policy) AllAllowedDomains() []string {
 	var ads []string

config/policy_test.go

@@ -55,6 +55,7 @@ func Test_PolicyValidate(t *testing.T) {
 		{"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},
+		{"UDP To URLs", Policy{From: "udp+https://httpbin.corp.example:4000", To: mustParseWeightedURLs(t, "udp://one.example.com:5000", "udp://two.example.com:5000")}, false},
 	}
 
 	for _, tt := range tests {

internal/urlutil/url.go

@@ -114,10 +114,12 @@ func GetDomainsForURL(u *url.URL, includeDefaultPort bool) []string {
 	}
 
 	// tcp+https://ssh.example.com:22
+	// udp+https://ssh.example.com:22
 	// => ssh.example.com:22
 	// tcp+https://proxy.example.com/ssh.example.com:22
+	// udp+https://proxy.example.com/ssh.example.com:22
 	// => ssh.example.com:22
-	if strings.HasPrefix(u.Scheme, "tcp+") {
+	if strings.HasPrefix(u.Scheme, "tcp+") || strings.HasPrefix(u.Scheme, "udp+") {
 		hosts := strings.Split(u.Path, "/")[1:]
 		// if there are no domains in the path part of the URL, use the host
 		if len(hosts) == 0 {