resource: Make resource listbyowner tenancy aware

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

@@ -10,39 +10,71 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/internal/resource"
 	"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 {
+	reg, err := s.validateListByOwnerRequest(req)
+	if err != nil {
 		return nil, err
 	}
 
-	_, err := s.resolveType(req.Owner.Type)
+	// Convert v2 request tenancy to v1 for ACL subsystem.
+	entMeta := v2TenancyToV1EntMeta(req.Owner.Tenancy)
+	token := tokenFromContext(ctx)
+
+	// Fill entMeta with token tenancy when empty.
+	authz, authzContext, err := s.getAuthorizer(token, entMeta)
 	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)
+	// Handle defaulting empty tenancy units from request.
+	v1EntMetaToV2Tenancy(reg, entMeta, req.Owner.Tenancy)
+
+	// Check list ACL before verifying tenancy exists to not leak tenancy existence.
+	err = reg.ACLs.List(authz, authzContext)
+	switch {
+	case acl.IsErrPermissionDenied(err):
+		return nil, status.Error(codes.PermissionDenied, err.Error())
+	case err != nil:
+		return nil, status.Errorf(codes.Internal, "failed list acl: %v", err)
 	}
 
-	// TODO(spatel): Refactor _ and entMeta in NET-4917
-	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
-	if err != nil {
+	// Check v1 tenancy exists for the v2 resource.
+	if err = v1TenancyExists(reg, s.V1TenancyBridge, req.Owner.Tenancy, codes.InvalidArgument); err != nil {
 		return nil, err
 	}
 
+	// Get owned resources.
+	children, err := s.Backend.ListByOwner(ctx, req.Owner)
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "failed list by owner: %v", err)
+	}
+
 	result := make([]*pbresource.Resource, 0)
 	for _, child := range children {
-		reg, err := s.resolveType(child.Id.Type)
+		// Retrieve child type's registration to access read ACL hook.
+		childReg, err := s.resolveType(child.Id.Type)
 		if err != nil {
 			return nil, err
 		}
 
-		// ACL filter
-		err = reg.ACLs.Read(authz, authzContext, child.Id)
+		// Rebuild authorizer if tenancy not identical between owner and child (child scope
+		// may be narrower).
+		childAuthz := authz
+		childAuthzContext := authzContext
+		if !resource.EqualTenancy(req.Owner.Tenancy, child.Id.Tenancy) {
+			childEntMeta := v2TenancyToV1EntMeta(child.Id.Tenancy)
+			childAuthz, childAuthzContext, err = s.getAuthorizer(token, childEntMeta)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		// Filter out children that fail real ACL.
+		err = childReg.ACLs.Read(childAuthz, childAuthzContext, child.Id)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue
@@ -55,17 +87,36 @@ func (s *Server) ListByOwner(ctx context.Context, req *pbresource.ListByOwnerReq
 	return &pbresource.ListByOwnerResponse{Resources: result}, nil
 }
 
-func validateListByOwnerRequest(req *pbresource.ListByOwnerRequest) error {
+func (s *Server) validateListByOwnerRequest(req *pbresource.ListByOwnerRequest) (*resource.Registration, error) {
 	if req.Owner == nil {
-		return status.Errorf(codes.InvalidArgument, "owner is required")
+		return nil, status.Errorf(codes.InvalidArgument, "owner is required")
 	}
 
 	if err := validateId(req.Owner, "owner"); err != nil {
-		return err
+		return nil, err
 	}
 
 	if req.Owner.Uid == "" {
-		return status.Errorf(codes.InvalidArgument, "owner uid is required")
+		return nil, status.Errorf(codes.InvalidArgument, "owner uid is required")
 	}
-	return nil
+
+	reg, err := s.resolveType(req.Owner.Type)
+	if err != nil {
+		return nil, err
+	}
+
+	// Lowercase
+	resource.Normalize(req.Owner.Tenancy)
+
+	// Error when partition scoped and namespace not empty.
+	if reg.Scope == resource.ScopePartition && req.Owner.Tenancy.Namespace != "" {
+		return nil, status.Errorf(
+			codes.InvalidArgument,
+			"partition scoped type %s cannot have a namespace. got: %s",
+			resource.ToGVK(req.Owner.Type),
+			req.Owner.Tenancy.Namespace,
+		)
+	}
+
+	return reg, nil
 }

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

@@ -9,6 +9,7 @@ import (
 	"testing"
 
 	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/demo"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/proto/private/prototest"
@@ -17,41 +18,49 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 )
 
 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 = ""
+	testCases := map[string]func(artistId, recordlabelId *pbresource.ID) *pbresource.ID{
+		"no owner": func(artistId, recordLabelId *pbresource.ID) *pbresource.ID {
+			return nil
+		},
+		"no type": func(artistId, _ *pbresource.ID) *pbresource.ID {
+			artistId.Type = nil
+			return artistId
+		},
+		"no tenancy": func(artistId, _ *pbresource.ID) *pbresource.ID {
+			artistId.Tenancy = nil
+			return artistId
 		},
-		"tenancy namespace not default": func(req *pbresource.ListByOwnerRequest) {
-			req.Owner.Tenancy = clone(req.Owner.Tenancy)
-			req.Owner.Tenancy.Namespace = ""
+		"no name": func(artistId, _ *pbresource.ID) *pbresource.ID {
+			artistId.Name = ""
+			return artistId
 		},
-		"tenancy peername not local": func(req *pbresource.ListByOwnerRequest) {
-			req.Owner.Tenancy = clone(req.Owner.Tenancy)
-			req.Owner.Tenancy.PeerName = ""
+		"no uid": func(artistId, _ *pbresource.ID) *pbresource.ID {
+			artistId.Uid = ""
+			return artistId
+		},
+		"partition scope with non-empty namespace": func(_, recordLabelId *pbresource.ID) *pbresource.ID {
+			recordLabelId.Tenancy.Namespace = "ishouldnothaveanamespace"
+			return recordLabelId
 		},
 	}
 	for desc, modFn := range testCases {
 		t.Run(desc, func(t *testing.T) {
-			res, err := demo.GenerateV2Artist()
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
 			require.NoError(t, err)
 
-			req := &pbresource.ListByOwnerRequest{Owner: res.Id}
-			modFn(req)
+			// Each test case picks which resource to use based on the resource type's scope.
+			req := &pbresource.ListByOwnerRequest{Owner: modFn(artist.Id, recordLabel.Id)}
 
 			_, err = client.ListByOwner(testContext(t), req)
 			require.Error(t, err)
@@ -67,7 +76,7 @@ func TestListByOwner_TypeNotRegistered(t *testing.T) {
 	_, err := client.ListByOwner(context.Background(), &pbresource.ListByOwnerRequest{
 		Owner: &pbresource.ID{
 			Type:    demo.TypeV2Artist,
-			Tenancy: demo.TenancyDefault,
+			Tenancy: resource.DefaultNamespacedTenancy(),
 			Uid:     "bogus",
 			Name:    "bogus",
 		},
@@ -125,8 +134,108 @@ func TestListByOwner_Many(t *testing.T) {
 	prototest.AssertElementsMatch(t, albums, rsp3.Resources)
 }
 
+func TestListByOwner_OwnerTenancyDoesNotExist(t *testing.T) {
+	tenancyCases := map[string]func(artistId, recordlabelId *pbresource.ID) *pbresource.ID{
+		"partition not found when namespace scoped": func(artistId, _ *pbresource.ID) *pbresource.ID {
+			id := clone(artistId)
+			id.Uid = "doesnotmatter"
+			id.Tenancy.Partition = "boguspartition"
+			return id
+		},
+		"namespace not found when namespace scoped": func(artistId, _ *pbresource.ID) *pbresource.ID {
+			id := clone(artistId)
+			id.Uid = "doesnotmatter"
+			id.Tenancy.Namespace = "bogusnamespace"
+			return id
+		},
+		"partition not found when partition scoped": func(_, recordLabelId *pbresource.ID) *pbresource.ID {
+			id := clone(recordLabelId)
+			id.Uid = "doesnotmatter"
+			id.Tenancy.Partition = "boguspartition"
+			return id
+		},
+	}
+	for desc, modFn := range tenancyCases {
+		t.Run(desc, func(t *testing.T) {
+			server := testServer(t)
+			demo.RegisterTypes(server.Registry)
+			client := testClient(t, server)
+
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
+			require.NoError(t, err)
+			recordLabel, err = server.Backend.WriteCAS(testContext(t), recordLabel)
+			require.NoError(t, err)
+
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+			artist, err = server.Backend.WriteCAS(testContext(t), artist)
+			require.NoError(t, err)
+
+			// Verify non-existant tenancy units in owner err with not found.
+			_, err = client.ListByOwner(testContext(t), &pbresource.ListByOwnerRequest{Owner: modFn(artist.Id, recordLabel.Id)})
+			require.Error(t, err)
+			require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+			require.Contains(t, err.Error(), "resource not found")
+		})
+	}
+}
+
+func TestListByOwner_Tenancy_Defaults_And_Normalization(t *testing.T) {
+	for tenancyDesc, modFn := range tenancyCases() {
+		t.Run(tenancyDesc, func(t *testing.T) {
+			server := testServer(t)
+			demo.RegisterTypes(server.Registry)
+			client := testClient(t, server)
+
+			// Create partition scoped recordLabel.
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
+			require.NoError(t, err)
+			rsp1, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: recordLabel})
+			require.NoError(t, err)
+			recordLabel = rsp1.Resource
+
+			// Create namespace scoped artist.
+			artist, err := demo.GenerateV2Artist()
+			require.NoError(t, err)
+			rsp2, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: artist})
+			require.NoError(t, err)
+			artist = rsp2.Resource
+
+			// Owner will be either partition scoped (recordLabel) or namespace scoped (artist) based on testcase.
+			moddedOwnerId := modFn(artist.Id, recordLabel.Id)
+			var ownerId *pbresource.ID
+
+			// Avoid using the modded id when linking owner to child.
+			switch {
+			case proto.Equal(moddedOwnerId.Type, demo.TypeV2Artist):
+				ownerId = artist.Id
+			case proto.Equal(moddedOwnerId.Type, demo.TypeV1RecordLabel):
+				ownerId = recordLabel.Id
+			default:
+				require.Fail(t, "unexpected resource type")
+			}
+
+			// Link owner to child.
+			album, err := demo.GenerateV2Album(ownerId)
+			require.NoError(t, err)
+			rsp3, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+			require.NoError(t, err)
+			album = rsp3.Resource
+
+			// Test
+			listRsp, err := client.ListByOwner(testContext(t), &pbresource.ListByOwnerRequest{
+				Owner: moddedOwnerId,
+			})
+			require.NoError(t, err)
+
+			// Verify child album always returned.
+			prototest.AssertDeepEqual(t, album, listRsp.Resources[0])
+		})
+	}
+}
+
 func TestListByOwner_ACL_PerTypeDenied(t *testing.T) {
-	authz := AuthorizerFrom(t, `key_prefix "resource/demo.v2.Album/" { policy = "deny" }`)
+	authz := AuthorizerFrom(t, `key_prefix "resource/demo.v2.Album/" { policy = "deny" }`, demo.ArtistV2ListPolicy)
 	_, rsp, err := roundTripListByOwner(t, authz)
 
 	// verify resource filtered out, hence no results
@@ -135,7 +244,7 @@ func TestListByOwner_ACL_PerTypeDenied(t *testing.T) {
 }
 
 func TestListByOwner_ACL_PerTypeAllowed(t *testing.T) {
-	authz := AuthorizerFrom(t, `key_prefix "resource/demo.v2.Album/" { policy = "read" }`)
+	authz := AuthorizerFrom(t, `key_prefix "resource/demo.v2.Album/" { policy = "read" }`, demo.ArtistV2ListPolicy)
 	album, rsp, err := roundTripListByOwner(t, authz)
 
 	// verify resource not filtered out

internal/resource/decode_test.go

@@ -29,7 +29,7 @@ func TestGetDecodedResource(t *testing.T) {
 
 	babypantsID := &pbresource.ID{
 		Type:    demo.TypeV2Artist,
-		Tenancy: demo.TenancyDefault,
+		Tenancy: resource.DefaultNamespacedTenancy(),
 		Name:    "babypants",
 	}
 
@@ -75,7 +75,7 @@ func TestDecode(t *testing.T) {
 		foo := &pbresource.Resource{
 			Id: &pbresource.ID{
 				Type:    demo.TypeV2Artist,
-				Tenancy: demo.TenancyDefault,
+				Tenancy: resource.DefaultNamespacedTenancy(),
 				Name:    "babypants",
 			},
 			Data: any,
@@ -95,7 +95,7 @@ func TestDecode(t *testing.T) {
 		foo := &pbresource.Resource{
 			Id: &pbresource.ID{
 				Type:    demo.TypeV2Artist,
-				Tenancy: demo.TenancyDefault,
+				Tenancy: resource.DefaultNamespacedTenancy(),
 				Name:    "babypants",
 			},
 			Data: &anypb.Any{

internal/resource/demo/demo.go

@@ -22,13 +22,6 @@ import (
 )
 
 var (
-	// TenancyDefault contains the default values for all tenancy units.
-	TenancyDefault = &pbresource.Tenancy{
-		Partition: resource.DefaultPartitionName,
-		PeerName:  "local",
-		Namespace: resource.DefaultNamespaceName,
-	}
-
 	// TypeV1RecordLabel represents a record label which artists are signed to.
 	// Used specifically as a resource to test partition only scoped resources.
 	TypeV1RecordLabel = &pbresource.Type{