diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index b29a21d710..172a67556a 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -77703cf32e58964906d87ee83ddda44b5b8d1b7e +5e7b1de9a07114e277a22e9e3c68b0809c06f18d diff --git a/backend/tests/untrusted_interpolation.rs b/backend/tests/untrusted_interpolation.rs index 2171797e18..1994013c11 100644 --- a/backend/tests/untrusted_interpolation.rs +++ b/backend/tests/untrusted_interpolation.rs @@ -1,7 +1,8 @@ -//! Handler, public-app and agent tool jobs run with more rights than whoever wrote some of their +//! Handler, trigger, public-app and agent tool jobs run with more rights than whoever wrote some of their //! arguments, so nothing in those arguments may be resolved by the worker with the job's identity. use axum::{routing::post, Json as AxumJson, Router}; +use futures::StreamExt; use serde_json::{json, value::RawValue, Value}; use sqlx::{types::Json, Pool, Postgres}; use std::time::Duration; @@ -300,6 +301,116 @@ async fn public_app_resolves_nothing_the_viewer_sent(db: Pool) -> anyh Ok(()) } +#[cfg(feature = "http_trigger")] +#[sqlx::test(fixtures("base"))] +#[serial_test::serial] +async fn http_route_resolves_nothing_the_caller_sent(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + seed(&db).await?; + let client = reqwest::Client::new(); + let resp = client + .post(format!( + "http://localhost:{port}/api/w/test-workspace/http_triggers/create" + )) + .bearer_auth("SECRET_TOKEN") + .json(&json!({ + "path": "u/test-user/route", + "script_path": "f/handlers/h", + "is_flow": false, + "route_path": "untrusted", + "request_type": "async", + "authentication_method": "none", + "http_method": "post", + "is_static_website": false, + "workspaced_route": false, + "wrap_body": false, + "raw_string": false, + })) + .send() + .await?; + assert!(resp.status().is_success(), "{}", resp.text().await?); + // No credentials: the job still runs as the route's owner. + let resp = client + .post(format!("http://localhost:{port}/api/r/untrusted")) + .header("content-type", "application/json") + .body( + r#"{"error": "$var:u/test-user/secret", "leak": ["$WM_TOKEN", "$var:u/test-user/secret"], + "cfg": "$res:u/test-user/cfg", "price": "$10.00"}"#, + ) + .send() + .await?; + assert!(resp.status().is_success(), "{}", resp.text().await?); + let id: Uuid = resp.text().await?.trim_matches('"').parse()?; + let out = run(&db, port, id).await; + assert!(!out.to_string().contains(SECRET), "{out}"); + assert!(!out.to_string().contains("set_by"), "{out}"); + assert_omitted(&out["error"]); + assert_omitted(&out["leak"][0]); + assert_eq!(out["price"], json!("$10.00")); + Ok(()) +} + +/// The flow resolves the approver's payload with its own token, and an approver controls what a +/// reference points to even when they can read it, so the payload may hold none. +#[sqlx::test(fixtures("base"))] +#[serial_test::serial] +async fn resume_takes_no_reference(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + seed(&db).await?; + let flow: FlowValue = serde_json::from_value(json!({"modules": [ + {"id": "a", "suspend": {"required_events": 1}, + "value": {"type": "rawscript", "language": "bun", "input_transforms": {}, + "content": "export function main() { return 1 }"}}, + {"id": "b", + "value": {"type": "rawscript", "language": "bun", + "input_transforms": {"resume": {"type": "javascript", "expr": "resume"}}, + "content": "export function main(resume: any) { return resume }"}}, + ]}))?; + let flow = RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None }) + .push(&db) + .await; + let mut completed = listen_for_completed_jobs(&db).await; + let db_ = db.clone(); + let test = async move { + let db = db_; + completed.next().await.unwrap(); + while !sqlx::query_scalar::<_, bool>("SELECT suspend > 0 FROM v2_job_queue WHERE id = $1") + .bind(flow) + .fetch_one(&db) + .await + .unwrap() + { + tokio::time::sleep(Duration::from_millis(100)).await; + } + let resume = |payload: Value| { + reqwest::Client::new() + .post(format!( + "http://localhost:{port}/api/w/test-workspace/jobs_u/flow/resume_suspended/{flow}" + )) + .bearer_auth("SECRET_TOKEN") + .json(&json!({"payload": payload})) + .send() + }; + // An admin, who can read the resource. + let refused = resume(json!({"db": "$res:u/test-user/cfg"})).await.unwrap(); + assert_eq!(refused.status(), 400, "{}", refused.text().await.unwrap()); + let ok = resume(json!({"note": "$5.00"})).await.unwrap(); + assert!(ok.status().is_success(), "{}", ok.text().await.unwrap()); + completed.find(&flow).await.unwrap(); + }; + tokio::time::timeout(Duration::from_secs(60), in_test_worker(&db, test, port)) + .await + .expect("flow did not complete"); + server.close().await?; + let out = completed_job(flow, &db).await.json_result().unwrap(); + assert_eq!(out, json!({"note": "$5.00"})); + Ok(()) +} + /// Read as the jsonb args column holds it: whatever the worker's own test finds in the stored text, /// the check finds too, and nothing more. #[sqlx::test(fixtures("base"))] diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index bf3d7ea24c..4f1d75c2f7 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -5906,6 +5906,15 @@ async fn insert_resume_job<'c>( approved: bool, tx: &mut Transaction<'c, Postgres>, ) -> error::Result<()> { + // The approver, possibly anonymous, writes this payload, and later steps receive it as + // `resume`, interpolated with the flow's own token. + if windmill_common::variables::untrusted_arg_interpolates(&value.to_string()) { + return Err(Error::BadRequest( + "The resume payload cannot hold $var:, $jsonvar:, $res: or $encrypted: references, \ + or a $WM_* variable name" + .to_string(), + )); + } sqlx::query!( r#" INSERT INTO resume_job @@ -7726,6 +7735,12 @@ pub async fn get_args_and_trigger_metadata( ) .await? }; + // A native trigger's payload is the provider's, whichever branch built the args. + let args = if run_query.service_name.is_some() { + args.omit_untrusted_interpolation() + } else { + args + }; Ok(WebhookRun::Run(args, trigger_metadata)) } diff --git a/backend/windmill-api/src/triggers/http/handler.rs b/backend/windmill-api/src/triggers/http/handler.rs index 5db6bf303d..349c1ad169 100644 --- a/backend/windmill-api/src/triggers/http/handler.rs +++ b/backend/windmill-api/src/triggers/http/handler.rs @@ -761,6 +761,7 @@ async fn route_job( runnable_format, trigger.wrap_body, ) + .map(windmill_queue::PushArgsOwned::omit_untrusted_interpolation) .map_err(|e| e.into_response())?; let trigger_info = TriggerMetadata::new(Some(trigger.path.clone()), JobTriggerKind::Http); diff --git a/backend/windmill-queue/src/asset_dispatch.rs b/backend/windmill-queue/src/asset_dispatch.rs index b44541dc9d..878ff56069 100644 --- a/backend/windmill-queue/src/asset_dispatch.rs +++ b/backend/windmill-queue/src/asset_dispatch.rs @@ -277,7 +277,6 @@ async fn try_dispatch(db: &DB, job: &MiniCompletedJob) -> Result return Ok(DispatchResult::default()); }; - let args = fetch_args(db, &job.workspace_id, job.id).await?; if read_skip_arg(args.as_ref()) { return Ok(DispatchResult::default()); @@ -768,9 +767,7 @@ async fn push_subscriber( .await? .prefetch_cached(db) .await?; - let on_behalf_of = script - .on_behalf_of(&producer.workspace_id, db) - .await?; + let on_behalf_of = script.on_behalf_of(&producer.workspace_id, db).await?; let hash = ScriptHash(script.hash); let tag = script.tag; let concurrency_settings = script.runnable_settings.concurrency_settings; @@ -866,6 +863,12 @@ async fn push_subscriber( if let Some(p) = partition { args.insert(PARTITION_ARG.to_string(), to_raw_value(&p)); } + // The partition and chain come from the producer's args, which its author controls, while + // the subscriber runs as its own deployer. + let args = windmill_common::variables::omit_untrusted_interpolation( + args, + "The following resource has been omitted for security reasons: ", + ); // Attribute the dispatched run to a synthetic user so audit logs reflect // it came from the asset cascade, not the original human runner. diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index b78da75edf..7167a56ada 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -6165,6 +6165,19 @@ impl PushArgsOwned { sanitize_untrusted_args(extra); } } + + /// For args built from a third party's payload, pushed as someone else: see + /// [`omit_untrusted_interpolation`]. Not part of [`Self::sanitize_untrusted`], because + /// some trigger pushes carry args their author wrote, whose references must resolve. + pub fn omit_untrusted_interpolation(self) -> Self { + let omit = |args| { + omit_untrusted_interpolation( + args, + "The following resource has been omitted for security reasons: ", + ) + }; + PushArgsOwned { args: omit(self.args), extra: self.extra.map(omit) } + } } pub fn empty_result() -> Box { diff --git a/backend/windmill-trigger/src/trigger_helpers.rs b/backend/windmill-trigger/src/trigger_helpers.rs index 4056ae85e6..d31ec549dd 100644 --- a/backend/windmill-trigger/src/trigger_helpers.rs +++ b/backend/windmill-trigger/src/trigger_helpers.rs @@ -489,7 +489,7 @@ pub trait TriggerJobArgs { } }; - Ok(job_args) + Ok(job_args.omit_untrusted_interpolation()) } }