resource: List resources by owner (#17190)

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

@@ -0,0 +1,70 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package resource
+
+import (
+	"context"
+
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func (s *Server) ListByOwner(ctx context.Context, req *pbresource.ListByOwnerRequest) (*pbresource.ListByOwnerResponse, error) {
+	if err := validateListByOwnerRequest(req); err != nil {
+		return nil, err
+	}
+
+	_, err := s.resolveType(req.Owner.Type)
+	if err != nil {
+		return nil, err
+	}
+
+	children, err := s.Backend.ListByOwner(ctx, req.Owner)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed list by owner: %v", err)
+	}
+
+	authz, err := s.getAuthorizer(tokenFromContext(ctx))
+	if err != nil {
+		return nil, err
+	}
+
+	result := make([]*pbresource.Resource, 0)
+	for _, child := range children {
+		reg, err := s.resolveType(child.Id.Type)
+		if err != nil {
+			return nil, err
+		}
+
+		// ACL filter
+		err = reg.ACLs.Read(authz, child.Id)
+		switch {
+		case acl.IsErrPermissionDenied(err):
+			continue
+		case err != nil:
+			return nil, status.Errorf(codes.Internal, "failed read acl: %v", err)
+		}
+
+		result = append(result, child)
+	}
+	return &pbresource.ListByOwnerResponse{Resources: result}, nil
+}
+
+func validateListByOwnerRequest(req *pbresource.ListByOwnerRequest) error {
+	if req.Owner == nil {
+		return status.Errorf(codes.InvalidArgument, "owner is required")
+	}
+
+	if err := validateId(req.Owner, "owner"); err != nil {
+		return err
+	}
+
+	if req.Owner.Uid == "" {
+		return status.Errorf(codes.InvalidArgument, "owner uid is required")
+	}
+	return nil
+}

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

@@ -0,0 +1,174 @@
+// // Copyright (c) HashiCorp, Inc.
+// // SPDX-License-Identifier: MPL-2.0
+
+package resource
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/internal/resource/demo"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+	"github.com/hashicorp/consul/proto/private/prototest"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+func TestListByOwner_InputValidation(t *testing.T) {
+	server := testServer(t)
+	client := testClient(t, server)
+
+	demo.RegisterTypes(server.Registry)
+
+	testCases := map[string]func(*pbresource.ListByOwnerRequest){
+		"no owner":   func(req *pbresource.ListByOwnerRequest) { req.Owner = nil },
+		"no type":    func(req *pbresource.ListByOwnerRequest) { req.Owner.Type = nil },
+		"no tenancy": func(req *pbresource.ListByOwnerRequest) { req.Owner.Tenancy = nil },
+		"no name":    func(req *pbresource.ListByOwnerRequest) { req.Owner.Name = "" },
+		"no uid":     func(req *pbresource.ListByOwnerRequest) { req.Owner.Uid = "" },
+		// clone necessary to not pollute DefaultTenancy
+		"tenancy partition not default": func(req *pbresource.ListByOwnerRequest) {
+			req.Owner.Tenancy = clone(req.Owner.Tenancy)
+			req.Owner.Tenancy.Partition = ""
+		},
+		"tenancy namespace not default": func(req *pbresource.ListByOwnerRequest) {
+			req.Owner.Tenancy = clone(req.Owner.Tenancy)
+			req.Owner.Tenancy.Namespace = ""
+		},
+		"tenancy peername not local": func(req *pbresource.ListByOwnerRequest) {
+			req.Owner.Tenancy = clone(req.Owner.Tenancy)
+			req.Owner.Tenancy.PeerName = ""
+		},
+	}
+	for desc, modFn := range testCases {
+		t.Run(desc, func(t *testing.T) {
+			res, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+
+			req := &pbresource.ListByOwnerRequest{Owner: res.Id}
+			modFn(req)
+
+			_, err = client.ListByOwner(testContext(t), req)
+			require.Error(t, err)
+			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+		})
+	}
+}
+
+func TestListByOwner_TypeNotRegistered(t *testing.T) {
+	server := testServer(t)
+	client := testClient(t, server)
+
+	_, err := client.ListByOwner(context.Background(), &pbresource.ListByOwnerRequest{
+		Owner: &pbresource.ID{
+			Type:    demo.TypeV2Artist,
+			Tenancy: demo.TenancyDefault,
+			Uid:     "bogus",
+			Name:    "bogus",
+		},
+	})
+	require.Error(t, err)
+	require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+	require.Contains(t, err.Error(), "resource type demo.v2.artist not registered")
+}
+
+func TestListByOwner_Empty(t *testing.T) {
+	server := testServer(t)
+	demo.RegisterTypes(server.Registry)
+	client := testClient(t, server)
+
+	res, err := demo.GenerateV2Artist()
+	require.NoError(t, err)
+
+	rsp1, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: res})
+	require.NoError(t, err)
+
+	rsp2, err := client.ListByOwner(testContext(t), &pbresource.ListByOwnerRequest{Owner: rsp1.Resource.Id})
+	require.NoError(t, err)
+	require.Empty(t, rsp2.Resources)
+}
+
+func TestListByOwner_Many(t *testing.T) {
+	server := testServer(t)
+	demo.RegisterTypes(server.Registry)
+	client := testClient(t, server)
+
+	res, err := demo.GenerateV2Artist()
+	require.NoError(t, err)
+
+	rsp1, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: res})
+	artist := rsp1.Resource
+	require.NoError(t, err)
+
+	albums := make([]*pbresource.Resource, 10)
+	for i := 0; i < len(albums); i++ {
+		album, err := demo.GenerateV2Album(artist.Id)
+		require.NoError(t, err)
+
+		// Prevent test flakes if the generated names collide.
+		album.Id.Name = fmt.Sprintf("%s-%d", artist.Id.Name, i)
+
+		rsp2, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+		require.NoError(t, err)
+		albums[i] = rsp2.Resource
+	}
+
+	rsp3, err := client.ListByOwner(testContext(t), &pbresource.ListByOwnerRequest{
+		Owner: artist.Id,
+	})
+	require.NoError(t, err)
+	prototest.AssertElementsMatch(t, albums, rsp3.Resources)
+}
+
+func TestListByOwner_ACL_PerTypeDenied(t *testing.T) {
+	authz := AuthorizerFrom(t, `key_prefix "resource/demo.v2.album/" { policy = "deny" }`)
+	_, rsp, err := roundTripListByOwner(t, authz)
+
+	// verify resource filtered out, hence no results
+	require.NoError(t, err)
+	require.Empty(t, rsp.Resources)
+}
+
+func TestListByOwner_ACL_PerTypeAllowed(t *testing.T) {
+	authz := AuthorizerFrom(t, `key_prefix "resource/demo.v2.album/" { policy = "read" }`)
+	album, rsp, err := roundTripListByOwner(t, authz)
+
+	// verify resource not filtered out
+	require.NoError(t, err)
+	require.Len(t, rsp.Resources, 1)
+	prototest.AssertDeepEqual(t, album, rsp.Resources[0])
+}
+
+// roundtrip a ListByOwner which attempts to return a single resource
+func roundTripListByOwner(t *testing.T, authz acl.Authorizer) (*pbresource.Resource, *pbresource.ListByOwnerResponse, error) {
+	server := testServer(t)
+	client := testClient(t, server)
+	demo.RegisterTypes(server.Registry)
+
+	artist, err := demo.GenerateV2Artist()
+	require.NoError(t, err)
+
+	rsp1, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: artist})
+	artist = rsp1.Resource
+	require.NoError(t, err)
+
+	album, err := demo.GenerateV2Album(artist.Id)
+	require.NoError(t, err)
+
+	rsp2, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+	album = rsp2.Resource
+	require.NoError(t, err)
+
+	mockACLResolver := &MockACLResolver{}
+	mockACLResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
+		Return(authz, nil)
+	server.ACLResolver = mockACLResolver
+
+	rsp3, err := client.ListByOwner(testContext(t), &pbresource.ListByOwnerRequest{Owner: artist.Id})
+	return album, rsp3, err
+}

agent/grpc-middleware/rate_limit_mappings.gen.go

@@ -28,6 +28,7 @@ var rpcRateLimitSpecs = map[string]rate.OperationSpec{
 	"/hashicorp.consul.internal.storage.raft.ForwardingService/Write":            {Type: rate.OperationTypeExempt, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/Delete":                          {Type: rate.OperationTypeWrite, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/List":                            {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
+	"/hashicorp.consul.resource.ResourceService/ListByOwner":                     {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/Read":                            {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/WatchList":                       {Type: rate.OperationTypeRead, Category: rate.OperationCategoryResource},
 	"/hashicorp.consul.resource.ResourceService/Write":                           {Type: rate.OperationTypeWrite, Category: rate.OperationCategoryResource},

internal/storage/conformance/conformance.go

@@ -39,7 +39,7 @@ func Test(t *testing.T, opts TestOptions) {
 	t.Run("Read", func(t *testing.T) { testRead(t, opts) })
 	t.Run("CAS Write", func(t *testing.T) { testCASWrite(t, opts) })
 	t.Run("CAS Delete", func(t *testing.T) { testCASDelete(t, opts) })
-	t.Run("OwnerReferences", func(t *testing.T) { testOwnerReferences(t, opts) })
+	t.Run("ListByOwner", func(t *testing.T) { testListByOwner(t, opts) })
 
 	testListWatch(t, opts)
 }
@@ -518,7 +518,7 @@ func testListWatch(t *testing.T, opts TestOptions) {
 	})
 }
 
-func testOwnerReferences(t *testing.T, opts TestOptions) {
+func testListByOwner(t *testing.T, opts TestOptions) {
 	backend := opts.NewBackend(t)
 	ctx := testContext(t)
 
@@ -555,39 +555,39 @@ func testOwnerReferences(t *testing.T, opts TestOptions) {
 	require.NoError(t, err)
 
 	eventually(t, func(t testingT) {
-		refs, err := backend.OwnerReferences(ctx, owner.Id)
+		res, err := backend.ListByOwner(ctx, owner.Id)
 		require.NoError(t, err)
-		prototest.AssertElementsMatch(t, refs, []*pbresource.ID{r1.Id, r2.Id})
+		prototest.AssertElementsMatch(t, res, []*pbresource.Resource{r1, r2})
 	})
 
 	t.Run("references are anchored to a specific uid", func(t *testing.T) {
 		id := clone(owner.Id)
 		id.Uid = "different"
 
 		eventually(t, func(t testingT) {
-			refs, err := backend.OwnerReferences(ctx, id)
+			res, err := backend.ListByOwner(ctx, id)
 			require.NoError(t, err)
-			require.Empty(t, refs)
+			require.Empty(t, res)
 		})
 	})
 
 	t.Run("deleting the owner doesn't remove the references", func(t *testing.T) {
 		require.NoError(t, backend.DeleteCAS(ctx, owner.Id, owner.Version))
 
 		eventually(t, func(t testingT) {
-			refs, err := backend.OwnerReferences(ctx, owner.Id)
+			res, err := backend.ListByOwner(ctx, owner.Id)
 			require.NoError(t, err)
-			prototest.AssertElementsMatch(t, refs, []*pbresource.ID{r1.Id, r2.Id})
+			prototest.AssertElementsMatch(t, res, []*pbresource.Resource{r1, r2})
 		})
 	})
 
 	t.Run("deleting the owned resource removes its reference", func(t *testing.T) {
 		require.NoError(t, backend.DeleteCAS(ctx, r2.Id, r2.Version))
 
 		eventually(t, func(t testingT) {
-			refs, err := backend.OwnerReferences(ctx, owner.Id)
+			res, err := backend.ListByOwner(ctx, owner.Id)
 			require.NoError(t, err)
-			prototest.AssertElementsMatch(t, refs, []*pbresource.ID{r1.Id})
+			prototest.AssertElementsMatch(t, res, []*pbresource.Resource{r1})
 		})
 	})
 }

internal/storage/inmem/backend.go

@@ -70,7 +70,7 @@ func (b *Backend) WatchList(_ context.Context, resType storage.UnversionedType,
 	return b.store.WatchList(resType, tenancy, namePrefix)
 }
 
-// OwnerReferences implements the storage.Backend interface.
-func (b *Backend) OwnerReferences(_ context.Context, id *pbresource.ID) ([]*pbresource.ID, error) {
-	return b.store.OwnerReferences(id)
+// ListByOwner implements the storage.Backend interface.
+func (b *Backend) ListByOwner(_ context.Context, id *pbresource.ID) ([]*pbresource.Resource, error) {
+	return b.store.ListByOwner(id)
 }

internal/storage/inmem/store.go

@@ -259,11 +259,10 @@ func (s *Store) WatchList(typ storage.UnversionedType, ten *pbresource.Tenancy,
 	}, nil
 }
 
-// OwnerReferences returns the IDs of resources owned by the resource with the
-// given ID.
+// ListByOwner returns resources owned by the resource with the given ID.
 //
 // For more information, see the storage.Backend documentation.
-func (s *Store) OwnerReferences(id *pbresource.ID) ([]*pbresource.ID, error) {
+func (s *Store) ListByOwner(id *pbresource.ID) ([]*pbresource.Resource, error) {
 	tx := s.txn(false)
 	defer tx.Abort()
 
@@ -272,11 +271,11 @@ func (s *Store) OwnerReferences(id *pbresource.ID) ([]*pbresource.ID, error) {
 		return nil, err
 	}
 
-	var refs []*pbresource.ID
+	var res []*pbresource.Resource
 	for v := iter.Next(); v != nil; v = iter.Next() {
-		refs = append(refs, v.(*pbresource.Resource).Id)
+		res = append(res, v.(*pbresource.Resource))
 	}
-	return refs, nil
+	return res, nil
 }
 
 func (s *Store) txn(write bool) *memdb.Txn {

internal/storage/raft/backend.go

@@ -217,9 +217,9 @@ func (b *Backend) WatchList(_ context.Context, resType storage.UnversionedType,
 	return b.store.WatchList(resType, tenancy, namePrefix)
 }
 
-// OwnerReferences implements the storage.Backend interface.
-func (b *Backend) OwnerReferences(_ context.Context, id *pbresource.ID) ([]*pbresource.ID, error) {
-	return b.store.OwnerReferences(id)
+// ListByOwner implements the storage.Backend interface.
+func (b *Backend) ListByOwner(_ context.Context, id *pbresource.ID) ([]*pbresource.Resource, error) {
+	return b.store.ListByOwner(id)
 }
 
 // Apply is called by the FSM with the bytes of a Raft log entry, with Consul's