catalog, mesh: implement missing ACL hooks (#19143)

This change adds ACL hooks to the remaining catalog and mesh resources, excluding any computed ones. Those will for now continue using the default operator:x permissions.

It refactors a lot of the common testing functions so that they can be re-used between resources.

There are also some types that we don't yet support (e.g. virtual IPs) that this change adds ACL hooks to for future-proofing.

agent/grpc-external/services/resource/read.go

@@ -50,7 +50,7 @@ func (s *Server) Read(ctx context.Context, req *pbresource.ReadRequest) (*pbreso
 	authzNeedsData := false
 	err = reg.ACLs.Read(authz, authzContext, req.Id, nil)
 	switch {
-	case errors.Is(err, resource.ErrNeedData):
+	case errors.Is(err, resource.ErrNeedResource):
 		authzNeedsData = true
 		err = nil
 	case acl.IsErrPermissionDenied(err):

internal/auth/internal/types/computed_traffic_permissions.go

@@ -19,7 +19,7 @@ func RegisterComputedTrafficPermission(r resource.Registry) {
 		ACLs: &resource.ACLHooks{
 			Read:  aclReadHookComputedTrafficPermissions,
 			Write: aclWriteHookComputedTrafficPermissions,
-			List:  aclListHookComputedTrafficPermissions,
+			List:  resource.NoOpACLListHook,
 		},
 		Validate: ValidateComputedTrafficPermissions,
 		Scope:    resource.ScopeNamespace,
@@ -71,9 +71,3 @@ func aclReadHookComputedTrafficPermissions(authorizer acl.Authorizer, authzConte
 func aclWriteHookComputedTrafficPermissions(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
 	return authorizer.ToAllowAuthorizer().TrafficPermissionsWriteAllowed(res.Id.Name, authzContext)
 }
-
-func aclListHookComputedTrafficPermissions(_ acl.Authorizer, _ *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/auth/internal/types/traffic_permissions.go

@@ -19,7 +19,7 @@ func RegisterTrafficPermissions(r resource.Registry) {
 		ACLs: &resource.ACLHooks{
 			Read:  aclReadHookTrafficPermissions,
 			Write: aclWriteHookTrafficPermissions,
-			List:  aclListHookTrafficPermissions,
+			List:  resource.NoOpACLListHook,
 		},
 		Validate: ValidateTrafficPermissions,
 		Mutate:   MutateTrafficPermissions,
@@ -273,7 +273,7 @@ func isLocalPeer(p string) bool {
 
 func aclReadHookTrafficPermissions(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, _ *pbresource.ID, res *pbresource.Resource) error {
 	if res == nil {
-		return resource.ErrNeedData
+		return resource.ErrNeedResource
 	}
 	return authorizeDestination(res, func(dest string) error {
 		return authorizer.ToAllowAuthorizer().TrafficPermissionsReadAllowed(dest, authzContext)
@@ -286,12 +286,6 @@ func aclWriteHookTrafficPermissions(authorizer acl.Authorizer, authzContext *acl
 	})
 }
 
-func aclListHookTrafficPermissions(_ acl.Authorizer, _ *acl.AuthorizerContext) error {
-	// No-op List permission as we want to default to filtering resources
-	// from the list using the Read enforcement
-	return nil
-}
-
 func authorizeDestination(res *pbresource.Resource, intentionAllowed func(string) error) error {
 	tp, err := resource.Decode[*pbauth.TrafficPermissions](res)
 	if err != nil {

internal/auth/internal/types/workload_identity.go

@@ -18,7 +18,7 @@ func RegisterWorkloadIdentity(r resource.Registry) {
 		ACLs: &resource.ACLHooks{
 			Read:  aclReadHookWorkloadIdentity,
 			Write: aclWriteHookWorkloadIdentity,
-			List:  aclListHookWorkloadIdentity,
+			List:  resource.NoOpACLListHook,
 		},
 		Validate: nil,
 	})
@@ -36,7 +36,7 @@ func aclReadHookWorkloadIdentity(
 	if res != nil {
 		return authorizer.ToAllowAuthorizer().IdentityReadAllowed(res.Id.Name, authzCtx)
 	}
-	return resource.ErrNeedData
+	return resource.ErrNeedResource
 }
 
 func aclWriteHookWorkloadIdentity(
@@ -44,13 +44,7 @@ func aclWriteHookWorkloadIdentity(
 	authzCtx *acl.AuthorizerContext,
 	res *pbresource.Resource) error {
 	if res == nil {
-		return resource.ErrNeedData
+		return resource.ErrNeedResource
 	}
 	return authorizer.ToAllowAuthorizer().IdentityWriteAllowed(res.Id.Name, authzCtx)
 }
-
-func aclListHookWorkloadIdentity(authorizer acl.Authorizer, context *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/auth/internal/types/workload_identity_test.go

@@ -82,8 +82,8 @@ func TestWorkloadIdentityACLs(t *testing.T) {
 			checkF(t, tc.listOK, err)
 		})
 		t.Run("errors", func(t *testing.T) {
-			require.ErrorIs(t, reg.ACLs.Read(authz, &acl.AuthorizerContext{}, nil, nil), resource.ErrNeedData)
-			require.ErrorIs(t, reg.ACLs.Write(authz, &acl.AuthorizerContext{}, nil), resource.ErrNeedData)
+			require.ErrorIs(t, reg.ACLs.Read(authz, &acl.AuthorizerContext{}, nil, nil), resource.ErrNeedResource)
+			require.ErrorIs(t, reg.ACLs.Write(authz, &acl.AuthorizerContext{}, nil), resource.ErrNeedResource)
 		})
 	}
 

internal/catalog/catalogtest/helpers/acl_hooks_test_helpers.go

@@ -0,0 +1,21 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package helpers
+
+import (
+	"testing"
+
+	"github.com/hashicorp/consul/internal/catalog"
+	"github.com/hashicorp/consul/internal/catalog/internal/testhelpers"
+	"github.com/hashicorp/consul/internal/resource"
+	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func RunWorkloadSelectingTypeACLsTests[T catalog.WorkloadSelecting](t *testing.T, typ *pbresource.Type,
+	getData func(selector *pbcatalog.WorkloadSelector) T,
+	registerFunc func(registry resource.Registry),
+) {
+	testhelpers.RunWorkloadSelectingTypeACLsTests[T](t, typ, getData, registerFunc)
+}

internal/catalog/exports.go

@@ -48,6 +48,12 @@ var (
 	FailoverStatusConditionAcceptedUsingMeshDestinationPortReason  = failover.UsingMeshDestinationPortReason
 )
 
+type WorkloadSelecting = types.WorkloadSelecting
+
+func ACLHooksForWorkloadSelectingType[T WorkloadSelecting]() *resource.ACLHooks {
+	return types.ACLHooksForWorkloadSelectingType[T]()
+}
+
 // RegisterTypes adds all resource types within the "catalog" API group
 // to the given type registry
 func RegisterTypes(r resource.Registry) {

internal/catalog/internal/controllers/nodehealth/controller_test.go

@@ -73,7 +73,7 @@ type nodeHealthControllerTestSuite struct {
 }
 
 func (suite *nodeHealthControllerTestSuite) SetupTest() {
-	suite.resourceClient = svctest.RunResourceService(suite.T(), types.Register)
+	suite.resourceClient = svctest.RunResourceService(suite.T(), types.Register, types.RegisterDNSPolicy)
 	suite.runtime = controller.Runtime{Client: suite.resourceClient, Logger: testutil.Logger(suite.T())}
 
 	// The rest of the setup will be to prime the resource service with some data

internal/catalog/internal/testhelpers/acl_hooks_test_helpers.go

@@ -0,0 +1,113 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package testhelpers
+
+import (
+	"testing"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/resource/resourcetest"
+	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+// WorkloadSelecting denotes a resource type that uses workload selectors.
+type WorkloadSelecting interface {
+	proto.Message
+	GetWorkloads() *pbcatalog.WorkloadSelector
+}
+
+func RunWorkloadSelectingTypeACLsTests[T WorkloadSelecting](t *testing.T, typ *pbresource.Type,
+	getData func(selector *pbcatalog.WorkloadSelector) T,
+	registerFunc func(registry resource.Registry),
+) {
+	// Wire up a registry to generically invoke hooks
+	registry := resource.NewRegistry()
+	registerFunc(registry)
+
+	cases := map[string]resourcetest.ACLTestCase{
+		"no rules": {
+			Rules:   ``,
+			Data:    getData(&pbcatalog.WorkloadSelector{Names: []string{"workload"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test read": {
+			Rules:   `service "test" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Names: []string{"workload"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with named selectors and insufficient policy": {
+			Rules:   `service "test" { policy = "write" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Names: []string{"workload"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with prefixed selectors and insufficient policy": {
+			Rules:   `service "test" { policy = "write" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with named selectors": {
+			Rules:   `service "test" { policy = "write" } service "workload" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Names: []string{"workload"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with prefixed selectors": {
+			Rules:   `service "test" { policy = "write" } service_prefix "workload-" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload-"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with prefixed selectors and a policy with more specific than the selector": {
+			Rules:   `service "test" { policy = "write" } service_prefix "workload-" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"wor"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with prefixed selectors and a policy with less specific than the selector": {
+			Rules:   `service "test" { policy = "write" } service_prefix "wor" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload-"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with prefixed selectors and a policy with a specific service": {
+			Rules:  `service "test" { policy = "write" } service "workload" { policy = "read" }`,
+			Data:   getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload"}}),
+			Typ:    typ,
+			ReadOK: resourcetest.ALLOW,
+			// TODO (ishustava): this is wrong and should be fixed in a follow up PR. We should not allow
+			// a policy for a specific service when only prefixes are specified in the selector.
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			resourcetest.RunACLTestCase(t, tc, registry)
+		})
+	}
+}

internal/catalog/internal/types/acl_hooks.go

@@ -0,0 +1,56 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package types
+
+import (
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func aclReadHookResourceWithWorkloadSelector(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) error {
+	return authorizer.ToAllowAuthorizer().ServiceReadAllowed(id.GetName(), authzContext)
+}
+
+func aclWriteHookResourceWithWorkloadSelector[T WorkloadSelecting](authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
+	if res == nil {
+		return resource.ErrNeedResource
+	}
+
+	decodedService, err := resource.Decode[T](res)
+	if err != nil {
+		return resource.ErrNeedResource
+	}
+
+	// First check service:write on the name.
+	err = authorizer.ToAllowAuthorizer().ServiceWriteAllowed(res.GetId().GetName(), authzContext)
+	if err != nil {
+		return err
+	}
+
+	// Then also check whether we're allowed to select a service.
+	for _, name := range decodedService.GetData().GetWorkloads().GetNames() {
+		err = authorizer.ToAllowAuthorizer().ServiceReadAllowed(name, authzContext)
+		if err != nil {
+			return err
+		}
+	}
+
+	for _, prefix := range decodedService.GetData().GetWorkloads().GetPrefixes() {
+		err = authorizer.ToAllowAuthorizer().ServiceReadAllowed(prefix, authzContext)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func ACLHooksForWorkloadSelectingType[T WorkloadSelecting]() *resource.ACLHooks {
+	return &resource.ACLHooks{
+		Read:  aclReadHookResourceWithWorkloadSelector,
+		Write: aclWriteHookResourceWithWorkloadSelector[T],
+		List:  resource.NoOpACLListHook,
+	}
+}

internal/catalog/internal/types/dns_policy.go

@@ -19,6 +19,7 @@ func RegisterDNSPolicy(r resource.Registry) {
 		Proto:    &pbcatalog.DNSPolicy{},
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateDNSPolicy,
+		ACLs:     ACLHooksForWorkloadSelectingType[*pbcatalog.DNSPolicy](),
 	})
 }
 

internal/catalog/internal/types/dns_policy_test.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/protobuf/reflect/protoreflect"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/hashicorp/consul/internal/catalog/internal/testhelpers"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -161,3 +162,19 @@ func TestValidateDNSPolicy_EmptySelector(t *testing.T) {
 	require.ErrorAs(t, err, &actual)
 	require.Equal(t, expected, actual)
 }
+
+func TestDNSPolicyACLs(t *testing.T) {
+	// Wire up a registry to generically invoke hooks
+	registry := resource.NewRegistry()
+	RegisterDNSPolicy(registry)
+
+	testhelpers.RunWorkloadSelectingTypeACLsTests[*pbcatalog.DNSPolicy](t, pbcatalog.DNSPolicyType,
+		func(selector *pbcatalog.WorkloadSelector) *pbcatalog.DNSPolicy {
+			return &pbcatalog.DNSPolicy{
+				Workloads: selector,
+				Weights:   &pbcatalog.Weights{Passing: 1, Warning: 0},
+			}
+		},
+		RegisterDNSPolicy,
+	)
+}

internal/catalog/internal/types/failover_policy.go

@@ -25,7 +25,7 @@ func RegisterFailoverPolicy(r resource.Registry) {
 		ACLs: &resource.ACLHooks{
 			Read:  aclReadHookFailoverPolicy,
 			Write: aclWriteHookFailoverPolicy,
-			List:  aclListHookFailoverPolicy,
+			List:  resource.NoOpACLListHook,
 		},
 	})
 }
@@ -371,9 +371,3 @@ func aclWriteHookFailoverPolicy(authorizer acl.Authorizer, authzContext *acl.Aut
 	return nil
 
 }
-
-func aclListHookFailoverPolicy(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/catalog/internal/types/failover_policy_test.go

@@ -10,8 +10,6 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 
-	"github.com/hashicorp/consul/acl"
-	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
@@ -685,105 +683,52 @@ func TestFailoverPolicyACLs(t *testing.T) {
 	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(pbcatalog.FailoverPolicyType)
-	require.True(t, ok)
-
-	run := func(t *testing.T, tc testcase) {
-		failoverData := &pbcatalog.FailoverPolicy{
-			Config: &pbcatalog.FailoverConfig{
-				Destinations: []*pbcatalog.FailoverDestination{
-					{Ref: newRef(pbcatalog.ServiceType, "api-backup")},
-				},
+	failoverData := &pbcatalog.FailoverPolicy{
+		Config: &pbcatalog.FailoverConfig{
+			Destinations: []*pbcatalog.FailoverDestination{
+				{Ref: newRef(pbcatalog.ServiceType, "api-backup")},
 			},
-		}
-		res := resourcetest.Resource(pbcatalog.FailoverPolicyType, "api").
-			WithTenancy(resource.DefaultNamespacedTenancy()).
-			WithData(t, failoverData).
-			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{
+	cases := map[string]resourcetest.ACLTestCase{
 		"no rules": {
-			rules:   ``,
-			readOK:  DENY,
-			writeOK: DENY,
-			listOK:  DEFAULT,
+			Rules:   ``,
+			Data:    failoverData,
+			Typ:     pbcatalog.FailoverPolicyType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
 		},
-		"service api read": {
-			rules:   `service "api" { policy = "read" }`,
-			readOK:  ALLOW,
-			writeOK: DENY,
-			listOK:  DEFAULT,
+		"service test read": {
+			Rules:   `service "test" { policy = "read" }`,
+			Data:    failoverData,
+			Typ:     pbcatalog.FailoverPolicyType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
 		},
-		"service api write": {
-			rules:   `service "api" { policy = "write" }`,
-			readOK:  ALLOW,
-			writeOK: DENY,
-			listOK:  DEFAULT,
+		"service test write": {
+			Rules:   `service "test" { policy = "write" }`,
+			Data:    failoverData,
+			Typ:     pbcatalog.FailoverPolicyType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
 		},
-		"service api write and api-backup read": {
-			rules:   `service "api" { policy = "write" } service "api-backup" { policy = "read" }`,
-			readOK:  ALLOW,
-			writeOK: ALLOW,
-			listOK:  DEFAULT,
+		"service test write and api-backup read": {
+			Rules:   `service "test" { policy = "write" } service "api-backup" { policy = "read" }`,
+			Data:    failoverData,
+			Typ:     pbcatalog.FailoverPolicyType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
 		},
 	}
 
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
-			run(t, tc)
+			resourcetest.RunACLTestCase(t, tc, registry)
 		})
 	}
 }

internal/catalog/internal/types/health_checks.go

@@ -17,6 +17,7 @@ func RegisterHealthChecks(r resource.Registry) {
 		Proto:    &pbcatalog.HealthChecks{},
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateHealthChecks,
+		ACLs:     ACLHooksForWorkloadSelectingType[*pbcatalog.HealthChecks](),
 	})
 }
 

internal/catalog/internal/types/health_checks_test.go

@@ -12,6 +12,7 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/durationpb"
 
+	"github.com/hashicorp/consul/internal/catalog/internal/testhelpers"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -196,3 +197,12 @@ func TestValidateHealthChecks_EmptySelector(t *testing.T) {
 	require.ErrorAs(t, err, &actual)
 	require.Equal(t, expected, actual)
 }
+
+func TestHealthChecksACLs(t *testing.T) {
+	testhelpers.RunWorkloadSelectingTypeACLsTests[*pbcatalog.HealthChecks](t, pbcatalog.HealthChecksType,
+		func(selector *pbcatalog.WorkloadSelector) *pbcatalog.HealthChecks {
+			return &pbcatalog.HealthChecks{Workloads: selector}
+		},
+		RegisterHealthChecks,
+	)
+}

internal/catalog/internal/types/health_status.go

@@ -6,6 +6,7 @@ package types
 import (
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -17,6 +18,11 @@ func RegisterHealthStatus(r resource.Registry) {
 		Proto:    &pbcatalog.HealthStatus{},
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateHealthStatus,
+		ACLs: &resource.ACLHooks{
+			Read:  aclReadHookHealthStatus,
+			Write: aclWriteHookHealthStatus,
+			List:  resource.NoOpACLListHook,
+		},
 	})
 }
 
@@ -66,3 +72,32 @@ func ValidateHealthStatus(res *pbresource.Resource) error {
 
 	return err
 }
+
+func aclReadHookHealthStatus(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, _ *pbresource.ID, res *pbresource.Resource) error {
+	if res == nil {
+		return resource.ErrNeedResource
+	}
+	// For a health status of a workload we need to check service:read perms.
+	if res.GetOwner() != nil && resource.EqualType(res.GetOwner().GetType(), pbcatalog.WorkloadType) {
+		return authorizer.ToAllowAuthorizer().ServiceReadAllowed(res.GetOwner().GetName(), authzContext)
+	}
+
+	if res.GetOwner() != nil && resource.EqualType(res.GetOwner().GetType(), pbcatalog.NodeType) {
+		return authorizer.ToAllowAuthorizer().NodeReadAllowed(res.GetOwner().GetName(), authzContext)
+	}
+
+	return acl.PermissionDenied("cannot read catalog.HealthStatus because there is no owner")
+}
+
+func aclWriteHookHealthStatus(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
+	// For a health status of a workload we need to check service:write perms.
+	if res.GetOwner() != nil && resource.EqualType(res.GetOwner().GetType(), pbcatalog.WorkloadType) {
+		return authorizer.ToAllowAuthorizer().ServiceWriteAllowed(res.GetOwner().GetName(), authzContext)
+	}
+
+	if res.GetOwner() != nil && resource.EqualType(res.GetOwner().GetType(), pbcatalog.NodeType) {
+		return authorizer.ToAllowAuthorizer().NodeWriteAllowed(res.GetOwner().GetName(), authzContext)
+	}
+
+	return acl.PermissionDenied("cannot write catalog.HealthStatus because there is no owner")
+}

internal/catalog/internal/types/health_status_test.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
@@ -214,3 +215,106 @@ func TestValidateHealthStatus_InvalidOwner(t *testing.T) {
 		})
 	}
 }
+
+func TestHealthStatusACLs(t *testing.T) {
+	registry := resource.NewRegistry()
+	Register(registry)
+
+	workload := resourcetest.Resource(pbcatalog.WorkloadType, "test").ID()
+	node := resourcetest.Resource(pbcatalog.NodeType, "test").ID()
+
+	healthStatusData := &pbcatalog.HealthStatus{
+		Type:   "tcp",
+		Status: pbcatalog.Health_HEALTH_PASSING,
+	}
+
+	cases := map[string]resourcetest.ACLTestCase{
+		"no rules": {
+			Rules:   ``,
+			Data:    healthStatusData,
+			Owner:   workload,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test read": {
+			Rules:   `service "test" { policy = "read" }`,
+			Data:    healthStatusData,
+			Owner:   workload,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write": {
+			Rules:   `service "test" { policy = "write" }`,
+			Data:    healthStatusData,
+			Owner:   workload,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test read with node owner": {
+			Rules:   `service "test" { policy = "read" }`,
+			Data:    healthStatusData,
+			Owner:   node,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with node owner": {
+			Rules:   `service "test" { policy = "write" }`,
+			Data:    healthStatusData,
+			Owner:   node,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"node test read with node owner": {
+			Rules:   `node "test" { policy = "read" }`,
+			Data:    healthStatusData,
+			Owner:   node,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"node test write with node owner": {
+			Rules:   `node "test" { policy = "write" }`,
+			Data:    healthStatusData,
+			Owner:   node,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"node test read with workload owner": {
+			Rules:   `node "test" { policy = "read" }`,
+			Data:    healthStatusData,
+			Owner:   workload,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"node test write with workload owner": {
+			Rules:   `node "test" { policy = "write" }`,
+			Data:    healthStatusData,
+			Owner:   workload,
+			Typ:     pbcatalog.HealthStatusType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			resourcetest.RunACLTestCase(t, tc, registry)
+		})
+	}
+}

internal/catalog/internal/types/node.go

@@ -6,6 +6,7 @@ package types
 import (
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -22,6 +23,11 @@ func RegisterNode(r resource.Registry) {
 		// Until that time, Node will remain namespace scoped.
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateNode,
+		ACLs: &resource.ACLHooks{
+			Read:  aclReadHookNode,
+			Write: aclWriteHookNode,
+			List:  resource.NoOpACLListHook,
+		},
 	})
 }
 
@@ -80,3 +86,11 @@ func validateNodeAddress(addr *pbcatalog.NodeAddress) error {
 
 	return nil
 }
+
+func aclReadHookNode(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) error {
+	return authorizer.ToAllowAuthorizer().NodeReadAllowed(id.GetName(), authzContext)
+}
+
+func aclWriteHookNode(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
+	return authorizer.ToAllowAuthorizer().NodeWriteAllowed(res.GetId().GetName(), authzContext)
+}

internal/catalog/internal/types/node_test.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
@@ -127,3 +128,48 @@ func TestValidateNode_AddressMissingHost(t *testing.T) {
 	require.ErrorAs(t, err, &actual)
 	require.Equal(t, expected, actual)
 }
+
+func TestNodeACLs(t *testing.T) {
+	registry := resource.NewRegistry()
+	Register(registry)
+
+	nodeData := &pbcatalog.Node{
+		Addresses: []*pbcatalog.NodeAddress{
+			{
+				Host: "1.1.1.1",
+			},
+		},
+	}
+	cases := map[string]resourcetest.ACLTestCase{
+		"no rules": {
+			Rules:   ``,
+			Data:    nodeData,
+			Typ:     pbcatalog.NodeType,
+			ReadOK:  resourcetest.DENY,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"node test read": {
+			Rules:   `node "test" { policy = "read" }`,
+			Data:    nodeData,
+			Typ:     pbcatalog.NodeType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"node test write": {
+			Rules:   `node "test" { policy = "write" }`,
+			Data:    nodeData,
+			Typ:     pbcatalog.NodeType,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			resourcetest.RunACLTestCase(t, tc, registry)
+		})
+	}
+}

internal/catalog/internal/types/service.go

@@ -20,6 +20,7 @@ func RegisterService(r resource.Registry) {
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateService,
 		Mutate:   MutateService,
+		ACLs:     ACLHooksForWorkloadSelectingType[*pbcatalog.Service](),
 	})
 }
 

internal/catalog/internal/types/service_endpoints.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -20,6 +21,15 @@ func RegisterServiceEndpoints(r resource.Registry) {
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateServiceEndpoints,
 		Mutate:   MutateServiceEndpoints,
+		ACLs: &resource.ACLHooks{
+			Read: func(authorizer acl.Authorizer, context *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) error {
+				return authorizer.ToAllowAuthorizer().ServiceReadAllowed(id.GetName(), context)
+			},
+			Write: func(authorizer acl.Authorizer, context *acl.AuthorizerContext, p *pbresource.Resource) error {
+				return authorizer.ToAllowAuthorizer().ServiceWriteAllowed(p.GetId().GetName(), context)
+			},
+			List: resource.NoOpACLListHook,
+		},
 	})
 }
 

internal/catalog/internal/types/service_endpoints_test.go

@@ -258,3 +258,47 @@ func TestMutateServiceEndpoints_PopulateOwner(t *testing.T) {
 	require.True(t, resource.EqualTenancy(res.Owner.Tenancy, defaultEndpointTenancy))
 	require.Equal(t, res.Owner.Name, res.Id.Name)
 }
+
+func TestServiceEndpointsACLs(t *testing.T) {
+	registry := resource.NewRegistry()
+	Register(registry)
+
+	service := rtest.Resource(pbcatalog.ServiceType, "test").
+		WithTenancy(resource.DefaultNamespacedTenancy()).ID()
+	serviceEndpointsData := &pbcatalog.ServiceEndpoints{}
+	cases := map[string]rtest.ACLTestCase{
+		"no rules": {
+			Rules:   ``,
+			Data:    serviceEndpointsData,
+			Owner:   service,
+			Typ:     pbcatalog.ServiceEndpointsType,
+			ReadOK:  rtest.DENY,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test read": {
+			Rules:   `service "test" { policy = "read" }`,
+			Data:    serviceEndpointsData,
+			Owner:   service,
+			Typ:     pbcatalog.ServiceEndpointsType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write": {
+			Rules:   `service "test" { policy = "write" }`,
+			Data:    serviceEndpointsData,
+			Owner:   service,
+			Typ:     pbcatalog.ServiceEndpointsType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.ALLOW,
+			ListOK:  rtest.DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			rtest.RunACLTestCase(t, tc, registry)
+		})
+	}
+}

internal/catalog/internal/types/service_test.go

@@ -10,6 +10,7 @@ import (
 	"google.golang.org/protobuf/reflect/protoreflect"
 	"google.golang.org/protobuf/types/known/anypb"
 
+	"github.com/hashicorp/consul/internal/catalog/internal/testhelpers"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
@@ -275,3 +276,12 @@ func TestValidateService_InvalidVIP(t *testing.T) {
 	require.Error(t, err)
 	require.ErrorIs(t, err, errNotIPAddress)
 }
+
+func TestServiceACLs(t *testing.T) {
+	testhelpers.RunWorkloadSelectingTypeACLsTests[*pbcatalog.Service](t, pbcatalog.ServiceType,
+		func(selector *pbcatalog.WorkloadSelector) *pbcatalog.Service {
+			return &pbcatalog.Service{Workloads: selector}
+		},
+		RegisterService,
+	)
+}

internal/catalog/internal/types/types.go

@@ -13,8 +13,10 @@ func Register(r resource.Registry) {
 	RegisterServiceEndpoints(r)
 	RegisterNode(r)
 	RegisterHealthStatus(r)
-	RegisterHealthChecks(r)
-	RegisterDNSPolicy(r)
-	RegisterVirtualIPs(r)
 	RegisterFailoverPolicy(r)
+
+	// todo (v2): re-register once these resources are implemented.
+	//RegisterHealthChecks(r)
+	//RegisterDNSPolicy(r)
+	//RegisterVirtualIPs(r)
 }

internal/catalog/internal/types/types_test.go

@@ -24,9 +24,9 @@ func TestTypeRegistration(t *testing.T) {
 		pbcatalog.ServiceEndpointsKind,
 		pbcatalog.NodeKind,
 		pbcatalog.HealthStatusKind,
-		pbcatalog.HealthChecksKind,
-		pbcatalog.DNSPolicyKind,
 		// todo (ishustava): uncomment once we implement these
+		//pbcatalog.HealthChecksKind,
+		//pbcatalog.DNSPolicyKind,
 		//pbcatalog.VirtualIPsKind,
 	}
 

internal/catalog/internal/types/virtual_ips.go

@@ -6,19 +6,28 @@ package types
 import (
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
 func RegisterVirtualIPs(r resource.Registry) {
-	// todo (ishustava): uncomment when we implement it
-	//r.Register(resource.Registration{
-	//	Type:     pbcatalog.VirtualIPsV2Beta1Type,
-	//	Proto:    &pbcatalog.VirtualIPs{},
-	//	Scope:    resource.ScopeNamespace,
-	//	Validate: ValidateVirtualIPs,
-	//})
+	r.Register(resource.Registration{
+		Type:     pbcatalog.VirtualIPsType,
+		Proto:    &pbcatalog.VirtualIPs{},
+		Scope:    resource.ScopeNamespace,
+		Validate: ValidateVirtualIPs,
+		ACLs: &resource.ACLHooks{
+			Read: func(authorizer acl.Authorizer, context *acl.AuthorizerContext, id *pbresource.ID, p *pbresource.Resource) error {
+				return authorizer.ToAllowAuthorizer().ServiceReadAllowed(id.GetName(), context)
+			},
+			Write: func(authorizer acl.Authorizer, context *acl.AuthorizerContext, p *pbresource.Resource) error {
+				return authorizer.ToAllowAuthorizer().ServiceWriteAllowed(p.GetId().GetName(), context)
+			},
+			List: resource.NoOpACLListHook,
+		},
+	})
 }
 
 func ValidateVirtualIPs(res *pbresource.Resource) error {

internal/catalog/internal/types/virtual_ips_test.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/hashicorp/consul/internal/resource"
+	rtest "github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
@@ -81,3 +82,47 @@ func TestValidateVirtualIPs_InvalidIP(t *testing.T) {
 	require.Error(t, err)
 	require.ErrorIs(t, err, errNotIPAddress)
 }
+
+func TestVirtualIPsACLs(t *testing.T) {
+	registry := resource.NewRegistry()
+	RegisterVirtualIPs(registry)
+
+	service := rtest.Resource(pbcatalog.ServiceType, "test").
+		WithTenancy(resource.DefaultNamespacedTenancy()).ID()
+	virtualIPsData := &pbcatalog.VirtualIPs{}
+	cases := map[string]rtest.ACLTestCase{
+		"no rules": {
+			Rules:   ``,
+			Data:    virtualIPsData,
+			Owner:   service,
+			Typ:     pbcatalog.VirtualIPsType,
+			ReadOK:  rtest.DENY,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test read": {
+			Rules:   `service "test" { policy = "read" }`,
+			Data:    virtualIPsData,
+			Owner:   service,
+			Typ:     pbcatalog.VirtualIPsType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write": {
+			Rules:   `service "test" { policy = "write" }`,
+			Data:    virtualIPsData,
+			Owner:   service,
+			Typ:     pbcatalog.VirtualIPsType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.ALLOW,
+			ListOK:  rtest.DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			rtest.RunACLTestCase(t, tc, registry)
+		})
+	}
+}

internal/catalog/internal/types/workload.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -20,6 +21,11 @@ func RegisterWorkload(r resource.Registry) {
 		Proto:    &pbcatalog.Workload{},
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateWorkload,
+		ACLs: &resource.ACLHooks{
+			Read:  aclReadHookWorkload,
+			Write: aclWriteHookWorkload,
+			List:  resource.NoOpACLListHook,
+		},
 	})
 }
 
@@ -145,3 +151,32 @@ func ValidateWorkload(res *pbresource.Resource) error {
 
 	return err
 }
+
+func aclReadHookWorkload(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) error {
+	return authorizer.ToAllowAuthorizer().ServiceReadAllowed(id.GetName(), authzContext)
+}
+
+func aclWriteHookWorkload(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
+	decodedWorkload, err := resource.Decode[*pbcatalog.Workload](res)
+	if err != nil {
+		return resource.ErrNeedResource
+	}
+
+	// First check service:write on the workload name.
+	err = authorizer.ToAllowAuthorizer().ServiceWriteAllowed(res.GetId().GetName(), authzContext)
+	if err != nil {
+		return err
+	}
+
+	// Check node:read permissions if node is specified.
+	if decodedWorkload.GetData().GetNodeName() != "" {
+		return authorizer.ToAllowAuthorizer().NodeReadAllowed(decodedWorkload.GetData().GetNodeName(), authzContext)
+	}
+
+	// Check identity:read permissions if identity is specified.
+	if decodedWorkload.GetData().GetIdentity() != "" {
+		return authorizer.ToAllowAuthorizer().IdentityReadAllowed(decodedWorkload.GetData().GetIdentity(), authzContext)
+	}
+
+	return nil
+}

internal/catalog/internal/types/workload_selecting.go

@@ -0,0 +1,16 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package types
+
+import (
+	"google.golang.org/protobuf/proto"
+
+	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
+)
+
+// WorkloadSelecting denotes a resource type that uses workload selectors.
+type WorkloadSelecting interface {
+	proto.Message
+	GetWorkloads() *pbcatalog.WorkloadSelector
+}

internal/catalog/internal/types/workload_test.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/protobuf/types/known/anypb"
 
 	"github.com/hashicorp/consul/internal/resource"
+	rtest "github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
@@ -304,3 +305,160 @@ func TestValidateWorkload_Locality(t *testing.T) {
 	require.ErrorAs(t, err, &actual)
 	require.Equal(t, expected, actual)
 }
+
+func TestWorkloadACLs(t *testing.T) {
+	registry := resource.NewRegistry()
+	Register(registry)
+
+	cases := map[string]rtest.ACLTestCase{
+		"no rules": {
+			Rules: ``,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.DENY,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test read": {
+			Rules: `service "test" { policy = "read" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write": {
+			Rules: `service "test" { policy = "write" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.ALLOW,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write with node": {
+			Rules: `service "test" { policy = "write" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+				NodeName: "test-node",
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write with workload identity": {
+			Rules: `service "test" { policy = "write" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+				Identity: "test-identity",
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write with workload identity and node": {
+			Rules: `service "test" { policy = "write" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+				NodeName: "test-node",
+				Identity: "test-identity",
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.DENY,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write with node and node policy": {
+			Rules: `service "test" { policy = "write" } node "test-node" { policy = "read" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+				NodeName: "test-node",
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.ALLOW,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write with workload identity and identity policy ": {
+			Rules: `service "test" { policy = "write" } identity "test-identity" { policy = "read" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+				Identity: "test-identity",
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.ALLOW,
+			ListOK:  rtest.DEFAULT,
+		},
+		"service test write with workload identity and node with both node and identity policy": {
+			Rules: `service "test" { policy = "write" } identity "test-identity" { policy = "read" } node "test-node" { policy = "read" }`,
+			Data: &pbcatalog.Workload{
+				Addresses: []*pbcatalog.WorkloadAddress{
+					{Host: "1.1.1.1"},
+				},
+				Ports: map[string]*pbcatalog.WorkloadPort{
+					"tcp": {Port: 8080},
+				},
+				NodeName: "test-node",
+				Identity: "test-identity",
+			},
+			Typ:     pbcatalog.WorkloadType,
+			ReadOK:  rtest.ALLOW,
+			WriteOK: rtest.ALLOW,
+			ListOK:  rtest.DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			rtest.RunACLTestCase(t, tc, registry)
+		})
+	}
+}

internal/mesh/internal/mappers/workloadselectionmapper/workload_selection_mapper.go

@@ -6,29 +6,21 @@ package workloadselectionmapper
 import (
 	"context"
 
-	"google.golang.org/protobuf/proto"
-
+	"github.com/hashicorp/consul/internal/catalog"
 	"github.com/hashicorp/consul/internal/controller"
 	"github.com/hashicorp/consul/internal/mesh/internal/mappers/common"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/mappers/selectiontracker"
 	"github.com/hashicorp/consul/lib/stringslice"
-	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
-// WorkloadSelecting denotes a resource type that uses workload selectors.
-type WorkloadSelecting interface {
-	proto.Message
-	GetWorkloads() *pbcatalog.WorkloadSelector
-}
-
-type Mapper[T WorkloadSelecting] struct {
+type Mapper[T catalog.WorkloadSelecting] struct {
 	workloadSelectionTracker *selectiontracker.WorkloadSelectionTracker
 	computedType             *pbresource.Type
 }
 
-func New[T WorkloadSelecting](computedType *pbresource.Type) *Mapper[T] {
+func New[T catalog.WorkloadSelecting](computedType *pbresource.Type) *Mapper[T] {
 	if computedType == nil {
 		panic("computed type is required")
 	}

internal/mesh/internal/types/destination_policy.go

@@ -24,7 +24,7 @@ func RegisterDestinationPolicy(r resource.Registry) {
 		ACLs: &resource.ACLHooks{
 			Read:  aclReadHookDestinationPolicy,
 			Write: aclWriteHookDestinationPolicy,
-			List:  aclListHookDestinationPolicy,
+			List:  resource.NoOpACLListHook,
 		},
 	})
 }
@@ -233,9 +233,3 @@ func aclWriteHookDestinationPolicy(authorizer acl.Authorizer, authzContext *acl.
 	// 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/destinations.go

@@ -22,6 +22,7 @@ func RegisterDestinations(r resource.Registry) {
 		Scope:    resource.ScopeNamespace,
 		Mutate:   MutateDestinations,
 		Validate: ValidateDestinations,
+		ACLs:     catalog.ACLHooksForWorkloadSelectingType[*pbmesh.Destinations](),
 	})
 }
 

internal/mesh/internal/types/destinations_configuration.go

@@ -7,17 +7,19 @@ import (
 	"github.com/hashicorp/go-multierror"
 
 	"github.com/hashicorp/consul/internal/catalog"
+
 	"github.com/hashicorp/consul/internal/resource"
 	pbmesh "github.com/hashicorp/consul/proto-public/pbmesh/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
-func RegisterUpstreamsConfiguration(r resource.Registry) {
+func RegisterDestinationsConfiguration(r resource.Registry) {
 	r.Register(resource.Registration{
 		Type:     pbmesh.DestinationsConfigurationType,
 		Proto:    &pbmesh.DestinationsConfiguration{},
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateDestinationsConfiguration,
+		ACLs:     catalog.ACLHooksForWorkloadSelectingType[*pbmesh.DestinationsConfiguration](),
 	})
 }
 

internal/mesh/internal/types/destinations_configuration_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	catalogtesthelpers "github.com/hashicorp/consul/internal/catalog/catalogtest/helpers"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
@@ -16,6 +17,15 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
+func TestDestinationsConfigurationACLs(t *testing.T) {
+	catalogtesthelpers.RunWorkloadSelectingTypeACLsTests[*pbmesh.DestinationsConfiguration](t, pbmesh.DestinationsConfigurationType,
+		func(selector *pbcatalog.WorkloadSelector) *pbmesh.DestinationsConfiguration {
+			return &pbmesh.DestinationsConfiguration{Workloads: selector}
+		},
+		RegisterDestinationsConfiguration,
+	)
+}
+
 func TestValidateDestinationsConfiguration(t *testing.T) {
 	type testcase struct {
 		data      *pbmesh.DestinationsConfiguration

internal/mesh/internal/types/destinations_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	catalogtesthelpers "github.com/hashicorp/consul/internal/catalog/catalogtest/helpers"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
@@ -415,3 +416,12 @@ func TestValidateDestinations(t *testing.T) {
 		})
 	}
 }
+
+func TestDestinationsACLs(t *testing.T) {
+	catalogtesthelpers.RunWorkloadSelectingTypeACLsTests[*pbmesh.Destinations](t, pbmesh.DestinationsType,
+		func(selector *pbcatalog.WorkloadSelector) *pbmesh.Destinations {
+			return &pbmesh.Destinations{Workloads: selector}
+		},
+		RegisterDestinations,
+	)
+}

internal/mesh/internal/types/proxy_configuration.go

@@ -6,10 +6,10 @@ package types
 import (
 	"math"
 
-	"github.com/hashicorp/go-multierror"
-
 	"github.com/hashicorp/consul/internal/catalog"
 
+	"github.com/hashicorp/go-multierror"
+
 	"github.com/hashicorp/consul/internal/resource"
 	pbmesh "github.com/hashicorp/consul/proto-public/pbmesh/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -23,6 +23,7 @@ func RegisterProxyConfiguration(r resource.Registry) {
 		Scope:    resource.ScopeNamespace,
 		Mutate:   MutateProxyConfiguration,
 		Validate: ValidateProxyConfiguration,
+		ACLs:     catalog.ACLHooksForWorkloadSelectingType[*pbmesh.ProxyConfiguration](),
 	})
 }
 

internal/mesh/internal/types/proxy_configuration_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/types/known/structpb"
 
+	catalogtesthelpers "github.com/hashicorp/consul/internal/catalog/catalogtest/helpers"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v2beta1"
@@ -20,6 +21,18 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
+func TestProxyConfigurationACLs(t *testing.T) {
+	catalogtesthelpers.RunWorkloadSelectingTypeACLsTests[*pbmesh.ProxyConfiguration](t, pbmesh.ProxyConfigurationType,
+		func(selector *pbcatalog.WorkloadSelector) *pbmesh.ProxyConfiguration {
+			return &pbmesh.ProxyConfiguration{
+				Workloads:     selector,
+				DynamicConfig: &pbmesh.DynamicConfig{},
+			}
+		},
+		RegisterProxyConfiguration,
+	)
+}
+
 func TestMutateProxyConfiguration(t *testing.T) {
 	cases := map[string]struct {
 		data    *pbmesh.ProxyConfiguration

internal/mesh/internal/types/proxy_state_template.go

@@ -44,11 +44,7 @@ func RegisterProxyStateTemplate(r resource.Registry) {
 				// managed by a controller.
 				return authorizer.ToAllowAuthorizer().OperatorWriteAllowed(authzContext)
 			},
-			List: func(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
-			},
+			List: resource.NoOpACLListHook,
 		},
 	})
 }

internal/mesh/internal/types/types.go

@@ -12,11 +12,12 @@ func Register(r resource.Registry) {
 	RegisterComputedProxyConfiguration(r)
 	RegisterDestinations(r)
 	RegisterComputedExplicitDestinations(r)
-	RegisterUpstreamsConfiguration(r)
 	RegisterProxyStateTemplate(r)
 	RegisterHTTPRoute(r)
 	RegisterTCPRoute(r)
 	RegisterGRPCRoute(r)
 	RegisterDestinationPolicy(r)
 	RegisterComputedRoutes(r)
+	// todo (v2): uncomment once we implement it.
+	//RegisterDestinationsConfiguration(r)
 }

internal/mesh/internal/types/types_test.go

@@ -21,13 +21,14 @@ func TestTypeRegistration(t *testing.T) {
 	requiredKinds := []string{
 		pbmesh.ProxyConfigurationKind,
 		pbmesh.DestinationsKind,
-		pbmesh.DestinationsConfigurationKind,
 		pbmesh.ProxyStateTemplateKind,
 		pbmesh.HTTPRouteKind,
 		pbmesh.TCPRouteKind,
 		pbmesh.GRPCRouteKind,
 		pbmesh.DestinationPolicyKind,
 		pbmesh.ComputedRoutesKind,
+		// todo (v2): re-enable once we implement it.
+		//pbmesh.DestinationsConfigurationKind,
 	}
 
 	r := resource.NewRegistry()

internal/mesh/internal/types/xroute.go

@@ -290,7 +290,7 @@ func xRouteACLHooks[R XRouteData]() *resource.ACLHooks {
 	hooks := &resource.ACLHooks{
 		Read:  aclReadHookXRoute[R],
 		Write: aclWriteHookXRoute[R],
-		List:  aclListHookXRoute[R],
+		List:  resource.NoOpACLListHook,
 	}
 
 	return hooks
@@ -298,7 +298,7 @@ func xRouteACLHooks[R XRouteData]() *resource.ACLHooks {
 
 func aclReadHookXRoute[R XRouteData](authorizer acl.Authorizer, _ *acl.AuthorizerContext, _ *pbresource.ID, res *pbresource.Resource) error {
 	if res == nil {
-		return resource.ErrNeedData
+		return resource.ErrNeedResource
 	}
 
 	dec, err := resource.Decode[R](res)
@@ -351,9 +351,3 @@ func aclWriteHookXRoute[R XRouteData](authorizer acl.Authorizer, _ *acl.Authoriz
 
 	return nil
 }
-
-func aclListHookXRoute[R XRouteData](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/xroute_test.go

@@ -458,7 +458,7 @@ func testXRouteACLs[R XRouteData](t *testing.T, newRoute func(t *testing.T, pare
 			require.True(t, ok)
 
 			err = reg.ACLs.Read(authz, &acl.AuthorizerContext{}, tc.res.Id, nil)
-			require.ErrorIs(t, err, resource.ErrNeedData, "read hook should require the data payload")
+			require.ErrorIs(t, err, resource.ErrNeedResource, "read hook should require the data payload")
 
 			checkF(t, "read", tc.readOK, reg.ACLs.Read(authz, &acl.AuthorizerContext{}, tc.res.Id, tc.res))
 			checkF(t, "write", tc.writeOK, reg.ACLs.Write(authz, &acl.AuthorizerContext{}, tc.res))

internal/resource/acls.go

@@ -0,0 +1,13 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package resource
+
+import "github.com/hashicorp/consul/acl"
+
+// NoOpACLListHook is a common function that can be used if no special list permission is required for a resource.
+func NoOpACLListHook(_ acl.Authorizer, _ *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/resource/demo/demo.go

@@ -97,7 +97,7 @@ func RegisterTypes(r resource.Registry) {
 	readACL := func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, res *pbresource.Resource) error {
 		if resource.EqualType(TypeV1RecordLabel, id.Type) {
 			if res == nil {
-				return resource.ErrNeedData
+				return resource.ErrNeedResource
 			}
 		}
 		key := fmt.Sprintf("resource/%s/%s", resource.ToGVK(id.Type), id.Name)

internal/resource/errors.go

@@ -137,6 +137,20 @@ type ErrOwnerTenantInvalid struct {
 }
 
 func (err ErrOwnerTenantInvalid) Error() string {
+	if err.ResourceTenancy == nil && err.OwnerTenancy != nil {
+		return fmt.Sprintf(
+			"empty resource tenancy cannot be owned by a resource in partition %s, namespace %s and peer %s",
+			err.OwnerTenancy.Partition, err.OwnerTenancy.Namespace, err.OwnerTenancy.PeerName,
+		)
+	}
+
+	if err.ResourceTenancy != nil && err.OwnerTenancy == nil {
+		return fmt.Sprintf(
+			"resource in partition %s, namespace %s and peer %s cannot be owned by a resource with empty tenancy",
+			err.ResourceTenancy.Partition, err.ResourceTenancy.Namespace, err.ResourceTenancy.PeerName,
+		)
+	}
+
 	return fmt.Sprintf(
 		"resource in partition %s, namespace %s and peer %s cannot be owned by a resource in partition %s, namespace %s and peer %s",
 		err.ResourceTenancy.Partition, err.ResourceTenancy.Namespace, err.ResourceTenancy.PeerName,

internal/resource/registry.go

@@ -68,14 +68,14 @@ type Registration struct {
 	Scope Scope
 }
 
-var ErrNeedData = errors.New("authorization check requires resource data")
+var ErrNeedResource = errors.New("authorization check requires the entire resource")
 
 type ACLHooks struct {
 	// Read is used to authorize Read RPCs and to filter results in List
 	// RPCs.
 	//
 	// It can be called an ID and possibly a Resource. The check will first
-	// attempt to use the ID and if the hook returns ErrNeedData, then the
+	// attempt to use the ID and if the hook returns ErrNeedResource, then the
 	// check will be deferred until the data is fetched from the storage layer.
 	//
 	// If it is omitted, `operator:read` permission is assumed.

internal/resource/resourcetest/acls.go

@@ -0,0 +1,85 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package resourcetest
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/reflect/protoreflect"
+
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/structs"
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+const (
+	DENY    = "deny"
+	ALLOW   = "allow"
+	DEFAULT = "default"
+)
+
+var 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)
+	}
+}
+
+type ACLTestCase struct {
+	Rules   string
+	Data    protoreflect.ProtoMessage
+	Owner   *pbresource.ID
+	Typ     *pbresource.Type
+	ReadOK  string
+	WriteOK string
+	ListOK  string
+}
+
+func RunACLTestCase(t *testing.T, tc ACLTestCase, registry resource.Registry) {
+	reg, ok := registry.Resolve(tc.Typ)
+	require.True(t, ok)
+
+	resolvedType, ok := registry.Resolve(tc.Typ)
+	require.True(t, ok)
+
+	res := Resource(tc.Typ, "test").
+		WithTenancy(DefaultTenancyForType(t, resolvedType)).
+		WithOwner(tc.Owner).
+		WithData(t, tc.Data).
+		Build()
+
+	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, res)
+		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)
+	})
+}

internal/resource/resourcetest/tenancy.go

@@ -5,6 +5,7 @@ package resourcetest
 
 import (
 	"strings"
+	"testing"
 
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -35,3 +36,17 @@ func Tenancy(s string) *pbresource.Tenancy {
 		return &pbresource.Tenancy{Partition: "BAD", Namespace: "BAD", PeerName: "BAD"}
 	}
 }
+
+func DefaultTenancyForType(t *testing.T, reg resource.Registration) *pbresource.Tenancy {
+	switch reg.Scope {
+	case resource.ScopeNamespace:
+		return resource.DefaultNamespacedTenancy()
+	case resource.ScopePartition:
+		return resource.DefaultPartitionedTenancy()
+	case resource.ScopeCluster:
+		return resource.DefaultClusteredTenancy()
+	default:
+		t.Fatalf("unsupported resource scope: %v", reg.Scope)
+		return nil
+	}
+}