From 63cc0371107c5f84afcdbde80b24abdebaee67ef Mon Sep 17 00:00:00 2001
From: Semir Patel <semir.patel@hashicorp.com>
Date: Mon, 7 Aug 2023 16:37:03 -0500
Subject: [PATCH] resource: Make resource read tenancy aware (#18397)

---
 agent/consul/server.go                        |  21 +-
 agent/consul/tenancy_bridge.go                |  15 ++
 agent/consul/tenancy_bridge_oss.go            |  21 ++
 .../grpc-external/services/resource/delete.go |   3 +-
 .../services/resource/delete_test.go          |  27 +--
 agent/grpc-external/services/resource/list.go |   5 +-
 .../services/resource/list_by_owner.go        |   5 +-
 .../services/resource/mock_TenancyBridge.go   |  73 +++++++
 agent/grpc-external/services/resource/read.go |  61 ++++--
 .../services/resource/read_test.go            | 183 ++++++++++++----
 .../grpc-external/services/resource/server.go |  50 +++--
 .../services/resource/server_oss.go           |  27 +++
 .../services/resource/server_oss_test.go      |  17 ++
 .../services/resource/server_test.go          |  32 ++-
 .../services/resource/testing/testing.go      |  16 +-
 .../grpc-external/services/resource/watch.go  |   5 +-
 .../grpc-external/services/resource/write.go  |   3 +-
 .../services/resource/write_status.go         |   3 +-
 .../services/resource/write_test.go           |  76 ++++---
 .../internal/types/proxy_state_template.go    |   6 +-
 internal/resource/demo/demo.go                |  49 ++++-
 internal/resource/http/http_test.go           |  16 --
 internal/resource/registry.go                 |  34 +--
 internal/resource/registry_test.go            |   4 +-
 internal/resource/resourcetest/builder.go     |   4 +-
 internal/resource/tenancy.go                  |  53 +++++
 proto/private/pbdemo/v1/demo.pb.binary.go     |  10 +
 proto/private/pbdemo/v1/demo.pb.go            | 202 ++++++++++++------
 proto/private/pbdemo/v1/demo.proto            |   5 +
 29 files changed, 755 insertions(+), 271 deletions(-)
 create mode 100644 agent/consul/tenancy_bridge.go
 create mode 100644 agent/consul/tenancy_bridge_oss.go
 create mode 100644 agent/grpc-external/services/resource/mock_TenancyBridge.go
 create mode 100644 agent/grpc-external/services/resource/server_oss.go
 create mode 100644 agent/grpc-external/services/resource/server_oss_test.go
 create mode 100644 internal/resource/tenancy.go

diff --git a/agent/consul/server.go b/agent/consul/server.go
index 032eb00bd9c2..fba8e8cfd919 100644
--- a/agent/consul/server.go
+++ b/agent/consul/server.go
@@ -36,6 +36,7 @@ import (
 	"golang.org/x/time/rate"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/reflection"
 
 	"github.com/hashicorp/consul-net-rpc/net/rpc"
 
@@ -1343,20 +1344,24 @@ func (s *Server) setupExternalGRPC(config *Config, typeRegistry resource.Registr
 	s.peerStreamServer.Register(s.externalGRPCServer)
 
 	s.resourceServiceServer = resourcegrpc.NewServer(resourcegrpc.Config{
-		Registry:    typeRegistry,
-		Backend:     s.raftStorageBackend,
-		ACLResolver: s.ACLResolver,
-		Logger:      logger.Named("grpc-api.resource"),
+		Registry:        typeRegistry,
+		Backend:         s.raftStorageBackend,
+		ACLResolver:     s.ACLResolver,
+		Logger:          logger.Named("grpc-api.resource"),
+		V1TenancyBridge: NewV1TenancyBridge(s),
 	})
 	s.resourceServiceServer.Register(s.externalGRPCServer)
+
+	reflection.Register(s.externalGRPCServer)
 }
 
 func (s *Server) setupInsecureResourceServiceClient(typeRegistry resource.Registry, logger hclog.Logger) error {
 	server := resourcegrpc.NewServer(resourcegrpc.Config{
-		Registry:    typeRegistry,
-		Backend:     s.raftStorageBackend,
-		ACLResolver: resolver.DANGER_NO_AUTH{},
-		Logger:      logger.Named("grpc-api.resource"),
+		Registry:        typeRegistry,
+		Backend:         s.raftStorageBackend,
+		ACLResolver:     resolver.DANGER_NO_AUTH{},
+		Logger:          logger.Named("grpc-api.resource"),
+		V1TenancyBridge: NewV1TenancyBridge(s),
 	})
 
 	conn, err := s.runInProcessGRPCServer(server.Register)
diff --git a/agent/consul/tenancy_bridge.go b/agent/consul/tenancy_bridge.go
new file mode 100644
index 000000000000..99f8ca660a3d
--- /dev/null
+++ b/agent/consul/tenancy_bridge.go
@@ -0,0 +1,15 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package consul
+
+// V1TenancyBridge is used by the resource service to access V1 implementations of
+// partitions and namespaces. This bridge will be removed when V2 implemenations
+// of partitions and namespaces are available.
+type V1TenancyBridge struct {
+	server *Server
+}
+
+func NewV1TenancyBridge(server *Server) *V1TenancyBridge {
+	return &V1TenancyBridge{server: server}
+}
diff --git a/agent/consul/tenancy_bridge_oss.go b/agent/consul/tenancy_bridge_oss.go
new file mode 100644
index 000000000000..09004c2ba603
--- /dev/null
+++ b/agent/consul/tenancy_bridge_oss.go
@@ -0,0 +1,21 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+//go:build !consulent
+// +build !consulent
+
+package consul
+
+func (b *V1TenancyBridge) NamespaceExists(partition, namespace string) (bool, error) {
+	if partition == "default" && namespace == "default" {
+		return true, nil
+	}
+	return false, nil
+}
+
+func (b *V1TenancyBridge) PartitionExists(partition string) (bool, error) {
+	if partition == "default" {
+		return true, nil
+	}
+	return false, nil
+}
diff --git a/agent/grpc-external/services/resource/delete.go b/agent/grpc-external/services/resource/delete.go
index 88987b6a69ff..597d184d2d6b 100644
--- a/agent/grpc-external/services/resource/delete.go
+++ b/agent/grpc-external/services/resource/delete.go
@@ -36,7 +36,8 @@ func (s *Server) Delete(ctx context.Context, req *pbresource.DeleteRequest) (*pb
 		return nil, err
 	}
 
-	authz, err := s.getAuthorizer(tokenFromContext(ctx))
+	// TODO(spatel): Refactor _ and entMeta in NET-4919
+	authz, _, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
 	if err != nil {
 		return nil, err
 	}
diff --git a/agent/grpc-external/services/resource/delete_test.go b/agent/grpc-external/services/resource/delete_test.go
index 0e98d3fd57a7..fd74d427eec1 100644
--- a/agent/grpc-external/services/resource/delete_test.go
+++ b/agent/grpc-external/services/resource/delete_test.go
@@ -30,19 +30,22 @@ func TestDelete_InputValidation(t *testing.T) {
 		"no type":    func(req *pbresource.DeleteRequest) { req.Id.Type = nil },
 		"no tenancy": func(req *pbresource.DeleteRequest) { req.Id.Tenancy = nil },
 		"no name":    func(req *pbresource.DeleteRequest) { req.Id.Name = "" },
+
+		// TODO(spatel): Refactor tenancy as part of NET-4919
+		//
 		// clone necessary to not pollute DefaultTenancy
-		"tenancy partition not default": func(req *pbresource.DeleteRequest) {
-			req.Id.Tenancy = clone(req.Id.Tenancy)
-			req.Id.Tenancy.Partition = ""
-		},
-		"tenancy namespace not default": func(req *pbresource.DeleteRequest) {
-			req.Id.Tenancy = clone(req.Id.Tenancy)
-			req.Id.Tenancy.Namespace = ""
-		},
-		"tenancy peername not local": func(req *pbresource.DeleteRequest) {
-			req.Id.Tenancy = clone(req.Id.Tenancy)
-			req.Id.Tenancy.PeerName = ""
-		},
+		// "tenancy partition not default": func(req *pbresource.DeleteRequest) {
+		// 	req.Id.Tenancy = clone(req.Id.Tenancy)
+		// 	req.Id.Tenancy.Partition = ""
+		// },
+		// "tenancy namespace not default": func(req *pbresource.DeleteRequest) {
+		// 	req.Id.Tenancy = clone(req.Id.Tenancy)
+		// 	req.Id.Tenancy.Namespace = ""
+		// },
+		// "tenancy peername not local": func(req *pbresource.DeleteRequest) {
+		// 	req.Id.Tenancy = clone(req.Id.Tenancy)
+		// 	req.Id.Tenancy.PeerName = ""
+		// },
 	}
 	for desc, modFn := range testCases {
 		t.Run(desc, func(t *testing.T) {
diff --git a/agent/grpc-external/services/resource/list.go b/agent/grpc-external/services/resource/list.go
index 77269e74688f..7332c0c8a0a2 100644
--- a/agent/grpc-external/services/resource/list.go
+++ b/agent/grpc-external/services/resource/list.go
@@ -25,7 +25,8 @@ func (s *Server) List(ctx context.Context, req *pbresource.ListRequest) (*pbreso
 		return nil, err
 	}
 
-	authz, err := s.getAuthorizer(tokenFromContext(ctx))
+	// TODO(spatel): Refactor _ and entMeta in NET-4915
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
 	if err != nil {
 		return nil, err
 	}
@@ -58,7 +59,7 @@ func (s *Server) List(ctx context.Context, req *pbresource.ListRequest) (*pbreso
 		}
 
 		// filter out items that don't pass read ACLs
-		err = reg.ACLs.Read(authz, resource.Id)
+		err = reg.ACLs.Read(authz, authzContext, resource.Id)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue
diff --git a/agent/grpc-external/services/resource/list_by_owner.go b/agent/grpc-external/services/resource/list_by_owner.go
index 2cc203e72c30..02b513011fdd 100644
--- a/agent/grpc-external/services/resource/list_by_owner.go
+++ b/agent/grpc-external/services/resource/list_by_owner.go
@@ -28,7 +28,8 @@ func (s *Server) ListByOwner(ctx context.Context, req *pbresource.ListByOwnerReq
 		return nil, status.Errorf(codes.Internal, "failed list by owner: %v", err)
 	}
 
-	authz, err := s.getAuthorizer(tokenFromContext(ctx))
+	// TODO(spatel): Refactor _ and entMeta in NET-4917
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
 	if err != nil {
 		return nil, err
 	}
@@ -41,7 +42,7 @@ func (s *Server) ListByOwner(ctx context.Context, req *pbresource.ListByOwnerReq
 		}
 
 		// ACL filter
-		err = reg.ACLs.Read(authz, child.Id)
+		err = reg.ACLs.Read(authz, authzContext, child.Id)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue
diff --git a/agent/grpc-external/services/resource/mock_TenancyBridge.go b/agent/grpc-external/services/resource/mock_TenancyBridge.go
new file mode 100644
index 000000000000..f2dcc6e0b46a
--- /dev/null
+++ b/agent/grpc-external/services/resource/mock_TenancyBridge.go
@@ -0,0 +1,73 @@
+// Code generated by mockery v2.20.0. DO NOT EDIT.
+
+package resource
+
+import mock "github.com/stretchr/testify/mock"
+
+// MockTenancyBridge is an autogenerated mock type for the TenancyBridge type
+type MockTenancyBridge struct {
+	mock.Mock
+}
+
+// NamespaceExists provides a mock function with given fields: partition, namespace
+func (_m *MockTenancyBridge) NamespaceExists(partition string, namespace string) (bool, error) {
+	ret := _m.Called(partition, namespace)
+
+	var r0 bool
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string, string) (bool, error)); ok {
+		return rf(partition, namespace)
+	}
+	if rf, ok := ret.Get(0).(func(string, string) bool); ok {
+		r0 = rf(partition, namespace)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	if rf, ok := ret.Get(1).(func(string, string) error); ok {
+		r1 = rf(partition, namespace)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// PartitionExists provides a mock function with given fields: partition
+func (_m *MockTenancyBridge) PartitionExists(partition string) (bool, error) {
+	ret := _m.Called(partition)
+
+	var r0 bool
+	var r1 error
+	if rf, ok := ret.Get(0).(func(string) (bool, error)); ok {
+		return rf(partition)
+	}
+	if rf, ok := ret.Get(0).(func(string) bool); ok {
+		r0 = rf(partition)
+	} else {
+		r0 = ret.Get(0).(bool)
+	}
+
+	if rf, ok := ret.Get(1).(func(string) error); ok {
+		r1 = rf(partition)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+type mockConstructorTestingTNewMockTenancyBridge interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewMockTenancyBridge creates a new instance of MockTenancyBridge. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewMockTenancyBridge(t mockConstructorTestingTNewMockTenancyBridge) *MockTenancyBridge {
+	mock := &MockTenancyBridge{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}
diff --git a/agent/grpc-external/services/resource/read.go b/agent/grpc-external/services/resource/read.go
index c75779183cb8..490909114b3f 100644
--- a/agent/grpc-external/services/resource/read.go
+++ b/agent/grpc-external/services/resource/read.go
@@ -11,28 +11,41 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/storage"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 )
 
 func (s *Server) Read(ctx context.Context, req *pbresource.ReadRequest) (*pbresource.ReadResponse, error) {
-	if err := validateReadRequest(req); err != nil {
-		return nil, err
-	}
-
-	// check type exists
-	reg, err := s.resolveType(req.Id.Type)
+	// Light first pass validation based on what user passed in and not much more.
+	reg, err := s.validateReadRequest(req)
 	if err != nil {
 		return nil, err
 	}
 
-	authz, err := s.getAuthorizer(tokenFromContext(ctx))
+	// acl.EnterpriseMeta acl.AuthorizerContext follow rules for V1 resources since they integrate with the V1 acl subsystem.
+	// pbresource.Tenacy follows rules for V2 resources and the Resource service.
+	// Example:
+	//
+	//    An OSS namespace scoped resource:
+	//      V1: EnterpriseMeta{}
+	//      V2: Tenancy {Partition: "default", Namespace: "default"}
+	//
+	//   An ENT namespace scoped resource:
+	//      V1: EnterpriseMeta{Partition: "default", Namespace: "default"}
+	//      V2: Tenancy {Partition: "default", Namespace: "default"}
+	//
+	// It is necessary to convert back and forth depending on which component supports which version, V1 or V2.
+	entMeta := v2TenancyToV1EntMeta(req.Id.Tenancy)
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), entMeta)
 	if err != nil {
 		return nil, err
 	}
 
-	// check acls
-	err = reg.ACLs.Read(authz, req.Id)
+	v1EntMetaToV2Tenancy(reg, entMeta, req.Id.Tenancy)
+
+	// ACL check comes before tenancy existence checks to not leak tenancy "existence".
+	err = reg.ACLs.Read(authz, authzContext, req.Id)
 	switch {
 	case acl.IsErrPermissionDenied(err):
 		return nil, status.Error(codes.PermissionDenied, err.Error())
@@ -40,6 +53,11 @@ func (s *Server) Read(ctx context.Context, req *pbresource.ReadRequest) (*pbreso
 		return nil, status.Errorf(codes.Internal, "failed read acl: %v", err)
 	}
 
+	// Check V1 tenancy exists for the V2 resource.
+	if err = v1TenancyExists(reg, s.V1TenancyBridge, req.Id.Tenancy); err != nil {
+		return nil, err
+	}
+
 	resource, err := s.Backend.Read(ctx, readConsistencyFrom(ctx), req.Id)
 	switch {
 	case err == nil:
@@ -53,13 +71,30 @@ func (s *Server) Read(ctx context.Context, req *pbresource.ReadRequest) (*pbreso
 	}
 }
 
-func validateReadRequest(req *pbresource.ReadRequest) error {
+func (s *Server) validateReadRequest(req *pbresource.ReadRequest) (*resource.Registration, error) {
 	if req.Id == nil {
-		return status.Errorf(codes.InvalidArgument, "id is required")
+		return nil, status.Errorf(codes.InvalidArgument, "id is required")
 	}
 
 	if err := validateId(req.Id, "id"); err != nil {
-		return err
+		return nil, err
 	}
-	return nil
+
+	// Check type exists.
+	reg, err := s.resolveType(req.Id.Type)
+	if err != nil {
+		return nil, err
+	}
+
+	// Check scope
+	if reg.Scope == resource.ScopePartition && req.Id.Tenancy.Namespace != "" {
+		return nil, status.Errorf(
+			codes.InvalidArgument,
+			"partition scoped resource %s cannot have a namespace. got: %s",
+			resource.ToGVK(req.Id.Type),
+			req.Id.Tenancy.Namespace,
+		)
+	}
+
+	return reg, nil
 }
diff --git a/agent/grpc-external/services/resource/read_test.go b/agent/grpc-external/services/resource/read_test.go
index cca911ec15b5..82085b451ff5 100644
--- a/agent/grpc-external/services/resource/read_test.go
+++ b/agent/grpc-external/services/resource/read_test.go
@@ -5,6 +5,7 @@ package resource
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -12,6 +13,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/hashicorp/consul/acl/resolver"
 	"github.com/hashicorp/consul/internal/resource"
@@ -24,35 +26,37 @@ import (
 func TestRead_InputValidation(t *testing.T) {
 	server := testServer(t)
 	client := testClient(t, server)
-
 	demo.RegisterTypes(server.Registry)
 
-	testCases := map[string]func(*pbresource.ReadRequest){
-		"no id":      func(req *pbresource.ReadRequest) { req.Id = nil },
-		"no type":    func(req *pbresource.ReadRequest) { req.Id.Type = nil },
-		"no tenancy": func(req *pbresource.ReadRequest) { req.Id.Tenancy = nil },
-		"no name":    func(req *pbresource.ReadRequest) { req.Id.Name = "" },
-		// clone necessary to not pollute DefaultTenancy
-		"tenancy partition not default": func(req *pbresource.ReadRequest) {
-			req.Id.Tenancy = clone(req.Id.Tenancy)
-			req.Id.Tenancy.Partition = ""
+	testCases := map[string]func(artistId, recordlabelId *pbresource.ID) *pbresource.ID{
+		"no id": 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.ReadRequest) {
-			req.Id.Tenancy = clone(req.Id.Tenancy)
-			req.Id.Tenancy.Namespace = ""
+		"no name": func(artistId, _ *pbresource.ID) *pbresource.ID {
+			artistId.Name = ""
+			return artistId
 		},
-		"tenancy peername not local": func(req *pbresource.ReadRequest) {
-			req.Id.Tenancy = clone(req.Id.Tenancy)
-			req.Id.Tenancy.PeerName = ""
+		"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)
 
-			req := &pbresource.ReadRequest{Id: res.Id}
-			modFn(req)
+			recordLabel, err := demo.GenerateV1RecordLabel("LoonyTunes")
+			require.NoError(t, err)
+
+			// Each test case picks which resource to use based on the resource type's scope.
+			req := &pbresource.ReadRequest{Id: modFn(artist.Id, recordLabel.Id)}
 
 			_, err = client.Read(testContext(t), req)
 			require.Error(t, err)
@@ -77,18 +81,50 @@ func TestRead_TypeNotFound(t *testing.T) {
 func TestRead_ResourceNotFound(t *testing.T) {
 	for desc, tc := range readTestCases() {
 		t.Run(desc, func(t *testing.T) {
-			server := testServer(t)
-
-			demo.RegisterTypes(server.Registry)
-			client := testClient(t, server)
-
-			artist, err := demo.GenerateV2Artist()
-			require.NoError(t, err)
-
-			_, err = client.Read(tc.ctx, &pbresource.ReadRequest{Id: artist.Id})
-			require.Error(t, err)
-			require.Equal(t, codes.NotFound.String(), status.Code(err).String())
-			require.Contains(t, err.Error(), "resource not found")
+			tenancyCases := map[string]func(artistId, recordlabelId *pbresource.ID) *pbresource.ID{
+				"resource not found by name": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					artistId.Name = "bogusname"
+					return artistId
+				},
+				"partition not found when namespace scoped": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					id := clone(artistId)
+					id.Tenancy.Partition = "boguspartition"
+					return id
+				},
+				"namespace not found when namespace scoped": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					id := clone(artistId)
+					id.Tenancy.Namespace = "bogusnamespace"
+					return id
+				},
+				"partition not found when partition scoped": func(_, recordLabelId *pbresource.ID) *pbresource.ID {
+					id := clone(recordLabelId)
+					id.Tenancy.Partition = "boguspartition"
+					return id
+				},
+			}
+			for tenancyDesc, modFn := range tenancyCases {
+				t.Run(tenancyDesc, 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(tc.ctx, recordLabel)
+					require.NoError(t, err)
+
+					artist, err := demo.GenerateV2Artist()
+					require.NoError(t, err)
+					artist, err = server.Backend.WriteCAS(tc.ctx, artist)
+					require.NoError(t, err)
+
+					// Each tenancy test case picks which resource to use based on the resource type's scope.
+					_, err = client.Read(tc.ctx, &pbresource.ReadRequest{Id: modFn(artist.Id, recordLabel.Id)})
+					require.Error(t, err)
+					require.Equal(t, codes.NotFound.String(), status.Code(err).String())
+					require.Contains(t, err.Error(), "resource not found")
+				})
+			}
 		})
 	}
 }
@@ -121,20 +157,77 @@ func TestRead_GroupVersionMismatch(t *testing.T) {
 func TestRead_Success(t *testing.T) {
 	for desc, tc := range readTestCases() {
 		t.Run(desc, func(t *testing.T) {
-			server := testServer(t)
-
-			demo.RegisterTypes(server.Registry)
-			client := testClient(t, server)
-
-			artist, err := demo.GenerateV2Artist()
-			require.NoError(t, err)
-
-			resource1, err := server.Backend.WriteCAS(tc.ctx, artist)
-			require.NoError(t, err)
-
-			rsp, err := client.Read(tc.ctx, &pbresource.ReadRequest{Id: artist.Id})
-			require.NoError(t, err)
-			prototest.AssertDeepEqual(t, resource1, rsp.Resource)
+			tenancyCases := map[string]func(artistId, recordlabelId *pbresource.ID) *pbresource.ID{
+				"namespaced resource provides nonempty partition and namespace": func(artistId, recordLabelId *pbresource.ID) *pbresource.ID {
+					return artistId
+				},
+				"namespaced resource provides uppercase namespace and partition": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					id := clone(artistId)
+					id.Tenancy.Partition = strings.ToUpper(artistId.Tenancy.Partition)
+					id.Tenancy.Namespace = strings.ToUpper(artistId.Tenancy.Namespace)
+					return id
+				},
+				"namespaced resource inherits tokens namespace when empty": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					id := clone(artistId)
+					id.Tenancy.Namespace = ""
+					return id
+				},
+				"namespaced resource inherits tokens partition when empty": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					id := clone(artistId)
+					id.Tenancy.Partition = ""
+					return id
+				},
+				"namespaced resource inherits tokens partition and namespace when empty": func(artistId, _ *pbresource.ID) *pbresource.ID {
+					id := clone(artistId)
+					id.Tenancy.Partition = ""
+					id.Tenancy.Namespace = ""
+					return id
+				},
+				"partitioned resource provides nonempty partition": func(_, recordLabelId *pbresource.ID) *pbresource.ID {
+					return recordLabelId
+				},
+				"partitioned resource provides uppercase partition": func(_, recordLabelId *pbresource.ID) *pbresource.ID {
+					id := clone(recordLabelId)
+					id.Tenancy.Partition = strings.ToUpper(recordLabelId.Tenancy.Partition)
+					return id
+				},
+				"partitioned resource inherits tokens partition when empty": func(_, recordLabelId *pbresource.ID) *pbresource.ID {
+					id := clone(recordLabelId)
+					id.Tenancy.Partition = ""
+					return id
+				},
+			}
+			for tenancyDesc, modFn := range tenancyCases {
+				t.Run(tenancyDesc, 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(tc.ctx, recordLabel)
+					require.NoError(t, err)
+
+					artist, err := demo.GenerateV2Artist()
+					require.NoError(t, err)
+					artist, err = server.Backend.WriteCAS(tc.ctx, artist)
+					require.NoError(t, err)
+
+					// Each tenancy test case picks which resource to use based on the resource type's scope.
+					req := &pbresource.ReadRequest{Id: modFn(artist.Id, recordLabel.Id)}
+					rsp, err := client.Read(tc.ctx, req)
+					require.NoError(t, err)
+
+					switch {
+					case proto.Equal(rsp.Resource.Id.Type, demo.TypeV2Artist):
+						prototest.AssertDeepEqual(t, artist, rsp.Resource)
+					case proto.Equal(rsp.Resource.Id.Type, demo.TypeV1RecordLabel):
+						prototest.AssertDeepEqual(t, recordLabel, rsp.Resource)
+					default:
+						require.Fail(t, "unexpected resource type")
+					}
+				})
+			}
 		})
 	}
 }
diff --git a/agent/grpc-external/services/resource/server.go b/agent/grpc-external/services/resource/server.go
index 51bb4610d527..e011c81c0fd0 100644
--- a/agent/grpc-external/services/resource/server.go
+++ b/agent/grpc-external/services/resource/server.go
@@ -31,6 +31,9 @@ type Config struct {
 	// Backend is the storage backend that will be used for resource persistence.
 	Backend     Backend
 	ACLResolver ACLResolver
+	// V1TenancyBridge temporarily allows us to use V1 implementations of
+	// partitions and namespaces until V2 implementations are available.
+	V1TenancyBridge TenancyBridge
 }
 
 //go:generate mockery --name Registry --inpackage
@@ -48,6 +51,12 @@ type ACLResolver interface {
 	ResolveTokenAndDefaultMeta(string, *acl.EnterpriseMeta, *acl.AuthorizerContext) (resolver.Result, error)
 }
 
+//go:generate mockery --name TenancyBridge --inpackage
+type TenancyBridge interface {
+	PartitionExists(partition string) (bool, error)
+	NamespaceExists(partition string, namespace string) (bool, error)
+}
+
 func NewServer(cfg Config) *Server {
 	return &Server{cfg}
 }
@@ -100,12 +109,13 @@ func readConsistencyFrom(ctx context.Context) storage.ReadConsistency {
 	return storage.EventualConsistency
 }
 
-func (s *Server) getAuthorizer(token string) (acl.Authorizer, error) {
-	authz, err := s.ACLResolver.ResolveTokenAndDefaultMeta(token, nil, nil)
+func (s *Server) getAuthorizer(token string, entMeta *acl.EnterpriseMeta) (acl.Authorizer, *acl.AuthorizerContext, error) {
+	authzContext := &acl.AuthorizerContext{}
+	authz, err := s.ACLResolver.ResolveTokenAndDefaultMeta(token, entMeta, authzContext)
 	if err != nil {
-		return nil, status.Errorf(codes.Internal, "failed getting authorizer: %v", err)
+		return nil, nil, status.Errorf(codes.Internal, "failed getting authorizer: %v", err)
 	}
-	return authz, nil
+	return authz, authzContext, nil
 }
 
 func isGRPCStatusError(err error) bool {
@@ -130,20 +140,30 @@ func validateId(id *pbresource.ID, errorPrefix string) error {
 	if field != "" {
 		return status.Errorf(codes.InvalidArgument, "%s.%s is required", errorPrefix, field)
 	}
+	resource.Normalize(id.Tenancy)
 
-	// Revisit defaulting and non-namespaced resources post-1.16
-	var expected string
-	switch {
-	case id.Tenancy.Partition != "default":
-		field, expected = "partition", "default"
-	case id.Tenancy.Namespace != "default":
-		field, expected = "namespace", "default"
-	case id.Tenancy.PeerName != "local":
-		field, expected = "peername", "local"
+	return nil
+}
+
+func v1TenancyExists(reg *resource.Registration, v1Bridge TenancyBridge, tenancy *pbresource.Tenancy) error {
+	if reg.Scope == resource.ScopePartition || reg.Scope == resource.ScopeNamespace {
+		exists, err := v1Bridge.PartitionExists(tenancy.Partition)
+		switch {
+		case err != nil:
+			return err
+		case !exists:
+			return status.Errorf(codes.NotFound, "partition resource not found: %v", tenancy.Partition)
+		}
 	}
 
-	if field != "" {
-		return status.Errorf(codes.InvalidArgument, "%s.tenancy.%s must be %s", errorPrefix, field, expected)
+	if reg.Scope == resource.ScopeNamespace {
+		exists, err := v1Bridge.NamespaceExists(tenancy.Partition, tenancy.Namespace)
+		switch {
+		case err != nil:
+			return err
+		case !exists:
+			return status.Errorf(codes.NotFound, "namespace resource not found: %v", tenancy.Namespace)
+		}
 	}
 	return nil
 }
diff --git a/agent/grpc-external/services/resource/server_oss.go b/agent/grpc-external/services/resource/server_oss.go
new file mode 100644
index 000000000000..1871aab205aa
--- /dev/null
+++ b/agent/grpc-external/services/resource/server_oss.go
@@ -0,0 +1,27 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+//go:build !consulent
+// +build !consulent
+
+package resource
+
+import (
+	"github.com/hashicorp/consul/acl"
+	"github.com/hashicorp/consul/internal/resource"
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+func v2TenancyToV1EntMeta(tenancy *pbresource.Tenancy) *acl.EnterpriseMeta {
+	return acl.DefaultEnterpriseMeta()
+}
+
+func v1EntMetaToV2Tenancy(reg *resource.Registration, entMeta *acl.EnterpriseMeta, tenancy *pbresource.Tenancy) {
+	if (reg.Scope == resource.ScopeNamespace || reg.Scope == resource.ScopePartition) && tenancy.Partition == "" {
+		tenancy.Partition = entMeta.PartitionOrDefault()
+	}
+
+	if reg.Scope == resource.ScopeNamespace && tenancy.Namespace == "" {
+		tenancy.Namespace = entMeta.NamespaceOrDefault()
+	}
+}
diff --git a/agent/grpc-external/services/resource/server_oss_test.go b/agent/grpc-external/services/resource/server_oss_test.go
new file mode 100644
index 000000000000..5f92f5226c74
--- /dev/null
+++ b/agent/grpc-external/services/resource/server_oss_test.go
@@ -0,0 +1,17 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+//go:build !consulent
+// +build !consulent
+
+package resource
+
+import "github.com/hashicorp/consul/acl"
+
+func fillEntMeta(entMeta *acl.EnterpriseMeta) {
+	return
+}
+
+func fillAuthorizerContext(authzContext *acl.AuthorizerContext) {
+	return
+}
diff --git a/agent/grpc-external/services/resource/server_test.go b/agent/grpc-external/services/resource/server_test.go
index a92fff38a326..0e14f73292d2 100644
--- a/agent/grpc-external/services/resource/server_test.go
+++ b/agent/grpc-external/services/resource/server_test.go
@@ -57,16 +57,36 @@ func testServer(t *testing.T) *Server {
 	require.NoError(t, err)
 	go backend.Run(testContext(t))
 
-	// Mock the ACL Resolver to allow everything for testing
+	// Mock the ACL Resolver to "allow all" for testing.
 	mockACLResolver := &MockACLResolver{}
 	mockACLResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
-		Return(testutils.ACLsDisabled(t), nil)
+		Return(testutils.ACLsDisabled(t), nil).
+		Run(func(args mock.Arguments) {
+			// Caller expecting passed in tokenEntMeta and authorizerContext to be filled in.
+			tokenEntMeta := args.Get(1).(*acl.EnterpriseMeta)
+			if tokenEntMeta != nil {
+				fillEntMeta(tokenEntMeta)
+			}
+
+			authzContext := args.Get(2).(*acl.AuthorizerContext)
+			if authzContext != nil {
+				fillAuthorizerContext(authzContext)
+			}
+		})
+
+	// Mock the V1 tenancy bridge since we can't use the real thing.
+	mockTenancyBridge := &MockTenancyBridge{}
+	mockTenancyBridge.On("PartitionExists", resource.DefaultPartitionName).Return(true, nil)
+	mockTenancyBridge.On("NamespaceExists", resource.DefaultPartitionName, resource.DefaultNamespaceName).Return(true, nil)
+	mockTenancyBridge.On("PartitionExists", mock.Anything).Return(false, nil)
+	mockTenancyBridge.On("NamespaceExists", mock.Anything, mock.Anything).Return(false, nil)
 
 	return NewServer(Config{
-		Logger:      testutil.Logger(t),
-		Registry:    resource.NewRegistry(),
-		Backend:     backend,
-		ACLResolver: mockACLResolver,
+		Logger:          testutil.Logger(t),
+		Registry:        resource.NewRegistry(),
+		Backend:         backend,
+		ACLResolver:     mockACLResolver,
+		V1TenancyBridge: mockTenancyBridge,
 	})
 }
 
diff --git a/agent/grpc-external/services/resource/testing/testing.go b/agent/grpc-external/services/resource/testing/testing.go
index e049b229b085..d6eb6c069200 100644
--- a/agent/grpc-external/services/resource/testing/testing.go
+++ b/agent/grpc-external/services/resource/testing/testing.go
@@ -46,7 +46,8 @@ func AuthorizerFrom(t *testing.T, policyStrs ...string) resolver.Result {
 }
 
 // RunResourceService runs a Resource Service for the duration of the test and
-// returns a client to interact with it. ACLs will be disabled.
+// returns a client to interact with it. ACLs will be disabled and only the
+// default partition and namespace are available.
 func RunResourceService(t *testing.T, registerFns ...func(resource.Registry)) pbresource.ResourceServiceClient {
 	return RunResourceServiceWithACL(t, resolver.DANGER_NO_AUTH{}, registerFns...)
 }
@@ -68,11 +69,16 @@ func RunResourceServiceWithACL(t *testing.T, aclResolver svc.ACLResolver, regist
 
 	server := grpc.NewServer()
 
+	mockTenancyBridge := &svc.MockTenancyBridge{}
+	mockTenancyBridge.On("PartitionExists", resource.DefaultPartitionName).Return(true, nil)
+	mockTenancyBridge.On("NamespaceExists", resource.DefaultPartitionName, resource.DefaultNamespaceName).Return(true, nil)
+
 	svc.NewServer(svc.Config{
-		Backend:     backend,
-		Registry:    registry,
-		Logger:      testutil.Logger(t),
-		ACLResolver: aclResolver,
+		Backend:         backend,
+		Registry:        registry,
+		Logger:          testutil.Logger(t),
+		ACLResolver:     aclResolver,
+		V1TenancyBridge: mockTenancyBridge,
 	}).Register(server)
 
 	pipe := internal.NewPipeListener()
diff --git a/agent/grpc-external/services/resource/watch.go b/agent/grpc-external/services/resource/watch.go
index 35ec14513ac3..140ff78fbdcb 100644
--- a/agent/grpc-external/services/resource/watch.go
+++ b/agent/grpc-external/services/resource/watch.go
@@ -25,7 +25,8 @@ func (s *Server) WatchList(req *pbresource.WatchListRequest, stream pbresource.R
 		return err
 	}
 
-	authz, err := s.getAuthorizer(tokenFromContext(stream.Context()))
+	// TODO(spatel): Refactor _ and entMeta as part of NET-4914
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(stream.Context()), acl.DefaultEnterpriseMeta())
 	if err != nil {
 		return err
 	}
@@ -66,7 +67,7 @@ func (s *Server) WatchList(req *pbresource.WatchListRequest, stream pbresource.R
 		}
 
 		// filter out items that don't pass read ACLs
-		err = reg.ACLs.Read(authz, event.Resource.Id)
+		err = reg.ACLs.Read(authz, authzContext, event.Resource.Id)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue
diff --git a/agent/grpc-external/services/resource/write.go b/agent/grpc-external/services/resource/write.go
index 3900612f07e1..f55645ae945d 100644
--- a/agent/grpc-external/services/resource/write.go
+++ b/agent/grpc-external/services/resource/write.go
@@ -46,7 +46,8 @@ func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbre
 		return nil, err
 	}
 
-	authz, err := s.getAuthorizer(tokenFromContext(ctx))
+	// TODO(spatel): Refactor _ and entMeta as part of NET-4911
+	authz, _, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
 	if err != nil {
 		return nil, err
 	}
diff --git a/agent/grpc-external/services/resource/write_status.go b/agent/grpc-external/services/resource/write_status.go
index 205918e1dc2b..67a2eff05387 100644
--- a/agent/grpc-external/services/resource/write_status.go
+++ b/agent/grpc-external/services/resource/write_status.go
@@ -20,7 +20,8 @@ import (
 )
 
 func (s *Server) WriteStatus(ctx context.Context, req *pbresource.WriteStatusRequest) (*pbresource.WriteStatusResponse, error) {
-	authz, err := s.getAuthorizer(tokenFromContext(ctx))
+	// TODO(spatel): Refactor _ and entMeta as part of NET-4912
+	authz, _, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
 	if err != nil {
 		return nil, err
 	}
diff --git a/agent/grpc-external/services/resource/write_test.go b/agent/grpc-external/services/resource/write_test.go
index 4ec25ee26c0c..8c886d9bf876 100644
--- a/agent/grpc-external/services/resource/write_test.go
+++ b/agent/grpc-external/services/resource/write_test.go
@@ -36,19 +36,22 @@ func TestWrite_InputValidation(t *testing.T) {
 		"no tenancy":  func(req *pbresource.WriteRequest) { req.Resource.Id.Tenancy = nil },
 		"no name":     func(req *pbresource.WriteRequest) { req.Resource.Id.Name = "" },
 		"no data":     func(req *pbresource.WriteRequest) { req.Resource.Data = nil },
-		// clone necessary to not pollute DefaultTenancy
-		"tenancy partition not default": func(req *pbresource.WriteRequest) {
-			req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
-			req.Resource.Id.Tenancy.Partition = ""
-		},
-		"tenancy namespace not default": func(req *pbresource.WriteRequest) {
-			req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
-			req.Resource.Id.Tenancy.Namespace = ""
-		},
-		"tenancy peername not local": func(req *pbresource.WriteRequest) {
-			req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
-			req.Resource.Id.Tenancy.PeerName = ""
-		},
+
+		// TODO(spatel): Refactor tenancy as part of NET-4911
+		//
+		// // clone necessary to not pollute DefaultTenancy
+		// "tenancy partition not default": func(req *pbresource.WriteRequest) {
+		// 	req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
+		// 	req.Resource.Id.Tenancy.Partition = ""
+		// },
+		// "tenancy namespace not default": func(req *pbresource.WriteRequest) {
+		// 	req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
+		// 	req.Resource.Id.Tenancy.Namespace = ""
+		// },
+		// "tenancy peername not local": func(req *pbresource.WriteRequest) {
+		// 	req.Resource.Id.Tenancy = clone(req.Resource.Id.Tenancy)
+		// 	req.Resource.Id.Tenancy.PeerName = ""
+		// },
 		"wrong data type": func(req *pbresource.WriteRequest) {
 			var err error
 			req.Resource.Data, err = anypb.New(&pbdemov2.Album{})
@@ -99,28 +102,31 @@ func TestWrite_OwnerValidation(t *testing.T) {
 			modReqFn:      func(req *pbresource.WriteRequest) { req.Resource.Owner.Name = "" },
 			errorContains: "resource.owner.name",
 		},
-		// clone necessary to not pollute DefaultTenancy
-		"owner tenancy partition not default": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
-				req.Resource.Owner.Tenancy.Partition = ""
-			},
-			errorContains: "resource.owner.tenancy.partition",
-		},
-		"owner tenancy namespace not default": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
-				req.Resource.Owner.Tenancy.Namespace = ""
-			},
-			errorContains: "resource.owner.tenancy.namespace",
-		},
-		"owner tenancy peername not local": {
-			modReqFn: func(req *pbresource.WriteRequest) {
-				req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
-				req.Resource.Owner.Tenancy.PeerName = ""
-			},
-			errorContains: "resource.owner.tenancy.peername",
-		},
+
+		// TODO(spatel): Refactor tenancy as part of NET-4911
+		//
+		// // clone necessary to not pollute DefaultTenancy
+		// "owner tenancy partition not default": {
+		// 	modReqFn: func(req *pbresource.WriteRequest) {
+		// 		req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
+		// 		req.Resource.Owner.Tenancy.Partition = ""
+		// 	},
+		// 	errorContains: "resource.owner.tenancy.partition",
+		// },
+		// "owner tenancy namespace not default": {
+		// 	modReqFn: func(req *pbresource.WriteRequest) {
+		// 		req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
+		// 		req.Resource.Owner.Tenancy.Namespace = ""
+		// 	},
+		// 	errorContains: "resource.owner.tenancy.namespace",
+		// },
+		// "owner tenancy peername not local": {
+		// 	modReqFn: func(req *pbresource.WriteRequest) {
+		// 		req.Resource.Owner.Tenancy = clone(req.Resource.Owner.Tenancy)
+		// 		req.Resource.Owner.Tenancy.PeerName = ""
+		// 	},
+		// 	errorContains: "resource.owner.tenancy.peername",
+		// },
 	}
 	for desc, tc := range testCases {
 		t.Run(desc, func(t *testing.T) {
diff --git a/internal/mesh/internal/types/proxy_state_template.go b/internal/mesh/internal/types/proxy_state_template.go
index 7f46190ea015..84195ed2c665 100644
--- a/internal/mesh/internal/types/proxy_state_template.go
+++ b/internal/mesh/internal/types/proxy_state_template.go
@@ -27,13 +27,13 @@ func RegisterProxyStateTemplate(r resource.Registry) {
 		Proto:    &pbmesh.ProxyStateTemplate{},
 		Validate: nil,
 		ACLs: &resource.ACLHooks{
-			Read: func(authorizer acl.Authorizer, id *pbresource.ID) error {
+			Read: func(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
 				// Check service:read and operator:read permissions.
 				// If service:read is not allowed, check operator:read. We want to allow both as this
 				// resource is mostly useful for debuggability and we want to cover
 				// the most cases that serve that purpose.
-				serviceReadErr := authorizer.ToAllowAuthorizer().ServiceReadAllowed(id.Name, resource.AuthorizerContext(id.Tenancy))
-				operatorReadErr := authorizer.ToAllowAuthorizer().OperatorReadAllowed(resource.AuthorizerContext(id.Tenancy))
+				serviceReadErr := authorizer.ToAllowAuthorizer().ServiceReadAllowed(id.Name, authzContext)
+				operatorReadErr := authorizer.ToAllowAuthorizer().OperatorReadAllowed(authzContext)
 
 				switch {
 				case serviceReadErr != nil:
diff --git a/internal/resource/demo/demo.go b/internal/resource/demo/demo.go
index 88fe7134c074..6fd79f0b6cfb 100644
--- a/internal/resource/demo/demo.go
+++ b/internal/resource/demo/demo.go
@@ -24,9 +24,17 @@ import (
 var (
 	// TenancyDefault contains the default values for all tenancy units.
 	TenancyDefault = &pbresource.Tenancy{
-		Partition: "default",
+		Partition: resource.DefaultPartitionName,
 		PeerName:  "local",
-		Namespace: "default",
+		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{
+		Group:        "demo",
+		GroupVersion: "v1",
+		Kind:         "RecordLabel",
 	}
 
 	// TypeV1Artist represents a musician or group of musicians.
@@ -72,9 +80,9 @@ const (
 // TODO(spatel): We're standing-in key ACLs for demo resources until our ACL
 // system can be more modularly extended (or support generic resource permissions).
 func RegisterTypes(r resource.Registry) {
-	readACL := func(authz acl.Authorizer, id *pbresource.ID) error {
+	readACL := func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
 		key := fmt.Sprintf("resource/%s/%s", resource.ToGVK(id.Type), id.Name)
-		return authz.ToAllowAuthorizer().KeyReadAllowed(key, &acl.AuthorizerContext{})
+		return authz.ToAllowAuthorizer().KeyReadAllowed(key, authzContext)
 	}
 
 	writeACL := func(authz acl.Authorizer, res *pbresource.Resource) error {
@@ -124,6 +132,17 @@ func RegisterTypes(r resource.Registry) {
 		return nil
 	}
 
+	r.Register(resource.Registration{
+		Type:  TypeV1RecordLabel,
+		Proto: &pbdemov1.RecordLabel{},
+		ACLs: &resource.ACLHooks{
+			Read:  readACL,
+			Write: writeACL,
+			List:  makeListACL(TypeV1RecordLabel),
+		},
+		Scope: resource.ScopePartition,
+	})
+
 	r.Register(resource.Registration{
 		Type:  TypeV1Artist,
 		Proto: &pbdemov1.Artist{},
@@ -172,6 +191,28 @@ func RegisterTypes(r resource.Registry) {
 	})
 }
 
+// GenerateV1RecordLabel generates a named RecordLabel resource.
+func GenerateV1RecordLabel(name string) (*pbresource.Resource, error) {
+	data, err := anypb.New(&pbdemov1.RecordLabel{Name: name})
+	if err != nil {
+		return nil, err
+	}
+
+	return &pbresource.Resource{
+		Id: &pbresource.ID{
+			Type: TypeV1RecordLabel,
+			Tenancy: &pbresource.Tenancy{
+				Partition: resource.DefaultPartitionName,
+			},
+			Name: name,
+		},
+		Data: data,
+		Metadata: map[string]string{
+			"generated_at": time.Now().Format(time.RFC3339),
+		},
+	}, nil
+}
+
 // GenerateV2Artist generates a random Artist resource.
 func GenerateV2Artist() (*pbresource.Resource, error) {
 	adjective := adjectives[rand.Intn(len(adjectives))]
diff --git a/internal/resource/http/http_test.go b/internal/resource/http/http_test.go
index d385a861c3d0..80e506cd92de 100644
--- a/internal/resource/http/http_test.go
+++ b/internal/resource/http/http_test.go
@@ -80,22 +80,6 @@ func TestResourceHandler_InputValidation(t *testing.T) {
 			response:             httptest.NewRecorder(),
 			expectedResponseCode: http.StatusBadRequest,
 		},
-		{
-			description: "missing tenancy info",
-			request: httptest.NewRequest("PUT", "/keith-urban?partition=default&peer_name=local", strings.NewReader(`
-				{
-					"metadata": {
-						"foo": "bar"
-					},
-					"data": {
-						"name": "Keith Urban",
-						"genre": "GENRE_COUNTRY"
-					}
-				}
-			`)),
-			response:             httptest.NewRecorder(),
-			expectedResponseCode: http.StatusBadRequest,
-		},
 	}
 
 	for _, tc := range testCases {
diff --git a/internal/resource/registry.go b/internal/resource/registry.go
index 3bedbdebf893..afc3f25bcd4c 100644
--- a/internal/resource/registry.go
+++ b/internal/resource/registry.go
@@ -20,34 +20,6 @@ var (
 	kindRegexp         = regexp.MustCompile(`^[A-Z][A-Za-z\d]+$`)
 )
 
-// Scope describes the tenancy scope of a resource.
-type Scope int
-
-const (
-	// There is no default scope, it must be set explicitly.
-	ScopeUndefined Scope = iota
-	// ScopeCluster describes a resource that is scoped to a cluster.
-	ScopeCluster
-	// ScopePartition describes a resource that is scoped to a partition.
-	ScopePartition
-	// ScopeNamespace applies to a resource that is scoped to a partition and namespace.
-	ScopeNamespace
-)
-
-func (s Scope) String() string {
-	switch s {
-	case ScopeUndefined:
-		return "undefined"
-	case ScopeCluster:
-		return "cluster"
-	case ScopePartition:
-		return "partition"
-	case ScopeNamespace:
-		return "namespace"
-	}
-	panic(fmt.Sprintf("string mapping missing for scope %v", int(s)))
-}
-
 type Registry interface {
 	// Register the given resource type and its hooks.
 	Register(reg Registration)
@@ -85,7 +57,7 @@ type ACLHooks struct {
 	// RPCs.
 	//
 	// If it is omitted, `operator:read` permission is assumed.
-	Read func(acl.Authorizer, *pbresource.ID) error
+	Read func(acl.Authorizer, *acl.AuthorizerContext, *pbresource.ID) error
 
 	// Write is used to authorize Write and Delete RPCs.
 	//
@@ -147,8 +119,8 @@ func (r *TypeRegistry) Register(registration Registration) {
 		registration.ACLs = &ACLHooks{}
 	}
 	if registration.ACLs.Read == nil {
-		registration.ACLs.Read = func(authz acl.Authorizer, id *pbresource.ID) error {
-			return authz.ToAllowAuthorizer().OperatorReadAllowed(&acl.AuthorizerContext{})
+		registration.ACLs.Read = func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
+			return authz.ToAllowAuthorizer().OperatorReadAllowed(authzContext)
 		}
 	}
 	if registration.ACLs.Write == nil {
diff --git a/internal/resource/registry_test.go b/internal/resource/registry_test.go
index 7d8bb2433921..d717f46f59ce 100644
--- a/internal/resource/registry_test.go
+++ b/internal/resource/registry_test.go
@@ -43,8 +43,8 @@ func TestRegister_Defaults(t *testing.T) {
 	require.True(t, ok)
 
 	// verify default read hook requires operator:read
-	require.NoError(t, reg.ACLs.Read(testutils.ACLOperatorRead(t), artist.Id))
-	require.True(t, acl.IsErrPermissionDenied(reg.ACLs.Read(testutils.ACLNoPermissions(t), artist.Id)))
+	require.NoError(t, reg.ACLs.Read(testutils.ACLOperatorRead(t), nil, artist.Id))
+	require.True(t, acl.IsErrPermissionDenied(reg.ACLs.Read(testutils.ACLNoPermissions(t), nil, artist.Id)))
 
 	// verify default write hook requires operator:write
 	require.NoError(t, reg.ACLs.Write(testutils.ACLOperatorWrite(t), artist))
diff --git a/internal/resource/resourcetest/builder.go b/internal/resource/resourcetest/builder.go
index 395cba57b28f..38f1a6e3ec4a 100644
--- a/internal/resource/resourcetest/builder.go
+++ b/internal/resource/resourcetest/builder.go
@@ -37,8 +37,8 @@ func Resource(rtype *pbresource.Type, name string) *resourceBuilder {
 					Kind:         rtype.Kind,
 				},
 				Tenancy: &pbresource.Tenancy{
-					Partition: "default",
-					Namespace: "default",
+					Partition: resource.DefaultPartitionName,
+					Namespace: resource.DefaultNamespaceName,
 					PeerName:  "local",
 				},
 				Name: name,
diff --git a/internal/resource/tenancy.go b/internal/resource/tenancy.go
new file mode 100644
index 000000000000..e783a004fc3f
--- /dev/null
+++ b/internal/resource/tenancy.go
@@ -0,0 +1,53 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: MPL-2.0
+
+package resource
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/hashicorp/consul/proto-public/pbresource"
+)
+
+const (
+	DefaultPartitionName = "default"
+	DefaultNamespaceName = "default"
+)
+
+// Scope describes the tenancy scope of a resource.
+type Scope int
+
+const (
+	// There is no default scope, it must be set explicitly.
+	ScopeUndefined Scope = iota
+	// ScopeCluster describes a resource that is scoped to a cluster.
+	ScopeCluster
+	// ScopePartition describes a resource that is scoped to a partition.
+	ScopePartition
+	// ScopeNamespace applies to a resource that is scoped to a partition and namespace.
+	ScopeNamespace
+)
+
+func (s Scope) String() string {
+	switch s {
+	case ScopeUndefined:
+		return "undefined"
+	case ScopeCluster:
+		return "cluster"
+	case ScopePartition:
+		return "partition"
+	case ScopeNamespace:
+		return "namespace"
+	}
+	panic(fmt.Sprintf("string mapping missing for scope %v", int(s)))
+}
+
+// Normalize lowercases partition and namespace.
+func Normalize(tenancy *pbresource.Tenancy) {
+	if tenancy == nil {
+		return
+	}
+	tenancy.Partition = strings.ToLower(tenancy.Partition)
+	tenancy.Namespace = strings.ToLower(tenancy.Namespace)
+}
diff --git a/proto/private/pbdemo/v1/demo.pb.binary.go b/proto/private/pbdemo/v1/demo.pb.binary.go
index 45a3c34e5536..b52d5f33496e 100644
--- a/proto/private/pbdemo/v1/demo.pb.binary.go
+++ b/proto/private/pbdemo/v1/demo.pb.binary.go
@@ -7,6 +7,16 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+// MarshalBinary implements encoding.BinaryMarshaler
+func (msg *RecordLabel) MarshalBinary() ([]byte, error) {
+	return proto.Marshal(msg)
+}
+
+// UnmarshalBinary implements encoding.BinaryUnmarshaler
+func (msg *RecordLabel) UnmarshalBinary(b []byte) error {
+	return proto.Unmarshal(b, msg)
+}
+
 // MarshalBinary implements encoding.BinaryMarshaler
 func (msg *Artist) MarshalBinary() ([]byte, error) {
 	return proto.Marshal(msg)
diff --git a/proto/private/pbdemo/v1/demo.pb.go b/proto/private/pbdemo/v1/demo.pb.go
index 9ee119d76f54..a69008fce767 100644
--- a/proto/private/pbdemo/v1/demo.pb.go
+++ b/proto/private/pbdemo/v1/demo.pb.go
@@ -105,6 +105,61 @@ func (Genre) EnumDescriptor() ([]byte, []int) {
 	return file_private_pbdemo_v1_demo_proto_rawDescGZIP(), []int{0}
 }
 
+type RecordLabel struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	Name        string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
+	Description string `protobuf:"bytes,2,opt,name=description,proto3" json:"description,omitempty"`
+}
+
+func (x *RecordLabel) Reset() {
+	*x = RecordLabel{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_private_pbdemo_v1_demo_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *RecordLabel) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*RecordLabel) ProtoMessage() {}
+
+func (x *RecordLabel) ProtoReflect() protoreflect.Message {
+	mi := &file_private_pbdemo_v1_demo_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use RecordLabel.ProtoReflect.Descriptor instead.
+func (*RecordLabel) Descriptor() ([]byte, []int) {
+	return file_private_pbdemo_v1_demo_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *RecordLabel) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *RecordLabel) GetDescription() string {
+	if x != nil {
+		return x.Description
+	}
+	return ""
+}
+
 type Artist struct {
 	state         protoimpl.MessageState
 	sizeCache     protoimpl.SizeCache
@@ -119,7 +174,7 @@ type Artist struct {
 func (x *Artist) Reset() {
 	*x = Artist{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_private_pbdemo_v1_demo_proto_msgTypes[0]
+		mi := &file_private_pbdemo_v1_demo_proto_msgTypes[1]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -132,7 +187,7 @@ func (x *Artist) String() string {
 func (*Artist) ProtoMessage() {}
 
 func (x *Artist) ProtoReflect() protoreflect.Message {
-	mi := &file_private_pbdemo_v1_demo_proto_msgTypes[0]
+	mi := &file_private_pbdemo_v1_demo_proto_msgTypes[1]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -145,7 +200,7 @@ func (x *Artist) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Artist.ProtoReflect.Descriptor instead.
 func (*Artist) Descriptor() ([]byte, []int) {
-	return file_private_pbdemo_v1_demo_proto_rawDescGZIP(), []int{0}
+	return file_private_pbdemo_v1_demo_proto_rawDescGZIP(), []int{1}
 }
 
 func (x *Artist) GetName() string {
@@ -190,7 +245,7 @@ type Album struct {
 func (x *Album) Reset() {
 	*x = Album{}
 	if protoimpl.UnsafeEnabled {
-		mi := &file_private_pbdemo_v1_demo_proto_msgTypes[1]
+		mi := &file_private_pbdemo_v1_demo_proto_msgTypes[2]
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		ms.StoreMessageInfo(mi)
 	}
@@ -203,7 +258,7 @@ func (x *Album) String() string {
 func (*Album) ProtoMessage() {}
 
 func (x *Album) ProtoReflect() protoreflect.Message {
-	mi := &file_private_pbdemo_v1_demo_proto_msgTypes[1]
+	mi := &file_private_pbdemo_v1_demo_proto_msgTypes[2]
 	if protoimpl.UnsafeEnabled && x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -216,7 +271,7 @@ func (x *Album) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Album.ProtoReflect.Descriptor instead.
 func (*Album) Descriptor() ([]byte, []int) {
-	return file_private_pbdemo_v1_demo_proto_rawDescGZIP(), []int{1}
+	return file_private_pbdemo_v1_demo_proto_rawDescGZIP(), []int{2}
 }
 
 func (x *Album) GetName() string {
@@ -254,59 +309,63 @@ var file_private_pbdemo_v1_demo_proto_rawDesc = []byte{
 	0x2f, 0x76, 0x31, 0x2f, 0x64, 0x65, 0x6d, 0x6f, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x21,
 	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c,
 	0x2e, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x64, 0x65, 0x6d, 0x6f, 0x2e, 0x76,
-	0x31, 0x22, 0xa3, 0x01, 0x0a, 0x06, 0x41, 0x72, 0x74, 0x69, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04,
-	0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65,
-	0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x18,
-	0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69,
-	0x6f, 0x6e, 0x12, 0x3e, 0x0a, 0x05, 0x67, 0x65, 0x6e, 0x72, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28,
-	0x0e, 0x32, 0x28, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f,
-	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x64, 0x65,
-	0x6d, 0x6f, 0x2e, 0x76, 0x31, 0x2e, 0x47, 0x65, 0x6e, 0x72, 0x65, 0x52, 0x05, 0x67, 0x65, 0x6e,
-	0x72, 0x65, 0x12, 0x23, 0x0a, 0x0d, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x6d, 0x65, 0x6d, 0x62,
-	0x65, 0x72, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0c, 0x67, 0x72, 0x6f, 0x75, 0x70,
-	0x4d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x73, 0x22, 0x8c, 0x01, 0x0a, 0x05, 0x41, 0x6c, 0x62, 0x75,
-	0x6d, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
-	0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x26, 0x0a, 0x0f, 0x79, 0x65, 0x61, 0x72, 0x5f, 0x6f, 0x66,
-	0x5f, 0x72, 0x65, 0x6c, 0x65, 0x61, 0x73, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0d,
-	0x79, 0x65, 0x61, 0x72, 0x4f, 0x66, 0x52, 0x65, 0x6c, 0x65, 0x61, 0x73, 0x65, 0x12, 0x2f, 0x0a,
-	0x13, 0x63, 0x72, 0x69, 0x74, 0x69, 0x63, 0x61, 0x6c, 0x6c, 0x79, 0x5f, 0x61, 0x63, 0x6c, 0x61,
-	0x69, 0x6d, 0x65, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x12, 0x63, 0x72, 0x69, 0x74,
-	0x69, 0x63, 0x61, 0x6c, 0x6c, 0x79, 0x41, 0x63, 0x6c, 0x61, 0x69, 0x6d, 0x65, 0x64, 0x12, 0x16,
-	0x0a, 0x06, 0x74, 0x72, 0x61, 0x63, 0x6b, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x09, 0x52, 0x06,
-	0x74, 0x72, 0x61, 0x63, 0x6b, 0x73, 0x2a, 0xe9, 0x01, 0x0a, 0x05, 0x47, 0x65, 0x6e, 0x72, 0x65,
-	0x12, 0x15, 0x0a, 0x11, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x55, 0x4e, 0x53, 0x50, 0x45, 0x43,
-	0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a, 0x47, 0x45, 0x4e, 0x52, 0x45,
-	0x5f, 0x4a, 0x41, 0x5a, 0x5a, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a, 0x47, 0x45, 0x4e, 0x52, 0x45,
-	0x5f, 0x46, 0x4f, 0x4c, 0x4b, 0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09, 0x47, 0x45, 0x4e, 0x52, 0x45,
-	0x5f, 0x50, 0x4f, 0x50, 0x10, 0x03, 0x12, 0x0f, 0x0a, 0x0b, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f,
-	0x4d, 0x45, 0x54, 0x41, 0x4c, 0x10, 0x04, 0x12, 0x0e, 0x0a, 0x0a, 0x47, 0x45, 0x4e, 0x52, 0x45,
-	0x5f, 0x50, 0x55, 0x4e, 0x4b, 0x10, 0x05, 0x12, 0x0f, 0x0a, 0x0b, 0x47, 0x45, 0x4e, 0x52, 0x45,
-	0x5f, 0x42, 0x4c, 0x55, 0x45, 0x53, 0x10, 0x06, 0x12, 0x11, 0x0a, 0x0d, 0x47, 0x45, 0x4e, 0x52,
-	0x45, 0x5f, 0x52, 0x5f, 0x41, 0x4e, 0x44, 0x5f, 0x42, 0x10, 0x07, 0x12, 0x11, 0x0a, 0x0d, 0x47,
-	0x45, 0x4e, 0x52, 0x45, 0x5f, 0x43, 0x4f, 0x55, 0x4e, 0x54, 0x52, 0x59, 0x10, 0x08, 0x12, 0x0f,
-	0x0a, 0x0b, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x44, 0x49, 0x53, 0x43, 0x4f, 0x10, 0x09, 0x12,
-	0x0d, 0x0a, 0x09, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x53, 0x4b, 0x41, 0x10, 0x0a, 0x12, 0x11,
-	0x0a, 0x0d, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x48, 0x49, 0x50, 0x5f, 0x48, 0x4f, 0x50, 0x10,
-	0x0b, 0x12, 0x0f, 0x0a, 0x0b, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x49, 0x4e, 0x44, 0x49, 0x45,
-	0x10, 0x0c, 0x42, 0x97, 0x02, 0x0a, 0x25, 0x63, 0x6f, 0x6d, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69,
-	0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x69, 0x6e, 0x74, 0x65,
-	0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x64, 0x65, 0x6d, 0x6f, 0x2e, 0x76, 0x31, 0x42, 0x09, 0x44, 0x65,
-	0x6d, 0x6f, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x3a, 0x67, 0x69, 0x74, 0x68, 0x75,
-	0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2f,
-	0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x2f, 0x70, 0x72, 0x69,
-	0x76, 0x61, 0x74, 0x65, 0x2f, 0x70, 0x62, 0x64, 0x65, 0x6d, 0x6f, 0x2f, 0x76, 0x31, 0x3b, 0x64,
-	0x65, 0x6d, 0x6f, 0x76, 0x31, 0xa2, 0x02, 0x04, 0x48, 0x43, 0x49, 0x44, 0xaa, 0x02, 0x21, 0x48,
-	0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e,
-	0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x44, 0x65, 0x6d, 0x6f, 0x2e, 0x56, 0x31,
-	0xca, 0x02, 0x21, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e,
-	0x73, 0x75, 0x6c, 0x5c, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x5c, 0x44, 0x65, 0x6d,
-	0x6f, 0x5c, 0x56, 0x31, 0xe2, 0x02, 0x2d, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
-	0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c,
-	0x5c, 0x44, 0x65, 0x6d, 0x6f, 0x5c, 0x56, 0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61,
-	0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x25, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70,
-	0x3a, 0x3a, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x3a, 0x3a, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e,
-	0x61, 0x6c, 0x3a, 0x3a, 0x44, 0x65, 0x6d, 0x6f, 0x3a, 0x3a, 0x56, 0x31, 0x62, 0x06, 0x70, 0x72,
-	0x6f, 0x74, 0x6f, 0x33,
+	0x31, 0x22, 0x43, 0x0a, 0x0b, 0x52, 0x65, 0x63, 0x6f, 0x72, 0x64, 0x4c, 0x61, 0x62, 0x65, 0x6c,
+	0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04,
+	0x6e, 0x61, 0x6d, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74,
+	0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72,
+	0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0xa3, 0x01, 0x0a, 0x06, 0x41, 0x72, 0x74, 0x69, 0x73,
+	0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52,
+	0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70,
+	0x74, 0x69, 0x6f, 0x6e, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63,
+	0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x3e, 0x0a, 0x05, 0x67, 0x65, 0x6e, 0x72, 0x65,
+	0x18, 0x03, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x28, 0x2e, 0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f,
+	0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e,
+	0x61, 0x6c, 0x2e, 0x64, 0x65, 0x6d, 0x6f, 0x2e, 0x76, 0x31, 0x2e, 0x47, 0x65, 0x6e, 0x72, 0x65,
+	0x52, 0x05, 0x67, 0x65, 0x6e, 0x72, 0x65, 0x12, 0x23, 0x0a, 0x0d, 0x67, 0x72, 0x6f, 0x75, 0x70,
+	0x5f, 0x6d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x73, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0c,
+	0x67, 0x72, 0x6f, 0x75, 0x70, 0x4d, 0x65, 0x6d, 0x62, 0x65, 0x72, 0x73, 0x22, 0x8c, 0x01, 0x0a,
+	0x05, 0x41, 0x6c, 0x62, 0x75, 0x6d, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01,
+	0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x26, 0x0a, 0x0f, 0x79, 0x65,
+	0x61, 0x72, 0x5f, 0x6f, 0x66, 0x5f, 0x72, 0x65, 0x6c, 0x65, 0x61, 0x73, 0x65, 0x18, 0x02, 0x20,
+	0x01, 0x28, 0x05, 0x52, 0x0d, 0x79, 0x65, 0x61, 0x72, 0x4f, 0x66, 0x52, 0x65, 0x6c, 0x65, 0x61,
+	0x73, 0x65, 0x12, 0x2f, 0x0a, 0x13, 0x63, 0x72, 0x69, 0x74, 0x69, 0x63, 0x61, 0x6c, 0x6c, 0x79,
+	0x5f, 0x61, 0x63, 0x6c, 0x61, 0x69, 0x6d, 0x65, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52,
+	0x12, 0x63, 0x72, 0x69, 0x74, 0x69, 0x63, 0x61, 0x6c, 0x6c, 0x79, 0x41, 0x63, 0x6c, 0x61, 0x69,
+	0x6d, 0x65, 0x64, 0x12, 0x16, 0x0a, 0x06, 0x74, 0x72, 0x61, 0x63, 0x6b, 0x73, 0x18, 0x04, 0x20,
+	0x03, 0x28, 0x09, 0x52, 0x06, 0x74, 0x72, 0x61, 0x63, 0x6b, 0x73, 0x2a, 0xe9, 0x01, 0x0a, 0x05,
+	0x47, 0x65, 0x6e, 0x72, 0x65, 0x12, 0x15, 0x0a, 0x11, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x55,
+	0x4e, 0x53, 0x50, 0x45, 0x43, 0x49, 0x46, 0x49, 0x45, 0x44, 0x10, 0x00, 0x12, 0x0e, 0x0a, 0x0a,
+	0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x4a, 0x41, 0x5a, 0x5a, 0x10, 0x01, 0x12, 0x0e, 0x0a, 0x0a,
+	0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x46, 0x4f, 0x4c, 0x4b, 0x10, 0x02, 0x12, 0x0d, 0x0a, 0x09,
+	0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x50, 0x4f, 0x50, 0x10, 0x03, 0x12, 0x0f, 0x0a, 0x0b, 0x47,
+	0x45, 0x4e, 0x52, 0x45, 0x5f, 0x4d, 0x45, 0x54, 0x41, 0x4c, 0x10, 0x04, 0x12, 0x0e, 0x0a, 0x0a,
+	0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x50, 0x55, 0x4e, 0x4b, 0x10, 0x05, 0x12, 0x0f, 0x0a, 0x0b,
+	0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x42, 0x4c, 0x55, 0x45, 0x53, 0x10, 0x06, 0x12, 0x11, 0x0a,
+	0x0d, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x52, 0x5f, 0x41, 0x4e, 0x44, 0x5f, 0x42, 0x10, 0x07,
+	0x12, 0x11, 0x0a, 0x0d, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x43, 0x4f, 0x55, 0x4e, 0x54, 0x52,
+	0x59, 0x10, 0x08, 0x12, 0x0f, 0x0a, 0x0b, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x44, 0x49, 0x53,
+	0x43, 0x4f, 0x10, 0x09, 0x12, 0x0d, 0x0a, 0x09, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x53, 0x4b,
+	0x41, 0x10, 0x0a, 0x12, 0x11, 0x0a, 0x0d, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f, 0x48, 0x49, 0x50,
+	0x5f, 0x48, 0x4f, 0x50, 0x10, 0x0b, 0x12, 0x0f, 0x0a, 0x0b, 0x47, 0x45, 0x4e, 0x52, 0x45, 0x5f,
+	0x49, 0x4e, 0x44, 0x49, 0x45, 0x10, 0x0c, 0x42, 0x97, 0x02, 0x0a, 0x25, 0x63, 0x6f, 0x6d, 0x2e,
+	0x68, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c,
+	0x2e, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x64, 0x65, 0x6d, 0x6f, 0x2e, 0x76,
+	0x31, 0x42, 0x09, 0x44, 0x65, 0x6d, 0x6f, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x3a,
+	0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x68, 0x61, 0x73, 0x68, 0x69,
+	0x63, 0x6f, 0x72, 0x70, 0x2f, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x2f, 0x70, 0x72, 0x6f, 0x74,
+	0x6f, 0x2f, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x2f, 0x70, 0x62, 0x64, 0x65, 0x6d, 0x6f,
+	0x2f, 0x76, 0x31, 0x3b, 0x64, 0x65, 0x6d, 0x6f, 0x76, 0x31, 0xa2, 0x02, 0x04, 0x48, 0x43, 0x49,
+	0x44, 0xaa, 0x02, 0x21, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72, 0x70, 0x2e, 0x43, 0x6f,
+	0x6e, 0x73, 0x75, 0x6c, 0x2e, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x44, 0x65,
+	0x6d, 0x6f, 0x2e, 0x56, 0x31, 0xca, 0x02, 0x21, 0x48, 0x61, 0x73, 0x68, 0x69, 0x63, 0x6f, 0x72,
+	0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61,
+	0x6c, 0x5c, 0x44, 0x65, 0x6d, 0x6f, 0x5c, 0x56, 0x31, 0xe2, 0x02, 0x2d, 0x48, 0x61, 0x73, 0x68,
+	0x69, 0x63, 0x6f, 0x72, 0x70, 0x5c, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x5c, 0x49, 0x6e, 0x74,
+	0x65, 0x72, 0x6e, 0x61, 0x6c, 0x5c, 0x44, 0x65, 0x6d, 0x6f, 0x5c, 0x56, 0x31, 0x5c, 0x47, 0x50,
+	0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x25, 0x48, 0x61, 0x73, 0x68,
+	0x69, 0x63, 0x6f, 0x72, 0x70, 0x3a, 0x3a, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6c, 0x3a, 0x3a, 0x49,
+	0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x3a, 0x3a, 0x44, 0x65, 0x6d, 0x6f, 0x3a, 0x3a, 0x56,
+	0x31, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
 }
 
 var (
@@ -322,11 +381,12 @@ func file_private_pbdemo_v1_demo_proto_rawDescGZIP() []byte {
 }
 
 var file_private_pbdemo_v1_demo_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
-var file_private_pbdemo_v1_demo_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
+var file_private_pbdemo_v1_demo_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
 var file_private_pbdemo_v1_demo_proto_goTypes = []interface{}{
-	(Genre)(0),     // 0: hashicorp.consul.internal.demo.v1.Genre
-	(*Artist)(nil), // 1: hashicorp.consul.internal.demo.v1.Artist
-	(*Album)(nil),  // 2: hashicorp.consul.internal.demo.v1.Album
+	(Genre)(0),          // 0: hashicorp.consul.internal.demo.v1.Genre
+	(*RecordLabel)(nil), // 1: hashicorp.consul.internal.demo.v1.RecordLabel
+	(*Artist)(nil),      // 2: hashicorp.consul.internal.demo.v1.Artist
+	(*Album)(nil),       // 3: hashicorp.consul.internal.demo.v1.Album
 }
 var file_private_pbdemo_v1_demo_proto_depIdxs = []int32{
 	0, // 0: hashicorp.consul.internal.demo.v1.Artist.genre:type_name -> hashicorp.consul.internal.demo.v1.Genre
@@ -344,7 +404,7 @@ func file_private_pbdemo_v1_demo_proto_init() {
 	}
 	if !protoimpl.UnsafeEnabled {
 		file_private_pbdemo_v1_demo_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
-			switch v := v.(*Artist); i {
+			switch v := v.(*RecordLabel); i {
 			case 0:
 				return &v.state
 			case 1:
@@ -356,6 +416,18 @@ func file_private_pbdemo_v1_demo_proto_init() {
 			}
 		}
 		file_private_pbdemo_v1_demo_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*Artist); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+		file_private_pbdemo_v1_demo_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} {
 			switch v := v.(*Album); i {
 			case 0:
 				return &v.state
@@ -374,7 +446,7 @@ func file_private_pbdemo_v1_demo_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: file_private_pbdemo_v1_demo_proto_rawDesc,
 			NumEnums:      1,
-			NumMessages:   2,
+			NumMessages:   3,
 			NumExtensions: 0,
 			NumServices:   0,
 		},
diff --git a/proto/private/pbdemo/v1/demo.proto b/proto/private/pbdemo/v1/demo.proto
index 60ce1d0a4f50..4fcbe7155c48 100644
--- a/proto/private/pbdemo/v1/demo.proto
+++ b/proto/private/pbdemo/v1/demo.proto
@@ -7,6 +7,11 @@ syntax = "proto3";
 // Consul's generic storage APIs.
 package hashicorp.consul.internal.demo.v1;
 
+message RecordLabel {
+  string name = 1;
+  string description = 2;
+}
+
 message Artist {
   string name = 1;
   string description = 2;