mesh: add more validations to Destinations resource (#19202)

command/resource/testdata/nested_data.hcl

@@ -21,6 +21,7 @@ Data {
       DestinationPort = "tcp"
 
       IpPort = {
+        Ip   = "127.0.0.1"
         Port = 1234
       }
     }

internal/catalog/exports.go

@@ -110,3 +110,11 @@ func ValidateLocalServiceRefNoSection(ref *pbresource.Reference, wrapErr func(er
 func ValidateSelector(sel *pbcatalog.WorkloadSelector, allowEmpty bool) error {
 	return types.ValidateSelector(sel, allowEmpty)
 }
+
+func ValidatePortName(name string) error {
+	return types.ValidatePortName(name)
+}
+
+func IsValidUnixSocketPath(host string) bool {
+	return types.IsValidUnixSocketPath(host)
+}

internal/catalog/internal/types/failover_policy.go

@@ -145,7 +145,7 @@ func ValidateFailoverPolicy(res *pbresource.Resource) error {
 				Wrapped: err,
 			}
 		}
-		if portNameErr := validatePortName(portName); portNameErr != nil {
+		if portNameErr := ValidatePortName(portName); portNameErr != nil {
 			merr = multierror.Append(merr, resource.ErrInvalidMapKey{
 				Map:     "port_configs",
 				Key:     portName,
@@ -245,7 +245,7 @@ func validateFailoverPolicyDestination(dest *pbcatalog.FailoverDestination, port
 	// assumed and will be reconciled.
 	if dest.Port != "" {
 		if ported {
-			if portNameErr := validatePortName(dest.Port); portNameErr != nil {
+			if portNameErr := ValidatePortName(dest.Port); portNameErr != nil {
 				merr = multierror.Append(merr, wrapErr(resource.ErrInvalidField{
 					Name:    "port",
 					Wrapped: portNameErr,

internal/catalog/internal/types/service.go

@@ -89,7 +89,7 @@ func ValidateService(res *pbresource.Resource) error {
 		}
 
 		// validate the target port
-		if nameErr := validatePortName(port.TargetPort); nameErr != nil {
+		if nameErr := ValidatePortName(port.TargetPort); nameErr != nil {
 			err = multierror.Append(err, resource.ErrInvalidListElement{
 				Name:  "ports",
 				Index: idx,

internal/catalog/internal/types/service_endpoints.go

@@ -126,7 +126,7 @@ func validateEndpoint(endpoint *pbcatalog.Endpoint, res *pbresource.Resource) er
 	// Validate the endpoints ports
 	for portName, port := range endpoint.Ports {
 		// Port names must be DNS labels
-		if portNameErr := validatePortName(portName); portNameErr != nil {
+		if portNameErr := ValidatePortName(portName); portNameErr != nil {
 			err = multierror.Append(err, resource.ErrInvalidMapKey{
 				Map:     "ports",
 				Key:     portName,

internal/catalog/internal/types/validators.go

@@ -56,7 +56,7 @@ func isValidDNSLabel(label string) bool {
 	return dnsLabelMatcher.Match([]byte(label))
 }
 
-func isValidUnixSocketPath(host string) bool {
+func IsValidUnixSocketPath(host string) bool {
 	if len(host) > maxUnixSocketPathLen || !strings.HasPrefix(host, "unix://") || strings.Contains(host, "\000") {
 		return false
 	}
@@ -71,7 +71,7 @@ func validateWorkloadHost(host string) error {
 	}
 
 	// Check if the host represents an IP address, unix socket path or a DNS name
-	if !isValidIPAddress(host) && !isValidUnixSocketPath(host) && !isValidDNSName(host) {
+	if !isValidIPAddress(host) && !IsValidUnixSocketPath(host) && !isValidDNSName(host) {
 		return errInvalidWorkloadHostFormat{Host: host}
 	}
 
@@ -139,7 +139,7 @@ func validateIPAddress(ip string) error {
 	return nil
 }
 
-func validatePortName(name string) error {
+func ValidatePortName(name string) error {
 	if name == "" {
 		return resource.ErrEmpty
 	}
@@ -184,7 +184,7 @@ func validateWorkloadAddress(addr *pbcatalog.WorkloadAddress, ports map[string]*
 
 	// Ensure that unix sockets reference exactly 1 port. They may also indirectly reference 1 port
 	// by the workload having only a single port and omitting any explicit port assignment.
-	if isValidUnixSocketPath(addr.Host) &&
+	if IsValidUnixSocketPath(addr.Host) &&
 		(len(addr.Ports) > 1 || (len(addr.Ports) == 0 && len(ports) > 1)) {
 		err = multierror.Append(err, errUnixSocketMultiport)
 	}

internal/catalog/internal/types/validators_test.go

@@ -178,7 +178,7 @@ func TestIsValidUnixSocketPath(t *testing.T) {
 
 	for name, tcase := range cases {
 		t.Run(name, func(t *testing.T) {
-			require.Equal(t, tcase.valid, isValidUnixSocketPath(tcase.name))
+			require.Equal(t, tcase.valid, IsValidUnixSocketPath(tcase.name))
 		})
 	}
 }
@@ -361,15 +361,15 @@ func TestValidatePortName(t *testing.T) {
 	// test for the isValidDNSLabel function.
 
 	t.Run("empty", func(t *testing.T) {
-		require.Equal(t, resource.ErrEmpty, validatePortName(""))
+		require.Equal(t, resource.ErrEmpty, ValidatePortName(""))
 	})
 
 	t.Run("invalid", func(t *testing.T) {
-		require.Equal(t, errNotDNSLabel, validatePortName("foo.com"))
+		require.Equal(t, errNotDNSLabel, ValidatePortName("foo.com"))
 	})
 
 	t.Run("ok", func(t *testing.T) {
-		require.NoError(t, validatePortName("http"))
+		require.NoError(t, ValidatePortName("http"))
 	})
 }
 

internal/catalog/internal/types/workload.go

@@ -44,7 +44,7 @@ func ValidateWorkload(res *pbresource.Resource) error {
 
 	// Validate the Workload Ports
 	for portName, port := range workload.Ports {
-		if portNameErr := validatePortName(portName); portNameErr != nil {
+		if portNameErr := ValidatePortName(portName); portNameErr != nil {
 			err = multierror.Append(err, resource.ErrInvalidMapKey{
 				Map:     "ports",
 				Key:     portName,

internal/mesh/internal/controllers/sidecarproxy/controller_test.go

@@ -553,7 +553,7 @@ func (suite *controllerTestSuite) TestController() {
 		}).Write(suite.T(), suite.client)
 
 	testutil.RunStep(suite.T(), "add explicit destinations and check that new proxy state is generated", func(t *testing.T) {
-		webProxyStateTemplate = suite.client.WaitForNewVersion(t, webProxyStateTemplateID, webProxyStateTemplate.Version)
+		webProxyStateTemplate = suite.client.WaitForNewVersion(suite.T(), webProxyStateTemplateID, webProxyStateTemplate.Version)
 
 		requireExplicitDestinationsFound(t, "api", webProxyStateTemplate)
 	})
@@ -613,7 +613,7 @@ func (suite *controllerTestSuite) TestController() {
 		})
 
 		// We should get a new web proxy template resource because this destination should be removed.
-		webProxyStateTemplate = suite.client.WaitForNewVersion(t, webProxyStateTemplateID, webProxyStateTemplate.Version)
+		webProxyStateTemplate = suite.client.WaitForNewVersion(suite.T(), webProxyStateTemplateID, webProxyStateTemplate.Version)
 
 		requireExplicitDestinationsNotFound(t, "api", webProxyStateTemplate)
 	})

internal/mesh/internal/types/destinations.go

@@ -4,6 +4,8 @@
 package types
 
 import (
+	"net"
+
 	"github.com/hashicorp/go-multierror"
 	"google.golang.org/protobuf/proto"
 
@@ -73,7 +75,6 @@ func ValidateDestinations(res *pbresource.Resource) error {
 
 	var merr error
 
-	// Validate the workload selector
 	if selErr := catalog.ValidateSelector(destinations.Workloads, false); selErr != nil {
 		merr = multierror.Append(merr, resource.ErrInvalidField{
 			Name:    "workloads",
@@ -81,10 +82,17 @@ func ValidateDestinations(res *pbresource.Resource) error {
 		})
 	}
 
+	if destinations.GetPqDestinations() != nil {
+		merr = multierror.Append(merr, resource.ErrInvalidField{
+			Name:    "pq_destinations",
+			Wrapped: resource.ErrUnsupported,
+		})
+	}
+
 	for i, dest := range destinations.Destinations {
 		wrapDestErr := func(err error) error {
 			return resource.ErrInvalidListElement{
-				Name:    "upstreams",
+				Name:    "destinations",
 				Index:   i,
 				Wrapped: err,
 			}
@@ -101,8 +109,73 @@ func ValidateDestinations(res *pbresource.Resource) error {
 			merr = multierror.Append(merr, refErr)
 		}
 
-		// TODO(v2): validate port name using catalog validator
+		if portErr := catalog.ValidatePortName(dest.DestinationPort); portErr != nil {
-		// TODO(v2): validate ListenAddr
+			merr = multierror.Append(merr, wrapDestErr(resource.ErrInvalidField{
+				Name:    "destination_port",
+				Wrapped: portErr,
+			}))
+		}
+
+		if dest.GetDatacenter() != "" {
+			merr = multierror.Append(merr, wrapDestErr(resource.ErrInvalidField{
+				Name:    "datacenter",
+				Wrapped: resource.ErrUnsupported,
+			}))
+		}
+
+		if listenAddrErr := validateListenAddr(dest); listenAddrErr != nil {
+			merr = multierror.Append(merr, wrapDestErr(listenAddrErr))
+		}
+	}
+
+	return merr
+}
+
+func validateListenAddr(dest *pbmesh.Destination) error {
+	var merr error
+
+	if dest.GetListenAddr() == nil {
+		return multierror.Append(merr, resource.ErrInvalidFields{
+			Names:   []string{"ip_port", "unix"},
+			Wrapped: resource.ErrMissingOneOf,
+		})
+	}
+
+	switch listenAddr := dest.GetListenAddr().(type) {
+	case *pbmesh.Destination_IpPort:
+		if ipPortErr := validateIPPort(listenAddr.IpPort); ipPortErr != nil {
+			merr = multierror.Append(merr, resource.ErrInvalidField{
+				Name:    "ip_port",
+				Wrapped: ipPortErr,
+			})
+		}
+	case *pbmesh.Destination_Unix:
+		if !catalog.IsValidUnixSocketPath(listenAddr.Unix.GetPath()) {
+			merr = multierror.Append(merr, resource.ErrInvalidField{
+				Name: "unix",
+				Wrapped: resource.ErrInvalidField{
+					Name:    "path",
+					Wrapped: errInvalidUnixSocketPath,
+				},
+			})
+		}
+	}
+
+	return merr
+}
+
+func validateIPPort(ipPort *pbmesh.IPPortAddress) error {
+	var merr error
+
+	if listenPortErr := validatePort(ipPort.GetPort(), "port"); listenPortErr != nil {
+		merr = multierror.Append(merr, listenPortErr)
+	}
+
+	if net.ParseIP(ipPort.GetIp()) == nil {
+		merr = multierror.Append(merr, resource.ErrInvalidField{
+			Name:    "ip",
+			Wrapped: errInvalidIP,
+		})
 	}
 
 	return merr

internal/mesh/internal/types/destinations_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
-func TestMutateUpstreams(t *testing.T) {
+func TestMutateDestinations(t *testing.T) {
 	type testcase struct {
 		tenancy   *pbresource.Tenancy
 		data      *pbmesh.Destinations
@@ -86,7 +86,7 @@ func TestMutateUpstreams(t *testing.T) {
 	}
 }
 
-func TestValidateUpstreams(t *testing.T) {
+func TestValidateDestinations(t *testing.T) {
 	type testcase struct {
 		data       *pbmesh.Destinations
 		skipMutate bool
@@ -151,7 +151,7 @@ func TestValidateUpstreams(t *testing.T) {
 					{DestinationRef: nil},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "upstreams": invalid "destination_ref" field: missing required field`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "destination_ref" field: missing required field`,
 		},
 		"dest/bad type": {
 			skipMutate: true,
@@ -163,7 +163,7 @@ func TestValidateUpstreams(t *testing.T) {
 					{DestinationRef: newRefWithTenancy(pbcatalog.WorkloadType, "default.default", "api")},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "upstreams": invalid "destination_ref" field: invalid "type" field: reference must have type catalog.v2beta1.Service`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "destination_ref" field: invalid "type" field: reference must have type catalog.v2beta1.Service`,
 		},
 		"dest/nil tenancy": {
 			skipMutate: true,
@@ -175,7 +175,7 @@ func TestValidateUpstreams(t *testing.T) {
 					{DestinationRef: &pbresource.Reference{Type: pbcatalog.ServiceType, Name: "api"}},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "upstreams": invalid "destination_ref" field: invalid "tenancy" field: missing required field`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "destination_ref" field: invalid "tenancy" field: missing required field`,
 		},
 		"dest/bad dest tenancy/partition": {
 			skipMutate: true,
@@ -187,7 +187,7 @@ func TestValidateUpstreams(t *testing.T) {
 					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, ".bar", "api")},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "upstreams": invalid "destination_ref" field: invalid "tenancy" field: invalid "partition" field: cannot be empty`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "destination_ref" field: invalid "tenancy" field: invalid "partition" field: cannot be empty`,
 		},
 		"dest/bad dest tenancy/namespace": {
 			skipMutate: true,
@@ -199,7 +199,7 @@ func TestValidateUpstreams(t *testing.T) {
 					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "foo", "api")},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "upstreams": invalid "destination_ref" field: invalid "tenancy" field: invalid "namespace" field: cannot be empty`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "destination_ref" field: invalid "tenancy" field: invalid "namespace" field: cannot be empty`,
 		},
 		"dest/bad dest tenancy/peer_name": {
 			skipMutate: true,
@@ -213,17 +213,158 @@ func TestValidateUpstreams(t *testing.T) {
 						Reference("")},
 				},
 			},
-			expectErr: `invalid element at index 0 of list "upstreams": invalid "destination_ref" field: invalid "tenancy" field: invalid "peer_name" field: must be set to "local"`,
+			expectErr: `invalid element at index 0 of list "destinations": invalid "destination_ref" field: invalid "tenancy" field: invalid "peer_name" field: must be set to "local"`,
+		},
+		"unsupported pq_destinations": {
+			skipMutate: true,
+			data: &pbmesh.Destinations{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
+				PqDestinations: []*pbmesh.PreparedQueryDestination{
+					{Name: "foo-query"},
+				},
+			},
+			expectErr: `invalid "pq_destinations" field: field is currently not supported`,
+		},
+		"missing destination port": {
+			skipMutate: true,
+			data: &pbmesh.Destinations{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
+				Destinations: []*pbmesh.Destination{
+					{
+						DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "127.0.0.1",
+								Port: 1234,
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid element at index 0 of list "destinations": invalid "destination_port" field: cannot be empty`,
+		},
+		"unsupported datacenter": {
+			skipMutate: true,
+			data: &pbmesh.Destinations{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
+				Destinations: []*pbmesh.Destination{
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
+						DestinationPort: "p1",
+						Datacenter:      "dc2",
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "127.0.0.1",
+								Port: 1234,
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid element at index 0 of list "destinations": invalid "datacenter" field: field is currently not supported`,
+		},
+		"missing listen addr": {
+			skipMutate: true,
+			data: &pbmesh.Destinations{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
+				Destinations: []*pbmesh.Destination{
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
+						DestinationPort: "p1",
+					},
+				},
+			},
+			expectErr: `invalid "ip_port,unix" fields: missing one of the required fields`,
+		},
+		"invalid ip for listen addr": {
+			skipMutate: true,
+			data: &pbmesh.Destinations{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
+				Destinations: []*pbmesh.Destination{
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
+						DestinationPort: "p1",
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "invalid",
+								Port: 1234,
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid "ip" field: IP address is not valid`,
+		},
+		"invalid port for listen addr": {
+			skipMutate: true,
+			data: &pbmesh.Destinations{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
+				Destinations: []*pbmesh.Destination{
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
+						DestinationPort: "p1",
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "127.0.0.1",
+								Port: 0,
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid "port" field: port number is outside the range 1 to 65535`,
+		},
+		"invalid unix path for listen addr": {
+			skipMutate: true,
+			data: &pbmesh.Destinations{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
+				Destinations: []*pbmesh.Destination{
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
+						DestinationPort: "p1",
+						ListenAddr: &pbmesh.Destination_Unix{
+							Unix: &pbmesh.UnixSocketAddress{
+								Path: "foo",
+							},
+						},
+					},
+				},
+			},
+			expectErr: `invalid "unix" field: invalid "path" field: unix socket path is not valid`,
 		},
 		"normal": {
 			data: &pbmesh.Destinations{
-				Workloads: &pbcatalog.WorkloadSelector{
+				Workloads: &pbcatalog.WorkloadSelector{Names: []string{"foo"}},
-					Names: []string{"blah"},
-				},
 				Destinations: []*pbmesh.Destination{
-					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api")},
+					{
-					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "foo.zim", "api")},
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
-					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "gir.zim", "api")},
+						DestinationPort: "p1",
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "127.0.0.1",
+								Port: 1234,
+							},
+						},
+					},
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.zim", "api"),
+						DestinationPort: "p2",
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "127.0.0.1",
+								Port: 1235,
+							},
+						},
+					},
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "gir.zim", "api"),
+						DestinationPort: "p3",
+						ListenAddr: &pbmesh.Destination_Unix{
+							Unix: &pbmesh.UnixSocketAddress{
+								Path: "unix://foo/bar",
+							},
+						},
+					},
 				},
 			},
 		},
@@ -234,9 +375,35 @@ func TestValidateUpstreams(t *testing.T) {
 					Filter: "metadata.foo == bar",
 				},
 				Destinations: []*pbmesh.Destination{
-					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api")},
+					{
-					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "foo.zim", "api")},
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.bar", "api"),
-					{DestinationRef: newRefWithTenancy(pbcatalog.ServiceType, "gir.zim", "api")},
+						DestinationPort: "p1",
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "127.0.0.1",
+								Port: 1234,
+							},
+						},
+					},
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "foo.zim", "api"),
+						DestinationPort: "p2",
+						ListenAddr: &pbmesh.Destination_IpPort{
+							IpPort: &pbmesh.IPPortAddress{
+								Ip:   "127.0.0.1",
+								Port: 1235,
+							},
+						},
+					},
+					{
+						DestinationRef:  newRefWithTenancy(pbcatalog.ServiceType, "gir.zim", "api"),
+						DestinationPort: "p3",
+						ListenAddr: &pbmesh.Destination_Unix{
+							Unix: &pbmesh.UnixSocketAddress{
+								Path: "unix://foo/bar",
+							},
+						},
+					},
 				},
 			},
 		},

internal/mesh/internal/types/errors.go

@@ -9,6 +9,8 @@ import (
 
 var (
 	errInvalidPort               = errors.New("port number is outside the range 1 to 65535")
+	errInvalidIP                 = errors.New("IP address is not valid")
+	errInvalidUnixSocketPath     = errors.New("unix socket path is not valid")
 	errInvalidExposePathProtocol = errors.New("invalid protocol: only HTTP and HTTP2 protocols are allowed")
 	errMissingProxyConfigData    = errors.New("at least one of \"bootstrap_config\" or \"dynamic_config\" fields must be set")
 )

internal/resource/errors.go

@@ -14,6 +14,7 @@ import (
 
 var (
 	ErrMissing                  = NewConstError("missing required field")
+	ErrMissingOneOf             = NewConstError("missing one of the required fields")
 	ErrEmpty                    = NewConstError("cannot be empty")
 	ErrReferenceTenancyNotEqual = NewConstError("resource tenancy and reference tenancy differ")
 	ErrUnsupported              = NewConstError("field is currently not supported")