fix: enforce caller ACL on flow step/subflow references

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/.sqlx/query-0bfdd2a6bb7e58d97542b4961674c42cfec966e68073041cc4b5d23d98cdab79.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT hash FROM script WHERE workspace_id = $1 AND hash = ANY($2::bigint[])",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "hash",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8Array"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "0bfdd2a6bb7e58d97542b4961674c42cfec966e68073041cc4b5d23d98cdab79"
+}

backend/.sqlx/query-3dccc8e745f4a0973541088f66172e74af6828c1ab52cf7a6fd10b305deade85.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM flow WHERE workspace_id = $1 AND path = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "3dccc8e745f4a0973541088f66172e74af6828c1ab52cf7a6fd10b305deade85"
+}

backend/.sqlx/query-ad6ea0915624aa7895f594f784a350fa949db241fb2dc5c7ec16c829739081a5.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT DISTINCT path FROM script WHERE workspace_id = $1 AND path = ANY($2::text[]) AND deleted = false",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "ad6ea0915624aa7895f594f784a350fa949db241fb2dc5c7ec16c829739081a5"
+}

backend/.sqlx/query-b159cc3ab6fddcfb561c90d20bee2ff605a1c3721a4cbcc080f98fbb87277c37.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND hash = $2)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "b159cc3ab6fddcfb561c90d20bee2ff605a1c3721a4cbcc080f98fbb87277c37"
+}

backend/.sqlx/query-e8c412affbf7c0808d252f61310ff385c7212522ab7f9fab79c3bdcced1a4911.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND path = $2 AND deleted = false)",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "exists",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "e8c412affbf7c0808d252f61310ff385c7212522ab7f9fab79c3bdcced1a4911"
+}

backend/.sqlx/query-f22aaca69608b8769bf3fb581fee8705ee35717d42163c464ec281e695a341ef.json

@@ -0,0 +1,23 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT path FROM flow WHERE workspace_id = $1 AND path = ANY($2::text[])",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "TextArray"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "f22aaca69608b8769bf3fb581fee8705ee35717d42163c464ec281e695a341ef"
+}

backend/tests/fixtures/flow_step_refs_authz.sql

@@ -0,0 +1,29 @@
+-- Fixture for the flow step/sub-flow reference authorization test (WIN-2412).
+--
+-- A private folder only the admin can read holds a script and a flow. A plain
+-- workspace member (test-user-3) has write access to `f/shared`, so they can
+-- deploy and preview flows there, but must not be able to point a step at
+-- anything inside `f/private`.
+
+INSERT INTO folder (workspace_id, name, display_name, owners, extra_perms, created_by)
+VALUES ('test-workspace', 'private', 'Private Folder', '{"u/test-user"}', '{}', 'test-user');
+
+INSERT INTO folder (workspace_id, name, display_name, owners, extra_perms, created_by)
+VALUES ('test-workspace', 'shared', 'Shared Folder', '{"u/test-user"}', '{"u/test-user-3": true}', 'test-user');
+
+INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, summary, description, lock, extra_perms)
+VALUES ('test-workspace', 2412001, 'f/private/hidden',
+        'export function main() { return "hidden"; }',
+        'deno', 'script', 'test-user', '{}', 'Hidden script', '', '', '{}');
+
+INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, summary, description, lock, extra_perms)
+VALUES ('test-workspace', 2412002, 'f/shared/visible',
+        'export function main() { return "visible"; }',
+        'deno', 'script', 'test-user', '{}', 'Visible script', '', '', '{}');
+
+INSERT INTO flow (workspace_id, path, summary, description, value, edited_by, edited_at, schema, extra_perms, versions)
+VALUES ('test-workspace', 'f/private/hiddenflow', 'Hidden flow', '',
+        '{"modules": []}', 'test-user', NOW(), '{}', '{}', ARRAY[2412003::bigint]);
+
+INSERT INTO flow_version (id, workspace_id, path, value, schema, created_by, created_at)
+VALUES (2412003, 'test-workspace', 'f/private/hiddenflow', '{"modules": []}', '{}', 'test-user', NOW());

backend/tests/flow_step_refs_authz.rs

@@ -0,0 +1,154 @@
+//! Regression test for the flow step/sub-flow reference ACL bypass (WIN-2412).
+//!
+//! Step references (`script` by path or hash, `flow` sub-flows) are resolved with a
+//! privileged connection when the flow runs, so whoever saves the flow decides which
+//! runnable a step reaches. Without a gate at save time, any workspace member could
+//! deploy — or preview — a flow whose step points at a folder-protected script and read
+//! its output, or inherit its `on_behalf_of` identity, through the flow.
+//!
+//! Pinned against the `flow_step_refs_authz` fixture (a script and a flow in a folder only
+//! the admin can read):
+//!   - a plain member is denied on create, update and preview, including when the
+//!     reference is nested inside a branch or is the failure module, and the response
+//!     names the runnable it refused;
+//!   - the admin, who can read it, is not blocked (no over-blocking);
+//!   - a member referencing a runnable they *can* read still deploys.
+
+use serde_json::json;
+use sqlx::{Pool, Postgres};
+use windmill_test_utils::*;
+
+const WS: &str = "test-workspace";
+const ADMIN: &str = "SECRET_TOKEN";
+const MEMBER: &str = "SECRET_TOKEN_3";
+
+async fn post(
+    base: &str,
+    path: &str,
+    token: &str,
+    body: serde_json::Value,
+) -> (reqwest::StatusCode, String) {
+    let resp = reqwest::Client::new()
+        .post(format!("{base}/api/w/{WS}/{path}"))
+        .header("Authorization", format!("Bearer {token}"))
+        .json(&body)
+        .send()
+        .await
+        .expect("request");
+    let status = resp.status();
+    (status, resp.text().await.expect("body"))
+}
+
+fn new_flow(path: &str, value: serde_json::Value) -> serde_json::Value {
+    json!({ "path": path, "summary": "", "description": "", "value": value, "schema": {} })
+}
+
+fn script_step(path: &str) -> serde_json::Value {
+    json!({ "modules": [{ "id": "a", "value": { "type": "script", "path": path, "input_transforms": {} } }] })
+}
+
+#[sqlx::test(fixtures("base", "flow_step_refs_authz"))]
+async fn test_flow_step_refs_require_caller_access(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let base = format!("http://localhost:{}", server.addr.port());
+
+    // ---- CORE REGRESSION: a member cannot deploy a step pointing at a script they
+    //      cannot read, wherever in the flow the reference sits.
+    for (case, value) in [
+        ("top-level script step", script_step("f/private/hidden")),
+        (
+            "script step nested in a branch",
+            json!({ "modules": [{ "id": "a", "value": { "type": "branchall", "branches": [
+                { "summary": "", "modules": [{ "id": "b", "value": { "type": "script", "path": "f/private/hidden", "input_transforms": {} } }] }
+            ] } }] }),
+        ),
+        (
+            "failure module",
+            json!({
+                "modules": [],
+                "failure_module": { "id": "failure", "value": { "type": "script", "path": "f/private/hidden", "input_transforms": {} } }
+            }),
+        ),
+        (
+            "sub-flow step",
+            json!({ "modules": [{ "id": "a", "value": { "type": "flow", "path": "f/private/hiddenflow", "input_transforms": {} } }] }),
+        ),
+    ] {
+        let (status, body) = post(
+            &base,
+            "flows/create",
+            MEMBER,
+            new_flow("f/shared/attempt", value),
+        )
+        .await;
+        assert_eq!(
+            status,
+            reqwest::StatusCode::UNAUTHORIZED,
+            "member must be denied deploying a {case} they cannot read: {body}"
+        );
+        assert!(
+            body.contains("f/private/hidden"),
+            "the denial should name the runnable it refused ({case}): {body}"
+        );
+    }
+
+    // Preview runs a request-supplied flow value, which never goes through the deploy
+    // path — it must be gated on its own.
+    let (status, body) = post(
+        &base,
+        "jobs/run/preview_flow",
+        MEMBER,
+        json!({ "value": script_step("f/private/hidden"), "path": "f/shared/attempt", "args": {} }),
+    )
+    .await;
+    assert_eq!(
+        status,
+        reqwest::StatusCode::UNAUTHORIZED,
+        "member must be denied previewing a step they cannot read: {body}"
+    );
+
+    // ---- NO OVER-BLOCKING: a readable reference still deploys, for the member and for
+    //      the admin who can read the private folder.
+    let (status, body) = post(
+        &base,
+        "flows/create",
+        MEMBER,
+        new_flow("f/shared/allowed", script_step("f/shared/visible")),
+    )
+    .await;
+    assert_eq!(
+        status,
+        reqwest::StatusCode::CREATED,
+        "a member referencing a script they can read must still deploy: {body}"
+    );
+
+    let (status, body) = post(
+        &base,
+        "flows/create",
+        ADMIN,
+        new_flow("f/private/delegating", script_step("f/private/hidden")),
+    )
+    .await;
+    assert_eq!(
+        status,
+        reqwest::StatusCode::CREATED,
+        "the admin can read the private script and must not be blocked: {body}"
+    );
+
+    // An update re-points an already-deployed flow, so it is gated like a create.
+    let (status, body) = post(
+        &base,
+        "flows/update/f/shared/allowed",
+        MEMBER,
+        new_flow("f/shared/allowed", script_step("f/private/hidden")),
+    )
+    .await;
+    assert_eq!(
+        status,
+        reqwest::StatusCode::UNAUTHORIZED,
+        "member must be denied re-pointing a flow at a script they cannot read: {body}"
+    );
+
+    Ok(())
+}

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

@@ -47,7 +47,10 @@ use windmill_common::HUB_BASE_URL;
 use windmill_common::{
     db::UserDB,
     error::{self, to_anyhow, Error, JsonResult, Result},
-    flows::{EditFlow, Flow, FlowWithStarred, ListFlowQuery, ListableFlow, NewFlow},
+    flows::{
+        require_readable_flow_step_refs, EditFlow, Flow, FlowWithStarred, ListFlowQuery,
+        ListableFlow, NewFlow,
+    },
     jobs::JobPayload,
     schedule::Schedule,
     triggers::MovedNativeTrigger,
@@ -533,6 +536,24 @@ async fn list_paths_from_workspace_runnable(
     Ok(Json(runnables))
 }
 
+/// Gate the runnables the flow's steps point at on the deployer's own ACL.
+///
+/// Runs after `maybe_refresh_folders` so it sees the folder set the deploy itself is authorized
+/// with, and before the flow value is written: a reference persisted here is resolved with a
+/// privileged connection at run time.
+async fn check_flow_step_refs_access(
+    authed: &ApiAuthed,
+    user_db: &UserDB,
+    db: &DB,
+    w_id: &str,
+    new_flow: &NewFlow,
+) -> error::Result<()> {
+    let value = new_flow
+        .parse_flow_value()
+        .map_err(|e| Error::BadRequest(format!("Invalid flow value: {e:#}")))?;
+    require_readable_flow_step_refs(authed, user_db, db, w_id, &value).await
+}
+
 async fn validate_flow(new_flow: &NewFlow) -> error::Result<()> {
     #[cfg(not(feature = "enterprise"))]
     if new_flow.ws_error_handler_muted.is_some_and(|val| val) {
@@ -610,6 +631,8 @@ async fn create_flow(
     // cron::Schedule::from_str(&ns.schedule).map_err(|e| error::Error::BadRequest(e.to_string()))?;
     let authed = maybe_refresh_folders(&nf.path, &w_id, authed, &db).await;
 
+    check_flow_step_refs_access(&authed, &user_db, &db, &w_id, &nf).await?;
+
     // Apply folder default_permissioned_as on create when the caller did not
     // explicitly preserve a value and the user can preserve.
     let explicit_preserve = (nf.on_behalf_of_email.is_some()
@@ -1144,6 +1167,9 @@ async fn update_flow(
     validate_flow(&nf).await?;
 
     let authed = maybe_refresh_folders(&flow_path, &w_id, authed, &db).await;
+
+    check_flow_step_refs_access(&authed, &user_db, &db, &w_id, &nf).await?;
+
     let mut tx = user_db.clone().begin(&authed).await?;
 
     check_schedule_conflict(&mut tx, &w_id, flow_path).await?;

backend/windmill-api/src/jobs.rs

@@ -9186,6 +9186,16 @@ async fn run_preview_flow_job(
     // jobs:run scope so a narrowly-scoped token cannot escape its scope. See run_preview_script.
     check_scopes(&authed, || format!("jobs:run"))?;
     require_path_read_access_for_preview(&authed, &raw_flow.path)?;
+    // The preview value is request-supplied, so its step references never went through the
+    // deploy-time gate in `create_flow`/`update_flow`.
+    windmill_common::flows::require_readable_flow_step_refs(
+        &authed,
+        &user_db,
+        &db,
+        &w_id,
+        &raw_flow.value,
+    )
+    .await?;
     let scheduled_for = run_query.get_scheduled_for(&db).await?;
     let tag = run_query.tag.clone().or(raw_flow.tag.clone());
     check_tag_available_for_workspace(&db, &w_id, &tag, &authed).await?;

backend/windmill-common/src/flows.rs

@@ -15,7 +15,7 @@ use sqlx::types::JsonRawValue;
 
 use crate::{
     cache::{self, FlowExtras},
-    db::DB,
+    db::{Authable, UserDB, DB},
     error::{to_anyhow, Error},
     utils::{http_get_from_hub, StripPath},
     worker::{to_raw_value, Connection},
@@ -246,6 +246,204 @@ pub async fn resolve_modules(
     Ok(())
 }
 
+/// A deployed workspace runnable that a flow step points at.
+///
+/// Inline steps (`rawscript`, `flowscript`) and `hub/` references carry their own code or come
+/// from the public hub, so neither is subject to a workspace ACL.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub enum FlowStepRef {
+    /// `script` step resolved by path at run time (latest deployed version).
+    ScriptPath(String),
+    /// `script` step pinned to a version.
+    ScriptHash { path: String, hash: i64 },
+    /// `flow` step (sub-flow).
+    FlowPath(String),
+}
+
+impl std::fmt::Display for FlowStepRef {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            FlowStepRef::ScriptPath(path) => write!(f, "script {path}"),
+            FlowStepRef::ScriptHash { path, hash } => {
+                write!(
+                    f,
+                    "script {path} (hash {})",
+                    crate::scripts::ScriptHash(*hash)
+                )
+            }
+            FlowStepRef::FlowPath(path) => write!(f, "flow {path}"),
+        }
+    }
+}
+
+fn is_workspace_owned_path(path: &str) -> bool {
+    path.starts_with("u/") || path.starts_with("f/") || path.starts_with("g/")
+}
+
+/// Every deployed runnable the steps of `value` point at, deduplicated.
+///
+/// Descends into loops, branches and agent tools, and covers the failure and preprocessor
+/// modules — the API is reachable directly, so anything that resolves a path at run time must be
+/// collected here.
+pub fn collect_flow_step_refs(value: &FlowValue) -> anyhow::Result<Vec<FlowStepRef>> {
+    let mut refs: Vec<FlowStepRef> = Vec::new();
+    let mut seen: std::collections::HashSet<FlowStepRef> = std::collections::HashSet::new();
+    let mut collect = |module: &FlowModule| -> anyhow::Result<()> {
+        let step_ref = match module.get_value() {
+            Ok(FlowModuleValue::Script { path, hash, .. }) if is_workspace_owned_path(&path) => {
+                match hash {
+                    Some(hash) => FlowStepRef::ScriptHash { path, hash: hash.0 },
+                    None => FlowStepRef::ScriptPath(path),
+                }
+            }
+            Ok(FlowModuleValue::Flow { path, .. }) if is_workspace_owned_path(&path) => {
+                FlowStepRef::FlowPath(path)
+            }
+            _ => return Ok(()),
+        };
+        if seen.insert(step_ref.clone()) {
+            refs.push(step_ref);
+        }
+        Ok(())
+    };
+
+    FlowModule::traverse_modules(&value.modules, &mut collect)?;
+    let extra_modules: Vec<FlowModule> = value
+        .failure_module
+        .iter()
+        .chain(value.preprocessor_module.iter())
+        .map(|m| (**m).clone())
+        .collect();
+    FlowModule::traverse_modules(&extra_modules, &mut collect)?;
+    Ok(refs)
+}
+
+/// Reject a flow whose steps point at a runnable that exists in the workspace but is not
+/// readable by `authed`.
+///
+/// Without this, any workspace member could save a step referencing a folder-protected script
+/// and read its output — or inherit its `on_behalf_of` identity — through the flow, since step
+/// references are resolved with a privileged connection at run time.
+///
+/// A reference to a runnable that does not exist is deliberately tolerated: flows are routinely
+/// deployed before the scripts they call (CLI/git-sync push order), and run-time resolution
+/// enforces the ACL again on whatever the reference ends up pointing at.
+pub async fn require_readable_flow_step_refs<T: Authable + Sync>(
+    authed: &T,
+    user_db: &UserDB,
+    db: &DB,
+    w_id: &str,
+    value: &FlowValue,
+) -> Result<(), Error> {
+    let refs = collect_flow_step_refs(value)
+        .map_err(|e| Error::BadRequest(format!("Invalid flow value: {e:#}")))?;
+    if refs.is_empty() {
+        return Ok(());
+    }
+
+    let script_paths = refs
+        .iter()
+        .filter_map(|r| match r {
+            FlowStepRef::ScriptPath(path) => Some(path.clone()),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    let script_hashes = refs
+        .iter()
+        .filter_map(|r| match r {
+            FlowStepRef::ScriptHash { hash, .. } => Some(*hash),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+    let flow_paths = refs
+        .iter()
+        .filter_map(|r| match r {
+            FlowStepRef::FlowPath(path) => Some(path.clone()),
+            _ => None,
+        })
+        .collect::<Vec<_>>();
+
+    let mut tx = user_db.clone().begin(authed).await?;
+    let visible_script_paths = sqlx::query_scalar!(
+        "SELECT DISTINCT path FROM script WHERE workspace_id = $1 AND path = ANY($2::text[]) AND deleted = false",
+        w_id,
+        &script_paths[..]
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+    let visible_script_hashes = sqlx::query_scalar!(
+        "SELECT hash FROM script WHERE workspace_id = $1 AND hash = ANY($2::bigint[])",
+        w_id,
+        &script_hashes[..]
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+    let visible_flow_paths = sqlx::query_scalar!(
+        "SELECT path FROM flow WHERE workspace_id = $1 AND path = ANY($2::text[])",
+        w_id,
+        &flow_paths[..]
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+    tx.commit().await?;
+
+    let invisible = refs
+        .into_iter()
+        .filter(|r| match r {
+            FlowStepRef::ScriptPath(path) => !visible_script_paths.contains(path),
+            FlowStepRef::ScriptHash { hash, .. } => !visible_script_hashes.contains(hash),
+            FlowStepRef::FlowPath(path) => !visible_flow_paths.contains(path),
+        })
+        .collect::<Vec<_>>();
+    if invisible.is_empty() {
+        return Ok(());
+    }
+
+    // Re-probe the invisible ones without the RLS filter to tell "does not exist" (tolerated)
+    // from "exists but hidden" (the access the reference would otherwise borrow).
+    let mut denied: Vec<String> = Vec::new();
+    for step_ref in invisible {
+        let exists = match &step_ref {
+            FlowStepRef::ScriptPath(path) => sqlx::query_scalar!(
+                "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND path = $2 AND deleted = false)",
+                w_id,
+                path
+            )
+            .fetch_one(db)
+            .await?
+            .unwrap_or(false),
+            FlowStepRef::ScriptHash { hash, .. } => sqlx::query_scalar!(
+                "SELECT EXISTS(SELECT 1 FROM script WHERE workspace_id = $1 AND hash = $2)",
+                w_id,
+                hash
+            )
+            .fetch_one(db)
+            .await?
+            .unwrap_or(false),
+            FlowStepRef::FlowPath(path) => sqlx::query_scalar!(
+                "SELECT EXISTS(SELECT 1 FROM flow WHERE workspace_id = $1 AND path = $2)",
+                w_id,
+                path
+            )
+            .fetch_one(db)
+            .await?
+            .unwrap_or(false),
+        };
+        if exists {
+            denied.push(step_ref.to_string());
+        }
+    }
+
+    if denied.is_empty() {
+        return Ok(());
+    }
+
+    Err(Error::NotAuthorized(format!(
+        "You are not authorized to access the following runnable(s) referenced by this flow: {}",
+        denied.join(", ")
+    )))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -281,4 +479,46 @@ mod tests {
         let err = extract_hub_flow_id_from_path("hub/flows/0").unwrap_err();
         assert!(matches!(err, Error::BadRequest(_)));
     }
+
+    #[test]
+    fn collect_flow_step_refs_reaches_every_nested_step() {
+        let value: FlowValue = serde_json::from_value(serde_json::json!({
+            "modules": [
+                { "id": "a", "value": { "type": "script", "path": "f/protected/direct" } },
+                { "id": "b", "value": { "type": "script", "path": "f/protected/pinned", "hash": "000000000000002a" } },
+                { "id": "c", "value": { "type": "script", "path": "hub/1234/some_hub_script" } },
+                { "id": "d", "value": { "type": "rawscript", "content": "", "language": "bun" } },
+                { "id": "e", "value": { "type": "forloopflow", "iterator": { "type": "static", "value": [] }, "parallel": false, "modules": [
+                    { "id": "e1", "value": { "type": "flow", "path": "f/protected/subflow" } }
+                ] } },
+                { "id": "f", "value": { "type": "branchone", "default": [
+                    { "id": "f1", "value": { "type": "script", "path": "u/someone/in_default" } }
+                ], "branches": [ { "expr": "true", "modules": [
+                    { "id": "f2", "value": { "type": "script", "path": "g/all/in_branch" } }
+                ] } ] } }
+            ],
+            "failure_module": { "id": "failure", "value": { "type": "script", "path": "f/protected/on_failure" } },
+            "preprocessor_module": { "id": "preprocessor", "value": { "type": "script", "path": "f/protected/preprocess" } }
+        }))
+        .unwrap();
+
+        let mut refs = collect_flow_step_refs(&value)
+            .unwrap()
+            .into_iter()
+            .map(|r| r.to_string())
+            .collect::<Vec<_>>();
+        refs.sort();
+        assert_eq!(
+            refs,
+            vec![
+                "flow f/protected/subflow",
+                "script f/protected/direct",
+                "script f/protected/on_failure",
+                "script f/protected/pinned (hash 000000000000002a)",
+                "script f/protected/preprocess",
+                "script g/all/in_branch",
+                "script u/someone/in_default",
+            ]
+        );
+    }
 }

backend/windmill-common/src/lib.rs

@@ -1965,6 +1965,18 @@ pub async fn get_script_info_for_hash<'e, E: sqlx::PgExecutor<'e>>(
                         computed_hash.unwrap_or_else(|| PermsCache::compute_hash(db_authed.authed)),
                         ScriptHash(hash),
                     );
+                } else if get_script_info_for_hash_inner(db, w_id, hash)
+                    .await?
+                    .is_some()
+                {
+                    // Same "exists but not authorized" split as the path-keyed
+                    // `get_latest_deployed_hash_for_path`, so a pinned reference cannot be
+                    // mistaken for a stale one and silently tolerated by the caller.
+                    return Err(Error::NotAuthorized(format!(
+                        "You are not authorized to access this script: {} (but it exists). Your permissions are: {:?}",
+                        ScriptHash(hash),
+                        db_authed.authed
+                    )));
                 }
                 hash_info
             } else {

backend/windmill-worker/src/ai/tools.rs

@@ -371,6 +371,13 @@ async fn execute_windmill_tool(
 
     let is_ai_agent_tool = matches!(tool_value, FlowModuleValue::AIAgent { .. });
 
+    // The permissions the tool job is pushed with. Resolved before the payload because the
+    // tool's script reference is looked up under them.
+    let job_perms: Option<windmill_common::db::Authed> =
+        windmill_common::auth::get_job_perms(ctx.db, &ctx.job.id, &ctx.job.workspace_id)
+            .await?
+            .map(|x| x.into());
+
     let job_payload = match tool_value {
         FlowModuleValue::Script { path: script_path, hash: script_hash, tag_override, .. } => {
             script_to_payload(
@@ -381,6 +388,7 @@ async fn execute_windmill_tool(
                 tool_module,
                 tag_override,
                 tool_module.apply_preprocessor,
+                job_perms.as_ref(),
             )
             .await?
         }
@@ -459,12 +467,7 @@ async fn execute_windmill_tool(
         }
     };
 
-    let mut tx = ctx.db.begin().await?;
-
-    let job_perms =
-        windmill_common::auth::get_job_perms(&mut *tx, &ctx.job.id, &ctx.job.workspace_id)
-            .await?
-            .map(|x| x.into());
+    let tx = ctx.db.begin().await?;
 
     let (email, permissioned_as) = if let Some(on_behalf_of) = job_payload.on_behalf_of.as_ref() {
         (&on_behalf_of.email, on_behalf_of.permissioned_as.clone())

backend/windmill-worker/src/worker_flow.rs

@@ -36,7 +36,7 @@ use windmill_common::auth::get_job_perms;
 use windmill_common::bench::BenchmarkIter;
 use windmill_common::cache::{self, RawData};
 use windmill_common::client::AuthedClient;
-use windmill_common::db::Authed;
+use windmill_common::db::{Authed, AuthedRef, UserDB, UserDbWithAuthed};
 use windmill_common::flow_conversations::{add_message_to_conversation_tx, MessageType};
 use windmill_common::flow_status::{
     ApprovalConditions, FlowJobDuration, FlowJobsDuration, FlowStatusModuleWParent,
@@ -4023,6 +4023,21 @@ async fn push_next_flow_job(
     };
     tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow job args computed");
 
+    let flow_root_job = get_root_job_id(&flow_job);
+    // The permissions every step of this run is pushed with. Resolved before the transform
+    // because the step's script/sub-flow reference is looked up under them, so a step cannot
+    // reach a runnable the run itself is not authorized for.
+    let root_job_perms = get_job_perms(db, &flow_root_job, &flow_job.workspace_id).await?;
+    // The end user is a property of whoever triggered the root run, so every step (and
+    // transitively every subflow's step) must see the same WM_END_USER_EMAIL as the run.
+    // It is read from the root's `job_perms` rather than off `flow_job`: only a freshly
+    // pulled job carries `permissioned_as_end_user_email`, and every step past the first
+    // is pushed from a flow job re-fetched by `get_mini_pulled_job`, which does not.
+    let end_user_email = root_job_perms
+        .as_ref()
+        .and_then(|x| x.end_user_email.clone());
+    let job_perms: Option<Authed> = root_job_perms.map(|x| x.into());
+
     let next_flow_transform = compute_next_flow_transform(
         arc_flow_job_args.clone(),
         arc_last_job_result.clone(),
@@ -4040,6 +4055,7 @@ async fn push_next_flow_job(
         resume.clone(),
         approvers.clone(),
         is_skipped,
+        job_perms.as_ref(),
     )
     .warn_after_seconds(3)
     .await?;
@@ -4379,21 +4395,6 @@ async fn push_next_flow_job(
                 .or_else(|| Some(flow_job.id))
         };
 
-        let flow_root_job = get_root_job_id(&flow_job);
-
-        // forward root job permissions to the new job
-        let root_job_perms =
-            get_job_perms(&mut *tx, &flow_root_job, &flow_job.workspace_id).await?;
-        // The end user is a property of whoever triggered the root run, so every step (and
-        // transitively every subflow's step) must see the same WM_END_USER_EMAIL as the run.
-        // It is read from the root's `job_perms` rather than off `flow_job`: only a freshly
-        // pulled job carries `permissioned_as_end_user_email`, and every step past the first
-        // is pushed from a flow job re-fetched by `get_mini_pulled_job`, which does not.
-        let end_user_email = root_job_perms
-            .as_ref()
-            .and_then(|x| x.end_user_email.clone());
-        let job_perms: Option<Authed> = root_job_perms.map(|x| x.into());
-
         tracing::debug!(id = %flow_job.id, root_id = %job_root, "computed perms for job {i} of {len}");
         let tag = resolve_flow_step_tag(
             step.is_preprocessor_step(),
@@ -4487,7 +4488,7 @@ async fn push_next_flow_job(
             new_job_priority_override,
             job_perms.as_ref(),
             continue_with_runners,
-            end_user_email,
+            end_user_email.clone(),
             None,
             None,
         )
@@ -5159,6 +5160,7 @@ async fn compute_next_flow_transform(
     resume: Arc<Box<RawValue>>,
     approvers: Arc<Box<RawValue>>,
     is_skipped: bool,
+    step_authed: Option<&Authed>,
 ) -> error::Result<NextFlowTransform> {
     if module.mock.is_some() && module.mock.as_ref().unwrap().enabled {
         return Ok(NextFlowTransform::Continue(
@@ -5203,6 +5205,7 @@ async fn compute_next_flow_transform(
                 delete_after_secs,
                 &flow_job.workspace_id,
                 db,
+                step_authed,
             )
             .await?;
             if let Some(nested) = nested_restart_payload(status, &module.id, None) {
@@ -5237,6 +5240,7 @@ async fn compute_next_flow_transform(
                 module,
                 tag_override,
                 module.apply_preprocessor,
+                step_authed,
             )
             .await?;
             Ok(NextFlowTransform::Continue(
@@ -5344,6 +5348,7 @@ async fn compute_next_flow_transform(
                 module,
                 delete_after_use,
                 start_runners,
+                step_authed,
             )
             .await
         }
@@ -5403,6 +5408,7 @@ async fn compute_next_flow_transform(
                         module,
                         delete_after_use,
                         start_runners,
+                        step_authed,
                     )
                     .await
                 }
@@ -5723,6 +5729,7 @@ async fn next_loop_iteration(
     module: &FlowModule,
     delete_after_use: bool,
     start_runners: bool,
+    step_authed: Option<&Authed>,
 ) -> Result<NextFlowTransform, Error> {
     let inner_path = || format!("{}/loop-{}", flow_job.runnable_path(), ns.index);
     if is_simple {
@@ -5748,7 +5755,8 @@ async fn next_loop_iteration(
         };
         return Ok(NextFlowTransform::Continue(
             ContinuePayload::SingleJob(
-                payload_from_simple_module(value, db, flow_job, module, inner_path()).await?,
+                payload_from_simple_module(value, db, flow_job, module, inner_path(), step_authed)
+                    .await?,
             ),
             NextStatus::NextLoopIteration { next: ns, simple_input_transforms, start_runners },
         ));
@@ -5968,6 +5976,7 @@ async fn payload_from_simple_module(
     flow_job: &MiniPulledJob,
     module: &FlowModule,
     inner_path: String,
+    step_authed: Option<&Authed>,
 ) -> Result<JobPayloadWithTag, Error> {
     let delete_after_use = module.delete_after_use.unwrap_or(false);
     let delete_after_secs = module.delete_after_secs;
@@ -5979,6 +5988,7 @@ async fn payload_from_simple_module(
                 delete_after_secs,
                 &flow_job.workspace_id,
                 db,
+                step_authed,
             )
             .await?
         }
@@ -5991,6 +6001,7 @@ async fn payload_from_simple_module(
                 module,
                 tag_override,
                 module.apply_preprocessor,
+                step_authed,
             )
             .await?
         }
@@ -6074,14 +6085,37 @@ pub fn raw_script_to_payload(
     }
 }
 
+/// The connection a step's runnable reference is resolved through.
+///
+/// `Some` binds the lookup to the identity the step will run as, so a stored reference cannot
+/// reach a script or sub-flow that identity is not authorized for. `None` (no `job_perms` row for
+/// the root job) falls back to an unfiltered lookup.
+fn step_db_authed<'a>(
+    db: &DB,
+    authed_ref: &'a Option<AuthedRef<'a>>,
+) -> Option<UserDbWithAuthed<'a, AuthedRef<'a>>> {
+    authed_ref
+        .as_ref()
+        .map(|authed| UserDbWithAuthed { db: UserDB::new(db.clone()), authed })
+}
+
 async fn flow_to_payload(
     path: String,
     delete_after_use: bool,
     delete_after_secs: Option<i32>,
     w_id: &str,
     db: &DB,
+    step_authed: Option<&Authed>,
 ) -> Result<JobPayloadWithTag, Error> {
-    let flow_info = get_latest_flow_version_info_for_path(None, &db, w_id, &path, true).await?;
+    let authed_ref = step_authed.map(|x| x.to_authed_ref());
+    let flow_info = get_latest_flow_version_info_for_path(
+        step_db_authed(db, &authed_ref),
+        &db,
+        w_id,
+        &path,
+        true,
+    )
+    .await?;
     let on_behalf_of = flow_info.on_behalf_of(w_id, &db).await?;
     let FlowVersionInfo { version, tag, .. } = flow_info;
     let payload = JobPayload::Flow {
@@ -6121,18 +6155,20 @@ pub async fn script_to_payload(
     module: &FlowModule,
     tag_override: Option<String>,
     apply_preprocessor: Option<bool>,
+    step_authed: Option<&Authed>,
 ) -> Result<JobPayloadWithTag, Error> {
     let tag_override = if tag_override.as_ref().is_some_and(|x| x.trim().is_empty()) {
         None
     } else {
         tag_override
     };
+    let authed_ref = step_authed.map(|x| x.to_authed_ref());
     let (payload, tag, delete_after_use, delete_after_secs, script_timeout, on_behalf_of) =
         if script_hash.is_none() {
             let (jp, tag, delete_after_use, delete_after_secs, script_timeout, on_behalf_of) =
                 script_path_to_payload(
                     &script_path,
-                    None,
+                    step_db_authed(db, &authed_ref),
                     db.clone(),
                     &flow_job.workspace_id,
                     Some(true),
@@ -6149,13 +6185,16 @@ pub async fn script_to_payload(
         } else {
             let hash = script_hash.unwrap();
 
-            let script_info = get_script_info_for_hash(None, db, &flow_job.workspace_id, hash.0)
-                .await?
-                .prefetch_cached(&db)
-                .await?;
-            let on_behalf_of = script_info
-                .on_behalf_of(&flow_job.workspace_id, db)
-                .await?;
+            let script_info = get_script_info_for_hash(
+                step_db_authed(db, &authed_ref),
+                db,
+                &flow_job.workspace_id,
+                hash.0,
+            )
+            .await?
+            .prefetch_cached(&db)
+            .await?;
+            let on_behalf_of = script_info.on_behalf_of(&flow_job.workspace_id, db).await?;
             let ScriptHashInfo {
                 tag,
                 cache_ttl,