fix: re-point cloned fork identities that name nobody in the fork

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/windmill-api-integration-tests/tests/fork_clone_on_behalf_of.rs

@@ -311,3 +311,121 @@ async fn test_fork_keeps_only_resolvable_on_behalf_of(db: Pool<Postgres>) -> any
 
     Ok(())
 }
+
+/// Apps, schedules, triggers and their drafts cannot drop an identity the way scripts and flows
+/// do, so one naming nobody in the fork goes to its creator while one that still resolves stays.
+/// Forked as an admin, whose app policies the clone otherwise keeps.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_fork_repoints_unresolvable_identities(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+
+    let server = ApiServer::start(db.clone()).await?;
+    let base_url = format!("http://localhost:{}/api", server.addr.port());
+
+    let stranger = json!({
+        "on_behalf_of": "u/test-user-2",
+        "on_behalf_of_email": "test2@windmill.dev",
+        "execution_mode": "publisher",
+    });
+    sqlx::query(
+        "INSERT INTO app (workspace_id, path, summary, policy, versions)
+         VALUES ('test-workspace', 'u/test-user/stranger', '', $1, '{}'),
+                ('test-workspace', 'u/test-user/group', '', $2, '{}')",
+    )
+    .bind(&stranger)
+    .bind(json!({
+        "on_behalf_of": "g/all",
+        "on_behalf_of_email": "group-all@windmill.dev",
+        "execution_mode": "publisher",
+    }))
+    .execute(&db)
+    .await?;
+    // The clone re-aggregates `versions` from `app_version`, and the column is NOT NULL.
+    sqlx::query(
+        "WITH v AS (
+            INSERT INTO app_version (app_id, value, created_by)
+            SELECT id, '{}'::json, 'test-user' FROM app WHERE workspace_id = 'test-workspace'
+            RETURNING id, app_id
+         )
+         UPDATE app SET versions = ARRAY[v.id] FROM v WHERE app.id = v.app_id",
+    )
+    .execute(&db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO draft (workspace_id, path, typ, value, created_at, email)
+         VALUES ('test-workspace', 'u/test-user/stranger', 'raw_app', $1::json, NOW(), 'test@windmill.dev'),
+                ('test-workspace', 'u/test-user/stranger', 'trigger_websocket', $2::json, NOW(), 'test@windmill.dev'),
+                ('test-workspace', 'u/test-user/nul', 'raw_app', $3::json, NOW(), 'test@windmill.dev')",
+    )
+    .bind(json!({ "policy": stranger }))
+    .bind(json!({ "permissioned_as": "u/test-user-2" }))
+    // Saved before drafts were stripped of NULs: any jsonb parse of it raises, so it must be
+    // skipped rather than abort the fork. Built from parts because a NUL escape can't sit in source.
+    .bind(format!(
+        r#"{{"policy":{{"on_behalf_of":"u/test-user-2"}},"files":{{"f":"a{}u0000"}}}}"#,
+        "\\"
+    ))
+    .execute(&db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO schedule (workspace_id, path, edited_by, schedule, script_path, email, permissioned_as, enabled)
+         VALUES ('test-workspace', 'u/test-user/stranger', 'test-user', '0 0 * * * *', 'u/test-user/s', 'test2@windmill.dev', 'u/test-user-2', false)",
+    )
+    .execute(&db)
+    .await?;
+    sqlx::query(
+        "INSERT INTO websocket_trigger (workspace_id, path, url, script_path, is_flow, edited_by, permissioned_as, mode)
+         VALUES ('test-workspace', 'u/test-user/stranger', 'ws://localhost', 'u/test-user/s', false, 'test-user', 'u/test-user-2', 'disabled')",
+    )
+    .execute(&db)
+    .await?;
+
+    let resp = reqwest::Client::new()
+        .post(format!(
+            "{base_url}/w/test-workspace/workspaces/create_fork"
+        ))
+        .header("Authorization", "Bearer SECRET_TOKEN")
+        .json(&json!({ "id": "wm-fork-repoint", "name": "Fork", "color": "#0000ff" }))
+        .send()
+        .await?;
+    assert!(
+        resp.status().is_success(),
+        "creating the fork: {}",
+        resp.text().await?
+    );
+
+    let text = |sql: &'static str| sqlx::query_scalar::<_, String>(sql).fetch_one(&db);
+    assert_eq!(
+        text("SELECT (policy->>'on_behalf_of') || ' ' || (policy->>'on_behalf_of_email') FROM app WHERE workspace_id = 'wm-fork-repoint' AND path = 'u/test-user/stranger'").await?,
+        "u/test-user test@windmill.dev"
+    );
+    assert_eq!(
+        text("SELECT policy->>'on_behalf_of' FROM app WHERE workspace_id = 'wm-fork-repoint' AND path = 'u/test-user/group'").await?,
+        "g/all"
+    );
+    assert_eq!(
+        text("SELECT value->'policy'->>'on_behalf_of' FROM draft WHERE workspace_id = 'wm-fork-repoint' AND path = 'u/test-user/stranger' AND typ = 'raw_app'").await?,
+        "u/test-user"
+    );
+    assert_eq!(
+        text("SELECT CASE WHEN strpos(value::text, 'u/test-user-2') > 0 THEN 'kept' ELSE 'rewritten' END FROM draft WHERE workspace_id = 'wm-fork-repoint' AND path = 'u/test-user/nul'").await?,
+        "kept"
+    );
+    assert_eq!(
+        text("SELECT value->>'permissioned_as' FROM draft WHERE workspace_id = 'wm-fork-repoint' AND typ = 'trigger_websocket'").await?,
+        "u/test-user"
+    );
+    assert_eq!(
+        text("SELECT permissioned_as || ' ' || email FROM schedule WHERE workspace_id = 'wm-fork-repoint'").await?,
+        "u/test-user test@windmill.dev"
+    );
+    assert_eq!(
+        text(
+            "SELECT permissioned_as FROM websocket_trigger WHERE workspace_id = 'wm-fork-repoint'"
+        )
+        .await?,
+        "u/test-user"
+    );
+
+    Ok(())
+}

backend/windmill-api-workspaces/src/workspaces.rs

@@ -6927,6 +6927,127 @@ async fn clear_orphaned_compat_address(
     Ok(())
 }
 
+/// SQL boolean: the principal the `principal` expression yields resolves in the workspace bound as
+/// `$1`. The same predicate `clone_scripts` and `clone_flows` inline, whose `query!` macros cannot
+/// take a composed string, so keep the three in step.
+fn principal_resolves_sql(principal: &str) -> String {
+    format!(
+        "CASE WHEN {principal} LIKE 'u/%' THEN EXISTS (
+                SELECT 1 FROM usr u WHERE u.workspace_id = $1
+                  AND u.username = substring({principal} from 3)
+                UNION ALL
+                SELECT 1 FROM password pw WHERE pw.super_admin
+                  AND (pw.username = substring({principal} from 3)
+                       OR pw.email = substring({principal} from 3)))
+              WHEN {principal} LIKE 'g/%' THEN EXISTS (
+                SELECT 1 FROM group_ g WHERE g.workspace_id = $1
+                  AND g.name = substring({principal} from 3))
+              ELSE EXISTS (
+                SELECT 1 FROM usr u WHERE u.workspace_id = $1 AND u.username = {principal}
+                UNION ALL
+                SELECT 1 FROM password pw WHERE pw.email = {principal} AND pw.super_admin)
+         END"
+    )
+}
+
+/// Re-point the identities a fork clones verbatim at its creator when they name nobody in the fork,
+/// once its membership is final so copied members keep theirs. Unlike scripts and flows these
+/// cannot drop the identity: an app deploy rejects a preserved one that does not resolve, and
+/// publisher apps, schedules and triggers need one to run.
+async fn repoint_unresolvable_cloned_identities(
+    tx: &mut Transaction<'_, Postgres>,
+    target_workspace_id: &str,
+    authed: &ApiAuthed,
+) -> Result<()> {
+    let principal = username_to_permissioned_as(&authed.username);
+
+    sqlx::query(&format!(
+        "UPDATE app SET policy = policy
+             || jsonb_build_object('on_behalf_of', $2::text, 'on_behalf_of_email', $3::text)
+         WHERE workspace_id = $1 AND policy->>'on_behalf_of' IS NOT NULL
+           AND NOT ({})",
+        principal_resolves_sql("(policy->>'on_behalf_of')")
+    ))
+    .bind(target_workspace_id)
+    .bind(&principal)
+    .bind(&authed.email)
+    .execute(&mut **tx)
+    .await?;
+
+    // A draft holding a genuine NUL escape (the rule of `json_text_has_nul_escape`) is left as it
+    // is, since parsing it would abort the fork. The check must stay in a CASE: json `->>` raises on
+    // a NUL anywhere in the value, and Postgres reorders plain AND conditions.
+    let nul_escape = r"(^|[^\\])(\\\\)*\\u0000";
+    sqlx::query(&format!(
+        "UPDATE draft SET value = to_json(jsonb_set(jsonb_set(to_jsonb(value),
+             ARRAY['policy', 'on_behalf_of'], to_jsonb($2::text)),
+             ARRAY['policy', 'on_behalf_of_email'], to_jsonb($3::text)))
+         WHERE workspace_id = $1 AND typ IN ('app', 'raw_app')
+           AND CASE WHEN value::text ~ $4 THEN false
+                    ELSE value->'policy'->>'on_behalf_of' IS NOT NULL AND NOT ({}) END",
+        principal_resolves_sql("(value->'policy'->>'on_behalf_of')")
+    ))
+    .bind(target_workspace_id)
+    .bind(&principal)
+    .bind(&authed.email)
+    .bind(nul_escape)
+    .execute(&mut **tx)
+    .await?;
+
+    sqlx::query(&format!(
+        "UPDATE draft SET value = to_json(jsonb_set(to_jsonb(value),
+             ARRAY['permissioned_as'], to_jsonb($2::text)))
+         WHERE workspace_id = $1 AND starts_with(typ::text, 'trigger_')
+           AND CASE WHEN value::text ~ $3 THEN false
+                    ELSE value->>'permissioned_as' IS NOT NULL AND NOT ({}) END",
+        principal_resolves_sql("(value->>'permissioned_as')")
+    ))
+    .bind(target_workspace_id)
+    .bind(&principal)
+    .bind(nul_escape)
+    .execute(&mut **tx)
+    .await?;
+
+    let column_resolves = principal_resolves_sql("permissioned_as");
+
+    // SAFETY: every table name is a literal from this list, never user input.
+    for table in [
+        "http_trigger",
+        "websocket_trigger",
+        "kafka_trigger",
+        "nats_trigger",
+        "postgres_trigger",
+        "mqtt_trigger",
+        "amqp_trigger",
+        "sqs_trigger",
+        "gcp_trigger",
+        "azure_trigger",
+        "email_trigger",
+    ] {
+        sqlx::query(&format!(
+            "UPDATE {table} SET permissioned_as = $2
+             WHERE workspace_id = $1 AND NOT ({column_resolves})"
+        ))
+        .bind(target_workspace_id)
+        .bind(&principal)
+        .execute(&mut **tx)
+        .await?;
+    }
+
+    // `email` is still written for workers that predate `permissioned_as`.
+    sqlx::query(&format!(
+        "UPDATE schedule SET permissioned_as = $2, email = $3
+         WHERE workspace_id = $1 AND NOT ({column_resolves})"
+    ))
+    .bind(target_workspace_id)
+    .bind(&principal)
+    .bind(&authed.email)
+    .execute(&mut **tx)
+    .await?;
+
+    Ok(())
+}
+
 /// Carries over the recorded principal under the rule spelled out on [`clone_scripts`].
 async fn clone_flows(
     tx: &mut Transaction<'_, Postgres>,
@@ -8083,6 +8204,8 @@ async fn create_workspace_fork(
     // re-enables in the fork, with parent-conflict warnings on enable.
     clone_triggers_and_schedules(&mut tx, &parent_workspace_id, &forked_id).await?;
 
+    repoint_unresolvable_cloned_identities(&mut tx, &forked_id, &authed).await?;
+
     // Update forked datatable settings to point to new databases
     for fdt in &nw.forked_datatables {
         apply_forked_datatable(&db, &mut tx, &parent_workspace_id, &forked_id, fdt).await?;