mesh: add ACL checks for DestinationPolicy resources (#18920)

DestinationPolicy resources are name-aligned with the Service they control.

The ACLs should be:

- list: (default)
- read: service:<resource_name>:read
- write: service:<resource_name>:write

internal/mesh/internal/types/destination_policy.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbmesh "github.com/hashicorp/consul/proto-public/pbmesh/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -34,6 +35,11 @@ func RegisterDestinationPolicy(r resource.Registry) {
 		Proto:    &pbmesh.DestinationPolicy{},
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateDestinationPolicy,
+		ACLs: &resource.ACLHooks{
+			Read:  aclReadHookDestinationPolicy,
+			Write: aclWriteHookDestinationPolicy,
+			List:  aclListHookDestinationPolicy,
+		},
 	})
 }
 
@@ -225,3 +231,25 @@ func ValidateDestinationPolicy(res *pbresource.Resource) error {
 
 	return merr
 }
+
+func aclReadHookDestinationPolicy(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) error {
+	// DestinationPolicy is name-aligned with Service
+	serviceName := id.Name
+
+	// Check service:read permissions.
+	return authorizer.ToAllowAuthorizer().ServiceReadAllowed(serviceName, authzContext)
+}
+
+func aclWriteHookDestinationPolicy(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
+	// DestinationPolicy is name-aligned with Service
+	serviceName := res.Id.Name
+
+	// Check service:write permissions on the service this is controlling.
+	return authorizer.ToAllowAuthorizer().ServiceWriteAllowed(serviceName, authzContext)
+}
+
+func aclListHookDestinationPolicy(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext) error {
+	// No-op List permission as we want to default to filtering resources
+	// from the list using the Read enforcement.
+	return nil
+}

internal/mesh/internal/types/destination_policy_test.go

@@ -10,8 +10,12 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/types/known/durationpb"
 
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbmesh "github.com/hashicorp/consul/proto-public/pbmesh/v2beta1"
+	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/proto/private/prototest"
 	"github.com/hashicorp/consul/sdk/testutil"
 )
@@ -508,3 +512,105 @@ func TestValidateDestinationPolicy(t *testing.T) {
 		})
 	}
 }
+
+func TestDestinationPolicyACLs(t *testing.T) {
+	// Wire up a registry to generically invoke hooks
+	registry := resource.NewRegistry()
+	Register(registry)
+
+	type testcase struct {
+		rules   string
+		check   func(t *testing.T, authz acl.Authorizer, res *pbresource.Resource)
+		readOK  string
+		writeOK string
+		listOK  string
+	}
+
+	const (
+		DENY    = "deny"
+		ALLOW   = "allow"
+		DEFAULT = "default"
+	)
+
+	checkF := func(t *testing.T, expect string, got error) {
+		switch expect {
+		case ALLOW:
+			if acl.IsErrPermissionDenied(got) {
+				t.Fatal("should be allowed")
+			}
+		case DENY:
+			if !acl.IsErrPermissionDenied(got) {
+				t.Fatal("should be denied")
+			}
+		case DEFAULT:
+			require.Nil(t, got, "expected fallthrough decision")
+		default:
+			t.Fatalf("unexpected expectation: %q", expect)
+		}
+	}
+
+	reg, ok := registry.Resolve(DestinationPolicyType)
+	require.True(t, ok)
+
+	run := func(t *testing.T, tc testcase) {
+		destData := &pbmesh.DestinationPolicy{
+			PortConfigs: map[string]*pbmesh.DestinationConfig{
+				"http": {
+					ConnectTimeout: durationpb.New(55 * time.Second),
+				},
+			},
+		}
+		res := resourcetest.Resource(DestinationPolicyType, "api").
+			WithTenancy(resource.DefaultNamespacedTenancy()).
+			WithData(t, destData).
+			Build()
+		resourcetest.ValidateAndNormalize(t, registry, res)
+
+		config := acl.Config{
+			WildcardName: structs.WildcardSpecifier,
+		}
+		authz, err := acl.NewAuthorizerFromRules(tc.rules, &config, nil)
+		require.NoError(t, err)
+		authz = acl.NewChainedAuthorizer([]acl.Authorizer{authz, acl.DenyAll()})
+
+		t.Run("read", func(t *testing.T) {
+			err := reg.ACLs.Read(authz, &acl.AuthorizerContext{}, res.Id, nil)
+			checkF(t, tc.readOK, err)
+		})
+		t.Run("write", func(t *testing.T) {
+			err := reg.ACLs.Write(authz, &acl.AuthorizerContext{}, res)
+			checkF(t, tc.writeOK, err)
+		})
+		t.Run("list", func(t *testing.T) {
+			err := reg.ACLs.List(authz, &acl.AuthorizerContext{})
+			checkF(t, tc.listOK, err)
+		})
+	}
+
+	cases := map[string]testcase{
+		"no rules": {
+			rules:   ``,
+			readOK:  DENY,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"service api read": {
+			rules:   `service "api" { policy = "read" }`,
+			readOK:  ALLOW,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"service api write": {
+			rules:   `service "api" { policy = "write" }`,
+			readOK:  ALLOW,
+			writeOK: ALLOW,
+			listOK:  DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}