catalog: add ACL checks for FailoverPolicy resources (#18919)

FailoverPolicy resources are name-aligned with the Service they control.
They also contain a list of possible failover destinations that are References
 to other Services.

The ACLs should be:

- list: (default)
- read: service:<resource_name>:read
- write: service:<resource_name>:write + service:<destination_name>:read (for any destination)

internal/catalog/internal/types/failover_policy.go

@@ -9,6 +9,7 @@ import (
 	"github.com/hashicorp/go-multierror"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbcatalog "github.com/hashicorp/consul/proto-public/pbcatalog/v1alpha1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -35,6 +36,11 @@ func RegisterFailoverPolicy(r resource.Registry) {
 		Scope:    resource.ScopeNamespace,
 		Mutate:   MutateFailoverPolicy,
 		Validate: ValidateFailoverPolicy,
+		ACLs: &resource.ACLHooks{
+			Read:  aclReadHookFailoverPolicy,
+			Write: aclWriteHookFailoverPolicy,
+			List:  aclListHookFailoverPolicy,
+		},
 	})
 }
 
@@ -316,3 +322,56 @@ func SimplifyFailoverPolicy(svc *pbcatalog.Service, failover *pbcatalog.Failover
 
 	return failover
 }
+
+func aclReadHookFailoverPolicy(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
+	// FailoverPolicy is name-aligned with Service
+	serviceName := id.Name
+
+	// Check service:read permissions.
+	return authorizer.ToAllowAuthorizer().ServiceReadAllowed(serviceName, authzContext)
+}
+
+func aclWriteHookFailoverPolicy(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
+	// FailoverPolicy is name-aligned with Service
+	serviceName := res.Id.Name
+
+	// Check service:write permissions on the service this is controlling.
+	if err := authorizer.ToAllowAuthorizer().ServiceWriteAllowed(serviceName, authzContext); err != nil {
+		return err
+	}
+
+	dec, err := resource.Decode[*pbcatalog.FailoverPolicy](res)
+	if err != nil {
+		return err
+	}
+
+	// Ensure you have service:read on any destination that may be affected by
+	// traffic FROM this config change.
+	if dec.Data.Config != nil {
+		for _, dest := range dec.Data.Config.Destinations {
+			destAuthzContext := resource.AuthorizerContext(dest.Ref.GetTenancy())
+			destServiceName := dest.Ref.GetName()
+			if err := authorizer.ToAllowAuthorizer().ServiceReadAllowed(destServiceName, destAuthzContext); err != nil {
+				return err
+			}
+		}
+	}
+	for _, pc := range dec.Data.PortConfigs {
+		for _, dest := range pc.Destinations {
+			destAuthzContext := resource.AuthorizerContext(dest.Ref.GetTenancy())
+			destServiceName := dest.Ref.GetName()
+			if err := authorizer.ToAllowAuthorizer().ServiceReadAllowed(destServiceName, destAuthzContext); err != nil {
+				return err
+			}
+		}
+	}
+
+	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,6 +10,8 @@ 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/v1alpha1"
@@ -663,6 +665,114 @@ func TestSimplifyFailoverPolicy(t *testing.T) {
 	}
 }
 
+func TestFailoverPolicyACLs(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(FailoverPolicyType)
+	require.True(t, ok)
+
+	run := func(t *testing.T, tc testcase) {
+		failoverData := &pbcatalog.FailoverPolicy{
+			Config: &pbcatalog.FailoverConfig{
+				Destinations: []*pbcatalog.FailoverDestination{
+					{Ref: newRef(ServiceType, "api-backup")},
+				},
+			},
+		}
+		res := resourcetest.Resource(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)
+			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: DENY,
+			listOK:  DEFAULT,
+		},
+		"service api write and api-backup read": {
+			rules:   `service "api" { policy = "write" } service "api-backup" { policy = "read" }`,
+			readOK:  ALLOW,
+			writeOK: ALLOW,
+			listOK:  DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
 func newRef(typ *pbresource.Type, name string) *pbresource.Reference {
 	return resourcetest.Resource(typ, name).
 		WithTenancy(resource.DefaultNamespacedTenancy()).