sync more acl enforcement

sync w ent at 32756f7

Signed-off-by: acpana <8968914+acpana@users.noreply.github.com>

agent/consul/internal_endpoint.go

@@ -595,17 +595,15 @@ func (m *Internal) ExportedPeeredServices(args *structs.DCSpecificRequest, reply
 		return err
 	}
 
-	authz, err := m.srv.ResolveTokenAndDefaultMeta(args.Token, &args.EnterpriseMeta, nil)
+	var authzCtx acl.AuthorizerContext
+	authz, err := m.srv.ResolveTokenAndDefaultMeta(args.Token, &args.EnterpriseMeta, &authzCtx)
 	if err != nil {
 		return err
 	}
-
 	if err := m.srv.validateEnterpriseRequest(&args.EnterpriseMeta, false); err != nil {
 		return err
 	}
 
-	// TODO(peering): acls: mesh gateway needs appropriate wildcard service:read
-
 	return m.srv.blockingQuery(
 		&args.QueryOptions,
 		&reply.QueryMeta,
@@ -632,11 +630,14 @@ func (m *Internal) PeeredUpstreams(args *structs.PartitionSpecificRequest, reply
 		return err
 	}
 
-	// TODO(peering): ACL for filtering
-	// authz, err := m.srv.ResolveTokenAndDefaultMeta(args.Token, &args.EnterpriseMeta, nil)
-	// if err != nil {
-	// 	return err
-	// }
+	var authzCtx acl.AuthorizerContext
+	authz, err := m.srv.ResolveTokenAndDefaultMeta(args.Token, &args.EnterpriseMeta, &authzCtx)
+	if err != nil {
+		return err
+	}
+	if err := authz.ToAllowAuthorizer().ServiceWriteAnyAllowed(&authzCtx); err != nil {
+		return err
+	}
 
 	if err := m.srv.validateEnterpriseRequest(&args.EnterpriseMeta, false); err != nil {
 		return err
@@ -657,9 +658,6 @@ func (m *Internal) PeeredUpstreams(args *structs.PartitionSpecificRequest, reply
 			}
 
 			reply.Index, reply.Services = index, result
-
-			// TODO(peering): low priority: consider ACL filtering
-			// m.srv.filterACLWithAuthorizer(authz, reply)
 			return nil
 		})
 }

agent/consul/internal_endpoint_test.go

@@ -3,6 +3,7 @@ package consul
 import (
 	"encoding/base64"
 	"fmt"
+	"math/rand"
 	"os"
 	"strings"
 	"testing"
@@ -3291,3 +3292,193 @@ func TestInternal_ServiceGatewayService_Terminating_Destination(t *testing.T) {
 	assert.Len(t, nodes, 1)
 	assert.Equal(t, expect, nodes)
 }
+
+func TestInternal_ExportedPeeredServices_ACLEnforcement(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+	t.Parallel()
+
+	_, s := testServerWithConfig(t, testServerACLConfig)
+	codec := rpcClient(t, s)
+
+	require.NoError(t, s.fsm.State().PeeringWrite(1, &pbpeering.Peering{
+		ID:   testUUID(),
+		Name: "peer-1",
+	}))
+	require.NoError(t, s.fsm.State().PeeringWrite(1, &pbpeering.Peering{
+		ID:   testUUID(),
+		Name: "peer-2",
+	}))
+	require.NoError(t, s.fsm.State().EnsureConfigEntry(1, &structs.ExportedServicesConfigEntry{
+		Name: "default",
+		Services: []structs.ExportedService{
+			{
+				Name: "web",
+				Consumers: []structs.ServiceConsumer{
+					{PeerName: "peer-1"},
+				},
+			},
+			{
+				Name: "db",
+				Consumers: []structs.ServiceConsumer{
+					{PeerName: "peer-2"},
+				},
+			},
+			{
+				Name: "api",
+				Consumers: []structs.ServiceConsumer{
+					{PeerName: "peer-1"},
+				},
+			},
+		},
+	}))
+
+	type testcase struct {
+		name      string
+		token     string
+		expect    map[string]structs.ServiceList
+		expectErr string
+	}
+	run := func(t *testing.T, tc testcase) {
+		var out *structs.IndexedExportedServiceList
+		req := structs.DCSpecificRequest{
+			Datacenter:   "dc1",
+			QueryOptions: structs.QueryOptions{Token: tc.token},
+		}
+		err := msgpackrpc.CallWithCodec(codec, "Internal.ExportedPeeredServices", &req, &out)
+
+		if tc.expectErr != "" {
+			require.Error(t, err)
+			require.Contains(t, err.Error(), tc.expectErr)
+			require.Nil(t, out)
+			return
+		}
+
+		require.NoError(t, err)
+
+		require.Len(t, out.Services, len(tc.expect))
+		for k, v := range tc.expect {
+			require.ElementsMatch(t, v, out.Services[k])
+		}
+	}
+	tcs := []testcase{
+		{
+			name: "can read all",
+			token: tokenWithRules(t, codec, TestDefaultInitialManagementToken,
+				`
+			service_prefix "" {
+				policy = "read"
+			}
+			`),
+			expect: map[string]structs.ServiceList{
+				"peer-1": {
+					structs.NewServiceName("api", nil),
+					structs.NewServiceName("web", nil),
+				},
+				"peer-2": {
+					structs.NewServiceName("db", nil),
+				},
+			},
+		},
+		{
+			name: "filtered",
+			token: tokenWithRules(t, codec, TestDefaultInitialManagementToken,
+				`
+			service "web" { policy = "read" }
+			service "api" { policy = "read" }
+			service "db"  { policy = "deny" }	
+			`),
+			expect: map[string]structs.ServiceList{
+				"peer-1": {
+					structs.NewServiceName("api", nil),
+					structs.NewServiceName("web", nil),
+				},
+			},
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func tokenWithRules(t *testing.T, codec rpc.ClientCodec, mgmtToken, rules string) string {
+	t.Helper()
+
+	var tok *structs.ACLToken
+	var err error
+	retry.Run(t, func(r *retry.R) {
+		tok, err = upsertTestTokenWithPolicyRules(codec, mgmtToken, "dc1", rules)
+		require.NoError(r, err)
+	})
+	return tok.SecretID
+}
+
+func TestInternal_PeeredUpstreams_ACLEnforcement(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+	t.Parallel()
+
+	_, s := testServerWithConfig(t, testServerACLConfig)
+	codec := rpcClient(t, s)
+
+	type testcase struct {
+		name      string
+		token     string
+		expectErr string
+	}
+	run := func(t *testing.T, tc testcase) {
+		var out *structs.IndexedPeeredServiceList
+
+		req := structs.PartitionSpecificRequest{
+			Datacenter:   "dc1",
+			QueryOptions: structs.QueryOptions{Token: tc.token},
+		}
+		err := msgpackrpc.CallWithCodec(codec, "Internal.PeeredUpstreams", &req, &out)
+
+		if tc.expectErr != "" {
+			require.Error(t, err)
+			require.Contains(t, err.Error(), tc.expectErr)
+			require.Nil(t, out)
+		} else {
+			require.NoError(t, err)
+		}
+	}
+	tcs := []testcase{
+		{
+			name: "can write all",
+			token: tokenWithRules(t, codec, TestDefaultInitialManagementToken, `
+			service_prefix "" {
+				policy = "write"
+			}
+			`),
+		},
+		{
+			name:      "can't write",
+			token:     tokenWithRules(t, codec, TestDefaultInitialManagementToken, ``),
+			expectErr: "lacks permission 'service:write' on \"any service\"",
+		},
+	}
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+}
+
+func testUUID() string {
+	buf := make([]byte, 16)
+	if _, err := rand.Read(buf); err != nil {
+		panic(fmt.Errorf("failed to read random bytes: %v", err))
+	}
+
+	return fmt.Sprintf("%08x-%04x-%04x-%04x-%12x",
+		buf[0:4],
+		buf[4:6],
+		buf[6:8],
+		buf[8:10],
+		buf[10:16])
+}

agent/proxycfg-glue/exported_peered_services.go

@@ -3,14 +3,15 @@ package proxycfgglue
 import (
 	"context"
 
+	"github.com/hashicorp/consul/agent/structs/aclfilter"
 	"github.com/hashicorp/go-memdb"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/cache"
 	cachetype "github.com/hashicorp/consul/agent/cache-types"
 	"github.com/hashicorp/consul/agent/consul/watch"
 	"github.com/hashicorp/consul/agent/proxycfg"
 	"github.com/hashicorp/consul/agent/structs"
-	"github.com/hashicorp/consul/agent/structs/aclfilter"
 )
 
 // CacheExportedPeeredServices satisfies the proxycfg.ExportedPeeredServices
@@ -33,8 +34,8 @@ type serverExportedPeeredServices struct {
 func (s *serverExportedPeeredServices) Notify(ctx context.Context, req *structs.DCSpecificRequest, correlationID string, ch chan<- proxycfg.UpdateEvent) error {
 	return watch.ServerLocalNotify(ctx, correlationID, s.deps.GetStore,
 		func(ws memdb.WatchSet, store Store) (uint64, *structs.IndexedExportedServiceList, error) {
-			// TODO(peering): acls: mesh gateway needs appropriate wildcard service:read
-			authz, err := s.deps.ACLResolver.ResolveTokenAndDefaultMeta(req.Token, &req.EnterpriseMeta, nil)
+			var authzCtx acl.AuthorizerContext
+			authz, err := s.deps.ACLResolver.ResolveTokenAndDefaultMeta(req.Token, &req.EnterpriseMeta, &authzCtx)
 			if err != nil {
 				return 0, nil, err
 			}

agent/proxycfg-glue/exported_peered_services_test.go

@@ -4,13 +4,12 @@ import (
 	"context"
 	"testing"
 
-	"github.com/stretchr/testify/require"
-
 	"github.com/hashicorp/consul/agent/consul/state"
 	"github.com/hashicorp/consul/agent/proxycfg"
 	"github.com/hashicorp/consul/agent/structs"
 	"github.com/hashicorp/consul/proto/pbpeering"
 	"github.com/hashicorp/consul/sdk/testutil"
+	"github.com/stretchr/testify/require"
 )
 
 func TestServerExportedPeeredServices(t *testing.T) {

agent/proxycfg-glue/peered_upstreams.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/hashicorp/go-memdb"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/cache"
 	cachetype "github.com/hashicorp/consul/agent/cache-types"
 	"github.com/hashicorp/consul/agent/consul/watch"
@@ -29,9 +30,17 @@ type serverPeeredUpstreams struct {
 }
 
 func (s *serverPeeredUpstreams) Notify(ctx context.Context, req *structs.PartitionSpecificRequest, correlationID string, ch chan<- proxycfg.UpdateEvent) error {
-	// TODO(peering): ACL filtering.
 	return watch.ServerLocalNotify(ctx, correlationID, s.deps.GetStore,
 		func(ws memdb.WatchSet, store Store) (uint64, *structs.IndexedPeeredServiceList, error) {
+			var authzCtx acl.AuthorizerContext
+			authz, err := s.deps.ACLResolver.ResolveTokenAndDefaultMeta(req.Token, &req.EnterpriseMeta, &authzCtx)
+			if err != nil {
+				return 0, nil, err
+			}
+			if err := authz.ToAllowAuthorizer().ServiceWriteAnyAllowed(&authzCtx); err != nil {
+				return 0, nil, err
+			}
+
 			index, vips, err := store.VirtualIPsForAllImportedServices(ws, req.EnterpriseMeta)
 			if err != nil {
 				return 0, nil, err

agent/proxycfg-glue/peered_upstreams_test.go

@@ -14,6 +14,28 @@ import (
 	"github.com/hashicorp/consul/sdk/testutil"
 )
 
+func registerService(t *testing.T, index uint64, peerName, serviceName, nodeName string, store *state.Store) {
+	require.NoError(t, store.EnsureRegistration(index, &structs.RegisterRequest{
+		Node:           nodeName,
+		Service:        &structs.NodeService{Service: serviceName, ID: serviceName},
+		PeerName:       peerName,
+		EnterpriseMeta: *acl.DefaultEnterpriseMeta(),
+	}))
+
+	require.NoError(t, store.EnsureRegistration(index, &structs.RegisterRequest{
+		Node: nodeName,
+		Service: &structs.NodeService{
+			Service: fmt.Sprintf("%s-proxy", serviceName),
+			Kind:    structs.ServiceKindConnectProxy,
+			Proxy: structs.ConnectProxyConfig{
+				DestinationServiceName: serviceName,
+			},
+		},
+		PeerName:       peerName,
+		EnterpriseMeta: *acl.DefaultEnterpriseMeta(),
+	}))
+}
+
 func TestServerPeeredUpstreams(t *testing.T) {
 	const (
 		index    uint64 = 123
@@ -26,33 +48,12 @@ func TestServerPeeredUpstreams(t *testing.T) {
 	store := state.NewStateStore(nil)
 	enableVirtualIPs(t, store)
 
-	registerService := func(t *testing.T, index uint64, peerName, serviceName string) {
-		require.NoError(t, store.EnsureRegistration(index, &structs.RegisterRequest{
-			Node:           nodeName,
-			Service:        &structs.NodeService{Service: serviceName, ID: serviceName},
-			PeerName:       peerName,
-			EnterpriseMeta: *acl.DefaultEnterpriseMeta(),
-		}))
-
-		require.NoError(t, store.EnsureRegistration(index, &structs.RegisterRequest{
-			Node: nodeName,
-			Service: &structs.NodeService{
-				Service: fmt.Sprintf("%s-proxy", serviceName),
-				Kind:    structs.ServiceKindConnectProxy,
-				Proxy: structs.ConnectProxyConfig{
-					DestinationServiceName: serviceName,
-				},
-			},
-			PeerName:       peerName,
-			EnterpriseMeta: *acl.DefaultEnterpriseMeta(),
-		}))
-	}
-
-	registerService(t, index, "peer-1", "web")
+	registerService(t, index, "peer-1", "web", nodeName, store)
 
 	eventCh := make(chan proxycfg.UpdateEvent)
 	dataSource := ServerPeeredUpstreams(ServerDataSourceDeps{
-		GetStore: func() Store { return store },
+		GetStore:    func() Store { return store },
+		ACLResolver: newStaticResolver(acl.ManageAll()),
 	})
 	require.NoError(t, dataSource.Notify(ctx, &structs.PartitionSpecificRequest{EnterpriseMeta: *acl.DefaultEnterpriseMeta()}, "", eventCh))
 
@@ -64,7 +65,7 @@ func TestServerPeeredUpstreams(t *testing.T) {
 	})
 
 	testutil.RunStep(t, "register another service", func(t *testing.T) {
-		registerService(t, index+1, "peer-2", "db")
+		registerService(t, index+1, "peer-2", "db", nodeName, store)
 
 		result := getEventResult[*structs.IndexedPeeredServiceList](t, eventCh)
 		require.Len(t, result.Services, 2)
@@ -78,6 +79,52 @@ func TestServerPeeredUpstreams(t *testing.T) {
 	})
 }
 
+func TestServerPeeredUpstreams_ACLEnforcement(t *testing.T) {
+	const (
+		index    uint64 = 123
+		nodeName        = "node-1"
+	)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	store := state.NewStateStore(nil)
+	enableVirtualIPs(t, store)
+
+	registerService(t, index, "peer-1", "web", nodeName, store)
+
+	testutil.RunStep(t, "read web", func(t *testing.T) {
+		authz := policyAuthorizer(t, `
+		service "web" { policy = "write" }`)
+
+		eventCh := make(chan proxycfg.UpdateEvent)
+		dataSource := ServerPeeredUpstreams(ServerDataSourceDeps{
+			GetStore:    func() Store { return store },
+			ACLResolver: newStaticResolver(authz),
+		})
+		require.NoError(t, dataSource.Notify(ctx, &structs.PartitionSpecificRequest{EnterpriseMeta: *acl.DefaultEnterpriseMeta()}, "", eventCh))
+
+		result := getEventResult[*structs.IndexedPeeredServiceList](t, eventCh)
+		require.Len(t, result.Services, 1)
+		require.Equal(t, "peer-1", result.Services[0].Peer)
+		require.Equal(t, "web", result.Services[0].ServiceName.Name)
+	})
+
+	testutil.RunStep(t, "can't read web", func(t *testing.T) {
+		authz := policyAuthorizer(t, ``)
+
+		eventCh := make(chan proxycfg.UpdateEvent)
+		dataSource := ServerPeeredUpstreams(ServerDataSourceDeps{
+			GetStore:    func() Store { return store },
+			ACLResolver: newStaticResolver(authz),
+		})
+		require.NoError(t, dataSource.Notify(ctx, &structs.PartitionSpecificRequest{EnterpriseMeta: *acl.DefaultEnterpriseMeta()}, "", eventCh))
+
+		err := getEventError(t, eventCh)
+		require.Contains(t, err.Error(), "lacks permission 'service:write' on \"any service\"")
+	})
+}
+
 func enableVirtualIPs(t *testing.T, store *state.Store) {
 	t.Helper()
 

agent/proxycfg-glue/trust_bundle.go

@@ -33,9 +33,19 @@ type serverTrustBundle struct {
 }
 
 func (s *serverTrustBundle) Notify(ctx context.Context, req *cachetype.TrustBundleReadRequest, correlationID string, ch chan<- proxycfg.UpdateEvent) error {
-	// TODO(peering): ACL check.
+	entMeta := structs.NodeEnterpriseMetaInPartition(req.Request.Partition)
+
 	return watch.ServerLocalNotify(ctx, correlationID, s.deps.GetStore,
 		func(ws memdb.WatchSet, store Store) (uint64, *pbpeering.TrustBundleReadResponse, error) {
+			var authzCtx acl.AuthorizerContext
+			authz, err := s.deps.ACLResolver.ResolveTokenAndDefaultMeta(req.Token, entMeta, &authzCtx)
+			if err != nil {
+				return 0, nil, err
+			}
+			if err := authz.ToAllowAuthorizer().ServiceWriteAnyAllowed(&authzCtx); err != nil {
+				return 0, nil, err
+			}
+
 			index, bundle, err := store.PeeringTrustBundleRead(ws, state.Query{
 				Value:          req.Request.Name,
 				EnterpriseMeta: *structs.NodeEnterpriseMetaInPartition(req.Request.Partition),
@@ -71,13 +81,20 @@ type serverTrustBundleList struct {
 func (s *serverTrustBundleList) Notify(ctx context.Context, req *cachetype.TrustBundleListRequest, correlationID string, ch chan<- proxycfg.UpdateEvent) error {
 	entMeta := acl.NewEnterpriseMetaWithPartition(req.Request.Partition, req.Request.Namespace)
 
-	// TODO(peering): ACL check.
 	return watch.ServerLocalNotify(ctx, correlationID, s.deps.GetStore,
 		func(ws memdb.WatchSet, store Store) (uint64, *pbpeering.TrustBundleListByServiceResponse, error) {
+			var authzCtx acl.AuthorizerContext
+			authz, err := s.deps.ACLResolver.ResolveTokenAndDefaultMeta(req.Token, &entMeta, &authzCtx)
+			if err != nil {
+				return 0, nil, err
+			}
+			if err := authz.ToAllowAuthorizer().ServiceWriteAllowed(req.Request.ServiceName, &authzCtx); err != nil {
+				return 0, nil, err
+			}
+
 			var (
 				index   uint64
 				bundles []*pbpeering.PeeringTrustBundle
-				err     error
 			)
 			switch {
 			case req.Request.Kind == string(structs.ServiceKindMeshGateway):

agent/proxycfg-glue/trust_bundle_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	"github.com/hashicorp/consul/acl"
 	cachetype "github.com/hashicorp/consul/agent/cache-types"
 	"github.com/stretchr/testify/require"
 
@@ -29,7 +30,8 @@ func TestServerTrustBundle(t *testing.T) {
 	}))
 
 	dataSource := ServerTrustBundle(ServerDataSourceDeps{
-		GetStore: func() Store { return store },
+		GetStore:    func() Store { return store },
+		ACLResolver: newStaticResolver(acl.ManageAll()),
 	})
 
 	eventCh := make(chan proxycfg.UpdateEvent)
@@ -56,6 +58,59 @@ func TestServerTrustBundle(t *testing.T) {
 	})
 }
 
+func TestServerTrustBundle_ACLEnforcement(t *testing.T) {
+	const (
+		index    uint64 = 123
+		peerName        = "peer1"
+	)
+
+	store := state.NewStateStore(nil)
+
+	require.NoError(t, store.PeeringTrustBundleWrite(index, &pbpeering.PeeringTrustBundle{
+		PeerName:    peerName,
+		TrustDomain: "before.com",
+	}))
+
+	testutil.RunStep(t, "can read", func(t *testing.T) {
+		authz := policyAuthorizer(t, `
+		service "web" { policy = "write" }`)
+		dataSource := ServerTrustBundle(ServerDataSourceDeps{
+			GetStore:    func() Store { return store },
+			ACLResolver: newStaticResolver(authz),
+		})
+
+		eventCh := make(chan proxycfg.UpdateEvent)
+		err := dataSource.Notify(context.Background(), &cachetype.TrustBundleReadRequest{
+			Request: &pbpeering.TrustBundleReadRequest{
+				Name: peerName,
+			},
+		}, "", eventCh)
+		require.NoError(t, err)
+
+		result := getEventResult[*pbpeering.TrustBundleReadResponse](t, eventCh)
+		require.Equal(t, "before.com", result.Bundle.TrustDomain)
+	})
+
+	testutil.RunStep(t, "can't read", func(t *testing.T) {
+		authz := policyAuthorizer(t, ``)
+		dataSource := ServerTrustBundle(ServerDataSourceDeps{
+			GetStore:    func() Store { return store },
+			ACLResolver: newStaticResolver(authz),
+		})
+
+		eventCh := make(chan proxycfg.UpdateEvent)
+		err := dataSource.Notify(context.Background(), &cachetype.TrustBundleReadRequest{
+			Request: &pbpeering.TrustBundleReadRequest{
+				Name: peerName,
+			},
+		}, "", eventCh)
+		require.NoError(t, err)
+
+		err = getEventError(t, eventCh)
+		require.Contains(t, err.Error(), "provided token lacks permission 'service:write' on \"any service\"")
+	})
+}
+
 func TestServerTrustBundleList(t *testing.T) {
 	const index uint64 = 123
 
@@ -94,8 +149,9 @@ func TestServerTrustBundleList(t *testing.T) {
 		})
 
 		dataSource := ServerTrustBundleList(ServerDataSourceDeps{
-			Datacenter: "dc1",
-			GetStore:   func() Store { return store },
+			Datacenter:  "dc1",
+			GetStore:    func() Store { return store },
+			ACLResolver: newStaticResolver(acl.ManageAll()),
 		})
 
 		eventCh := make(chan proxycfg.UpdateEvent)
@@ -135,7 +191,8 @@ func TestServerTrustBundleList(t *testing.T) {
 		}))
 
 		dataSource := ServerTrustBundleList(ServerDataSourceDeps{
-			GetStore: func() Store { return store },
+			GetStore:    func() Store { return store },
+			ACLResolver: newStaticResolver(acl.ManageAll()),
 		})
 
 		eventCh := make(chan proxycfg.UpdateEvent)
@@ -152,6 +209,142 @@ func TestServerTrustBundleList(t *testing.T) {
 	})
 }
 
+func TestServerTrustBundleList_ACLEnforcement(t *testing.T) {
+	const index uint64 = 123
+	var (
+		authzWriteWeb = policyAuthorizer(t, `service "web" { policy = "write" }`)
+		authzWriteAll = policyAuthorizer(t, `service "" { policy = "write" }`)
+		authzNothing  = policyAuthorizer(t, ``)
+	)
+
+	t.Run("ACL enforcement: list by service", func(t *testing.T) {
+		const (
+			serviceName = "web"
+			us          = "default"
+			them        = "peer2"
+		)
+
+		store := state.NewStateStore(nil)
+		require.NoError(t, store.CASetConfig(index, &structs.CAConfiguration{ClusterID: "cluster-id"}))
+
+		testutil.RunStep(t, "export service to peer", func(t *testing.T) {
+			require.NoError(t, store.PeeringWrite(index, &pbpeering.Peering{
+				ID:    testUUID(t),
+				Name:  them,
+				State: pbpeering.PeeringState_ACTIVE,
+			}))
+
+			require.NoError(t, store.PeeringTrustBundleWrite(index, &pbpeering.PeeringTrustBundle{
+				PeerName: them,
+			}))
+
+			require.NoError(t, store.EnsureConfigEntry(index, &structs.ExportedServicesConfigEntry{
+				Name: us,
+				Services: []structs.ExportedService{
+					{
+						Name: serviceName,
+						Consumers: []structs.ServiceConsumer{
+							{PeerName: them},
+						},
+					},
+				},
+			}))
+		})
+
+		testutil.RunStep(t, "can read", func(t *testing.T) {
+			dataSource := ServerTrustBundleList(ServerDataSourceDeps{
+				Datacenter:  "dc1",
+				GetStore:    func() Store { return store },
+				ACLResolver: newStaticResolver(authzWriteWeb),
+			})
+
+			eventCh := make(chan proxycfg.UpdateEvent)
+			err := dataSource.Notify(context.Background(), &cachetype.TrustBundleListRequest{
+				Request: &pbpeering.TrustBundleListByServiceRequest{
+					ServiceName: serviceName,
+					Partition:   us,
+				},
+			}, "", eventCh)
+			require.NoError(t, err)
+
+			result := getEventResult[*pbpeering.TrustBundleListByServiceResponse](t, eventCh)
+			require.Len(t, result.Bundles, 1)
+		})
+
+		testutil.RunStep(t, "can't read", func(t *testing.T) {
+			dataSource := ServerTrustBundleList(ServerDataSourceDeps{
+				Datacenter:  "dc1",
+				GetStore:    func() Store { return store },
+				ACLResolver: newStaticResolver(authzNothing),
+			})
+
+			eventCh := make(chan proxycfg.UpdateEvent)
+			err := dataSource.Notify(context.Background(), &cachetype.TrustBundleListRequest{
+				Request: &pbpeering.TrustBundleListByServiceRequest{
+					ServiceName: serviceName,
+					Partition:   us,
+				},
+			}, "", eventCh)
+			require.NoError(t, err)
+
+			err = getEventError(t, eventCh)
+			require.Contains(t, err.Error(), "provided token lacks permission 'service:write' on \"web\"")
+		})
+	})
+
+	t.Run("ACL Enforcement: list for mesh gateway", func(t *testing.T) {
+		store := state.NewStateStore(nil)
+		require.NoError(t, store.CASetConfig(index, &structs.CAConfiguration{ClusterID: "cluster-id"}))
+
+		require.NoError(t, store.PeeringTrustBundleWrite(index, &pbpeering.PeeringTrustBundle{
+			PeerName: "peer1",
+		}))
+		require.NoError(t, store.PeeringTrustBundleWrite(index, &pbpeering.PeeringTrustBundle{
+			PeerName: "peer2",
+		}))
+
+		testutil.RunStep(t, "can read", func(t *testing.T) {
+			dataSource := ServerTrustBundleList(ServerDataSourceDeps{
+				Datacenter:  "dc1",
+				GetStore:    func() Store { return store },
+				ACLResolver: newStaticResolver(authzWriteAll),
+			})
+
+			eventCh := make(chan proxycfg.UpdateEvent)
+			err := dataSource.Notify(context.Background(), &cachetype.TrustBundleListRequest{
+				Request: &pbpeering.TrustBundleListByServiceRequest{
+					Kind:      string(structs.ServiceKindMeshGateway),
+					Partition: "default",
+				},
+			}, "", eventCh)
+			require.NoError(t, err)
+
+			result := getEventResult[*pbpeering.TrustBundleListByServiceResponse](t, eventCh)
+			require.Len(t, result.Bundles, 2)
+		})
+
+		testutil.RunStep(t, "can't read", func(t *testing.T) {
+			dataSource := ServerTrustBundleList(ServerDataSourceDeps{
+				Datacenter:  "dc1",
+				GetStore:    func() Store { return store },
+				ACLResolver: newStaticResolver(authzNothing),
+			})
+
+			eventCh := make(chan proxycfg.UpdateEvent)
+			err := dataSource.Notify(context.Background(), &cachetype.TrustBundleListRequest{
+				Request: &pbpeering.TrustBundleListByServiceRequest{
+					Kind:      string(structs.ServiceKindMeshGateway),
+					Partition: "default",
+				},
+			}, "", eventCh)
+			require.NoError(t, err)
+
+			err = getEventError(t, eventCh)
+			require.Contains(t, err.Error(), "provided token lacks permission 'service:write'")
+		})
+	})
+}
+
 func testUUID(t *testing.T) string {
 	v, err := lib.GenerateUUID(nil)
 	require.NoError(t, err)