TrafficPermissions: add ACL hooks (#19023)

* add ACL hooks

* add hooks for CTPs

acl/MockAuthorizer.go

@@ -256,6 +256,19 @@ func (m *MockAuthorizer) Snapshot(ctx *AuthorizerContext) EnforcementDecision {
 	return ret.Get(0).(EnforcementDecision)
 }
 
+// TrafficPermissionsRead determines if specific traffic permissions can be read.
+func (m *MockAuthorizer) TrafficPermissionsRead(segment string, ctx *AuthorizerContext) EnforcementDecision {
+	ret := m.Called(segment, ctx)
+	return ret.Get(0).(EnforcementDecision)
+}
+
+// TrafficPermissionsWrite determines if specific traffic permissions can be
+// created, modified, or deleted.
+func (m *MockAuthorizer) TrafficPermissionsWrite(segment string, ctx *AuthorizerContext) EnforcementDecision {
+	ret := m.Called(segment, ctx)
+	return ret.Get(0).(EnforcementDecision)
+}
+
 func (p *MockAuthorizer) ToAllowAuthorizer() AllowAuthorizer {
 	return AllowAuthorizer{Authorizer: p}
 }

acl/acl_test.go

@@ -164,6 +164,14 @@ func checkAllowSnapshot(t *testing.T, authz Authorizer, prefix string, entCtx *A
 	require.Equal(t, Allow, authz.Snapshot(entCtx))
 }
 
+func checkAllowTrafficPermissionsRead(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
+	require.Equal(t, Allow, authz.TrafficPermissionsRead(prefix, entCtx))
+}
+
+func checkAllowTrafficPermissionsWrite(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
+	require.Equal(t, Allow, authz.TrafficPermissionsWrite(prefix, entCtx))
+}
+
 func checkDenyACLRead(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
 	require.Equal(t, Deny, authz.ACLRead(entCtx))
 }
@@ -312,6 +320,14 @@ func checkDenySnapshot(t *testing.T, authz Authorizer, prefix string, entCtx *Au
 	require.Equal(t, Deny, authz.Snapshot(entCtx))
 }
 
+func checkDenyTrafficPermissionsRead(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
+	require.Equal(t, Deny, authz.TrafficPermissionsRead(prefix, entCtx))
+}
+
+func checkDenyTrafficPermissionsWrite(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
+	require.Equal(t, Deny, authz.TrafficPermissionsWrite(prefix, entCtx))
+}
+
 func checkDefaultACLRead(t *testing.T, authz Authorizer, prefix string, entCtx *AuthorizerContext) {
 	require.Equal(t, Default, authz.ACLRead(entCtx))
 }
@@ -555,6 +571,8 @@ func TestACL(t *testing.T) {
 				{name: "AllowSessionRead", check: checkAllowSessionRead},
 				{name: "AllowSessionWrite", check: checkAllowSessionWrite},
 				{name: "DenySnapshot", check: checkDenySnapshot},
+				{name: "AllowTrafficPermissionsRead", check: checkAllowTrafficPermissionsRead},
+				{name: "AllowTrafficPermissionsWrite", check: checkAllowTrafficPermissionsWrite},
 			},
 		},
 		{
@@ -596,6 +614,8 @@ func TestACL(t *testing.T) {
 				{name: "AllowSessionRead", check: checkAllowSessionRead},
 				{name: "AllowSessionWrite", check: checkAllowSessionWrite},
 				{name: "AllowSnapshot", check: checkAllowSnapshot},
+				{name: "AllowTrafficPermissionsRead", check: checkAllowTrafficPermissionsRead},
+				{name: "AllowTrafficPermissionsWrite", check: checkAllowTrafficPermissionsWrite},
 			},
 		},
 		{
@@ -1074,26 +1094,26 @@ func TestACL(t *testing.T) {
 			checks: []aclCheck{
 				{name: "IdentityReadAllowed", prefix: "foo", check: checkAllowIdentityRead},
 				{name: "IdentityWriteAllowed", prefix: "foo", check: checkAllowIdentityWrite},
-				{name: "IntentionReadAllowed", prefix: "foo", check: checkAllowIntentionRead},
-				{name: "IntentionWriteAllowed", prefix: "foo", check: checkAllowIntentionWrite},
+				{name: "TrafficPermissionsReadAllowed", prefix: "foo", check: checkAllowTrafficPermissionsRead},
+				{name: "TrafficPermissionsWriteAllowed", prefix: "foo", check: checkAllowTrafficPermissionsWrite},
 				{name: "IdentityReadAllowed", prefix: "football", check: checkAllowIdentityRead},
 				{name: "IdentityWriteDenied", prefix: "football", check: checkDenyIdentityWrite},
-				{name: "IntentionReadAllowed", prefix: "football", check: checkAllowIntentionRead},
+				{name: "TrafficPermissionsReadAllowed", prefix: "football", check: checkAllowTrafficPermissionsRead},
 				// This might be surprising but omitting intention rule gives at most intention:read
 				// if we have identity:write perms. This matches services as well.
-				{name: "IntentionWriteDenied", prefix: "football", check: checkDenyIntentionWrite},
+				{name: "TrafficPermissionsWriteDenied", prefix: "football", check: checkDenyTrafficPermissionsWrite},
 				{name: "IdentityReadAllowed", prefix: "prefix", check: checkAllowIdentityRead},
 				{name: "IdentityWriteDenied", prefix: "prefix", check: checkDenyIdentityWrite},
-				{name: "IntentionReadAllowed", prefix: "prefix", check: checkAllowIntentionRead},
-				{name: "IntentionWriteDenied", prefix: "prefix", check: checkAllowIntentionWrite},
+				{name: "TrafficPermissionsReadAllowed", prefix: "prefix", check: checkAllowTrafficPermissionsRead},
+				{name: "TrafficPermissionsWriteDenied", prefix: "prefix", check: checkAllowTrafficPermissionsWrite},
 				{name: "IdentityReadDenied", prefix: "prefix-forbidden", check: checkDenyIdentityRead},
 				{name: "IdentityWriteDenied", prefix: "prefix-forbidden", check: checkDenyIdentityWrite},
-				{name: "IntentionReadDenied", prefix: "prefix-forbidden", check: checkDenyIntentionRead},
-				{name: "IntentionWriteDenied", prefix: "prefix-forbidden", check: checkDenyIntentionWrite},
+				{name: "TrafficPermissionsReadDenied", prefix: "prefix-forbidden", check: checkDenyTrafficPermissionsRead},
+				{name: "TrafficPermissionsWriteDenied", prefix: "prefix-forbidden", check: checkDenyTrafficPermissionsWrite},
 				{name: "IdentityReadAllowed", prefix: "foozball", check: checkAllowIdentityRead},
 				{name: "IdentityWriteAllowed", prefix: "foozball", check: checkAllowIdentityWrite},
-				{name: "IntentionReadAllowed", prefix: "foozball", check: checkAllowIntentionRead},
-				{name: "IntentionWriteDenied", prefix: "foozball", check: checkDenyIntentionWrite},
+				{name: "TrafficPermissionsReadAllowed", prefix: "foozball", check: checkAllowTrafficPermissionsRead},
+				{name: "TrafficPermissionsWriteDenied", prefix: "foozball", check: checkDenyTrafficPermissionsWrite},
 			},
 		},
 		{

acl/authorizer.go

@@ -188,6 +188,13 @@ type Authorizer interface {
 	// Snapshot checks for permission to take and restore snapshots.
 	Snapshot(*AuthorizerContext) EnforcementDecision
 
+	// TrafficPermissionsRead determines if specific traffic permissions can be read.
+	TrafficPermissionsRead(string, *AuthorizerContext) EnforcementDecision
+
+	// TrafficPermissionsWrite determines if specific traffic permissions can be
+	// created, modified, or deleted.
+	TrafficPermissionsWrite(string, *AuthorizerContext) EnforcementDecision
+
 	// Embedded Interface for Consul Enterprise specific ACL enforcement
 	enterpriseAuthorizer
 
@@ -315,6 +322,23 @@ func (a AllowAuthorizer) IntentionWriteAllowed(name string, ctx *AuthorizerConte
 	return nil
 }
 
+// TrafficPermissionsReadAllowed determines if specific traffic permissions can be read.
+func (a AllowAuthorizer) TrafficPermissionsReadAllowed(name string, ctx *AuthorizerContext) error {
+	if a.Authorizer.TrafficPermissionsRead(name, ctx) != Allow {
+		return PermissionDeniedByACL(a, ctx, ResourceIntention, AccessRead, name)
+	}
+	return nil
+}
+
+// TrafficPermissionsWriteAllowed determines if specific traffic permissions can be
+// created, modified, or deleted.
+func (a AllowAuthorizer) TrafficPermissionsWriteAllowed(name string, ctx *AuthorizerContext) error {
+	if a.Authorizer.TrafficPermissionsWrite(name, ctx) != Allow {
+		return PermissionDeniedByACL(a, ctx, ResourceIntention, AccessWrite, name)
+	}
+	return nil
+}
+
 // KeyListAllowed checks for permission to list keys under a prefix
 func (a AllowAuthorizer) KeyListAllowed(name string, ctx *AuthorizerContext) error {
 	if a.Authorizer.KeyList(name, ctx) != Allow {

acl/chained_authorizer.go

@@ -312,6 +312,21 @@ func (c *ChainedAuthorizer) Snapshot(entCtx *AuthorizerContext) EnforcementDecis
 	})
 }
 
+// TrafficPermissionsRead determines if specific traffic permissions can be read.
+func (c *ChainedAuthorizer) TrafficPermissionsRead(prefix string, entCtx *AuthorizerContext) EnforcementDecision {
+	return c.executeChain(func(authz Authorizer) EnforcementDecision {
+		return authz.TrafficPermissionsRead(prefix, entCtx)
+	})
+}
+
+// TrafficPermissionsWrite determines if specific traffic permissions can be
+// created, modified, or deleted.
+func (c *ChainedAuthorizer) TrafficPermissionsWrite(prefix string, entCtx *AuthorizerContext) EnforcementDecision {
+	return c.executeChain(func(authz Authorizer) EnforcementDecision {
+		return authz.TrafficPermissionsWrite(prefix, entCtx)
+	})
+}
+
 func (c *ChainedAuthorizer) ToAllowAuthorizer() AllowAuthorizer {
 	return AllowAuthorizer{Authorizer: c}
 }

acl/chained_authorizer_test.go

@@ -122,6 +122,12 @@ func (authz testAuthorizer) SessionWrite(string, *AuthorizerContext) Enforcement
 func (authz testAuthorizer) Snapshot(*AuthorizerContext) EnforcementDecision {
 	return EnforcementDecision(authz)
 }
+func (authz testAuthorizer) TrafficPermissionsRead(string, *AuthorizerContext) EnforcementDecision {
+	return EnforcementDecision(authz)
+}
+func (authz testAuthorizer) TrafficPermissionsWrite(string, *AuthorizerContext) EnforcementDecision {
+	return EnforcementDecision(authz)
+}
 
 func (authz testAuthorizer) ToAllowAuthorizer() AllowAuthorizer {
 	return AllowAuthorizer{Authorizer: &authz}

acl/policy_authorizer.go

@@ -20,6 +20,9 @@ type policyAuthorizer struct {
 	// intentionRules contains the service intention exact-match policies
 	intentionRules *radix.Tree
 
+	// trafficPermissionsRules contains the service intention exact-match policies
+	trafficPermissionsRules *radix.Tree
+
 	// keyRules contains the key exact-match policies
 	keyRules *radix.Tree
 
@@ -199,7 +202,7 @@ func (p *policyAuthorizer) loadRules(policy *PolicyRules) error {
 			}
 		}
 
-		if err := insertPolicyIntoRadix(id.Name, intention, &id.EnterpriseRule, p.intentionRules, false); err != nil {
+		if err := insertPolicyIntoRadix(id.Name, intention, &id.EnterpriseRule, p.trafficPermissionsRules, false); err != nil {
 			return err
 		}
 	}
@@ -220,7 +223,7 @@ func (p *policyAuthorizer) loadRules(policy *PolicyRules) error {
 			}
 		}
 
-		if err := insertPolicyIntoRadix(id.Name, intention, &id.EnterpriseRule, p.intentionRules, true); err != nil {
+		if err := insertPolicyIntoRadix(id.Name, intention, &id.EnterpriseRule, p.trafficPermissionsRules, true); err != nil {
 			return err
 		}
 	}
@@ -396,6 +399,7 @@ func newPolicyAuthorizerFromRules(rules *PolicyRules, ent *Config) (*policyAutho
 		agentRules:              radix.New(),
 		identityRules:           radix.New(),
 		intentionRules:          radix.New(),
+		trafficPermissionsRules: radix.New(),
 		keyRules:                radix.New(),
 		nodeRules:               radix.New(),
 		serviceRules:            radix.New(),
@@ -608,8 +612,7 @@ func (p *policyAuthorizer) IntentionDefaultAllow(_ *AuthorizerContext) Enforceme
 	return Default
 }
 
-// IntentionRead checks if writing (creating, updating, or deleting) of an
-// intention is allowed.
+// IntentionRead checks if reading an intention is allowed.
 func (p *policyAuthorizer) IntentionRead(prefix string, _ *AuthorizerContext) EnforcementDecision {
 	if prefix == "*" {
 		return p.anyAllowed(p.intentionRules, AccessRead)
@@ -634,6 +637,31 @@ func (p *policyAuthorizer) IntentionWrite(prefix string, _ *AuthorizerContext) E
 	return Default
 }
 
+// TrafficPermissionsRead checks if reading of traffic permissions is allowed.
+func (p *policyAuthorizer) TrafficPermissionsRead(prefix string, _ *AuthorizerContext) EnforcementDecision {
+	if prefix == "*" {
+		return p.anyAllowed(p.trafficPermissionsRules, AccessRead)
+	}
+
+	if rule, ok := getPolicy(prefix, p.trafficPermissionsRules); ok {
+		return enforce(rule.access, AccessRead)
+	}
+	return Default
+}
+
+// TrafficPermissionsWrite checks if writing (creating, updating, or deleting) of traffic
+// permissions is allowed.
+func (p *policyAuthorizer) TrafficPermissionsWrite(prefix string, _ *AuthorizerContext) EnforcementDecision {
+	if prefix == "*" {
+		return p.allAllowed(p.trafficPermissionsRules, AccessWrite)
+	}
+
+	if rule, ok := getPolicy(prefix, p.trafficPermissionsRules); ok {
+		return enforce(rule.access, AccessWrite)
+	}
+	return Default
+}
+
 // KeyRead returns if a key is allowed to be read
 func (p *policyAuthorizer) KeyRead(key string, _ *AuthorizerContext) EnforcementDecision {
 	if rule, ok := getPolicy(key, p.keyRules); ok {

acl/static_authorizer.go

@@ -292,6 +292,20 @@ func (s *staticAuthorizer) Snapshot(_ *AuthorizerContext) EnforcementDecision {
 	return Deny
 }
 
+func (s *staticAuthorizer) TrafficPermissionsRead(string, *AuthorizerContext) EnforcementDecision {
+	if s.defaultAllow {
+		return Allow
+	}
+	return Deny
+}
+
+func (s *staticAuthorizer) TrafficPermissionsWrite(string, *AuthorizerContext) EnforcementDecision {
+	if s.defaultAllow {
+		return Allow
+	}
+	return Deny
+}
+
 func (s *staticAuthorizer) ToAllowAuthorizer() AllowAuthorizer {
 	return AllowAuthorizer{Authorizer: s}
 }

internal/auth/internal/types/computed_traffic_permissions.go

@@ -6,6 +6,7 @@ package types
 import (
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -15,8 +16,13 @@ func RegisterComputedTrafficPermission(r resource.Registry) {
 	r.Register(resource.Registration{
 		Type:  pbauth.ComputedTrafficPermissionsType,
 		Proto: &pbauth.ComputedTrafficPermissions{},
-		Scope:    resource.ScopeNamespace,
+		ACLs: &resource.ACLHooks{
+			Read:  aclReadHookComputedTrafficPermissions,
+			Write: aclWriteHookComputedTrafficPermissions,
+			List:  aclListHookComputedTrafficPermissions,
+		},
 		Validate: ValidateComputedTrafficPermissions,
+		Scope:    resource.ScopeNamespace,
 	})
 }
 
@@ -57,3 +63,17 @@ func ValidateComputedTrafficPermissions(res *pbresource.Resource) error {
 
 	return merr
 }
+
+func aclReadHookComputedTrafficPermissions(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) error {
+	return authorizer.ToAllowAuthorizer().TrafficPermissionsReadAllowed(id.Name, authzContext)
+}
+
+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/computed_traffic_permissions_test.go

@@ -8,9 +8,12 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
+	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
@@ -45,3 +48,129 @@ func TestValidateComputedTrafficPermissions_Permissions(t *testing.T) {
 		})
 	}
 }
+
+func TestComputedTrafficPermissionsACLs(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(pbauth.ComputedTrafficPermissionsType)
+	require.True(t, ok)
+
+	run := func(t *testing.T, tc testcase) {
+		ctpData := &pbauth.ComputedTrafficPermissions{}
+		res := resourcetest.Resource(pbauth.ComputedTrafficPermissionsType, "wi1").
+			WithTenancy(resource.DefaultNamespacedTenancy()).
+			WithData(t, ctpData).
+			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, 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)
+		})
+	}
+
+	cases := map[string]testcase{
+		"no rules": {
+			rules:   ``,
+			readOK:  DENY,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 read, no intentions": {
+			rules:   `identity "wi1" { policy = "read" }`,
+			readOK:  ALLOW,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 read, deny intentions": {
+			rules:   `identity "wi1" { policy = "read", intentions = "deny" }`,
+			readOK:  DENY,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 read, intentions read": {
+			rules:   `identity "wi1" { policy = "read", intentions = "read" }`,
+			readOK:  ALLOW,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, write intentions": {
+			rules:   `identity "wi1" { policy = "read", intentions = "write" }`,
+			readOK:  ALLOW,
+			writeOK: ALLOW,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, deny intentions": {
+			rules:   `identity "wi1" { policy = "write", intentions = "deny" }`,
+			readOK:  DENY,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, intentions read": {
+			rules:   `identity "wi1" { policy = "write", intentions = "read" }`,
+			readOK:  ALLOW,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, intentions write": {
+			rules:   `identity "wi1" { policy = "write", intentions = "write" }`,
+			readOK:  ALLOW,
+			writeOK: ALLOW,
+			listOK:  DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}

internal/auth/internal/types/traffic_permissions.go

@@ -6,6 +6,7 @@ package types
 import (
 	"github.com/hashicorp/go-multierror"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
 	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
 	"github.com/hashicorp/consul/proto-public/pbresource"
@@ -15,9 +16,14 @@ func RegisterTrafficPermissions(r resource.Registry) {
 	r.Register(resource.Registration{
 		Type:  pbauth.TrafficPermissionsType,
 		Proto: &pbauth.TrafficPermissions{},
-		Scope:    resource.ScopeNamespace,
+		ACLs: &resource.ACLHooks{
+			Read:  aclReadHookTrafficPermissions,
+			Write: aclWriteHookTrafficPermissions,
+			List:  aclListHookTrafficPermissions,
+		},
 		Validate: ValidateTrafficPermissions,
 		Mutate:   MutateTrafficPermissions,
+		Scope:    resource.ScopeNamespace,
 	})
 }
 
@@ -264,3 +270,37 @@ func sourceHasIncompatibleTenancies(src pbauth.SourceToSpiffe) bool {
 func isLocalPeer(p string) bool {
 	return p == "local" || p == ""
 }
+
+func aclReadHookTrafficPermissions(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, _ *pbresource.ID, res *pbresource.Resource) error {
+	if res == nil {
+		return resource.ErrNeedData
+	}
+	return authorizeDestination(res, func(dest string) error {
+		return authorizer.ToAllowAuthorizer().TrafficPermissionsReadAllowed(dest, authzContext)
+	})
+}
+
+func aclWriteHookTrafficPermissions(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, res *pbresource.Resource) error {
+	return authorizeDestination(res, func(dest string) error {
+		return authorizer.ToAllowAuthorizer().TrafficPermissionsWriteAllowed(dest, authzContext)
+	})
+}
+
+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 {
+		return err
+	}
+	// Check intention:x permissions for identity
+	err = intentionAllowed(tp.Data.Destination.IdentityName)
+	if err != nil {
+		return err
+	}
+	return nil
+}

internal/auth/internal/types/traffic_permissions_test.go

@@ -8,6 +8,8 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/resourcetest"
 	pbauth "github.com/hashicorp/consul/proto-public/pbauth/v2beta1"
@@ -644,3 +646,132 @@ func TestMutateTrafficPermissions(t *testing.T) {
 		})
 	}
 }
+
+func TestTrafficPermissionsACLs(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(pbauth.TrafficPermissionsType)
+	require.True(t, ok)
+
+	run := func(t *testing.T, tc testcase) {
+		tpData := &pbauth.TrafficPermissions{
+			Destination: &pbauth.Destination{IdentityName: "wi1"},
+			Action:      pbauth.Action_ACTION_ALLOW,
+		}
+		res := resourcetest.Resource(pbauth.TrafficPermissionsType, "tp1").
+			WithTenancy(resource.DefaultNamespacedTenancy()).
+			WithData(t, tpData).
+			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, 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)
+		})
+	}
+
+	cases := map[string]testcase{
+		"no rules": {
+			rules:   ``,
+			readOK:  DENY,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 read, no intentions": {
+			rules:   `identity "wi1" { policy = "read" }`,
+			readOK:  ALLOW,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 read, deny intentions": {
+			rules:   `identity "wi1" { policy = "read", intentions = "deny" }`,
+			readOK:  DENY,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 read, intentions read": {
+			rules:   `identity "wi1" { policy = "read", intentions = "read" }`,
+			readOK:  ALLOW,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, write intentions": {
+			rules:   `identity "wi1" { policy = "read", intentions = "write" }`,
+			readOK:  ALLOW,
+			writeOK: ALLOW,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, deny intentions": {
+			rules:   `identity "wi1" { policy = "write", intentions = "deny" }`,
+			readOK:  DENY,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, intentions read": {
+			rules:   `identity "wi1" { policy = "write", intentions = "read" }`,
+			readOK:  ALLOW,
+			writeOK: DENY,
+			listOK:  DEFAULT,
+		},
+		"workload identity w1 write, intentions write": {
+			rules:   `identity "wi1" { policy = "write", intentions = "write" }`,
+			readOK:  ALLOW,
+			writeOK: ALLOW,
+			listOK:  DEFAULT,
+		},
+	}
+
+	for name, tc := range cases {
+		t.Run(name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}