mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-19 00:02:03 +00:00
* fix(flows): reject corrupt step paths at deploy + atomic cache writes (#9751) A flow step could execute an unrelated (and in the reported case, destructive) script at runtime even though every stored definition looked correct. A forensic dump traced it to two issues: - Deploy accepted absolute/local step paths. `wmill sync push` from a feature- branch checkout under /tmp baked an absolute path (`/tmp/.../ops/scripts/clean_device/...`) into a step's `value.path`. Persisted verbatim, it mis-resolved to an unrelated script at runtime. - The on-disk cache write was neither truncating nor atomic. `FsBackedCache::put` used `write+create`, so a shorter overwrite left stale trailing bytes and concurrent writers could interleave into a torn file — a corrupt cached blob that a worker then scheduled from. Fixes: - Reject non-workspace flow step paths (must be u/, f/, g/ or hub/) in `validate_flow_value` (covers create_flow + update_flow, recursively through loops/branches/AI-agent tools) and early in the CLI `pushFlow`. - Make `FsBackedCache::put` write a unique temp file (truncate + fsync) then atomically rename it over the target, cleaning up on error. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(flows): validate failure/preprocessor module paths + sub-flow paths in CLI Address PR review (cubic + claude): - Backend `validate_flow_value` is the authoritative guard but only walked `modules`; extend it to also validate `failure_module` and `preprocessor_module` (which can themselves be sub-flows/loops/branches), so an absolute path there can't be persisted. - CLI preflight only collected `type: "script"` paths; now collects sub-flow (`type: "flow"`) step paths too (recursively, incl. failure/preprocessor), so the comment's claim matches the behavior. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(cli): include AI-agent tool step paths in flow path preflight Address Codex review: collectStepPaths skipped aiagent tools, so a bad path in a tool fell through to the API error instead of the local fail-fast. The backend already validates these (traverse_modules walks AIAgent tools); this aligns the CLI early-error with it. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * test(flows): make failure/preprocessor path test key explicit The test used `slot:` as a json! key. json! does interpolate an ident key to its variable's value (json!({slot:1}) with slot="failure_module" => {"failure_module":1}), so the test was correct and exercised the validation — but the behavior is subtle, so build the key explicitly via serde_json::Map to remove ambiguity (review nit). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(cache): use a UUID temp name for atomic put (shared-volume safe) Address Codex (P1): pid+counter temp names collide across container PID namespaces on a shared cache volume (same pid, PUT_SEQ resets to 0 per process), so two workers could truncate/clobber the same temp file before rename. Use a random UUID suffix (matching worker.rs's atomic-write helpers) — globally unique, so the cross-process temp-file hazard is closed. Also trims the comment to the AGENTS.md <=4-line limit (Pi nit). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
GitHub
@@ -1189,11 +1189,25 @@ const _: () = {
|
||||
use std::fs::OpenOptions;
|
||||
use std::io::Write;
|
||||
|
||||
// Atomic write: truncate+write a uniquely-named temp file (UUID, not pid — pids
|
||||
// collide across container PID namespaces on a shared cache volume), fsync, then
|
||||
// rename(2) over the target. Without this a shorter overwrite leaves a stale tail
|
||||
// and concurrent writers tear the file — a corrupt entry a reader would import.
|
||||
let final_path = item.path(self);
|
||||
let tmp_path = final_path.with_extension(format!("tmp.{}", Uuid::new_v4()));
|
||||
OpenOptions::new()
|
||||
.write(true)
|
||||
.create(true)
|
||||
.open(item.path(self))
|
||||
.and_then(|mut file| file.write_all(data.as_ref()))
|
||||
.truncate(true)
|
||||
.open(&tmp_path)
|
||||
.and_then(|mut file| {
|
||||
file.write_all(data.as_ref())?;
|
||||
file.sync_all()
|
||||
})
|
||||
.and_then(|()| std::fs::rename(&tmp_path, &final_path))
|
||||
.inspect_err(|_| {
|
||||
let _ = std::fs::remove_file(&tmp_path);
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1237,6 +1251,33 @@ mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn fs_cache_put_overwrites_without_stale_tail() {
|
||||
// Regression for the non-truncating, non-atomic `put`: overwriting a value with a
|
||||
// shorter one must not leave stale trailing bytes (which imported as corrupt/wrong
|
||||
// content — the #9751 worker-cache hazard).
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let root = dir.path();
|
||||
|
||||
root.put("k", b"a-long-cached-value-0123456789").unwrap();
|
||||
assert_eq!(root.get("k").unwrap(), b"a-long-cached-value-0123456789");
|
||||
|
||||
root.put("k", b"short").unwrap();
|
||||
assert_eq!(
|
||||
root.get("k").unwrap(),
|
||||
b"short",
|
||||
"shorter overwrite must fully replace, no stale tail"
|
||||
);
|
||||
|
||||
// No temp files left behind after a successful write.
|
||||
let leftover: Vec<_> = std::fs::read_dir(root)
|
||||
.unwrap()
|
||||
.filter_map(|e| e.ok())
|
||||
.filter(|e| e.file_name().to_string_lossy().contains(".tmp."))
|
||||
.collect();
|
||||
assert!(leftover.is_empty(), "temp files must be renamed/cleaned up");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flow_data_extras_preserves_notes_and_groups() {
|
||||
let raw = serde_json::value::to_raw_value(&json!({
|
||||
|
||||
@@ -155,6 +155,19 @@ fn validate_retry(retry: &Retry, module_id: &str) -> anyhow::Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Script/sub-flow step references must be workspace paths (`u/`, `f/`, `g/`) or a hub
|
||||
/// reference (`hub/`). Empty is tolerated for intermediate/incomplete steps. This blocks
|
||||
/// absolute or local filesystem paths (e.g. `/tmp/.../ops/scripts/...` baked in by a
|
||||
/// `wmill sync push` from a feature-branch checkout) from being persisted into a flow,
|
||||
/// where they silently mis-resolve to an unrelated script at runtime (#9751).
|
||||
fn is_workspace_runnable_path(path: &str) -> bool {
|
||||
path.is_empty()
|
||||
|| path.starts_with("u/")
|
||||
|| path.starts_with("f/")
|
||||
|| path.starts_with("g/")
|
||||
|| path.starts_with("hub/")
|
||||
}
|
||||
|
||||
fn validate_flow_value<'de, D>(deserializer: D) -> Result<Box<RawValue>, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
@@ -164,21 +177,38 @@ where
|
||||
let flow_value: FlowValue = serde_json::from_str(raw_value.get())
|
||||
.map_err(|e| serde::de::Error::custom(format!("Invalid flow value: {}", e)))?;
|
||||
|
||||
FlowModule::traverse_modules(&flow_value.modules, &mut |module| {
|
||||
let mut validate_module = |module: &FlowModule| -> anyhow::Result<()> {
|
||||
if let Some(ref retry) = module.retry {
|
||||
validate_retry(retry, &module.id)?;
|
||||
}
|
||||
return Ok(());
|
||||
})
|
||||
.map_err(|e| serde::de::Error::custom(e.to_string()))?;
|
||||
if let Ok(FlowModuleValue::Script { path, .. } | FlowModuleValue::Flow { path, .. }) =
|
||||
module.get_value()
|
||||
{
|
||||
if !is_workspace_runnable_path(&path) {
|
||||
return Err(anyhow::anyhow!(
|
||||
"step '{}' references '{}', which is not a workspace path (expected u/, \
|
||||
f/, g/ or hub/). Absolute or local filesystem paths are not allowed in \
|
||||
flow steps.",
|
||||
module.id,
|
||||
path
|
||||
));
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
};
|
||||
|
||||
if let Some(ref _failure_module) = flow_value.failure_module {
|
||||
//add validation logic here for failure module
|
||||
}
|
||||
|
||||
if let Some(ref _preprocessor_module) = flow_value.preprocessor_module {
|
||||
//add validation logic here for preprocessor module
|
||||
}
|
||||
// The API is the authoritative guard (it can be called directly, bypassing the CLI), so
|
||||
// it must cover every step that resolves a path: the main modules AND the failure /
|
||||
// preprocessor modules (which can themselves be sub-flows/loops/branches).
|
||||
let extra_modules: Vec<FlowModule> = flow_value
|
||||
.failure_module
|
||||
.iter()
|
||||
.chain(flow_value.preprocessor_module.iter())
|
||||
.map(|m| (**m).clone())
|
||||
.collect();
|
||||
FlowModule::traverse_modules(&flow_value.modules, &mut validate_module)
|
||||
.and_then(|()| FlowModule::traverse_modules(&extra_modules, &mut validate_module))
|
||||
.map_err(|e| serde::de::Error::custom(e.to_string()))?;
|
||||
|
||||
Ok(raw_value)
|
||||
}
|
||||
@@ -1228,6 +1258,108 @@ mod tests {
|
||||
assert_eq!(val.modules.len(), 1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flow_rejects_absolute_step_path() {
|
||||
// #9751: an absolute local path baked into a step must be rejected on deploy.
|
||||
let bad = json!({
|
||||
"path": "f/test/flow",
|
||||
"summary": "",
|
||||
"value": { "modules": [{
|
||||
"id": "validate_onboard_target",
|
||||
"value": {
|
||||
"type": "script",
|
||||
"path": "/tmp/tmp.X/f/ops/scripts/clean_device/pre_clean",
|
||||
"input_transforms": {}
|
||||
}
|
||||
}]}
|
||||
});
|
||||
let err = serde_json::from_value::<NewFlow>(bad)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(
|
||||
err.contains("not a workspace path"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
assert!(
|
||||
err.contains("validate_onboard_target"),
|
||||
"error should name the step: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flow_rejects_absolute_step_path_in_nested_module() {
|
||||
let bad = json!({
|
||||
"path": "f/test/flow",
|
||||
"summary": "",
|
||||
"value": { "modules": [{
|
||||
"id": "loop",
|
||||
"value": {
|
||||
"type": "forloopflow",
|
||||
"iterator": {"type": "javascript", "expr": "[1]"},
|
||||
"modules": [{
|
||||
"id": "inner",
|
||||
"value": {"type": "script", "path": "/abs/path", "input_transforms": {}}
|
||||
}]
|
||||
}
|
||||
}]}
|
||||
});
|
||||
let err = serde_json::from_value::<NewFlow>(bad)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(
|
||||
err.contains("not a workspace path"),
|
||||
"unexpected error: {err}"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flow_rejects_absolute_path_in_failure_and_preprocessor_modules() {
|
||||
for slot in ["failure_module", "preprocessor_module"] {
|
||||
// Build the value with the slot as an explicit (interpolated) key.
|
||||
let mut value = serde_json::Map::new();
|
||||
value.insert("modules".to_string(), json!([]));
|
||||
value.insert(
|
||||
slot.to_string(),
|
||||
json!({
|
||||
"id": slot,
|
||||
"value": {"type": "script", "path": "/abs/path", "input_transforms": {}}
|
||||
}),
|
||||
);
|
||||
let bad = json!({ "path": "f/test/flow", "summary": "", "value": value });
|
||||
let err = serde_json::from_value::<NewFlow>(bad)
|
||||
.unwrap_err()
|
||||
.to_string();
|
||||
assert!(
|
||||
err.contains("not a workspace path"),
|
||||
"{slot} should be validated, got: {err}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn flow_accepts_workspace_step_paths() {
|
||||
for p in [
|
||||
"f/ops/scripts/x",
|
||||
"u/me/y",
|
||||
"g/grp/z",
|
||||
"hub/123/foo",
|
||||
"", // tolerated for incomplete steps
|
||||
] {
|
||||
let ok = json!({
|
||||
"path": "f/test/flow",
|
||||
"summary": "",
|
||||
"value": { "modules": [{
|
||||
"id": "a",
|
||||
"value": {"type": "script", "path": p, "input_transforms": {}}
|
||||
}]}
|
||||
});
|
||||
assert!(
|
||||
serde_json::from_value::<NewFlow>(ok).is_ok(),
|
||||
"path {p:?} should be accepted"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ai_agent_omit_output_from_conversation_defaults_to_false() {
|
||||
let input = json!({
|
||||
|
||||
@@ -135,6 +135,31 @@ function warnAboutLocalPathScriptDivergence(
|
||||
|
||||
const alreadySynced: string[] = [];
|
||||
|
||||
// Collect every script/sub-flow step path in a flow value — recursively through loops,
|
||||
// branches, and the failure/preprocessor modules — for workspace-path validation. Unlike
|
||||
// `collectPathScriptPaths` this also includes `type: "flow"` sub-flow steps.
|
||||
function collectStepPaths(flowValue: any): string[] {
|
||||
const paths: string[] = [];
|
||||
const walk = (modules: any[] | undefined) => {
|
||||
for (const m of modules ?? []) {
|
||||
const v = m?.value;
|
||||
if (!v) continue;
|
||||
if ((v.type === "script" || v.type === "flow") && typeof v.path === "string") {
|
||||
paths.push(v.path);
|
||||
}
|
||||
walk(v.modules);
|
||||
walk(v.default);
|
||||
for (const b of v.branches ?? []) walk(b?.modules);
|
||||
// AI-agent tools are step-like and can carry script paths too.
|
||||
walk(v.tools);
|
||||
}
|
||||
};
|
||||
walk(flowValue?.modules);
|
||||
if (flowValue?.failure_module) walk([flowValue.failure_module]);
|
||||
if (flowValue?.preprocessor_module) walk([flowValue.preprocessor_module]);
|
||||
return paths;
|
||||
}
|
||||
|
||||
export async function pushFlow(
|
||||
workspace: string,
|
||||
remotePath: string,
|
||||
@@ -190,6 +215,21 @@ export async function pushFlow(
|
||||
);
|
||||
}
|
||||
|
||||
// Reject script/sub-flow steps whose path is not a workspace path (u/, f/, g/ or hub/).
|
||||
// A flow.yaml generated from a feature-branch checkout can carry absolute local paths
|
||||
// (e.g. /tmp/.../ops/scripts/...); pushed, they silently mis-resolve at runtime (#9751).
|
||||
// The backend re-validates the same rule for every step type, so this is a fail-fast.
|
||||
const badStepPaths = collectStepPaths(localFlow.value).filter(
|
||||
(p) => p !== "" && !/^(u|f|g|hub)\//.test(p)
|
||||
);
|
||||
if (badStepPaths.length > 0) {
|
||||
throw new Error(
|
||||
`Cannot push flow ${remotePath}: step(s) reference non-workspace path(s): ${badStepPaths.join(", ")}. ` +
|
||||
`Flow step paths must be workspace paths (u/, f/, g/ or hub/), not absolute or local filesystem paths. ` +
|
||||
`This usually means flow.yaml was generated with paths from a checkout directory.`
|
||||
);
|
||||
}
|
||||
|
||||
const hasOnBehalfOf = (localFlow as any).has_on_behalf_of ?? !!localFlow.on_behalf_of_email;
|
||||
delete (localFlow as any).has_on_behalf_of;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user