fix: set correct owner on project grants (#9089)

# Which Problems Are Solved

In versions previous to v2.66 it was possible to set a different
resource owner on project grants. This was introduced with the new
resource based API. The resource owner was possible to overwrite using
the x-zitadel-org header.

Because of this issue project grants got the wrong resource owner,
instead of the owner of the project it got the granted org which is
wrong because a resource owner of an aggregate is not allowed to change.

# How the Problems Are Solved

- The wrong owners of the events are set to the original owner of the
project.
- A new event is pushed to these aggregates `project.owner.corrected` 
- The projection updates the owners of the user grants if that event was
written

# Additional Changes

The eventstore push function (replaced in version 2.66) writes the
correct resource owner.

# Additional Context

closes https://github.com/zitadel/zitadel/issues/9072

cmd/setup/45.go

@@ -0,0 +1,111 @@
+package setup
+
+import (
+	"context"
+	_ "embed"
+	"encoding/json"
+	"fmt"
+
+	"github.com/zitadel/logging"
+
+	"github.com/zitadel/zitadel/internal/api/authz"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/query/projection"
+	"github.com/zitadel/zitadel/internal/repository/instance"
+	"github.com/zitadel/zitadel/internal/repository/owner"
+	"github.com/zitadel/zitadel/internal/repository/project"
+)
+
+var (
+	//go:embed 45.sql
+	correctProjectOwnerEvents string
+)
+
+type CorrectProjectOwners struct {
+	eventstore *eventstore.Eventstore
+}
+
+func (mig *CorrectProjectOwners) Execute(ctx context.Context, _ eventstore.Event) error {
+	instances, err := mig.eventstore.InstanceIDs(
+		ctx,
+		eventstore.NewSearchQueryBuilder(eventstore.ColumnsInstanceIDs).
+			OrderDesc().
+			AddQuery().
+			AggregateTypes("instance").
+			EventTypes(instance.InstanceAddedEventType).
+			Builder(),
+	)
+	if err != nil {
+		return err
+	}
+
+	ctx = authz.SetCtxData(ctx, authz.CtxData{UserID: "SETUP"})
+	for i, instance := range instances {
+		ctx = authz.WithInstanceID(ctx, instance)
+		logging.WithFields("instance_id", instance, "migration", mig.String(), "progress", fmt.Sprintf("%d/%d", i+1, len(instances))).Info("correct owners of projects")
+		didCorrect, err := mig.correctInstanceProjects(ctx, instance)
+		if err != nil {
+			return err
+		}
+		if !didCorrect {
+			continue
+		}
+		_, err = projection.ProjectGrantProjection.Trigger(ctx)
+		logging.OnError(err).Debug("failed triggering project grant projection to update owners")
+	}
+	return nil
+}
+
+func (mig *CorrectProjectOwners) correctInstanceProjects(ctx context.Context, instance string) (didCorrect bool, err error) {
+	var correctedOwners []eventstore.Command
+
+	tx, err := mig.eventstore.Client().BeginTx(ctx, nil)
+	if err != nil {
+		return false, err
+	}
+	defer func() {
+		if err != nil {
+			_ = tx.Rollback()
+			return
+		}
+		err = tx.Commit()
+	}()
+
+	rows, err := tx.QueryContext(ctx, correctProjectOwnerEvents, instance)
+	if err != nil {
+		return false, err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		aggregate := &eventstore.Aggregate{
+			InstanceID: instance,
+			Type:       project.AggregateType,
+			Version:    project.AggregateVersion,
+		}
+		var payload json.RawMessage
+		err := rows.Scan(
+			&aggregate.ID,
+			&aggregate.ResourceOwner,
+			&payload,
+		)
+		if err != nil {
+			return false, err
+		}
+		previousOwners := make(map[uint32]string)
+		if err := json.Unmarshal(payload, &previousOwners); err != nil {
+			return false, err
+		}
+		correctedOwners = append(correctedOwners, owner.NewCorrected(ctx, aggregate, previousOwners))
+	}
+	if rows.Err() != nil {
+		return false, rows.Err()
+	}
+
+	_, err = mig.eventstore.PushWithClient(ctx, tx, correctedOwners...)
+	return len(correctedOwners) > 0, err
+}
+
+func (*CorrectProjectOwners) String() string {
+	return "43_correct_project_owners"
+}

cmd/setup/45.sql

@@ -0,0 +1,79 @@
+WITH corrupt_streams AS (
+    select
+        e.instance_id
+        , e.aggregate_type
+        , e.aggregate_id
+        , min(e.sequence) as min_sequence
+        , count(distinct e.owner) as owner_count
+    from
+        eventstore.events2 e
+    where
+        e.instance_id = $1
+        and aggregate_type = 'project'
+    group by 
+        e.instance_id
+        , e.aggregate_type
+        , e.aggregate_id
+    having
+        count(distinct e.owner) > 1
+), correct_owners AS (
+    select
+        e.instance_id
+        , e.aggregate_type
+        , e.aggregate_id
+        , e.owner
+    from
+        eventstore.events2 e
+    join
+        corrupt_streams cs
+    on
+        e.instance_id = cs.instance_id
+        and e.aggregate_type = cs.aggregate_type
+        and e.aggregate_id = cs.aggregate_id
+        and e.sequence = cs.min_sequence
+), wrong_events AS (
+    select
+        e.instance_id
+        , e.aggregate_type
+        , e.aggregate_id
+        , e.sequence
+        , e.owner wrong_owner
+        , co.owner correct_owner
+    from
+        eventstore.events2 e
+    join
+        correct_owners co
+    on
+        e.instance_id = co.instance_id
+        and e.aggregate_type = co.aggregate_type
+        and e.aggregate_id = co.aggregate_id
+        and e.owner <> co.owner
+), updated_events AS (
+    UPDATE eventstore.events2 e
+        SET owner = we.correct_owner
+    FROM 
+        wrong_events we
+    WHERE
+        e.instance_id = we.instance_id
+        and e.aggregate_type = we.aggregate_type
+        and e.aggregate_id = we.aggregate_id
+        and e.sequence = we.sequence
+    RETURNING 
+        we.aggregate_id
+        , we.correct_owner
+        , we.sequence
+        , we.wrong_owner
+)
+SELECT
+    ue.aggregate_id
+    , ue.correct_owner
+    , jsonb_object_agg(
+        ue.sequence::TEXT --formant to string because crdb is not able to handle int
+        , ue.wrong_owner
+    ) payload
+FROM
+    updated_events ue
+GROUP BY
+    ue.aggregate_id
+    , ue.correct_owner
+;

cmd/setup/config.go

@@ -130,6 +130,7 @@ type Steps struct {
 	s42Apps7OIDCConfigsLoginVersion         *Apps7OIDCConfigsLoginVersion
 	s43CreateFieldsDomainIndex              *CreateFieldsDomainIndex
 	s44ReplaceCurrentSequencesIndex         *ReplaceCurrentSequencesIndex
+	s45CorrectProjectOwners                 *CorrectProjectOwners
 }
 
 func MustNewSteps(v *viper.Viper) *Steps {

cmd/setup/setup.go

@@ -173,6 +173,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 	steps.s42Apps7OIDCConfigsLoginVersion = &Apps7OIDCConfigsLoginVersion{dbClient: esPusherDBClient}
 	steps.s43CreateFieldsDomainIndex = &CreateFieldsDomainIndex{dbClient: queryDBClient}
 	steps.s44ReplaceCurrentSequencesIndex = &ReplaceCurrentSequencesIndex{dbClient: esPusherDBClient}
+	steps.s45CorrectProjectOwners = &CorrectProjectOwners{eventstore: eventstoreClient}
 
 	err = projection.Create(ctx, projectionDBClient, eventstoreClient, config.Projections, nil, nil, nil)
 	logging.OnError(err).Fatal("unable to start projections")
@@ -227,6 +228,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 		steps.s36FillV2Milestones,
 		steps.s38BackChannelLogoutNotificationStart,
 		steps.s44ReplaceCurrentSequencesIndex,
+		steps.s45CorrectProjectOwners,
 	} {
 		mustExecuteMigration(ctx, eventstoreClient, step, "migration failed")
 	}