resource: allow for the ACLs.Read hook to request the entire data payload to perform the authz check (#18925)

The ACLs.Read hook for a resource only allows for the identity of a 
resource to be passed in for use in authz consideration. For some 
resources we wish to allow for the current stored value to dictate how 
to enforce the ACLs (such as reading a list of applicable services from 
the payload and allowing service:read on any of them to control reading the enclosing resource).

This change update the interface to usually accept a *pbresource.ID, 
but if the hook decides it needs more data it returns a sentinel error 
and the resource service knows to defer the authz check until after
 fetching the data from storage.

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

@@ -69,7 +69,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, authzContext, resource.Id)
+		err = reg.ACLs.Read(authz, authzContext, resource.Id, resource)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue

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

@@ -74,7 +74,7 @@ func (s *Server) ListByOwner(ctx context.Context, req *pbresource.ListByOwnerReq
 		}
 
 		// Filter out children that fail real ACL.
-		err = childReg.ACLs.Read(childAuthz, childAuthzContext, child.Id)
+		err = childReg.ACLs.Read(childAuthz, childAuthzContext, child.Id, child)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue

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

@@ -44,9 +44,15 @@ func (s *Server) Read(ctx context.Context, req *pbresource.ReadRequest) (*pbreso
 
 	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)
+	// ACL check usually comes before tenancy existence checks to not leak
+	// tenancy "existence", unless the ACL check requires the data payload
+	// to function.
+	authzNeedsData := false
+	err = reg.ACLs.Read(authz, authzContext, req.Id, nil)
 	switch {
+	case errors.Is(err, resource.ErrNeedData):
+		authzNeedsData = true
+		err = nil
 	case acl.IsErrPermissionDenied(err):
 		return nil, status.Error(codes.PermissionDenied, err.Error())
 	case err != nil:
@@ -60,15 +66,25 @@ func (s *Server) Read(ctx context.Context, req *pbresource.ReadRequest) (*pbreso
 
 	resource, err := s.Backend.Read(ctx, readConsistencyFrom(ctx), req.Id)
 	switch {
-	case err == nil:
-		return &pbresource.ReadResponse{Resource: resource}, nil
 	case errors.Is(err, storage.ErrNotFound):
 		return nil, status.Error(codes.NotFound, err.Error())
 	case errors.As(err, &storage.GroupVersionMismatchError{}):
 		return nil, status.Error(codes.InvalidArgument, err.Error())
-	default:
+	case err != nil:
 		return nil, status.Errorf(codes.Internal, "failed read: %v", err)
 	}
+
+	if authzNeedsData {
+		err = reg.ACLs.Read(authz, authzContext, req.Id, resource)
+		switch {
+		case acl.IsErrPermissionDenied(err):
+			return nil, status.Error(codes.PermissionDenied, err.Error())
+		case err != nil:
+			return nil, status.Errorf(codes.Internal, "failed read acl: %v", err)
+		}
+	}
+
+	return &pbresource.ReadResponse{Resource: resource}, nil
 }
 
 func (s *Server) validateReadRequest(req *pbresource.ReadRequest) (*resource.Registration, error) {

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

@@ -5,6 +5,8 @@ package resource
 
 import (
 	"context"
+	"fmt"
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
@@ -14,12 +16,15 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/acl/resolver"
+	"github.com/hashicorp/consul/agent/grpc-external/testutils"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/demo"
 	"github.com/hashicorp/consul/internal/storage"
 	"github.com/hashicorp/consul/proto-public/pbresource"
 	"github.com/hashicorp/consul/proto/private/prototest"
+	"github.com/hashicorp/consul/sdk/testutil"
 )
 
 func TestRead_InputValidation(t *testing.T) {
@@ -213,18 +218,50 @@ func TestRead_VerifyReadConsistencyArg(t *testing.T) {
 // N.B. Uses key ACLs for now. See demo.RegisterTypes()
 func TestRead_ACLs(t *testing.T) {
 	type testCase struct {
-		authz resolver.Result
-		code  codes.Code
+		res          *pbresource.Resource
+		authz        resolver.Result
+		codeNotExist codes.Code
+		codeExists   codes.Code
 	}
+
+	artist, err := demo.GenerateV2Artist()
+	require.NoError(t, err)
+
+	label, err := demo.GenerateV1RecordLabel("blink1982")
+	require.NoError(t, err)
+
 	testcases := map[string]testCase{
-		"read hook denied": {
-			authz: AuthorizerFrom(t, demo.ArtistV1ReadPolicy),
-			code:  codes.PermissionDenied,
+		"artist-v1/read hook denied": {
+			res:          artist,
+			authz:        AuthorizerFrom(t, demo.ArtistV1ReadPolicy),
+			codeNotExist: codes.PermissionDenied,
+			codeExists:   codes.PermissionDenied,
 		},
-		"read hook allowed": {
-			authz: AuthorizerFrom(t, demo.ArtistV2ReadPolicy),
-			code:  codes.NotFound,
+		"artist-v2/read hook allowed": {
+			res:          artist,
+			authz:        AuthorizerFrom(t, demo.ArtistV2ReadPolicy),
+			codeNotExist: codes.NotFound,
+			codeExists:   codes.OK,
 		},
+		// Labels have the read ACL that requires reading the data.
+		"label-v1/read hook denied": {
+			res:          label,
+			authz:        AuthorizerFrom(t, demo.LabelV1ReadPolicy),
+			codeNotExist: codes.NotFound,
+			codeExists:   codes.PermissionDenied,
+		},
+	}
+
+	adminAuthz := AuthorizerFrom(t, `key_prefix "" { policy = "write" }`)
+
+	idx := 0
+	nextTokenContext := func(t *testing.T) context.Context {
+		// Each query should use a distinct token string to avoid caching so we can
+		// change the behavior each call.
+		token := fmt.Sprintf("token-%d", idx)
+		idx++
+		//nolint:staticcheck
+		return context.WithValue(testContext(t), "x-consul-token", token)
 	}
 
 	for desc, tc := range testcases {
@@ -232,23 +269,63 @@ func TestRead_ACLs(t *testing.T) {
 			server := testServer(t)
 			client := testClient(t, server)
 
-			mockACLResolver := &MockACLResolver{}
-			mockACLResolver.On("ResolveTokenAndDefaultMeta", mock.Anything, mock.Anything, mock.Anything).
-				Return(tc.authz, nil)
-			server.ACLResolver = mockACLResolver
+			dr := &dummyACLResolver{
+				result: testutils.ACLsDisabled(t),
+			}
+			server.ACLResolver = dr
+
 			demo.RegisterTypes(server.Registry)
 
-			artist, err := demo.GenerateV2Artist()
-			require.NoError(t, err)
+			dr.SetResult(tc.authz)
+			testutil.RunStep(t, "does not exist", func(t *testing.T) {
+				_, err = client.Read(nextTokenContext(t), &pbresource.ReadRequest{Id: tc.res.Id})
+				if tc.codeNotExist == codes.OK {
+					require.NoError(t, err)
+				} else {
+					require.Error(t, err)
+				}
+				require.Equal(t, tc.codeNotExist.String(), status.Code(err).String(), "%v", err)
+			})
 
-			// exercise ACL
-			_, err = client.Read(testContext(t), &pbresource.ReadRequest{Id: artist.Id})
-			require.Error(t, err)
-			require.Equal(t, tc.code.String(), status.Code(err).String())
+			// Create it.
+			dr.SetResult(adminAuthz)
+			_, err = client.Write(nextTokenContext(t), &pbresource.WriteRequest{Resource: tc.res})
+			require.NoError(t, err, "could not write resource")
+
+			dr.SetResult(tc.authz)
+			testutil.RunStep(t, "does exist", func(t *testing.T) {
+				// exercise ACL when the data does exist
+				_, err = client.Read(nextTokenContext(t), &pbresource.ReadRequest{Id: tc.res.Id})
+				if tc.codeExists == codes.OK {
+					require.NoError(t, err)
+				} else {
+					require.Error(t, err)
+				}
+				require.Equal(t, tc.codeExists.String(), status.Code(err).String())
+			})
 		})
 	}
 }
 
+type dummyACLResolver struct {
+	lock   sync.Mutex
+	result resolver.Result
+}
+
+var _ ACLResolver = (*dummyACLResolver)(nil)
+
+func (r *dummyACLResolver) SetResult(result resolver.Result) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	r.result = result
+}
+
+func (r *dummyACLResolver) ResolveTokenAndDefaultMeta(string, *acl.EnterpriseMeta, *acl.AuthorizerContext) (resolver.Result, error) {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+	return r.result, nil
+}
+
 type readTestCase struct {
 	consistency storage.ReadConsistency
 	ctx         context.Context

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

@@ -77,7 +77,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, authzContext, event.Resource.Id)
+		err = reg.ACLs.Read(authz, authzContext, event.Resource.Id, event.Resource)
 		switch {
 		case acl.IsErrPermissionDenied(err):
 			continue

docs/resources/guide.md

@@ -184,7 +184,7 @@ func RegisterTypes(r resource.Registry) {
 	})
 }
 
-func authzReadBar(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
+func authzReadBar(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID,  _ *pbresource.Resource) error {
 	return authz.ToAllowAuthorizer().
 		BarReadAllowed(id.Name, authzContext)
 }

internal/mesh/internal/types/proxy_state_template.go

@@ -36,7 +36,7 @@ func RegisterProxyStateTemplate(r resource.Registry) {
 		Scope:    resource.ScopeNamespace,
 		Validate: ValidateProxyStateTemplate,
 		ACLs: &resource.ACLHooks{
-			Read: func(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
+			Read: func(authorizer acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) 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

internal/resource/demo/demo.go

@@ -72,6 +72,8 @@ const (
 	ArtistV2ReadPolicy  = `key_prefix "resource/demo.v2.Artist/" { policy = "read" }`
 	ArtistV2WritePolicy = `key_prefix "resource/demo.v2.Artist/" { policy = "write" }`
 	ArtistV2ListPolicy  = `key_prefix "resource/" { policy = "list" }`
+	LabelV1ReadPolicy   = `key_prefix "resource/demo.v1.Label/" { policy = "read" }`
+	LabelV1WritePolicy  = `key_prefix "resource/demo.v1.Label/" { policy = "write" }`
 )
 
 // RegisterTypes registers the demo types. Should only be called in tests and
@@ -80,7 +82,12 @@ 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, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
+	readACL := func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, res *pbresource.Resource) error {
+		if resource.EqualType(TypeV1RecordLabel, id.Type) {
+			if res == nil {
+				return resource.ErrNeedData
+			}
+		}
 		key := fmt.Sprintf("resource/%s/%s", resource.ToGVK(id.Type), id.Name)
 		return authz.ToAllowAuthorizer().KeyReadAllowed(key, authzContext)
 	}

internal/resource/registry.go

@@ -4,6 +4,7 @@
 package resource
 
 import (
+	"errors"
 	"fmt"
 	"regexp"
 	"strings"
@@ -67,12 +68,18 @@ type Registration struct {
 	Scope Scope
 }
 
+var ErrNeedData = errors.New("authorization check requires resource data")
+
 type ACLHooks struct {
 	// Read is used to authorize Read RPCs and to filter results in List
 	// RPCs.
 	//
+	// It can be called an ID and possibly a Resource. The check will first
+	// attempt to use the ID and if the hook returns ErrNeedData, then the
+	// check will be deferred until the data is fetched from the storage layer.
+	//
 	// If it is omitted, `operator:read` permission is assumed.
-	Read func(acl.Authorizer, *acl.AuthorizerContext, *pbresource.ID) error
+	Read func(acl.Authorizer, *acl.AuthorizerContext, *pbresource.ID, *pbresource.Resource) error
 
 	// Write is used to authorize Write and Delete RPCs.
 	//
@@ -142,7 +149,7 @@ func (r *TypeRegistry) Register(registration Registration) {
 		registration.ACLs = &ACLHooks{}
 	}
 	if registration.ACLs.Read == nil {
-		registration.ACLs.Read = func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID) error {
+		registration.ACLs.Read = func(authz acl.Authorizer, authzContext *acl.AuthorizerContext, id *pbresource.ID, _ *pbresource.Resource) error {
 			return authz.ToAllowAuthorizer().OperatorReadAllowed(authzContext)
 		}
 	}

internal/resource/registry_test.go

@@ -6,17 +6,17 @@ package resource_test
 import (
 	"testing"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/grpc-external/testutils"
 	"github.com/hashicorp/consul/internal/resource"
 	"github.com/hashicorp/consul/internal/resource/demo"
 	"github.com/hashicorp/consul/proto-public/pbresource"
-
 	pbdemov1 "github.com/hashicorp/consul/proto/private/pbdemo/v1"
 	demov2 "github.com/hashicorp/consul/proto/private/pbdemo/v2"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"google.golang.org/protobuf/proto"
 )
 
 func TestRegister(t *testing.T) {
@@ -56,8 +56,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), nil, artist.Id))
-	require.True(t, acl.IsErrPermissionDenied(reg.ACLs.Read(testutils.ACLNoPermissions(t), nil, artist.Id)))
+	require.NoError(t, reg.ACLs.Read(testutils.ACLOperatorRead(t), nil, artist.Id, nil))
+	require.True(t, acl.IsErrPermissionDenied(reg.ACLs.Read(testutils.ACLNoPermissions(t), nil, artist.Id, nil)))
 
 	// verify default write hook requires operator:write
 	require.NoError(t, reg.ACLs.Write(testutils.ACLOperatorWrite(t), nil, artist))