Merge remote-tracking branch 'origin/main' into breadcrumb-leaf-navigation-popover

# Conflicts:
#	backend/ee-repo-ref.txt

backend/ee-repo-ref.txt

@@ -1 +1 @@
-5ceb7581a092ad9df1872a2743be2dba550e6129
+d5971fb924d4fdb849e73da6657021902acaacbd

backend/migrations/20261012180101_datatable_role_set.down.sql

@@ -0,0 +1 @@
+DROP TABLE IF EXISTS datatable_role_set;

backend/migrations/20261012180101_datatable_role_set.up.sql

@@ -0,0 +1,21 @@
+-- The logins behind a data table connection made as several roles at once: one Postgres login per
+-- distinct set of data table roles, a member of each. Which roles a login holds is its
+-- memberships on the cluster, and its name is derived from their catalog ids, so a row only has to
+-- carry the credential.
+--
+-- A row is written after its login exists and removed before the login is dropped, so a row always
+-- names a login that can authenticate with `pwd`.
+CREATE TABLE datatable_role_set (
+    cluster VARCHAR(20) NOT NULL CHECK (cluster IN ('instance', 'external_instance')),
+    name VARCHAR(63) NOT NULL,
+    -- Generated by Windmill, never entered by anyone, and never leaves the server.
+    pwd TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    -- Recorded at most once a day. A login unused for long enough is dropped, and made again if
+    -- its set of roles is asked for later.
+    last_used_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+    PRIMARY KEY (cluster, name)
+);
+
+GRANT ALL ON datatable_role_set TO windmill_user;
+GRANT ALL ON datatable_role_set TO windmill_admin;

backend/src/monitor.rs

@@ -4449,6 +4449,24 @@ pub async fn monitor_db(
         }
     };
 
+    // Drop the data table role set logins nobody has connected with for a month. Every ~40 min,
+    // spawned for the same reason as the credential maintenance above: it opens a connection per
+    // managed database for each login it drops.
+    let datatable_role_set_sweep_f = async {
+        if server_mode && iteration.is_some() && iteration.as_ref().unwrap().should_run(240) {
+            if let Some(db) = conn.as_sql() {
+                let db = db.clone();
+                tokio::spawn(async move {
+                    if let Err(e) =
+                        windmill_common::datatable_roles::sweep_unused_role_set_logins(&db).await
+                    {
+                        tracing::error!("Error sweeping unused data table role set logins: {e}");
+                    }
+                });
+            }
+        }
+    };
+
     // Re-encrypt the legacy-format secret variables once new values are written in AES-256-GCM.
     // Every ~5 min until one pass has gone through every workspace; spawned so a large instance
     // does not hold up the join below.
@@ -4538,6 +4556,7 @@ pub async fn monitor_db(
         cleanup_scheduled_job_deletions_f,
         git_auto_pull_f,
         git_credential_maintenance_f,
+        datatable_role_set_sweep_f,
         ai_session_retention_f,
         legacy_secret_upgrade_f,
         pipeline_freshness_watchdog_f,

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

@@ -1847,3 +1847,222 @@ async fn cleanup_waits_out_a_save_racing_it_for_an_instance_database(
     assert_eq!(users, vec!["test-workspace".to_string()]);
     Ok(())
 }
+
+#[cfg(all(feature = "private", feature = "enterprise"))]
+#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))]
+async fn a_connection_naming_no_role_holds_the_default_one_and_the_viewers(
+    db: Pool<Postgres>,
+) -> anyhow::Result<()> {
+    use windmill_common::workspaces::{get_datatable_resource_from_db, DatatableAccess};
+
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let suffix: String = uuid::Uuid::new_v4().simple().to_string()[..8].to_string();
+    let names = [
+        format!("wmtest_floor_{suffix}"),
+        format!("wmtest_emea_{suffix}"),
+        format!("wmtest_amer_{suffix}"),
+    ];
+    let roles_url = format!("http://localhost:{port}/api/settings/datatable_roles");
+    const MEMBER_LOGINS: &str = "SELECT m.rolname::text FROM pg_auth_members am
+         JOIN pg_roles r ON r.oid = am.roleid JOIN pg_roles m ON m.oid = am.member
+         WHERE r.rolname = $1 AND m.rolname LIKE 'wm\\_set\\_%'";
+
+    // The job is the app publisher's in every case: only the viewer the server recorded differs.
+    let connect = |role: Option<&'static str>, viewer: Option<&'static str>| {
+        get_datatable_resource_from_db(
+            &db,
+            "test-workspace",
+            "main",
+            role,
+            DatatableAccess::PermissionedAs {
+                permissioned_as: "u/test-user",
+                email: "test@windmill.dev",
+                end_user_email: viewer,
+            },
+        )
+    };
+    let set_permissions = |default_role: String, roles: Value| {
+        sqlx::query(
+            "UPDATE workspace_settings
+             SET datatable = jsonb_set(datatable, '{datatables,main,permissions}', $1)
+             WHERE workspace_id = 'test-workspace'",
+        )
+        .bind(json!({ "default_role": default_role, "roles": roles }))
+        .execute(&db)
+    };
+
+    let outcome = async {
+        let mut ids = vec![];
+        for name in &names {
+            let created: Value = authed(client().post(&roles_url), "SECRET_TOKEN")
+                .json(&json!({ "name": name }))
+                .send()
+                .await?
+                .error_for_status()?
+                .json()
+                .await?;
+            ids.push(created["id"].as_str().unwrap().to_string());
+        }
+        // `test-user-2` is named by both regional roles, one of them through a group, and by
+        // `admin`; `test-user-3` by one; the workspace admin `test-user` by neither.
+        let roles = json!({
+            "admin": {"tenants": ["u/test-user-2"]},
+            (ids[0].as_str()): {"tenants": ["*"]},
+            (ids[1].as_str()): {"tenants": ["u/test-user-2", "u/test-user-3"]},
+            (ids[2].as_str()): {"tenants": ["g/analysts"]},
+        });
+        set_permissions(ids[0].clone(), roles.clone()).await?;
+        sqlx::query(
+            "INSERT INTO usr_to_group (workspace_id, usr, group_)
+             VALUES ('test-workspace', 'test-user-2', 'analysts')",
+        )
+        .execute(&db)
+        .await?;
+        let member_of = |login: String| async {
+            let mut roles: Vec<String> = sqlx::query_scalar(
+                "SELECT r.rolname::text FROM pg_auth_members am
+                 JOIN pg_roles r ON r.oid = am.roleid JOIN pg_roles m ON m.oid = am.member
+                 WHERE m.rolname = $1",
+            )
+            .bind(login)
+            .fetch_all(&db)
+            .await?;
+            roles.sort();
+            Ok::<_, anyhow::Error>(roles)
+        };
+
+        let all = connect(None, Some("test2@windmill.dev")).await?;
+        let login = all["user"].as_str().unwrap().to_string();
+        assert!(login.starts_with("wm_set_"), "{login}");
+        let mut expected = names.to_vec();
+        expected.sort();
+        assert_eq!(
+            member_of(login.clone()).await?,
+            expected,
+            "the login holds the default role and the viewer's, and not `admin`"
+        );
+        assert_eq!(connect(None, Some("test2@windmill.dev")).await?, all);
+
+        let two = connect(None, Some("test3@windmill.dev")).await?;
+        let mut expected = names[..2].to_vec();
+        expected.sort();
+        assert_eq!(
+            member_of(two["user"].as_str().unwrap().to_string()).await?,
+            expected
+        );
+
+        // An admin may use every role, the one owning the tables included. As a viewer they add
+        // what names them, which here is nothing; nor does a job with no viewer recorded.
+        for viewer in [Some("test@windmill.dev"), None] {
+            assert_eq!(connect(None, viewer).await?["user"], names[0].as_str());
+        }
+        // A named role is that role alone, whoever is viewing.
+        let named: &'static str = Box::leak(names[1].clone().into_boxed_str());
+        assert_eq!(
+            connect(Some(named), Some("test2@windmill.dev")).await?["user"],
+            names[1].as_str()
+        );
+        // A default of `admin` is the connection it always was: the viewer's roles are not added.
+        set_permissions("admin".to_string(), roles.clone()).await?;
+        let admin = connect(None, Some("test2@windmill.dev")).await?;
+        assert_eq!(admin, connect(Some("admin"), None).await?);
+        assert!(!admin["user"].as_str().unwrap().starts_with("wm_"));
+
+        // With no default role the viewer's roles are all there is, the one open to everyone
+        // included, and a connection left with none is refused rather than resolved as anything.
+        set_permissions("$none".to_string(), roles.clone()).await?;
+        assert_eq!(
+            connect(None, Some("test3@windmill.dev")).await?,
+            two,
+            "no default changes nothing for a viewer the roles name"
+        );
+        assert!(connect(None, None).await.is_err());
+        // The same holds for a default the job's own identity is not a tenant of.
+        let mut closed = roles.clone();
+        closed[ids[0].as_str()] = json!({"tenants": []});
+        set_permissions(ids[0].clone(), closed).await?;
+        let as_member = |viewer: Option<&'static str>| {
+            get_datatable_resource_from_db(
+                &db,
+                "test-workspace",
+                "main",
+                None,
+                DatatableAccess::PermissionedAs {
+                    permissioned_as: "u/test-user-3",
+                    email: "test3@windmill.dev",
+                    end_user_email: viewer,
+                },
+            )
+        };
+        assert_eq!(
+            as_member(Some("test3@windmill.dev")).await?["user"],
+            names[1].as_str()
+        );
+        assert!(as_member(None).await.is_err());
+        set_permissions(ids[0].clone(), roles).await?;
+
+        // A set login nobody connects with any more is dropped, a recent one is kept, and a
+        // dropped one is made again for the next viewer who needs it.
+        let exists = || async {
+            sqlx::query_scalar::<_, bool>(
+                "SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)",
+            )
+            .bind(&login)
+            .fetch_one(&db)
+            .await
+        };
+        windmill_common::datatable_roles::sweep_unused_role_set_logins(&db).await?;
+        assert!(exists().await?, "a set login in use was swept");
+        sqlx::query("UPDATE datatable_role_set SET last_used_at = now() - interval '31 days'")
+            .execute(&db)
+            .await?;
+        windmill_common::datatable_roles::sweep_unused_role_set_logins(&db).await?;
+        assert!(!exists().await?, "an unused set login was kept");
+        assert_eq!(
+            connect(None, Some("test2@windmill.dev")).await?["user"],
+            login.as_str()
+        );
+
+        // A disabled role must not stay reachable through a login that is a member of it.
+        authed(
+            client().post(format!("{roles_url}/{}", ids[2])),
+            "SECRET_TOKEN",
+        )
+        .json(&json!({ "enabled": false }))
+        .send()
+        .await?
+        .error_for_status()?;
+        let can_login: bool =
+            sqlx::query_scalar("SELECT rolcanlogin FROM pg_roles WHERE rolname = $1")
+                .bind(&login)
+                .fetch_one(&db)
+                .await?;
+        assert!(!can_login, "{login} still logs in with a disabled role");
+        assert_eq!(
+            connect(None, Some("test2@windmill.dev")).await?["user"],
+            two["user"]
+        );
+        Ok::<_, anyhow::Error>(())
+    }
+    .await;
+
+    // Roles are cluster-wide, so they outlive this test's throwaway database.
+    for name in &names {
+        let logins: Vec<String> = sqlx::query_scalar(MEMBER_LOGINS)
+            .bind(name)
+            .fetch_all(&db)
+            .await
+            .unwrap_or_default();
+        for login in logins {
+            let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{login}\""))
+                .execute(&db)
+                .await;
+        }
+        let _ = sqlx::query(&format!("DROP ROLE IF EXISTS \"{name}\""))
+            .execute(&db)
+            .await;
+    }
+    outcome
+}

backend/windmill-api/openapi.yaml

@@ -5568,6 +5568,9 @@ paths:
                   type: boolean
                 default_role:
                   type: string
+                  description: >-
+                    The id of one of `roles`, `admin` (the default), or `$none` for no default
+                    role: a connection that names no role then holds its viewer's roles only.
                 roles:
                   type: array
                   items:
@@ -34782,6 +34785,7 @@ components:
           type: boolean
         default_role:
           type: string
+          description: A role id, `admin`, or `$none` when the data table has no default role.
         roles:
           type: array
           items:

backend/windmill-api/src/jobs.rs

@@ -8982,6 +8982,7 @@ async fn run_preview_script(
     check_tag_available_for_workspace(&db, &w_id, &tag, &push_args, &authed).await?;
 
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let (uuid, tx) = push(
         &db,
         tx,
@@ -9029,7 +9030,7 @@ async fn run_preview_script(
         None,
         Some(&authed.clone().into()),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(None),
         None,
         scope_ceiling.as_deref(),
@@ -9068,6 +9069,7 @@ async fn run_inline_preview_script(
     if let Some(job_id) = job_id {
         register_potential_assets_on_inline_execution(job_id, &w_id, &preview);
     }
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let utils = get_worker_internal_server_inline_utils()?;
     let result = utils.run_inline_preview_script.as_ref()(RunInlinePreviewScriptFnParams {
         content: preview.content,
@@ -9078,6 +9080,7 @@ async fn run_inline_preview_script(
         created_by: authed.display_username().to_string(),
         permissioned_as: username_to_permissioned_as(&authed.username),
         permissioned_as_email: authed.email.clone(),
+        end_user_email,
         lang: preview.language,
         job_dir: "".to_string(),
         worker_name: "".to_string(),
@@ -9217,6 +9220,7 @@ async fn run_inline_script_inner(
                 .to_string(),
         ));
     }
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let utils = get_worker_internal_server_inline_utils()?;
     let authed_owned: windmill_common::db::Authed = authed.clone().into();
     let result = utils.run_inline_script.as_ref()(RunInlineScriptFnParams {
@@ -9228,6 +9232,7 @@ async fn run_inline_script_inner(
         created_by: authed.display_username().to_string(),
         permissioned_as: username_to_permissioned_as(&authed.username),
         permissioned_as_email: authed.email.clone(),
+        end_user_email,
         job_dir: "".to_string(),
         worker_name: "".to_string(),
         worker_dir: "".to_string(),
@@ -9404,6 +9409,7 @@ async fn run_bundle_preview_script(
             // tracing::info!("is_tar 1: {is_tar}");
             // hmap.insert("")
             let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
+            let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
             let (uuid, ntx) = push(
                 &db,
                 ltx,
@@ -9447,7 +9453,7 @@ async fn run_bundle_preview_script(
                 None,
                 Some(&authed.clone().into()),
                 false,
-                None,
+                end_user_email,
                 authed.trigger_or_fallback(None),
                 None,
                 scope_ceiling.as_deref(),
@@ -10151,6 +10157,7 @@ async fn run_preview_flow_job(
         .await?;
 
     let scope_ceiling = windmill_api_auth::caller_scope_ceiling(&db, &authed).await?;
+    let end_user_email = run_end_user_email(&db, &w_id, &authed, None).await?;
     let (uuid, mut tx) = push(
         &db,
         tx,
@@ -10182,7 +10189,7 @@ async fn run_preview_flow_job(
         None,
         Some(&authed.clone().into()),
         false,
-        None,
+        end_user_email,
         authed.trigger_or_fallback(None),
         None,
         scope_ceiling.as_deref(),

backend/windmill-common/src/datatable_roles.rs

@@ -72,6 +72,11 @@ impl DatatableRoleCluster {
 /// created, renamed or dropped.
 pub const ADMIN_DATATABLE_ROLE: &str = "admin";
 
+/// Stands in a data table's `default_role` for "none": a connection that names no role then holds
+/// its viewer's roles only, and is refused without any. `$` is outside both the role-name charset
+/// and the catalog ids, and a server predating it refuses it as a role the data table lacks.
+pub const NO_DEFAULT_DATATABLE_ROLE: &str = "$none";
+
 /// The login the admin connection uses, and the role every created role is granted to — that
 /// membership is what later lets it `ALTER ... OWNER TO` a role and drop it.
 pub const CUSTOM_INSTANCE_USER: &str = "custom_instance_user";
@@ -121,8 +126,13 @@ fn is_reserved_role_name(name: &str) -> bool {
         || lower.starts_with("pg_")
         || lower.starts_with("windmill_")
         || lower.starts_with("custom_instance_")
+        || lower.starts_with(ROLE_SET_LOGIN_PREFIX)
 }
 
+/// Prefix of the logins Windmill creates for a connection made as several roles at once. Reserved,
+/// so no data table role can be created under, renamed to or taken over at such a name.
+pub const ROLE_SET_LOGIN_PREFIX: &str = "wm_set_";
+
 /// The charset is what makes every downstream interpolation safe: the name reaches Postgres as a
 /// quoted identifier, a `-- role <name>` annotation, and a `?role=` query parameter.
 pub fn validate_role_name(name: &str) -> Result<()> {
@@ -414,6 +424,13 @@ pub async fn drop_datatable_role(
     crate::datatable_roles_oss::drop_datatable_role(db, tx, cluster, name).await
 }
 
+/// Drop the logins made for a connection as several roles at once that nobody has connected with
+/// for a month, on every cluster. They are made again on demand, so this only bounds how many
+/// exist. Safe to call from any server at any time: it serializes on the role catalog lock.
+pub async fn sweep_unused_role_set_logins(db: &DB) -> Result<()> {
+    crate::datatable_roles_oss::sweep_unused_role_set_logins(db).await
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

backend/windmill-common/src/datatable_roles_oss.rs

@@ -29,7 +29,8 @@ pub(crate) use crate::datatable_roles_ee::{
     ensure_instance_db_grant_options_unchecked, forget_datatable_role_everywhere,
     insert_role_catalog_entry, read_role_catalog, read_role_catalog_tx,
     registered_instance_databases, rename_datatable_role, resolve_datatable_role_connection,
-    role_cluster, set_datatable_role_login, update_role_catalog_entry,
+    role_cluster, set_datatable_role_login, sweep_unused_role_set_logins,
+    update_role_catalog_entry,
 };
 
 #[cfg(not(all(feature = "private", feature = "enterprise")))]
@@ -235,6 +236,11 @@ mod ce {
         }
     }
 
+    /// No roles, so no set login to sweep.
+    pub(crate) async fn sweep_unused_role_set_logins(_db: &DB) -> Result<()> {
+        Ok(())
+    }
+
     pub(crate) async fn forget_datatable_role_everywhere(
         _tx: &mut Tx<'_>,
         _role_id: &str,

backend/windmill-common/src/jobs.rs

@@ -474,6 +474,8 @@ pub struct RunInlinePreviewScriptFnParams {
     pub created_by: String,
     pub permissioned_as: String,
     pub permissioned_as_email: String,
+    /// The caller's recorded end user: `job_perms.end_user_email` of the calling job.
+    pub end_user_email: Option<String>,
     pub base_internal_url: String,
     pub worker_name: String,
     pub conn: crate::worker::Connection,
@@ -496,6 +498,8 @@ pub struct RunInlineScriptFnParams {
     pub created_by: String,
     pub permissioned_as: String,
     pub permissioned_as_email: String,
+    /// The caller's recorded end user: `job_perms.end_user_email` of the calling job.
+    pub end_user_email: Option<String>,
     pub base_internal_url: String,
     pub worker_name: String,
     pub conn: crate::worker::Connection,

backend/windmill-common/src/workspaces.rs

@@ -1564,7 +1564,8 @@ pub struct DataTableReference {
 /// The access decision for one data table: which role a caller gets, and who may ask for each.
 #[derive(Deserialize, Serialize, Debug, Clone, Default)]
 pub struct DataTablePermissions {
-    /// A role id from the instance catalog, or `admin`. Absent = `admin`.
+    /// A role id from the instance catalog, `admin`, or
+    /// [`crate::datatable_roles::NO_DEFAULT_DATATABLE_ROLE`]. Absent = `admin`.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub default_role: Option<String>,
     /// Keyed by instance role id, plus the reserved `admin` key. A role absent from this map
@@ -2161,9 +2162,13 @@ pub enum DatatableAccess<'a> {
     PermissionedAs {
         permissioned_as: &'a str,
         email: &'a str,
+        /// The job's viewer: MUST be what the server recorded in `job_perms.end_user_email`,
+        /// never an argument or anything else the job or its caller can set.
+        end_user_email: Option<&'a str>,
     },
-    /// A job identified by id; its owner is read from `v2_job`. For agent workers and anything
-    /// else that authenticates as infrastructure rather than as the job's user.
+    /// A job identified by id; its owner is read from `v2_job` and its viewer from `job_perms`.
+    /// For agent workers and anything else that authenticates as infrastructure rather than as
+    /// the job's user.
     Job(uuid::Uuid),
     /// No identity established. Unpermissioned data tables resolve as before; permissioned ones
     /// are refused, so a caller predating this feature fails closed.
@@ -2211,6 +2216,13 @@ pub async fn can_use_datatable_role_in_governing_workspace(
 ///
 /// The resolved role **logs in as itself**. Never `SET ROLE`: a script could `RESET ROLE` its way
 /// back to admin.
+///
+/// A default role other than `admin` is a floor rather than the whole connection: a resolution
+/// naming no role holds it when the job's own identity may use it, plus the roles whose tenant
+/// lists name the job's viewer, as one login that is a member of each. With no role at all it is
+/// refused. Being an admin adds none of the viewer's and `admin` is never among them, so a viewer
+/// cannot be handed the connection that owns the tables and bypasses row-level security. A named
+/// role is that role alone.
 pub async fn get_datatable_resource_from_db(
     db: &DB,
     w_id: &str,

backend/windmill-worker/src/common.rs

@@ -1726,6 +1726,23 @@ pub async fn cached_result_path(
         hasher.update(b"wac_step:");
         hasher.update(step_key.as_bytes());
     }
+    // What a run returns can depend on who it ran for (`WM_END_USER_EMAIL`, a data table used
+    // as the viewer), so a result cached for one viewer is never served to another. Read from
+    // `job_perms`: a flow job re-fetched between steps does not carry it.
+    let end_user: Option<Option<String>> = sqlx::query_scalar(
+        "SELECT end_user_email FROM job_perms WHERE job_id = $1 AND workspace_id = $2",
+    )
+    .bind(job.id)
+    .bind(&job.workspace_id)
+    .fetch_optional(db)
+    .await?;
+    // Length-prefixed: the arguments follow undelimited, and the end of one viewer's address
+    // must not read as the start of another run's first argument name.
+    if let Some(end_user) = end_user.flatten() {
+        hasher.update(b"end_user:");
+        hasher.update((end_user.len() as u64).to_le_bytes());
+        hasher.update(end_user.as_bytes());
+    }
     hash_args(
         db,
         client,
@@ -2475,6 +2492,52 @@ mod tests {
         }
     }
 
+    // A cache hit returns before the job runs, so the key is all that keeps a result computed
+    // for one viewer (a data table used as the viewer, `WM_END_USER_EMAIL`) from another.
+    #[sqlx::test(migrations = "../migrations")]
+    async fn a_cached_result_is_keyed_by_the_recorded_viewer(pool: sqlx::PgPool) {
+        let client = test_client();
+
+        // The last two differ only in where the viewer's address ends and the first argument
+        // name starts.
+        let mut paths = vec![];
+        for (viewer, arg) in [
+            (Some("a@windmill.dev"), "mode"),
+            (Some("a@windmill.dev"), "mode"),
+            (Some("b@windmill.dev"), "mode"),
+            (None, "mode"),
+            (Some("a@windmill.co"), "mode"),
+            (Some("a@windmill.com"), "ode"),
+        ] {
+            let args = HashMap::from([(arg.to_string(), to_raw_value(&"all"))]);
+            let job = MiniPulledJob {
+                id: uuid::Uuid::new_v4(),
+                args: Some(sqlx::types::Json(args)),
+                ..test_job()
+            };
+            sqlx::query(
+                "INSERT INTO job_perms (job_id, email, username, is_admin, is_operator,
+                     workspace_id, groups, folders, end_user_email)
+                 VALUES ($1, 'publisher@windmill.dev', 'publisher', false, false, 'test', '{}',
+                     '{}', $2)",
+            )
+            .bind(job.id)
+            .bind(viewer)
+            .execute(&pool)
+            .await
+            .unwrap();
+            paths.push(cached_result_path(&pool, &client, &job, None).await.unwrap());
+        }
+        assert_eq!(
+            paths[0], paths[1],
+            "one viewer shares their own cached result"
+        );
+        assert_ne!(paths[0], paths[2], "two viewers share a cached result");
+        assert_ne!(paths[0], paths[3]);
+        assert_ne!(paths[2], paths[3]);
+        assert_ne!(paths[4], paths[5], "an address runs into the argument names");
+    }
+
     #[tokio::test]
     async fn test_transform_array_over_1000_passthrough() {
         let db_url = std::env::var("DATABASE_URL")

backend/windmill-worker/src/duckdb_executor.rs

@@ -2758,6 +2758,7 @@ async fn transform_attach_datatable(
                 DatatableAccess::PermissionedAs {
                     permissioned_as: &job.permissioned_as,
                     email: &job.permissioned_as_email,
+                    end_user_email: job.permissioned_as_end_user_email.as_deref(),
                 },
             )
             .await?

backend/windmill-worker/src/pg_executor.rs

@@ -1229,6 +1229,7 @@ pub async fn do_postgresql(
                             DatatableAccess::PermissionedAs {
                                 permissioned_as: &job.permissioned_as,
                                 email: &job.permissioned_as_email,
+                                end_user_email: job.permissioned_as_end_user_email.as_deref(),
                             },
                         )
                         .await?

backend/windmill-worker/src/worker.rs

@@ -7253,7 +7253,7 @@ pub fn init_worker_internal_server_inline_utils(
         base_internal_url,
         killpill_rx: Arc::new(killpill_rx),
         run_inline_preview_script: Arc::new(|params| {
-            let job = MiniPulledJob::new_inline(
+            let mut job = MiniPulledJob::new_inline(
                 params.workspace_id,
                 params.args,
                 params.created_by,
@@ -7265,6 +7265,7 @@ pub fn init_worker_internal_server_inline_utils(
                 "inline_preview".to_string(),
                 Some(params.lang),
             );
+            job.permissioned_as_end_user_email = params.end_user_email;
             Box::pin(async move {
                 let mut mem_peak: i32 = -1;
                 let mut canceled_by: Option<CanceledBy> = None;
@@ -7314,7 +7315,7 @@ pub fn init_worker_internal_server_inline_utils(
                 let content_info =
                     get_script_content_by_hash(&script_hash, &params.workspace_id, &params.conn)
                         .await?;
-                let job = MiniPulledJob::new_inline(
+                let mut job = MiniPulledJob::new_inline(
                     params.workspace_id,
                     params.args,
                     params.created_by,
@@ -7326,6 +7327,7 @@ pub fn init_worker_internal_server_inline_utils(
                     "inline_run".to_string(),
                     content_info.language,
                 );
+                job.permissioned_as_end_user_email = params.end_user_email;
                 let mut mem_peak: i32 = -1;
                 let mut canceled_by: Option<CanceledBy> = None;
                 let mut column_order: Option<Vec<String>> = None;