perf: role permissions in database (#9152)

# Which Problems Are Solved

Currently ZITADEL defines organization and instance member roles and
permissions in defaults.yaml. The permission check is done on API call
level. For example: "is this user allowed to make this call on this
org". This makes sense on the V1 API where the API is permission-level
shaped. For example, a search for users always happens in the context of
the organization. (Either the organization the calling user belongs to,
or through member ship and the x-zitadel-orgid header.

However, for resource based APIs we must be able to resolve permissions
by object. For example, an IAM_OWNER listing users should be able to get
all users in an instance based on the query filters. Alternatively a
user may have user.read permissions on one or more orgs. They should be
able to read just those users.

# How the Problems Are Solved

## Role permission mapping

The role permission mappings defined from `defaults.yaml` or local
config override are synchronized to the database on every run of
`zitadel setup`:

- A single query per **aggregate** builds a list of `add` and `remove`
actions needed to reach the desired state or role permission mappings
from the config.
- The required events based on the actions are pushed to the event
store.
- Events define search fields so that permission checking can use the
indices and is strongly consistent for both query and command sides.

The migration is split in the following aggregates:

- System aggregate for for roles prefixed with `SYSTEM`
- Each instance for roles not prefixed with `SYSTEM`. This is in
anticipation of instance level management over the API.

## Membership

Current instance / org / project membership events now have field table
definitions. Like the role permissions this ensures strong consistency
while still being able to use the indices of the fields table. A
migration is provided to fill the membership fields.

## Permission check

I aimed keeping the mental overhead to the developer to a minimal. The
provided implementation only provides a permission check for list
queries for org level resources, for example users. In the `query`
package there is a simple helper function `wherePermittedOrgs` which
makes sure the underlying database function is called as part of the
`SELECT` query and the permitted organizations are part of the `WHERE`
clause. This makes sure results from non-permitted organizations are
omitted. Under the hood:

- A Pg/PlSQL function searches for a list of organization IDs the passed
user has the passed permission.
- When the user has the permission on instance level, it returns early
with all organizations.
- The functions uses a number of views. The views help mapping the
fields entries into relational data and simplify the code use for the
function. The views provide some pre-filters which allow proper index
usage once the final `WHERE` clauses are set by the function.

# Additional Changes



# Additional Context

Closes #9032
Closes https://github.com/zitadel/zitadel/issues/9014

https://github.com/zitadel/zitadel/issues/9188 defines follow-ups for
the new permission framework based on this concept.

cmd/defaults.yaml

@@ -1132,6 +1132,7 @@ DefaultInstance:
     LoginDefaultOrg: true # ZITADEL_DEFAULTINSTANCE_FEATURES_LOGINDEFAULTORG
     # TriggerIntrospectionProjections: false # ZITADEL_DEFAULTINSTANCE_FEATURES_TRIGGERINTROSPECTIONPROJECTIONS
     # LegacyIntrospection: false # ZITADEL_DEFAULTINSTANCE_FEATURES_LEGACYINTROSPECTION
+    # PermissionCheckV2: false # ZITADEL_DEFAULTINSTANCE_FEATURES_PERMISSIONCHECKV2
   Limits:
     # AuditLogRetention limits the number of events that can be queried via the events API by their age.
     # A value of "0s" means that all events are available.
@@ -1195,6 +1196,9 @@ InternalAuthZ:
   # Configure the RolePermissionMappings by environment variable using JSON notation:
   # ZITADEL_INTERNALAUTHZ_ROLEPERMISSIONMAPPINGS='[{"role": "IAM_OWNER", "permissions": ["iam.write"]}, {"role": "ORG_OWNER", "permissions": ["org.write"]}]'
   # Beware that if you configure the RolePermissionMappings by environment variable, all the default RolePermissionMappings are lost.
+  #
+  # Warning: RolePermissionMappings are synhronized to the database.
+  # Changes here will only be applied after running `zitadel setup` or `zitadel start-from-setup`.
   RolePermissionMappings:
     - Role: "SYSTEM_OWNER"
       Permissions:

cmd/setup/46.go

@@ -0,0 +1,39 @@
+package setup
+
+import (
+	"context"
+	"embed"
+	"fmt"
+
+	"github.com/zitadel/logging"
+
+	"github.com/zitadel/zitadel/internal/database"
+	"github.com/zitadel/zitadel/internal/eventstore"
+)
+
+type InitPermissionFunctions struct {
+	eventstoreClient *database.DB
+}
+
+var (
+	//go:embed 46/*.sql
+	permissionFunctions embed.FS
+)
+
+func (mig *InitPermissionFunctions) Execute(ctx context.Context, _ eventstore.Event) error {
+	statements, err := readStatements(permissionFunctions, "46", "")
+	if err != nil {
+		return err
+	}
+	for _, stmt := range statements {
+		logging.WithFields("file", stmt.file, "migration", mig.String()).Info("execute statement")
+		if _, err := mig.eventstoreClient.ExecContext(ctx, stmt.query); err != nil {
+			return fmt.Errorf("%s %s: %w", mig.String(), stmt.file, err)
+		}
+	}
+	return nil
+}
+
+func (*InitPermissionFunctions) String() string {
+	return "46_init_permission_functions"
+}

cmd/setup/46/01-role_permissions_view.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE VIEW eventstore.role_permissions AS
+SELECT instance_id, aggregate_id, object_id as role, text_value as permission
+FROM eventstore.fields
+WHERE aggregate_type = 'permission'
+AND object_type = 'role_permission'
+AND field_name = 'permission';

cmd/setup/46/02-instance_orgs_view.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE VIEW eventstore.instance_orgs AS
+SELECT instance_id, aggregate_id as org_id
+FROM eventstore.fields
+WHERE aggregate_type = 'org'
+AND object_type = 'org'
+AND field_name = 'state';

cmd/setup/46/03-instance_members_view.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE VIEW eventstore.instance_members AS
+SELECT instance_id, object_id as user_id, text_value as role
+FROM eventstore.fields
+WHERE aggregate_type = 'instance'
+AND object_type = 'instance_member_role'
+AND field_name = 'instance_role';

cmd/setup/46/04-org_members_view.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE VIEW eventstore.org_members AS
+SELECT instance_id, aggregate_id as org_id, object_id as user_id, text_value as role
+FROM eventstore.fields
+WHERE aggregate_type = 'org'
+AND object_type = 'org_member_role'
+AND field_name = 'org_role';

cmd/setup/46/05-project_members_view.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE VIEW eventstore.project_members AS
+SELECT instance_id, aggregate_id as project_id, object_id as user_id, text_value as role
+FROM eventstore.fields
+WHERE aggregate_type = 'project'
+AND object_type = 'project_member_role'
+AND field_name = 'project_role';

cmd/setup/46/06-permitted_orgs_function.sql

@@ -0,0 +1,50 @@
+CREATE OR REPLACE FUNCTION eventstore.permitted_orgs(
+    instanceId TEXT
+    , userId TEXT
+    , perm TEXT
+
+    , org_ids OUT TEXT[]
+)
+	LANGUAGE 'plpgsql'
+	STABLE
+AS $$
+DECLARE
+	matched_roles TEXT[]; -- roles containing permission
+BEGIN
+	SELECT array_agg(rp.role) INTO matched_roles
+	FROM eventstore.role_permissions rp
+	WHERE rp.instance_id = instanceId
+	AND rp.permission = perm;
+	
+	-- First try if the permission was granted thru an instance-level role
+	DECLARE
+		has_instance_permission bool;
+	BEGIN
+		SELECT true INTO has_instance_permission
+			FROM eventstore.instance_members im
+			WHERE im.role = ANY(matched_roles)
+			AND im.instance_id = instanceId
+			AND im.user_id = userId
+			LIMIT 1;
+		
+		IF has_instance_permission THEN
+			-- Return all organizations
+			SELECT array_agg(o.org_id) INTO org_ids
+				FROM eventstore.instance_orgs o
+				WHERE o.instance_id = instanceId;
+			RETURN;
+		END IF;
+	END;
+	
+	-- Return the organizations where permission were granted thru org-level roles
+	SELECT array_agg(org_id) INTO org_ids
+	FROM (
+		SELECT DISTINCT om.org_id
+		FROM eventstore.org_members om
+		WHERE om.role = ANY(matched_roles)
+		AND om.instance_id = instanceID
+		AND om.user_id = userId
+	);
+    RETURN;
+END;
+$$;

cmd/setup/config.go

@@ -87,6 +87,9 @@ func MustNewConfig(v *viper.Viper) *Config {
 
 	id.Configure(config.Machine)
 
+	// Copy the global role permissions mappings to the instance until we allow instance-level configuration over the API.
+	config.DefaultInstance.RolePermissionMappings = config.InternalAuthZ.RolePermissionMappings
+
 	return config
 }
 
@@ -131,6 +134,7 @@ type Steps struct {
 	s43CreateFieldsDomainIndex              *CreateFieldsDomainIndex
 	s44ReplaceCurrentSequencesIndex         *ReplaceCurrentSequencesIndex
 	s45CorrectProjectOwners                 *CorrectProjectOwners
+	s46InitPermissionFunctions              *InitPermissionFunctions
 }
 
 func MustNewSteps(v *viper.Viper) *Steps {

cmd/setup/setup.go

@@ -174,6 +174,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 	steps.s43CreateFieldsDomainIndex = &CreateFieldsDomainIndex{dbClient: queryDBClient}
 	steps.s44ReplaceCurrentSequencesIndex = &ReplaceCurrentSequencesIndex{dbClient: esPusherDBClient}
 	steps.s45CorrectProjectOwners = &CorrectProjectOwners{eventstore: eventstoreClient}
+	steps.s46InitPermissionFunctions = &InitPermissionFunctions{eventstoreClient: esPusherDBClient}
 
 	err = projection.Create(ctx, projectionDBClient, eventstoreClient, config.Projections, nil, nil, nil)
 	logging.OnError(err).Fatal("unable to start projections")
@@ -196,6 +197,10 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 		&FillFieldsForInstanceDomains{
 			eventstore: eventstoreClient,
 		},
+		&SyncRolePermissions{
+			eventstore:             eventstoreClient,
+			rolePermissionMappings: config.InternalAuthZ.RolePermissionMappings,
+		},
 	}
 
 	for _, step := range []migration.Migration{
@@ -229,6 +234,7 @@ func Setup(ctx context.Context, config *Config, steps *Steps, masterKey string)
 		steps.s38BackChannelLogoutNotificationStart,
 		steps.s44ReplaceCurrentSequencesIndex,
 		steps.s45CorrectProjectOwners,
+		steps.s46InitPermissionFunctions,
 	} {
 		mustExecuteMigration(ctx, eventstoreClient, step, "migration failed")
 	}

cmd/setup/sync_role_permissions.go

@@ -0,0 +1,134 @@
+package setup
+
+import (
+	"context"
+	"database/sql"
+	_ "embed"
+	"fmt"
+	"strings"
+
+	"github.com/zitadel/logging"
+
+	"github.com/zitadel/zitadel/internal/api/authz"
+	"github.com/zitadel/zitadel/internal/database"
+	"github.com/zitadel/zitadel/internal/eventstore"
+	"github.com/zitadel/zitadel/internal/repository/instance"
+	"github.com/zitadel/zitadel/internal/repository/permission"
+)
+
+var (
+	//go:embed sync_role_permissions.sql
+	getRolePermissionOperationsQuery string
+)
+
+// SyncRolePermissions is a repeatable step which synchronizes the InternalAuthZ
+// RolePermissionMappings from the configuration to the database.
+// This is needed until role permissions are manageable over the API.
+type SyncRolePermissions struct {
+	eventstore             *eventstore.Eventstore
+	rolePermissionMappings []authz.RoleMapping
+}
+
+func (mig *SyncRolePermissions) Execute(ctx context.Context, _ eventstore.Event) error {
+	if err := mig.executeSystem(ctx); err != nil {
+		return err
+	}
+	return mig.executeInstances(ctx)
+}
+
+func (mig *SyncRolePermissions) executeSystem(ctx context.Context) error {
+	logging.WithFields("migration", mig.String()).Info("prepare system role permission sync events")
+
+	target := rolePermissionMappingsToDatabaseMap(mig.rolePermissionMappings, true)
+	cmds, err := mig.synchronizeCommands(ctx, "SYSTEM", target)
+	if err != nil {
+		return err
+	}
+	events, err := mig.eventstore.Push(ctx, cmds...)
+	if err != nil {
+		return err
+	}
+
+	logging.WithFields("migration", mig.String(), "pushed_events", len(events)).Info("pushed system role permission sync events")
+	return nil
+}
+
+func (mig *SyncRolePermissions) executeInstances(ctx context.Context) error {
+	instances, err := mig.eventstore.InstanceIDs(
+		ctx,
+		eventstore.NewSearchQueryBuilder(eventstore.ColumnsInstanceIDs).
+			OrderDesc().
+			AddQuery().
+			AggregateTypes(instance.AggregateType).
+			EventTypes(instance.InstanceAddedEventType).
+			Builder().
+			ExcludeAggregateIDs().
+			AggregateTypes(instance.AggregateType).
+			EventTypes(instance.InstanceRemovedEventType).
+			Builder(),
+	)
+	if err != nil {
+		return err
+	}
+	target := rolePermissionMappingsToDatabaseMap(mig.rolePermissionMappings, false)
+	for i, instanceID := range instances {
+		logging.WithFields("instance_id", instanceID, "migration", mig.String(), "progress", fmt.Sprintf("%d/%d", i+1, len(instances))).Info("prepare instance role permission sync events")
+		cmds, err := mig.synchronizeCommands(ctx, instanceID, target)
+		if err != nil {
+			return err
+		}
+		events, err := mig.eventstore.Push(ctx, cmds...)
+		if err != nil {
+			return err
+		}
+		logging.WithFields("instance_id", instanceID, "migration", mig.String(), "pushed_events", len(events)).Info("pushed instance role permission sync events")
+	}
+	return nil
+}
+
+// synchronizeCommands checks the current state of role permissions in the eventstore for the aggregate.
+// It returns the commands required to reach the desired state passed in target.
+// For system level permissions aggregateID must be set to `SYSTEM`,
+// else it is the instance ID.
+func (mig *SyncRolePermissions) synchronizeCommands(ctx context.Context, aggregateID string, target database.Map[[]string]) (cmds []eventstore.Command, err error) {
+	aggregate := permission.NewAggregate(aggregateID)
+	err = mig.eventstore.Client().QueryContext(ctx, func(rows *sql.Rows) error {
+		for rows.Next() {
+			var operation, role, perm string
+			if err := rows.Scan(&operation, &role, &perm); err != nil {
+				return err
+			}
+			logging.WithFields("aggregate_id", aggregateID, "migration", mig.String(), "operation", operation, "role", role, "permission", perm).Debug("sync role permission")
+			switch operation {
+			case "add":
+				cmds = append(cmds, permission.NewAddedEvent(ctx, aggregate, role, perm))
+			case "remove":
+				cmds = append(cmds, permission.NewRemovedEvent(ctx, aggregate, role, perm))
+			}
+		}
+		return rows.Close()
+
+	}, getRolePermissionOperationsQuery, aggregateID, target)
+	if err != nil {
+		return nil, err
+	}
+	return cmds, err
+}
+
+func (*SyncRolePermissions) String() string {
+	return "repeatable_sync_role_permissions"
+}
+
+func (*SyncRolePermissions) Check(lastRun map[string]interface{}) bool {
+	return true
+}
+
+func rolePermissionMappingsToDatabaseMap(mappings []authz.RoleMapping, system bool) database.Map[[]string] {
+	out := make(database.Map[[]string], len(mappings))
+	for _, m := range mappings {
+		if system == strings.HasPrefix(m.Role, "SYSTEM") {
+			out[m.Role] = m.Permissions
+		}
+	}
+	return out
+}

cmd/setup/sync_role_permissions.sql

@@ -0,0 +1,52 @@
+/*
+This query creates a change set of permissions that need to be added or removed.
+It compares the current state in the fields table (thru the role_permissions view)
+against a passed role permission mapping as JSON, created from Zitadel's config:
+
+{
+	"IAM_ADMIN_IMPERSONATOR": ["admin.impersonation", "impersonation"],
+	"IAM_END_USER_IMPERSONATOR": ["impersonation"],
+	"FOO_BAR": ["foo.bar", "bar.foo"]
+ }
+
+It uses an aggregate_id as first argument which may be an instance_id or 'SYSTEM'
+for system level permissions.
+*/
+WITH target AS (
+	-- unmarshal JSON representation into flattened tabular data
+	SELECT
+		key AS role,
+		jsonb_array_elements_text(value) AS permission
+	FROM jsonb_each($2::jsonb)
+), add AS (
+    -- find all role permissions that exist in `target` and not in `role_permissions`
+	SELECT t.role, t.permission
+	FROM eventstore.role_permissions p
+	RIGHT JOIN target t
+		ON p.aggregate_id = $1::text
+		AND p.role = t.role
+		AND p.permission = t.permission
+	WHERE p.role IS NULL
+), remove AS (
+    -- find all role permissions that exist `role_permissions` and not in `target`
+	SELECT p.role, p.permission
+	FROM eventstore.role_permissions p
+	LEFT JOIN target t
+		ON p.role = t.role
+		AND p.permission = t.permission
+	WHERE p.aggregate_id = $1::text
+	AND t.role IS NULL
+)
+-- return the required operations
+SELECT
+	'add' AS operation,
+	role,
+	permission
+FROM add
+UNION ALL
+SELECT
+	'remove' AS operation,
+	role,
+	permission
+FROM remove
+;

cmd/start/config.go

@@ -127,5 +127,8 @@ func MustNewConfig(v *viper.Viper) *Config {
 	id.Configure(config.Machine)
 	actions.SetHTTPConfig(&config.Actions.HTTP)
 
+	// Copy the global role permissions mappings to the instance until we allow instance-level configuration over the API.
+	config.DefaultInstance.RolePermissionMappings = config.InternalAuthZ.RolePermissionMappings
+
 	return config
 }