diff --git a/backend/.sqlx/query-0bfdd2a6bb7e58d97542b4961674c42cfec966e68073041cc4b5d23d98cdab79.json b/backend/.sqlx/query-0bfdd2a6bb7e58d97542b4961674c42cfec966e68073041cc4b5d23d98cdab79.json new file mode 100644 index 0000000000..ca9b7c3caf --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-3dccc8e745f4a0973541088f66172e74af6828c1ab52cf7a6fd10b305deade85.json b/backend/.sqlx/query-3dccc8e745f4a0973541088f66172e74af6828c1ab52cf7a6fd10b305deade85.json new file mode 100644 index 0000000000..c6e572863a --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-ad6ea0915624aa7895f594f784a350fa949db241fb2dc5c7ec16c829739081a5.json b/backend/.sqlx/query-ad6ea0915624aa7895f594f784a350fa949db241fb2dc5c7ec16c829739081a5.json new file mode 100644 index 0000000000..5a956f6846 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-b159cc3ab6fddcfb561c90d20bee2ff605a1c3721a4cbcc080f98fbb87277c37.json b/backend/.sqlx/query-b159cc3ab6fddcfb561c90d20bee2ff605a1c3721a4cbcc080f98fbb87277c37.json new file mode 100644 index 0000000000..d65f75636d --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-e8c412affbf7c0808d252f61310ff385c7212522ab7f9fab79c3bdcced1a4911.json b/backend/.sqlx/query-e8c412affbf7c0808d252f61310ff385c7212522ab7f9fab79c3bdcced1a4911.json new file mode 100644 index 0000000000..a91bfc9b73 --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-f22aaca69608b8769bf3fb581fee8705ee35717d42163c464ec281e695a341ef.json b/backend/.sqlx/query-f22aaca69608b8769bf3fb581fee8705ee35717d42163c464ec281e695a341ef.json new file mode 100644 index 0000000000..407244b0e9 --- /dev/null +++ b/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" +} diff --git a/backend/tests/fixtures/flow_step_refs_authz.sql b/backend/tests/fixtures/flow_step_refs_authz.sql new file mode 100644 index 0000000000..7e07d7de89 --- /dev/null +++ b/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()); diff --git a/backend/tests/flow_step_refs_authz.rs b/backend/tests/flow_step_refs_authz.rs new file mode 100644 index 0000000000..c9dd000cc8 --- /dev/null +++ b/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) -> 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(()) +} diff --git a/backend/windmill-api-flows/src/flows.rs b/backend/windmill-api-flows/src/flows.rs index 3c37523082..9c210336b0 100644 --- a/backend/windmill-api-flows/src/flows.rs +++ b/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?; diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 584c44bbec..d82b30d589 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/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?; diff --git a/backend/windmill-common/src/flows.rs b/backend/windmill-common/src/flows.rs index 11ffd5da76..22ddf18e04 100644 --- a/backend/windmill-common/src/flows.rs +++ b/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> { + let mut refs: Vec = Vec::new(); + let mut seen: std::collections::HashSet = 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 = 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( + 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::>(); + let script_hashes = refs + .iter() + .filter_map(|r| match r { + FlowStepRef::ScriptHash { hash, .. } => Some(*hash), + _ => None, + }) + .collect::>(); + let flow_paths = refs + .iter() + .filter_map(|r| match r { + FlowStepRef::FlowPath(path) => Some(path.clone()), + _ => None, + }) + .collect::>(); + + 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::>(); + 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 = 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::>(); + 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", + ] + ); + } } diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index b72d0ae1d9..9de9712515 100644 --- a/backend/windmill-common/src/lib.rs +++ b/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 { diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 9c15229ba6..b8e0a1e9ef 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/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::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()) diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 42246c6358..f2b8e42225 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/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 = 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 = 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>, approvers: Arc>, is_skipped: bool, + step_authed: Option<&Authed>, ) -> error::Result { 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 { 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 { 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>, +) -> Option>> { + 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, w_id: &str, db: &DB, + step_authed: Option<&Authed>, ) -> Result { - 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, apply_preprocessor: Option, + step_authed: Option<&Authed>, ) -> Result { 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,