resource: Make resource writestatus tenancy aware

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

@@ -8,40 +8,54 @@ import (
 	"errors"
 	"fmt"
 
+	"github.com/oklog/ulid/v2"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/types/known/timestamppb"
 
-	"github.com/oklog/ulid/v2"
-
 	"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) WriteStatus(ctx context.Context, req *pbresource.WriteStatusRequest) (*pbresource.WriteStatusResponse, error) {
-	// TODO(spatel): Refactor _ and entMeta as part of NET-4912
-	authz, _, err := s.getAuthorizer(tokenFromContext(ctx), acl.DefaultEnterpriseMeta())
+	reg, err := s.validateWriteStatusRequest(req)
 	if err != nil {
 		return nil, err
 	}
 
-	// check acls
-	err = authz.ToAllowAuthorizer().OperatorWriteAllowed(&acl.AuthorizerContext{})
-	switch {
-	case acl.IsErrPermissionDenied(err):
-		return nil, status.Error(codes.PermissionDenied, err.Error())
-	case err != nil:
-		return nil, status.Errorf(codes.Internal, "failed operator:write allowed acl: %v", err)
+	entMeta := v2TenancyToV1EntMeta(req.Id.Tenancy)
+	authz, authzContext, err := s.getAuthorizer(tokenFromContext(ctx), entMeta)
+	if err != nil {
+		return nil, err
 	}
 
-	if err := validateWriteStatusRequest(req); err != nil {
+	// Apply defaults when tenancy units empty.
+	v1EntMetaToV2Tenancy(reg, entMeta, req.Id.Tenancy)
+
+	// Check V1 tenancy exists for the V2 resource. Ignore "marked for deletion" since status updates
+	// should still work regardless.
+	if err = v1TenancyExists(reg, s.V1TenancyBridge, req.Id.Tenancy, codes.InvalidArgument); err != nil {
 		return nil, err
 	}
 
-	_, err = s.resolveType(req.Id.Type)
-	if err != nil {
-		return nil, err
+	// Retrieve resource since ACL hook requires it.
+	existing, err := s.Backend.Read(ctx, storage.EventualConsistency, req.Id)
+	switch {
+	case errors.Is(err, storage.ErrNotFound):
+		return nil, status.Errorf(codes.NotFound, err.Error())
+	case err != nil:
+		return nil, status.Errorf(codes.Internal, "failed read: %v", err)
+	}
+
+	// Check write ACL.
+	err = reg.ACLs.Write(authz, authzContext, existing)
+	switch {
+	case acl.IsErrPermissionDenied(err):
+		return nil, status.Error(codes.PermissionDenied, err.Error())
+	case err != nil:
+		return nil, status.Errorf(codes.Internal, "failed operator:write allowed acl: %v", err)
 	}
 
 	// At the storage backend layer, all writes are CAS operations.
@@ -99,7 +113,7 @@ func (s *Server) WriteStatus(ctx context.Context, req *pbresource.WriteStatusReq
 	return &pbresource.WriteStatusResponse{Resource: result}, nil
 }
 
-func validateWriteStatusRequest(req *pbresource.WriteStatusRequest) error {
+func (s *Server) validateWriteStatusRequest(req *pbresource.WriteStatusRequest) (*resource.Registration, error) {
 	var field string
 	switch {
 	case req.Id == nil:
@@ -144,16 +158,35 @@ func validateWriteStatusRequest(req *pbresource.WriteStatusRequest) error {
 		}
 	}
 	if field != "" {
-		return status.Errorf(codes.InvalidArgument, "%s is required", field)
+		return nil, status.Errorf(codes.InvalidArgument, "%s is required", field)
 	}
 
 	if req.Status.UpdatedAt != nil {
-		return status.Error(codes.InvalidArgument, "status.updated_at is automatically set and cannot be provided")
+		return nil, status.Error(codes.InvalidArgument, "status.updated_at is automatically set and cannot be provided")
 	}
 
 	if _, err := ulid.ParseStrict(req.Status.ObservedGeneration); err != nil {
-		return status.Error(codes.InvalidArgument, "status.observed_generation is not valid")
+		return nil, status.Error(codes.InvalidArgument, "status.observed_generation is not valid")
+	}
+
+	// Lowercase
+	resource.Normalize(req.Id.Tenancy)
+
+	// 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 nil
+	return reg, nil
 }