mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-03 08:02:19 +00:00
fix: drop caller-supplied _MODULES at push for every job (#11418)
* fix: drop caller-supplied _MODULES at push for every job Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> * fix: never interpolate _MODULES into a tag and drop unused mut Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
bb12bc766c
commit
c5f59d6ccc
@@ -0,0 +1,110 @@
|
||||
//! A caller's `_MODULES` never reaches a job: the worker builds that arg in as module code,
|
||||
//! so on a deployed runnable it would run caller code as the runnable (and its
|
||||
//! `on_behalf_of`). `push` drops it from `args` and `extra` alike and only sets it from a
|
||||
//! preview's own `RawCode::modules`.
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
use serde_json::{json, value::RawValue};
|
||||
use sqlx::{Pool, Postgres};
|
||||
use windmill_common::{
|
||||
jobs::{JobPayload, RawCode},
|
||||
runnable_settings::{ConcurrencySettings, DebouncingSettings},
|
||||
scripts::{ScriptHash, ScriptLang, ScriptModule},
|
||||
};
|
||||
use windmill_queue::{PushArgs, PushIsolationLevel};
|
||||
|
||||
fn modules(v: &str) -> serde_json::Value {
|
||||
json!({ "helper.ts": { "content": format!("export const v = \"{v}\""), "language": "bun" } })
|
||||
}
|
||||
|
||||
/// Pushes `payload` with a caller `_MODULES` both in `args` and in `extra` (where webhook
|
||||
/// query and headers land), and returns the stored `_MODULES`.
|
||||
async fn stored_modules(db: &Pool<Postgres>, payload: JobPayload) -> Option<serde_json::Value> {
|
||||
let caller: Box<RawValue> = serde_json::value::to_raw_value(&modules("caller")).unwrap();
|
||||
let args = HashMap::from([("_MODULES".to_string(), caller.clone())]);
|
||||
let extra = HashMap::from([("_MODULES".to_string(), caller)]);
|
||||
let (id, tx) = windmill_queue::push(
|
||||
db,
|
||||
PushIsolationLevel::IsolatedRoot(db.clone()),
|
||||
"test-workspace",
|
||||
payload,
|
||||
PushArgs { args: &args, extra: Some(extra) },
|
||||
"test-user",
|
||||
"test@windmill.dev",
|
||||
"u/test-user".to_string(),
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
false,
|
||||
false,
|
||||
None,
|
||||
true,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
false,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("push must succeed");
|
||||
tx.commit().await.unwrap();
|
||||
sqlx::query_scalar::<_, Option<serde_json::Value>>(
|
||||
"SELECT args->'_MODULES' FROM v2_job WHERE id = $1",
|
||||
)
|
||||
.bind(id)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
#[sqlx::test(fixtures("base"))]
|
||||
async fn caller_modules_never_reach_a_job(db: Pool<Postgres>) {
|
||||
let deployed = JobPayload::ScriptHash {
|
||||
hash: ScriptHash(123412),
|
||||
path: "f/system/hello".to_string(),
|
||||
cache_ttl: None,
|
||||
cache_ignore_s3_path: None,
|
||||
dedicated_worker: None,
|
||||
language: ScriptLang::Bun,
|
||||
priority: None,
|
||||
apply_preprocessor: false,
|
||||
concurrency_settings: ConcurrencySettings::default(),
|
||||
debouncing_settings: DebouncingSettings::default(),
|
||||
labels: None,
|
||||
};
|
||||
assert_eq!(stored_modules(&db, deployed).await, None);
|
||||
|
||||
let preview = |modules: Option<HashMap<String, ScriptModule>>| {
|
||||
JobPayload::Code(RawCode {
|
||||
hash: None,
|
||||
content: "import { v } from \"./helper\"; export function main() { return v }"
|
||||
.to_string(),
|
||||
path: None,
|
||||
language: ScriptLang::Bun,
|
||||
lock: None,
|
||||
concurrency_settings: ConcurrencySettings::default().into(),
|
||||
debouncing_settings: DebouncingSettings::default(),
|
||||
cache_ttl: None,
|
||||
cache_ignore_s3_path: None,
|
||||
dedicated_worker: None,
|
||||
modules,
|
||||
tag: None,
|
||||
})
|
||||
};
|
||||
assert_eq!(stored_modules(&db, preview(None)).await, None);
|
||||
let own = serde_json::from_value(modules("preview")).unwrap();
|
||||
assert_eq!(
|
||||
stored_modules(&db, preview(Some(own))).await,
|
||||
Some(modules("preview"))
|
||||
);
|
||||
}
|
||||
@@ -4362,12 +4362,10 @@ async fn execute_component(
|
||||
let resolved_delete_secs =
|
||||
resolve_delete_after_secs(None, policy_triggerables.delete_after_secs);
|
||||
|
||||
// `_MODULES` and `_TEMP_SCRIPT_REFS` are server-injected control keys (into
|
||||
// `extra`) that the worker reads back for a `Preview` job — which an inline run
|
||||
// is. A caller supplying them in `args` would inject module content/locks or
|
||||
// redirect relative-import resolution, unpinned, as the app identity. Drop them;
|
||||
// legitimate values ride in `extra`, never the request `args`.
|
||||
payload.args.remove("_MODULES");
|
||||
// `_TEMP_SCRIPT_REFS` is a server-injected control key (into `extra`) that the
|
||||
// worker reads back for a `Preview` job, which an inline run is. A caller
|
||||
// supplying it in `args` would redirect relative-import resolution, unpinned, as
|
||||
// the app identity. Drop it; the legitimate value rides in `extra`.
|
||||
payload.args.remove("_TEMP_SCRIPT_REFS");
|
||||
|
||||
let (mut args, job_id) = build_args(
|
||||
|
||||
@@ -7421,6 +7421,20 @@ pub async fn run_workflow_as_code(
|
||||
)
|
||||
.await?;
|
||||
|
||||
// A task re-runs a preview with the preview's modules: the ones `push` stored for it, never
|
||||
// the task's args. Read on their own, as `fetch_queued` swaps oversized args for a placeholder.
|
||||
let preview_modules = if job.job_kind == JobKind::Preview {
|
||||
sqlx::query_scalar::<_, Option<sqlx::types::Json<HashMap<String, ScriptModule>>>>(
|
||||
"SELECT args->'_MODULES' FROM v2_job WHERE id = $1",
|
||||
)
|
||||
.bind(job.id)
|
||||
.fetch_one(&db)
|
||||
.await?
|
||||
.map(|modules| modules.0)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
let (job_payload, tag, _delete_after_use, _delete_after_secs, timeout, on_behalf_of) =
|
||||
match job.job_kind {
|
||||
JobKind::Preview => (
|
||||
@@ -7444,7 +7458,7 @@ pub async fn run_workflow_as_code(
|
||||
dedicated_worker: None,
|
||||
// TODO(debouncing): enable for this mode
|
||||
debouncing_settings: DebouncingSettings::default(),
|
||||
modules: None,
|
||||
modules: preview_modules,
|
||||
tag: None,
|
||||
}),
|
||||
Some(job.tag.clone()),
|
||||
@@ -8630,9 +8644,6 @@ async fn run_preview_script(
|
||||
if let Some(fp) = &preview.flow_path {
|
||||
extra.insert("_FLOW_PATH".to_string(), to_raw_value(fp));
|
||||
}
|
||||
if let Some(ref modules) = preview.modules {
|
||||
extra.insert("_MODULES".to_string(), to_raw_value(modules));
|
||||
}
|
||||
if let Some(ref temp_script_refs) = preview.temp_script_refs {
|
||||
extra.insert(
|
||||
"_TEMP_SCRIPT_REFS".to_string(),
|
||||
|
||||
@@ -43,6 +43,7 @@ use windmill_common::auth::JobPerms;
|
||||
use windmill_common::bench::BenchmarkIter;
|
||||
use windmill_common::jobs::{
|
||||
script_path_to_payload, JobTriggerKind, TriggerKindLabel, EMAIL_ERROR_HANDLER_USER_EMAIL,
|
||||
MODULES_ARG,
|
||||
};
|
||||
use windmill_common::min_version::{
|
||||
MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2,
|
||||
@@ -5191,10 +5192,15 @@ pub fn interpolate_args(x: String, args: &PushArgs, workspace_id: &str) -> Strin
|
||||
for cap in RE_ARG_TAG.captures_iter(&workspaced) {
|
||||
let arg_name = cap.get(1).unwrap().as_str();
|
||||
let (root, rest) = arg_name.split_once('.').unwrap_or((arg_name, ""));
|
||||
let root_value = args
|
||||
.args
|
||||
.get(root)
|
||||
.or(args.extra.as_ref().and_then(|x| x.get(root)));
|
||||
// `push` strips a caller's `_MODULES` only after a run handler has authorized the
|
||||
// tag, so reading it here would let the authorized tag and the queued one differ.
|
||||
let root_value = (root != MODULES_ARG)
|
||||
.then(|| {
|
||||
args.args
|
||||
.get(root)
|
||||
.or(args.extra.as_ref().and_then(|x| x.get(root)))
|
||||
})
|
||||
.flatten();
|
||||
let arg_value = render_tag_path(root_value.map(|x| &**x), rest);
|
||||
interpolated =
|
||||
interpolated.replace(format!("$args[{}]", arg_name).as_str(), &arg_value);
|
||||
@@ -5977,7 +5983,7 @@ async fn push_inner<'c, 'd>(
|
||||
mut tx: PushIsolationLevel<'c>,
|
||||
workspace_id: &str,
|
||||
job_payload: JobPayload,
|
||||
mut args: PushArgs<'d>,
|
||||
args: PushArgs<'d>,
|
||||
user: &str,
|
||||
mut email: &str,
|
||||
mut permissioned_as: String,
|
||||
@@ -6003,6 +6009,27 @@ async fn push_inner<'c, 'd>(
|
||||
trigger: Option<TriggerMetadata>,
|
||||
suspended_mode: Option<bool>,
|
||||
) -> Result<(Uuid, Transaction<'c, Postgres>), Error> {
|
||||
// The worker builds a preview's `_MODULES` arg into the job as its module code. Every
|
||||
// caller-reachable value lands in `args` or `extra` (webhook query and headers go to
|
||||
// `extra`, WAC children copy their parent's args), so it is dropped from both and only
|
||||
// the `JobPayload::Code` arm below sets it, from the server-side `RawCode::modules`.
|
||||
let args_without_modules;
|
||||
let mut args = {
|
||||
let mut extra = args.extra;
|
||||
if let Some(extra) = extra.as_mut() {
|
||||
extra.remove(MODULES_ARG);
|
||||
}
|
||||
let args = if args.args.contains_key(MODULES_ARG) {
|
||||
let mut stripped = args.args.clone();
|
||||
stripped.remove(MODULES_ARG);
|
||||
args_without_modules = stripped;
|
||||
&args_without_modules
|
||||
} else {
|
||||
args.args
|
||||
};
|
||||
PushArgs { extra, args }
|
||||
};
|
||||
|
||||
#[cfg(feature = "cloud")]
|
||||
if *CLOUD_HOSTED {
|
||||
// A fork/dev workspace draws its plan and usage from the root (billing) workspace, so its
|
||||
@@ -6375,12 +6402,11 @@ async fn push_inner<'c, 'd>(
|
||||
language = ScriptLang::Bun;
|
||||
}
|
||||
}
|
||||
// Inject modules into job args as _MODULES so the worker can extract them
|
||||
if let Some(ref modules) = modules {
|
||||
match serde_json::to_string(modules).and_then(|s| RawValue::from_string(s)) {
|
||||
Ok(raw) => {
|
||||
let extra = args.extra.get_or_insert_with(HashMap::new);
|
||||
extra.insert("_MODULES".to_string(), raw);
|
||||
extra.insert(MODULES_ARG.to_string(), raw);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to serialize modules for preview job: {e}");
|
||||
@@ -8572,6 +8598,13 @@ mod render_tag_path_tests {
|
||||
interpolate_args("w-$args[cfg.lang]-$args[e]".to_string(), &push_args, "ws"),
|
||||
"w-eu-x"
|
||||
);
|
||||
|
||||
let args = HashMap::from([("_MODULES".to_string(), raw(r#""allowed-""#))]);
|
||||
let push_args = PushArgs { args: &args, extra: None };
|
||||
assert_eq!(
|
||||
interpolate_args("$args[_MODULES]private".to_string(), &push_args, "ws"),
|
||||
"private"
|
||||
);
|
||||
}
|
||||
|
||||
fn raw(json: &str) -> Box<RawValue> {
|
||||
|
||||
@@ -716,6 +716,10 @@ pub struct OnBehalfOf {
|
||||
|
||||
pub const ENTRYPOINT_OVERRIDE: &str = "_ENTRYPOINT_OVERRIDE";
|
||||
|
||||
/// Job-arg key carrying a preview's module code. Only `push` writes it, from
|
||||
/// `RawCode::modules`: a caller-supplied one is dropped there.
|
||||
pub const MODULES_ARG: &str = "_MODULES";
|
||||
|
||||
/// Reserved job-arg key holding the inbound W3C `traceparent` captured from the
|
||||
/// request that enqueued the job (run endpoints). It rides the `args` jsonb like
|
||||
/// [`ENTRYPOINT_OVERRIDE`]; normal scripts never see it because args are bound by
|
||||
|
||||
@@ -2940,7 +2940,9 @@ pub async fn handle_wac_v2_output(
|
||||
dedicated_worker: None,
|
||||
concurrency_settings: ConcurrencySettingsWithCustom::default(),
|
||||
debouncing_settings: DebouncingSettings::default(),
|
||||
modules: None,
|
||||
// `push` drops the `_MODULES` the children inherit with the
|
||||
// parent's args, so the preview's own modules are handed over here.
|
||||
modules: modules.clone(),
|
||||
tag: None,
|
||||
}))
|
||||
}
|
||||
|
||||
@@ -21,6 +21,7 @@ use windmill_common::get_latest_deployed_hash_for_path;
|
||||
use windmill_common::jobs::InlineScriptTarget;
|
||||
use windmill_common::jobs::RunInlineScriptFnParams;
|
||||
use windmill_common::jobs::WorkerInternalServerInlineUtils;
|
||||
use windmill_common::jobs::MODULES_ARG;
|
||||
use windmill_common::jobs::WORKER_INTERNAL_SERVER_INLINE_UTILS;
|
||||
use windmill_common::otel_oss::{
|
||||
otel_incr_worker_execution_count, otel_incr_worker_started,
|
||||
@@ -5545,13 +5546,14 @@ async fn handle_code_execution_job(
|
||||
|
||||
// Whatever is here is what gets written to the job dir and built in, so the agent-worker
|
||||
// server precomputing a cache name has to resolve modules the same way
|
||||
// (`windmill-api-agent-workers`, `get_code_and_lock`). Only a preview carries its modules
|
||||
// in its args: a deployed runnable's args are the caller's, so honoring `_MODULES` there
|
||||
// would run caller code as that runnable (and as its `on_behalf_of` identity).
|
||||
// (`windmill-api-agent-workers`, `get_code_and_lock`). `push` stores `_MODULES` for a
|
||||
// preview only, from its `RawCode`; a job queued by a server predating that may still carry
|
||||
// a caller's, and honoring it on a deployed runnable would run caller code as that runnable
|
||||
// (and as its `on_behalf_of` identity).
|
||||
let modules = modules_from_data.clone().or_else(|| {
|
||||
let args = job.args.as_ref().filter(|_| job.kind == JobKind::Preview);
|
||||
args.and_then(|args| {
|
||||
args.get("_MODULES").and_then(|raw| {
|
||||
args.get(MODULES_ARG).and_then(|raw| {
|
||||
serde_json::from_str::<std::collections::HashMap<String, ScriptModule>>(raw.get())
|
||||
.ok()
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user