diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 92f89c68fd..aae77f913d 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -3546b6b080e0234335ed2c8e2e9cc3d595bbadca +b35cb5b063667ad6a3aecdda550d6abffabc022e diff --git a/backend/tests/untrusted_interpolation.rs b/backend/tests/untrusted_interpolation.rs new file mode 100644 index 0000000000..2171797e18 --- /dev/null +++ b/backend/tests/untrusted_interpolation.rs @@ -0,0 +1,441 @@ +//! Handler, 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 serde_json::{json, value::RawValue, Value}; +use sqlx::{types::Json, Pool, Postgres}; +use std::time::Duration; +use uuid::Uuid; +use windmill_common::{flows::FlowValue, jobs::JobPayload}; +use windmill_test_utils::*; + +const SECRET: &str = "s3cr3t-value"; +const ENV_SECRET: &str = "env-s3cr3t"; +const READERS: &str = + r#"{"g/error_handler": false, "g/success_handler": false, "g/publishers": false}"#; + +async fn seed(db: &Pool) -> anyhow::Result<()> { + // Every test of this file is serial: the workspace env cache is process-wide. + sqlx::query( + "INSERT INTO workspace_env (workspace_id, name, value) VALUES ('test-workspace', 'API_KEY', $1)", + ) + .bind(ENV_SECRET) + .execute(db) + .await?; + windmill_common::variables::CUSTOM_ENVS_CACHE.remove("test-workspace"); + // How an app run as its publisher reads a secret: through the publisher's groups. + sqlx::query( + "INSERT INTO group_ (workspace_id, name, summary, extra_perms) + VALUES ('test-workspace', 'publishers', '', '{}')", + ) + .execute(db) + .await?; + sqlx::query( + "INSERT INTO usr_to_group (workspace_id, group_, usr) + VALUES ('test-workspace', 'publishers', 'test-user')", + ) + .execute(db) + .await?; + sqlx::query( + "INSERT INTO variable (workspace_id, path, value, is_secret, description, extra_perms) + VALUES ('test-workspace', 'u/test-user/secret', $1, false, '', $2::jsonb)", + ) + .bind(format!("\"{SECRET}\"")) + .bind(READERS) + .execute(db) + .await?; + sqlx::query( + "INSERT INTO resource_type (workspace_id, name, schema, description, created_by) + VALUES ('test-workspace', 'c_cfg', '{}', '', 'test-user')", + ) + .execute(db) + .await?; + sqlx::query( + "INSERT INTO resource (workspace_id, path, value, resource_type, extra_perms, created_by) + VALUES ('test-workspace', 'u/test-user/cfg', '{\"set_by\": \"admin\"}', 'c_cfg', $1::jsonb, + 'test-user')", + ) + .bind(READERS) + .execute(db) + .await?; + // The lock must be non-null for the script to count as deployed. + sqlx::query( + r#"INSERT INTO script (workspace_id, hash, path, content, language, kind, created_by, schema, + summary, description, lock) + VALUES ('test-workspace', 1234, 'f/handlers/h', $1, 'bun', 'script', 'test-user', '{}', + '', '', E'{}\n//bun.lock\n')"#, + ) + .bind( + "export async function main(error: any, cfg: any, success_result: any, pipeline: any, + price: any, note: any, leak: any, log: any) { + return { error, cfg, success_result, pipeline, price, note, leak, log }; + }", + ) + .execute(db) + .await?; + Ok(()) +} + +async fn run(db: &Pool, port: u16, id: Uuid) -> Value { + let done = listen_for_completed_jobs(db).await.find(&id); + tokio::time::timeout(Duration::from_secs(60), in_test_worker(db, done, port)) + .await + .expect("job did not complete"); + let job = completed_job(id, db).await; + assert!(job.success, "{:?}", job.result); + job.json_result().unwrap() +} + +fn assert_omitted(v: &Value) { + let s = v.as_str().unwrap_or_default(); + assert!( + s.starts_with("The following variable has been omitted for security reasons"), + "{v}" + ); +} + +/// What a failed job sends its handler: references in its error, and the `$WM_TOKEN` the worker +/// replaces with the handler's token in an argument that holds one. +fn failed_result() -> Box { + // Raw text: a number survives only if nothing on the way parses it. + RawValue::from_string( + r#"{"error": {"message": "$var:u/test-user/secret", "name": "$WM_TOKEN", + "stack": "$jsonvar:u/test-user/secret"}, + "meta": {"code": 1.10, "id": 18446744073709551617}, "log": "a\u0000b"}"# + .to_string(), + ) + .unwrap() +} + +fn extra_args() -> Json> { + Json(serde_json::value::to_raw_value(&json!({"cfg": "$res:u/test-user/cfg"})).unwrap()) +} + +/// Stored as jsonb stores it: numbers exact, the NUL jsonb refuses stripped. +async fn assert_handled(db: &Pool, port: u16, id: Uuid) -> Value { + let stored: String = sqlx::query_scalar("SELECT args::text FROM v2_job WHERE id = $1") + .bind(id) + .fetch_one(db) + .await + .unwrap(); + assert!( + stored.contains("1.10") && stored.contains("18446744073709551617"), + "{stored}" + ); + let out = run(db, port, id).await; + assert_eq!(out["log"], json!("ab")); + assert!(!out.to_string().contains(SECRET), "{out}"); + for k in ["message", "name", "stack"] { + assert_omitted(&out["error"][k]); + } + // The handler's own settings, written by an admin. + assert_eq!(out["cfg"], json!({"set_by": "admin"})); + out +} + +#[sqlx::test(fixtures("base"))] +#[serial_test::serial] +async fn error_handler_resolves_nothing_the_failed_job_wrote( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + seed(&db).await?; + let id = windmill_queue::push_error_handler( + &db, + Uuid::new_v4(), + None, + Some("u/test-user/failing".to_string()), + false, + "test-workspace", + "script/f/handlers/h", + Json(&failed_result()), + None, + None, + Some(extra_args()), + "test@windmill.dev", + None, + false, + None, + ) + .await?; + assert_handled(&db, server.addr.port(), id).await; + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +#[serial_test::serial] +async fn success_handler_resolves_nothing_the_job_returned( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + seed(&db).await?; + let result = json!({ + "pipeline": [{"$match": {"currency": "$USD", "$or": [{"a": "$category"}]}}], + "price": "$10.00", + "note": "cost $var:u/test-user/secret", + "leak": "\u{24}jsonvar:u/test-user/secret", + }); + let id = windmill_queue::push_success_handler( + &db, + Uuid::new_v4(), + None, + Some("u/test-user/succeeding".to_string()), + false, + "test-workspace", + "script/f/handlers/h", + Json(&result), + None, + Some(extra_args()), + "test@windmill.dev", + None, + ) + .await?; + let out = run(&db, server.addr.port(), id).await; + assert_omitted(&out["leak"]); + // Holds the prefix that makes the worker interpolate the whole argument. + assert!(out["note"] + .as_str() + .unwrap() + .starts_with("cost The following variable")); + for k in ["pipeline", "price"] { + assert_eq!(out[k], result[k], "{k}"); + } + assert_eq!(out["cfg"], json!({"set_by": "admin"})); + Ok(()) +} + +#[cfg(all(feature = "enterprise", feature = "private"))] +#[sqlx::test(fixtures("base"))] +#[serial_test::serial] +async fn recovery_handler_resolves_nothing_the_schedule_jobs_wrote( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + seed(&db).await?; + let job_id = Uuid::new_v4(); + let failed = failed_result(); + let success = json!({"leak": "$var:u/test-user/secret", "ok": {"$match": "$USD"}}); + windmill_queue::jobs_ee::handle_recovered_schedule( + &db, + job_id, + "u/test-user/schedule", + "u/test-user/flaky", + false, + "test-workspace", + "script/f/handlers/h", + Some(&failed), + sqlx::types::chrono::Utc::now(), + Json(&success), + 1, + sqlx::types::chrono::Utc::now(), + Some(extra_args()), + "u/test-user", + ) + .await?; + let id: Uuid = sqlx::query_scalar("SELECT id FROM v2_job WHERE parent_job = $1") + .bind(job_id) + .fetch_one(&db) + .await?; + let out = assert_handled(&db, server.addr.port(), id).await; + assert_omitted(&out["success_result"]["leak"]); + assert_eq!(out["success_result"]["ok"], success["ok"]); + Ok(()) +} + +#[sqlx::test(fixtures("base"))] +#[serial_test::serial] +async fn public_app_resolves_nothing_the_viewer_sent(db: Pool) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + seed(&db).await?; + let content = "export function main(a: any, b: any, c: any, d: any) { return { a, b, c, d } }"; + let ws = format!("http://localhost:{port}/api/w/test-workspace"); + let client = reqwest::Client::new(); + let resp = client + .post(format!("{ws}/apps/create")) + .bearer_auth("SECRET_TOKEN") + .json(&json!({ + "path": "u/test-user/app", + "summary": "", + "value": {"runnables": {"c": {"type": "inline", + "inlineScript": {"language": "bun", "content": content}}}}, + "policy": {"execution_mode": "anonymous", "triggerables_v2": { + format!("c:rawscript/{}", windmill_common::utils::calculate_hash(content)): + {"static_inputs": {}, "one_of_inputs": {}} + }}, + })) + .send() + .await?; + assert!(resp.status().is_success(), "{}", resp.text().await?); + // `\u0024` passes a text match on `$var:` until jsonb decodes it. + let body = r#"{"component": "c", + "raw_code": {"language": "bun", "content": "CONTENT"}, + "args": {"a": "\u0024var:u/test-user/secret", "b": {"x": ["$jsonvar:u/test-user/secret"]}, + "c": {"pipeline": [{"$match": {"currency": "$USD"}}]}, + "d": {"key": "$API_KEY", "note": "cost $var:x"}}}"# + .replace("CONTENT", content); + let resp = client + .post(format!("{ws}/apps_u/execute_component/u/test-user/app")) + .bearer_auth("SECRET_TOKEN_2") + .header("content-type", "application/json") + .body(body) + .send() + .await?; + assert!(resp.status().is_success(), "{}", resp.text().await?); + let id: Uuid = resp.text().await?.parse()?; + let out = run(&db, port, id).await; + assert!(!out.to_string().contains(SECRET), "{out}"); + assert!(!out.to_string().contains(ENV_SECRET), "{out}"); + assert_omitted(&out["a"]); + assert_omitted(&out["b"]["x"][0]); + assert_eq!( + out["c"], + json!({"pipeline": [{"$match": {"currency": "$USD"}}]}) + ); + assert_eq!(out["d"]["key"], json!("$API_KEY")); + 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"))] +#[serial_test::serial] +async fn reads_an_argument_as_jsonb_stores_it(db: Pool) -> anyhow::Result<()> { + use windmill_common::{utils::strip_json_nul, variables::*}; + for arg in [ + r#""\u0024var:x""#, + r#""$va\u0072:x""#, + r#""$va\\u0072:x""#, + r#""$va\u0000r:x""#, + r#""\ud800 $var:x""#, + r#"{"a": "\u0024res:x", "a": "ok"}"#, + r#"{"a": "ok", "a": "\u0024var:x"}"#, + r#"{"n": 1e400, "n": 0, "r": "\u0024var:x"}"#, + r#"{"\u0024encrypted:k": 1}"#, + r#""cost \u0024var:x""#, + r#"{"$match": {"c": "$USD"}, "p": 1.10}"#, + r#""$10.00""#, + ] { + let stored: Option = match sqlx::query_scalar("SELECT $1::jsonb::text") + .bind(strip_json_nul(arg).as_ref()) + .fetch_one(&db) + .await + { + Ok(s) => Some(s), + Err(sqlx::Error::Database(e)) if e.code().as_deref() == Some("22P02") => None, + Err(e) => return Err(e.into()), + }; + assert_eq!( + untrusted_arg_interpolates(arg), + stored.as_deref().map_or(true, arg_needs_interpolation), + "{arg} stored as {stored:?}" + ); + } + Ok(()) +} + +/// One tool call with `arguments`, then a final answer once the tool's reply is in. +fn model(arguments: &'static str) -> Router { + Router::new().route( + "/v1/chat/completions", + post(move |AxumJson(body): AxumJson| async move { + let answered = body["messages"] + .as_array() + .unwrap() + .iter() + .any(|m| m["role"] == "tool"); + let delta = if answered { + json!({"role": "assistant", "content": "done"}) + } else { + json!({"role": "assistant", "tool_calls": [{"index": 0, "id": "call_0", + "type": "function", "function": {"name": "echo", "arguments": arguments}}]}) + }; + let event = json!({"choices": [{"index": 0, "delta": delta, "finish_reason": null}]}); + let end = json!({"choices": [{"index": 0, "delta": {}, + "finish_reason": if answered {"stop"} else {"tool_calls"}}]}); + ( + [("content-type", "text/event-stream")], + format!("data: {event}\n\ndata: {end}\n\ndata: [DONE]\n\n"), + ) + }), + ) +} + +/// What the tool replied to the model. +async fn tool_reply(db: &Pool, port: u16, arguments: &'static str) -> String { + std::env::set_var("ALLOW_PRIVATE_AI_BASE_URLS", "true"); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base = format!("http://{}", listener.local_addr().unwrap()); + let stub = tokio::spawn(async move { axum::serve(listener, model(arguments)).await }); + let flow: FlowValue = serde_json::from_value(json!({"modules": [{"id": "agent", "value": { + "type": "aiagent", + "tools": [{"id": "echo", "summary": "echo", "value": { + "type": "rawscript", "language": "bun", "input_transforms": {}, + "content": "export async function main(x: any) { return x }"}}], + "input_transforms": { + "provider": {"type": "static", "value": {"kind": "customai", "model": "m", + "resource": {"base_url": format!("{base}/v1")}}}, + "user_message": {"type": "static", "value": "go"}, + "max_iterations": {"type": "static", "value": 2} + } + }}]})) + .unwrap(); + let id = RunJob::from(JobPayload::RawFlow { + value: flow, + path: Some("u/test-user/agent".into()), + restarted_from: None, + }) + .push(db) + .await; + let out = run(db, port, id).await; + stub.abort(); + out["messages"] + .as_array() + .unwrap() + .iter() + .find(|m| m["role"] == "tool") + .unwrap()["content"] + .as_str() + .unwrap() + .to_string() +} + +#[sqlx::test(fixtures("base"))] +#[serial_test::serial] +async fn agent_tool_call_resolves_nothing_the_model_wrote( + db: Pool, +) -> anyhow::Result<()> { + initialize_tracing().await; + let server = ApiServer::start(db.clone()).await?; + let port = server.addr.port(); + seed(&db).await?; + // jsonb keeps the last copy of a duplicate key, the one a Rust decode refuses to read past. + let reply = tool_reply( + &db, + port, + r#"{"x": {"n": 1e400, "n": 0, "r": "\u0024var:u/test-user/secret"}}"#, + ) + .await; + assert!(!reply.contains(SECRET), "{reply}"); + assert!(reply.contains("Tool call refused"), "{reply}"); + + // A tool the model calls without arguments, as Anthropic sends it. + let reply = tool_reply(&db, port, "").await; + assert!(!reply.contains("Tool call refused"), "{reply}"); + + let reply = tool_reply( + &db, + port, + r#"{"x": {"pipeline": [{"$match": {"currency": "$USD", "_id": "$category"}}]}}"#, + ) + .await; + assert_eq!( + serde_json::from_str::(&reply)?, + json!({"pipeline": [{"$match": {"currency": "$USD", "_id": "$category"}}]}) + ); + Ok(()) +} diff --git a/backend/windmill-api-integration-tests/tests/agent_on_behalf_of.rs b/backend/windmill-api-integration-tests/tests/agent_on_behalf_of.rs index 628614056c..540282de9a 100644 --- a/backend/windmill-api-integration-tests/tests/agent_on_behalf_of.rs +++ b/backend/windmill-api-integration-tests/tests/agent_on_behalf_of.rs @@ -153,6 +153,10 @@ async fn test_agent_run_identity_and_needs_read(db: Pool) -> anyhow::R for raw in [ r#"{"user_message":"$var:u/test-user-3/secret"}"#, r#"{"user_message":"\u0024var:u/test-user-3/secret"}"#, + // Resolved once a step combines it with an argument holding a reference. + r#"{"user_message":"$WM_TOKEN"}"#, + // jsonb drops the undecodable duplicate and keeps the reference. + r#"{"user_message":{"n":1e400,"n":0,"r":"\u0024var:u/test-user-3/secret"}}"#, ] { let resp = reqwest::Client::new() .post(format!("{base_url}/w/test-workspace/{run}")) @@ -163,6 +167,14 @@ async fn test_agent_run_identity_and_needs_read(db: Pool) -> anyhow::R .await?; assert_eq!(resp.status().as_u16(), 400, "a reference in {raw}"); } + let (status, body) = post( + &base_url, + "SECRET_TOKEN_2", + &run, + json!({ "user_message": {"$match": {"currency": "$USD", "_id": "$category"}} }), + ) + .await?; + assert_eq!(status, 201, "nothing the worker resolves: {body}"); // The run takes its conversation outside the caller's row policy, so a reader must not // continue another user's conversation; an admin still may, as the policy allows. diff --git a/backend/windmill-api/src/agent_runs.rs b/backend/windmill-api/src/agent_runs.rs index bec7180f19..f1f85985ea 100644 --- a/backend/windmill-api/src/agent_runs.rs +++ b/backend/windmill-api/src/agent_runs.rs @@ -21,7 +21,7 @@ use windmill_common::{ jobs::JobPayload, users::username_to_permissioned_as, utils::{not_found_if_none, StripPath}, - variables::arg_needs_interpolation, + variables::untrusted_arg_interpolates, }; use windmill_queue::{push, PushArgs, PushIsolationLevel}; @@ -136,11 +136,8 @@ pub(crate) async fn run_agent_as( }; if let Some(obo) = on_behalf_of.as_ref() { // The worker would interpolate such an input with the job's token, which is the - // identity's. Checked on the decoded value: the raw JSON can spell `$` as `\u0024`. - if args.values().any(|v| { - serde_json::from_str::(v.get()) - .is_ok_and(|v| arg_needs_interpolation(&v.to_string())) - }) { + // identity's. + if args.values().any(|v| untrusted_arg_interpolates(v.get())) { return Err(Error::BadRequest(format!( "Agent {path} runs on behalf of {}, so its inputs cannot reference variables or resources", obo.permissioned_as diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 944535dd84..d28005c2a0 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -76,7 +76,7 @@ use windmill_common::{ check_proper_path, http_get_from_hub, not_found_if_none, paginate, paginate_optional, query_elems_from_hub, require_admin, strip_json_nul, Pagination, RunnableKind, StripPath, }, - variables::{build_crypt, build_crypt_with_key_suffix, encrypt}, + variables::{build_crypt, build_crypt_with_key_suffix, encrypt, omit_untrusted_interpolation}, worker::{to_raw_value, CLOUD_HOSTED}, workspaces::{ check_deploy_rules, check_operator_can_build, check_user_against_rule, @@ -6553,6 +6553,7 @@ async fn build_args( } } + let mut untrusted = HashMap::new(); for (k, v) in args { if safe_args.contains_key(&k) { continue; @@ -6566,31 +6567,16 @@ async fn build_args( k.to_string(), ctx_value(prop, policy, authed, w_id, db).await?, ); - } else if !arg_str.contains("\"$var:") && !arg_str.contains("\"$res:") { - safe_args.insert(k.to_string(), v); } else { - safe_args.insert( - k.to_string(), - RawValue::from_string( - arg_str - .replace( - "$var:", - "The following variable has been omitted for security reasons: ", - ) - .replace( - "$res:", - "The following resource has been omitted for security reasons, to allow it, toggle: 'Allow resources from users' on that field input: ", - ), - ) - .map_err(|e| { - Error::internal_err(format!( - "failed to remove sensitive variable(s)/resource(s) with error: {}", - e - )) - })?, - ); + untrusted.insert(k, v); } } + safe_args.extend( + omit_untrusted_interpolation( + untrusted, + "The following resource has been omitted for security reasons, to allow it, toggle: 'Allow resources from users' on that field input: ", + ), + ); for k in sensitive_inputs { let Some(v) = safe_args.get(k) else { continue }; let raw = v.get(); diff --git a/backend/windmill-common/src/variables.rs b/backend/windmill-common/src/variables.rs index 65eb912ced..82039d71f5 100644 --- a/backend/windmill-common/src/variables.rs +++ b/backend/windmill-common/src/variables.rs @@ -10,21 +10,23 @@ use crate::db::{Authable, UserDB}; use crate::error::{self, Error}; use crate::scripts::ScriptHash; use crate::secret_backend::{get_secret_value, is_external_stored_value}; -use crate::utils::WarnAfterExt; -use crate::worker::Connection; +use crate::utils::{strip_json_nul, WarnAfterExt}; +use crate::worker::{to_raw_value, Connection}; use crate::workspace_crypt::CipherFormat; pub use crate::workspace_crypt::WorkspaceCrypt; use crate::{worker::WORKER_GROUP, BASE_URL, DB}; use chrono::{SecondsFormat, Utc}; use quick_cache::sync::Cache; use serde::{Deserialize, Serialize}; +use serde_json::value::RawValue; +use std::collections::HashMap; use std::sync::atomic::{AtomicBool, Ordering}; lazy_static::lazy_static! { pub static ref SECRET_SALT: Option = std::env::var("SECRET_SALT").ok(); static ref RESERVED_WM_VAR_NAME: regex::Regex = regex::Regex::new(r"^WM_[A-Z_]+$").unwrap(); static ref RE_RES_VAR: regex::Regex = - regex::Regex::new(r#"\$(?:var|jsonvar|res|encrypted)\:"#).unwrap(); + regex::Regex::new(r#"\$(var|jsonvar|res|encrypted)\:"#).unwrap(); } /// Whether a job argument, as JSON text, holds a `$var:`/`$jsonvar:`/`$res:`/`$encrypted:` @@ -34,6 +36,106 @@ pub fn arg_needs_interpolation(json: &str) -> bool { RE_RES_VAR.is_match(json) } +/// Whether the worker would interpolate `arg`, which an untrusted user controls on a job that +/// runs as someone else; `true` for one that cannot be read and may hold a reference. +pub fn untrusted_arg_interpolates(arg: &str) -> bool { + may_reference(arg) && decoded(arg).map_or(true, |v| interpolates(&v)) +} + +/// `args`, which an untrusted user controls on a job that runs as someone else, with every one the +/// worker would interpolate rewritten so it no longer is. The others keep their exact text. +pub fn omit_untrusted_interpolation( + args: HashMap>, + resource_omitted: &str, +) -> HashMap> { + args.into_iter() + .map(|(k, v)| { + if !may_reference(v.get()) { + return (k, v); + } + let v = match decoded(v.get()) { + Some(d) if !interpolates(&d) => v, + Some(mut d) => { + omit(&mut d, resource_omitted); + to_raw_value(&d) + } + None => to_raw_value( + &"This argument has been omitted for security reasons: it could not be checked", + ), + }; + (k, v) + }) + .collect() +} + +/// Without a `$`, spelled out or escaped, an argument holds no reference however jsonb rewrites +/// it, so it needs no decoding, even one serde_json cannot read. +fn may_reference(arg: &str) -> bool { + arg.contains('$') || arg.contains("\\u") +} + +/// `arg` as its jsonb column will hold it: the NUL stripped as `push` strips it, escapes decoded, +/// the last copy of a duplicate key kept. `None` for what serde_json cannot read, which jsonb may +/// still store (`1e400`). +fn decoded(arg: &str) -> Option { + serde_json::from_str(&strip_json_nul(arg)).ok() +} + +/// The worker's own test on the whole argument, plus a reserved `WM_*` key or value at any depth, +/// which a flow step can move into an argument the worker interpolates. A workspace or worker env +/// var name moved that way is still resolved: those names are not known on every path. +fn interpolates(v: &serde_json::Value) -> bool { + fn reserved_name(v: &serde_json::Value) -> bool { + match v { + serde_json::Value::String(s) => reserved(s).is_some(), + serde_json::Value::Array(a) => a.iter().any(reserved_name), + serde_json::Value::Object(m) => m + .iter() + .any(|(k, v)| reserved(k).is_some() || reserved_name(v)), + _ => false, + } + } + arg_needs_interpolation(&v.to_string()) || reserved_name(v) +} + +/// Keys included, since the worker's test reads them too. +fn omit(v: &mut serde_json::Value, resource_omitted: &str) { + match v { + serde_json::Value::String(s) => *s = omitted(s, resource_omitted), + serde_json::Value::Array(a) => a.iter_mut().for_each(|v| omit(v, resource_omitted)), + serde_json::Value::Object(m) => { + *m = std::mem::take(m) + .into_iter() + .map(|(k, mut v)| { + omit(&mut v, resource_omitted); + (omitted(&k, resource_omitted), v) + }) + .collect() + } + _ => {} + } +} + +/// The reserved variable a whole `$WM_*` value names, which the worker replaces with its value. +fn reserved(s: &str) -> Option<&str> { + s.strip_prefix('$') + .filter(|n| RESERVED_WM_VAR_NAME.is_match(n)) +} + +fn omitted(s: &str, resource_omitted: &str) -> String { + let variable = "The following variable has been omitted for security reasons: "; + if let Some(n) = reserved(s) { + return format!("{variable}{n}"); + } + RE_RES_VAR + .replace_all(s, |c: ®ex::Captures| match &c[1] { + "res" => resource_omitted, + "encrypted" => "The following encrypted value has been omitted for security reasons: ", + _ => variable, + }) + .into_owned() +} + /// Escape a string so it can be safely embedded inside a single-quoted JS /// string literal in a generated NativeTS/Bun prologue. pub fn escape_js_single_quoted(s: &str) -> String { @@ -986,9 +1088,48 @@ pub async fn get_variable_or_self_as( #[cfg(test)] mod tests { use super::{ - can_bind_as_prologue_const, decrypt, escape_js_single_quoted, is_valid_js_identifier, - DECRYPT_FAILED, + arg_needs_interpolation, can_bind_as_prologue_const, decrypt, escape_js_single_quoted, + is_valid_js_identifier, omit_untrusted_interpolation, DECRYPT_FAILED, }; + + #[test] + fn omits_every_argument_the_worker_would_interpolate() { + let omit = |s: &str| { + let arg = serde_json::value::RawValue::from_string(s.to_string()).unwrap(); + omit_untrusted_interpolation( + [("a".to_string(), arg)].into(), + "The following resource has been omitted for security reasons: ", + )["a"] + .get() + .to_string() + }; + for kept in [ + r#"{"$match": {"c": "$USD", "_id": "$category"}, "p": 1.10, "id": 18446744073709551617}"#, + r#""$10.00""#, + r#""$WM_TOKEN ""#, + ] { + assert_eq!(omit(kept), kept); + } + for hit in [ + r#""$var:u/a/s""#, + r#""\u0024res:u/a/r""#, + r#"{"key": "$API_KEY", "note": "cost $var:x"}"#, + r#"{"$encrypted:abc": 1}"#, + r#"["$WM_TOKEN"]"#, + r#"{"$WM_TOKEN": 1}"#, + r#""$va\u0000r:x""#, + r#"{"a": "ok", "a": "\u0024jsonvar:x"}"#, + ] { + let out = omit(hit); + assert!( + !arg_needs_interpolation(&out) && !out.contains("$WM_"), + "{hit}: {out}" + ); + assert!(out.contains("omitted for security reasons"), "{hit}: {out}"); + } + assert!(omit(r#"{"n": 1e400, "r": "$var:x"}"#).contains("could not be checked")); + assert_eq!(omit(r#"{"n": 1e400}"#), r#"{"n": 1e400}"#); + } use crate::error::Error; use crate::workspace_crypt::WorkspaceCrypt; use base64::Engine; diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index aea82eb6fd..3ec21f3c84 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -54,6 +54,7 @@ use windmill_common::runnable_settings::{ }; use windmill_common::triggers::TriggerMetadata; use windmill_common::utils::{calculate_hash, configure_client, now_from_db, strip_json_nul}; +use windmill_common::variables::omit_untrusted_interpolation; use windmill_common::worker::Connection; use windmill_common::otel_oss::{ @@ -3688,10 +3689,16 @@ pub async fn push_success_handler<'a, 'c, T: Serialize + Send + Sync>( return Ok(uuid); } +/// A handler's args from the job's result, which the job's author controls while the handler +/// runs as its own identity. fn sanitize_result(result: Json<&T>) -> HashMap> { let as_str = serde_json::to_string(result.0).unwrap_or_else(|_| "{}".to_string()); - serde_json::from_str::>>(&as_str) - .unwrap_or_else(|_| [("error".to_string(), RawValue::from_string(as_str).unwrap())].into()) + let args = serde_json::from_str::>>(&as_str) + .unwrap_or_else(|_| [("error".to_string(), RawValue::from_string(as_str).unwrap())].into()); + omit_untrusted_interpolation( + args, + "The following resource has been omitted for security reasons: ", + ) } // #[derive(Serialize)] diff --git a/backend/windmill-worker/src/ai/tools.rs b/backend/windmill-worker/src/ai/tools.rs index 0bc7bf6777..01b90922ff 100644 --- a/backend/windmill-worker/src/ai/tools.rs +++ b/backend/windmill-worker/src/ai/tools.rs @@ -19,7 +19,7 @@ use std::{collections::HashMap, sync::Arc, time::Duration}; use uuid::Uuid; use windmill_ai::{ai_types::OpenAIToolCall, query_builder::StreamEventSink, types::*}; use windmill_common::jobs::JobPayload; -use windmill_common::variables::arg_needs_interpolation; +use windmill_common::variables::untrusted_arg_interpolates; #[cfg(feature = "mcp")] use windmill_mcp::McpClient; @@ -199,8 +199,9 @@ pub async fn execute_tool_calls( /// which may be an identity it runs on behalf of. Only the step's own input transforms may name /// variables and resources; the model's arguments are text from whoever talks to it. fn interpolating_args(tool_call: &OpenAIToolCall) -> bool { - serde_json::from_str::(&tool_call.function.arguments) - .is_ok_and(|v| arg_needs_interpolation(&v.to_string())) + // Empty for a tool called without arguments, which runs as `{}`. + let args = &tool_call.function.arguments; + !args.is_empty() && untrusted_arg_interpolates(args) } async fn refuse_tool_call( @@ -209,8 +210,8 @@ async fn refuse_tool_call( messages: &mut Vec, final_events_str: &mut String, ) -> Result<(), Error> { - let error_msg = "Tool call refused: its arguments cannot reference variables or resources \ - ($var:, $jsonvar:, $res:, $encrypted:)" + let error_msg = "Tool call refused: its arguments could not be read, or reference variables, \ + resources or reserved names ($var:, $jsonvar:, $res:, $encrypted:, $WM_)" .to_string(); messages.push(OpenAIMessage { role: "tool".to_string(),