acls,catalog,mesh: properly authorize workload selectors on writes (#19260)

To properly enforce writes on resources that have workload selectors with prefixes, we need another service authorization rule that allows us to check whether read is allowed within a given prefix. Specifically we need to only allow writes if the policy prefix allows for a wider set of names than the prefix selector on the resource. We should also not allow policies with exact names for prefix matches.

Part of [NET-3993]

acl/MockAuthorizer.go

@@ -224,6 +224,11 @@ func (m *MockAuthorizer) ServiceReadAll(ctx *AuthorizerContext) EnforcementDecis
 	return ret.Get(0).(EnforcementDecision)
 }
 
+func (m *MockAuthorizer) ServiceReadPrefix(prefix string, ctx *AuthorizerContext) EnforcementDecision {
+	ret := m.Called(ctx)
+	return ret.Get(0).(EnforcementDecision)
+}
+
 // ServiceWrite checks for permission to create or update a given
 // service
 func (m *MockAuthorizer) ServiceWrite(segment string, ctx *AuthorizerContext) EnforcementDecision {

acl/acl_test.go

@@ -300,6 +300,14 @@ func checkDenyServiceReadAll(t *testing.T, authz Authorizer, _ string, entCtx *A
 	require.Equal(t, Deny, authz.ServiceReadAll(entCtx))
 }
 
+func checkAllowServiceReadPrefix(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
+	require.Equal(t, Allow, authz.ServiceReadPrefix(prefix, entCtx))
+}
+
+func checkDenyServiceReadPrefix(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
+	require.Equal(t, Deny, authz.ServiceReadPrefix(prefix, entCtx))
+}
+
 func checkDenyServiceWrite(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
 	require.Equal(t, Deny, authz.ServiceWrite(prefix, entCtx))
 }
@@ -456,6 +464,10 @@ func checkDefaultServiceReadAll(t *testing.T, authz Authorizer, _ string, entCtx
 	require.Equal(t, Default, authz.ServiceReadAll(entCtx))
 }
 
+func checkDefaultServiceReadPrefix(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
+	require.Equal(t, Default, authz.ServiceReadPrefix(prefix, entCtx))
+}
+
 func checkDefaultServiceWrite(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
 	require.Equal(t, Default, authz.ServiceWrite(prefix, entCtx))
 }

acl/authorizer.go

@@ -171,6 +171,9 @@ type Authorizer interface {
 	// ServiceReadAll checks for permission to read all services
 	ServiceReadAll(*AuthorizerContext) EnforcementDecision
 
+	// ServiceReadPrefix checks for permission to read services within the given prefix.
+	ServiceReadPrefix(string, *AuthorizerContext) EnforcementDecision
+
 	// ServiceWrite checks for permission to create or update a given
 	// service
 	ServiceWrite(string, *AuthorizerContext) EnforcementDecision
@@ -507,6 +510,14 @@ func (a AllowAuthorizer) ServiceReadAllAllowed(ctx *AuthorizerContext) error {
 	return nil
 }
 
+// ServiceReadPrefixAllowed checks for permission to read services within the given prefix
+func (a AllowAuthorizer) ServiceReadPrefixAllowed(prefix string, ctx *AuthorizerContext) error {
+	if a.Authorizer.ServiceReadPrefix(prefix, ctx) != Allow {
+		return PermissionDeniedByACL(a, ctx, ResourceService, AccessRead, prefix) // read
+	}
+	return nil
+}
+
 // ServiceWriteAllowed checks for permission to create or update a given
 // service
 func (a AllowAuthorizer) ServiceWriteAllowed(name string, ctx *AuthorizerContext) error {

acl/chained_authorizer.go

@@ -275,6 +275,12 @@ func (c *ChainedAuthorizer) ServiceReadAll(entCtx *AuthorizerContext) Enforcemen
 	})
 }
 
+func (c *ChainedAuthorizer) ServiceReadPrefix(prefix string, entCtx *AuthorizerContext) EnforcementDecision {
+	return c.executeChain(func(authz Authorizer) EnforcementDecision {
+		return authz.ServiceReadPrefix(prefix, entCtx)
+	})
+}
+
 // ServiceWrite checks for permission to create or update a given
 // service
 func (c *ChainedAuthorizer) ServiceWrite(name string, entCtx *AuthorizerContext) EnforcementDecision {

acl/chained_authorizer_test.go

@@ -107,6 +107,9 @@ func (authz testAuthorizer) ServiceRead(string, *AuthorizerContext) EnforcementD
 func (authz testAuthorizer) ServiceReadAll(*AuthorizerContext) EnforcementDecision {
 	return EnforcementDecision(authz)
 }
+func (authz testAuthorizer) ServiceReadPrefix(string, *AuthorizerContext) EnforcementDecision {
+	return EnforcementDecision(authz)
+}
 func (authz testAuthorizer) ServiceWrite(string, *AuthorizerContext) EnforcementDecision {
 	return EnforcementDecision(authz)
 }

acl/policy_authorizer.go

@@ -712,7 +712,7 @@ func (p *policyAuthorizer) KeyWritePrefix(prefix string, _ *AuthorizerContext) E
 	//     that do NOT grant AccessWrite.
 	//
 	// Conditions for Default:
-	//   * There is no prefix match rule that would appy to the given prefix.
+	//   * There is no prefix match rule that would apply to the given prefix.
 	//   AND
 	//   * There are no rules (exact or prefix match) within/under the given prefix
 	//     that would NOT grant AccessWrite.
@@ -916,6 +916,62 @@ func (p *policyAuthorizer) ServiceReadAll(_ *AuthorizerContext) EnforcementDecis
 	return p.allAllowed(p.serviceRules, AccessRead)
 }
 
+// ServiceReadPrefix determines whether service read is allowed within the given prefix.
+//
+// Access is allowed iff all the following are true:
+// - There's a read policy for the longest prefix that's shorter or equal to the provided prefix.
+// - There's no deny policy for any prefix that's longer than the given prefix.
+// - There's no deny policy for any exact match that's within the given prefix.
+func (p *policyAuthorizer) ServiceReadPrefix(prefix string, _ *AuthorizerContext) EnforcementDecision {
+	access := Default
+
+	// 1. Walk the prefix tree from root to the given prefix. Find the longest prefix matching ours,
+	//    and use that policy to determine our access as that is the most specific prefix, and it
+	//    should take precedence.
+	p.serviceRules.WalkPath(prefix, func(path string, leaf interface{}) bool {
+		rule := leaf.(*policyAuthorizerRadixLeaf)
+
+		if rule.prefix != nil {
+			switch rule.prefix.access {
+			case AccessRead, AccessWrite:
+				access = Allow
+			default:
+				access = Deny
+			}
+		}
+
+		// Don't stop iteration because we want to visit all nodes down to our leaf to find the more specific match
+		// as it should take precedence.
+		return false
+	})
+
+	// 2. Check rules "below" the given prefix. Access is allowed if there's no deny policy
+	//    for any prefix longer than ours or for any exact match that's within the prefix.
+	p.serviceRules.WalkPrefix(prefix, func(path string, leaf interface{}) bool {
+		rule := leaf.(*policyAuthorizerRadixLeaf)
+
+		if rule.prefix != nil && (rule.prefix.access != AccessRead && rule.prefix.access != AccessWrite) {
+			// If any prefix longer than the provided prefix has "deny" policy, then access is denied.
+			access = Deny
+
+			// We don't need to look at the rest of the tree in this case, so terminate early.
+			return true
+		}
+
+		if rule.exact != nil && (rule.exact.access != AccessRead && rule.exact.access != AccessWrite) {
+			// If any exact match policy has an explicit deny, then access is denied.
+			access = Deny
+
+			// We don't need to look at the rest of the tree in this case, so terminate early.
+			return true
+		}
+
+		return false
+	})
+
+	return access
+}
+
 // ServiceWrite checks if writing (registering) a service is allowed
 func (p *policyAuthorizer) ServiceWrite(name string, _ *AuthorizerContext) EnforcementDecision {
 	if rule, ok := getPolicy(name, p.serviceRules); ok {

acl/policy_authorizer_test.go

@@ -64,6 +64,8 @@ func TestPolicyAuthorizer(t *testing.T) {
 				{name: "DefaultPreparedQueryRead", prefix: "foo", check: checkDefaultPreparedQueryRead},
 				{name: "DefaultPreparedQueryWrite", prefix: "foo", check: checkDefaultPreparedQueryWrite},
 				{name: "DefaultServiceRead", prefix: "foo", check: checkDefaultServiceRead},
+				{name: "DefaultServiceReadAll", prefix: "foo", check: checkDefaultServiceReadAll},
+				{name: "DefaultServiceReadPrefix", prefix: "foo", check: checkDefaultServiceReadPrefix},
 				{name: "DefaultServiceWrite", prefix: "foo", check: checkDefaultServiceWrite},
 				{name: "DefaultServiceWriteAny", prefix: "", check: checkDefaultServiceWriteAny},
 				{name: "DefaultSessionRead", prefix: "foo", check: checkDefaultSessionRead},
@@ -396,6 +398,7 @@ func TestPolicyAuthorizer(t *testing.T) {
 				{name: "ServiceReadDenied", prefix: "football", check: checkDenyServiceRead},
 				{name: "ServiceWriteDenied", prefix: "football", check: checkDenyServiceWrite},
 				{name: "ServiceWriteAnyAllowed", prefix: "", check: checkAllowServiceWriteAny},
+				{name: "ServiceReadWithinPrefixDenied", prefix: "foot", check: checkDenyServiceReadPrefix},
 
 				{name: "IdentityReadPrefixAllowed", prefix: "fo", check: checkAllowIdentityRead},
 				{name: "IdentityWritePrefixDenied", prefix: "fo", check: checkDenyIdentityWrite},
@@ -570,6 +573,214 @@ func TestPolicyAuthorizer(t *testing.T) {
 				{name: "AllDenied", prefix: "*", check: checkDenyIntentionWrite},
 			},
 		},
+		"Service Read Prefix - read allowed with write policy and exact prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo",
+						Policy: PolicyWrite,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixAllowed", prefix: "foo", check: checkAllowServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - read allowed with read policy and exact prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo",
+						Policy: PolicyRead,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixAllowed", prefix: "foo", check: checkAllowServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - read denied with deny policy and exact prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo",
+						Policy: PolicyDeny,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDenied", prefix: "foo", check: checkDenyServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - read allowed with write policy and shorter prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo",
+						Policy: PolicyWrite,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixAllowed", prefix: "foo1", check: checkAllowServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - read allowed with read policy and shorter prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo",
+						Policy: PolicyRead,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixAllowed", prefix: "foo1", check: checkAllowServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - read denied with deny policy and shorter prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo",
+						Policy: PolicyDeny,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDenied", prefix: "foo1", check: checkDenyServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - default with write policy and longer prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo1",
+						Policy: PolicyWrite,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDefault", prefix: "foo", check: checkDefaultServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - default with read policy and longer prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo1",
+						Policy: PolicyRead,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDefault", prefix: "foo", check: checkDefaultServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - deny with deny policy and longer prefix": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "foo1",
+						Policy: PolicyDeny,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDenied", prefix: "foo", check: checkDenyServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - allow with two shorter prefixes - more specific one allowing read and less specific denying": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "fo",
+						Policy: PolicyDeny,
+					},
+					{
+						Name:   "foo",
+						Policy: PolicyRead,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixAllowed", prefix: "foo", check: checkAllowServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - deny with two shorter prefixes - more specific one denying and less specific allowing read": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "fo",
+						Policy: PolicyRead,
+					},
+					{
+						Name:   "foo",
+						Policy: PolicyDeny,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDenied", prefix: "foo", check: checkDenyServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - deny with exact match denying": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "fo",
+						Policy: PolicyRead,
+					},
+				},
+				Services: []*ServiceRule{
+					{
+						Name:   "foo-123",
+						Policy: PolicyDeny,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDenied", prefix: "foo", check: checkDenyServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - allow with exact match allowing read": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "fo",
+						Policy: PolicyRead,
+					},
+				},
+				Services: []*ServiceRule{
+					{
+						Name:   "foo-123",
+						Policy: PolicyRead,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixAllowed", prefix: "foo", check: checkAllowServiceReadPrefix},
+			},
+		},
+		"Service Read Prefix - deny with exact match allowing read but prefix match denying": {
+			policy: &Policy{PolicyRules: PolicyRules{
+				ServicePrefixes: []*ServiceRule{
+					{
+						Name:   "fo",
+						Policy: PolicyDeny,
+					},
+				},
+				Services: []*ServiceRule{
+					{
+						Name:   "foo-123",
+						Policy: PolicyRead,
+					},
+				},
+			}},
+			checks: []aclCheck{
+				{name: "ServiceReadPrefixDenied", prefix: "foo", check: checkDenyServiceReadPrefix},
+			},
+		},
 	}
 
 	for name, tcase := range cases {

acl/static_authorizer.go

@@ -257,6 +257,13 @@ func (s *staticAuthorizer) ServiceReadAll(*AuthorizerContext) EnforcementDecisio
 	return Deny
 }
 
+func (s *staticAuthorizer) ServiceReadPrefix(string, *AuthorizerContext) EnforcementDecision {
+	if s.defaultAllow {
+		return Allow
+	}
+	return Deny
+}
+
 func (s *staticAuthorizer) ServiceWrite(string, *AuthorizerContext) EnforcementDecision {
 	if s.defaultAllow {
 		return Allow

internal/catalog/internal/testhelpers/acl_hooks_test_helpers.go

@@ -69,6 +69,30 @@ func RunWorkloadSelectingTypeACLsTests[T WorkloadSelecting](t *testing.T, typ *p
 			WriteOK: resourcetest.ALLOW,
 			ListOK:  resourcetest.DEFAULT,
 		},
+		"service test write with multiple named selectors": {
+			Rules:   `service "test" { policy = "write" } service "workload1" { policy = "read" } service "workload2" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Names: []string{"workload1", "workload2"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with multiple named selectors and insufficient policy": {
+			Rules:   `service "test" { policy = "write" } service "workload1" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Names: []string{"workload1", "workload2"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with multiple named selectors and prefixed policy": {
+			Rules:   `service "test" { policy = "write" } service_prefix "workload" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Names: []string{"workload1", "workload2"}}),
+			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-"}}),
@@ -77,7 +101,7 @@ func RunWorkloadSelectingTypeACLsTests[T WorkloadSelecting](t *testing.T, typ *p
 			WriteOK: resourcetest.ALLOW,
 			ListOK:  resourcetest.DEFAULT,
 		},
-		"service test write with prefixed selectors and a policy with more specific than the selector": {
+		"service test write with prefixed selectors and a policy with more specific prefix than the selector": {
 			Rules:   `service "test" { policy = "write" } service_prefix "workload-" { policy = "read" }`,
 			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"wor"}}),
 			Typ:     typ,
@@ -85,7 +109,8 @@ func RunWorkloadSelectingTypeACLsTests[T WorkloadSelecting](t *testing.T, typ *p
 			WriteOK: resourcetest.DENY,
 			ListOK:  resourcetest.DEFAULT,
 		},
-		"service test write with prefixed selectors and a policy with less specific than the selector": {
+
+		"service test write with prefixed selectors and a policy with less specific prefix than the selector": {
 			Rules:   `service "test" { policy = "write" } service_prefix "wor" { policy = "read" }`,
 			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload-"}}),
 			Typ:     typ,
@@ -93,13 +118,73 @@ func RunWorkloadSelectingTypeACLsTests[T WorkloadSelecting](t *testing.T, typ *p
 			WriteOK: resourcetest.ALLOW,
 			ListOK:  resourcetest.DEFAULT,
 		},
+		// Prefix-based selectors should not allow writes when a policy only allows
+		// to read a specific service from that selector.
 		"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.
+			Rules:   `service "test" { policy = "write" } service "workload" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with multiple prefixed selectors": {
+			Rules:   `service "test" { policy = "write" } service_prefix "workload" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload-1", "workload-2"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with multiple prefixed selectors and insufficient policy": {
+			Rules:   `service "test" { policy = "write" } service_prefix "workload-1" { policy = "read" }`,
+			Data:    getData(&pbcatalog.WorkloadSelector{Prefixes: []string{"workload-1", "workload-2"}}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with a mix of named and prefixed selectors and insufficient policy": {
+			Rules: `service "test" { policy = "write" } service_prefix "workload" { policy = "read" }`,
+			Data: getData(&pbcatalog.WorkloadSelector{
+				Prefixes: []string{"workload-1", "workload-2"},
+				Names:    []string{"other-1", "other-2"},
+			}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.DENY,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with a mix of named and prefixed selectors and prefixed policy": {
+			Rules: `service "test" { policy = "write" } service_prefix "workload" { policy = "read" } service_prefix "other" { policy = "read" }`,
+			Data: getData(&pbcatalog.WorkloadSelector{
+				Prefixes: []string{"workload-1", "workload-2"},
+				Names:    []string{"other-1", "other-2"},
+			}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with a mix of named and prefixed selectors and both prefixed and specific policy": {
+			Rules: `service "test" { policy = "write" } service_prefix "workload" { policy = "read" } service "other-1" { policy = "read" } service "other-2" { policy = "read" }`,
+			Data: getData(&pbcatalog.WorkloadSelector{
+				Prefixes: []string{"workload-1", "workload-2"},
+				Names:    []string{"other-1", "other-2"},
+			}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
+			WriteOK: resourcetest.ALLOW,
+			ListOK:  resourcetest.DEFAULT,
+		},
+		"service test write with a mix of named and prefixed selectors and wildcard service read policy": {
+			Rules: `service "test" { policy = "write" } service_prefix "" { policy = "read" }`,
+			Data: getData(&pbcatalog.WorkloadSelector{
+				Prefixes: []string{"workload-1", "workload-2"},
+				Names:    []string{"other-1", "other-2"},
+			}),
+			Typ:     typ,
+			ReadOK:  resourcetest.ALLOW,
 			WriteOK: resourcetest.ALLOW,
 			ListOK:  resourcetest.DEFAULT,
 		},

internal/catalog/internal/types/acl_hooks.go

@@ -38,7 +38,7 @@ func aclWriteHookResourceWithWorkloadSelector[T WorkloadSelecting](authorizer ac
 	}
 
 	for _, prefix := range decodedService.GetData().GetWorkloads().GetPrefixes() {
-		err = authorizer.ToAllowAuthorizer().ServiceReadAllowed(prefix, authzContext)
+		err = authorizer.ToAllowAuthorizer().ServiceReadPrefixAllowed(prefix, authzContext)
 		if err != nil {
 			return err
 		}