Enforce Owner rules in `Write` endpoint (#16983)

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

@@ -108,4 +108,12 @@ func (s *Server) getAuthorizer(token string) (acl.Authorizer, error) {
 	return authz, nil
 }
 
+func isGRPCStatusError(err error) bool {
+	if err == nil {
+		return false
+	}
+	_, ok := status.FromError(err)
+	return ok
+}
+
 func clone[T proto.Message](v T) T { return proto.Clone(v).(T) }

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

@@ -9,6 +9,7 @@ import (
 	"github.com/oklog/ulid/v2"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/internal/resource"
@@ -110,10 +111,16 @@ func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbre
 		case errors.Is(err, storage.ErrNotFound):
 			input.Id.Uid = ulid.Make().String()
 
+			// Prevent setting statuses in this endpoint.
 			if len(input.Status) != 0 {
 				return errUseWriteStatus
 			}
 
+			// Enforce same tenancy for owner
+			if input.Owner != nil && !proto.Equal(input.Id.Tenancy, input.Owner.Tenancy) {
+				return status.Errorf(codes.InvalidArgument, "owner and resource tenancy must be the same")
+			}
+
 		// Update path.
 		case err == nil:
 			// Use the stored ID because it includes the Uid.
@@ -143,6 +150,12 @@ func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbre
 				return storage.ErrCASFailure
 			}
 
+			// Owner can only be set on creation. Enforce immutability.
+			if !proto.Equal(input.Owner, existing.Owner) {
+				return status.Errorf(codes.InvalidArgument, "owner cannot be changed")
+			}
+
+			// Carry over status and prevent updates
 			if input.Status == nil {
 				input.Status = existing.Status
 			} else if !resource.EqualStatus(input.Status, existing.Status) {
@@ -163,13 +176,11 @@ func (s *Server) Write(ctx context.Context, req *pbresource.WriteRequest) (*pbre
 		return nil, status.Error(codes.Aborted, err.Error())
 	case errors.Is(err, storage.ErrWrongUid):
 		return nil, status.Error(codes.FailedPrecondition, err.Error())
-	case err != nil:
-		if _, ok := status.FromError(err); !ok {
-			err = status.Errorf(codes.Internal, "failed to write resource: %v", err.Error())
-		}
+	case isGRPCStatusError(err):
 		return nil, err
+	case err != nil:
+		return nil, status.Errorf(codes.Internal, "failed to write resource: %v", err.Error())
 	}
-
 	return &pbresource.WriteResponse{Resource: result}, nil
 }
 

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

@@ -379,6 +379,64 @@ func TestWrite_NonCASUpdate_Retry(t *testing.T) {
 	require.NoError(t, <-errCh)
 }
 
+func TestWrite_Owner_Immutable(t *testing.T) {
+	// Use of proto.Equal(..) in implementation covers all permutations
+	// (nil -> non-nil, non-nil -> nil, owner1 -> owner2) so only the first one
+	// is tested.
+	server := testServer(t)
+	client := testClient(t, server)
+
+	demo.Register(server.Registry)
+
+	artist, err := demo.GenerateV2Artist()
+	require.NoError(t, err)
+	rsp1, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: artist})
+	require.NoError(t, err)
+	artist = rsp1.Resource
+
+	// create album with no owner
+	album, err := demo.GenerateV2Album(rsp1.Resource.Id)
+	require.NoError(t, err)
+	album.Owner = nil
+	rsp2, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+	require.NoError(t, err)
+
+	// setting owner on update should fail
+	album = rsp2.Resource
+	album.Owner = artist.Id
+	_, err = client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+	require.Error(t, err)
+	require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+	require.ErrorContains(t, err, "owner cannot be changed")
+}
+
+func TestWrite_Owner_RequireSameTenancy(t *testing.T) {
+	server := testServer(t)
+	client := testClient(t, server)
+
+	demo.Register(server.Registry)
+
+	artist, err := demo.GenerateV2Artist()
+	require.NoError(t, err)
+	rsp1, err := client.Write(testContext(t), &pbresource.WriteRequest{Resource: artist})
+	require.NoError(t, err)
+
+	// change album tenancy to be different from artist tenancy
+	album, err := demo.GenerateV2Album(rsp1.Resource.Id)
+	require.NoError(t, err)
+	album.Owner.Tenancy = &pbresource.Tenancy{
+		Partition: "some",
+		Namespace: "other",
+		PeerName:  "tenancy",
+	}
+
+	// verify create fails
+	_, err = client.Write(testContext(t), &pbresource.WriteRequest{Resource: album})
+	require.Error(t, err)
+	require.Equal(t, codes.InvalidArgument.String(), status.Code(err).String())
+	require.ErrorContains(t, err, "tenancy must be the same")
+}
+
 type blockOnceBackend struct {
 	storage.Backend