mod suspend_resume { #[cfg(feature = "deno_core")] use serde_json::json; #[cfg(feature = "deno_core")] use windmill_test_utils::*; #[cfg(feature = "deno_core")] use futures::{Stream, StreamExt}; #[cfg(feature = "deno_core")] use sqlx::types::Uuid; #[cfg(feature = "deno_core")] use sqlx::{Pool, Postgres}; #[cfg(feature = "deno_core")] use windmill_common::flows::FlowValue; #[cfg(feature = "deno_core")] use windmill_common::jobs::JobPayload; #[cfg(feature = "deno_core")] pub async fn initialize_tracing() { use std::sync::Once; static ONCE: Once = Once::new(); ONCE.call_once(|| { let _ = windmill_common::tracing_init::initialize_tracing( "test", &windmill_common::utils::Mode::Standalone, "test", ); }); } #[cfg(feature = "deno_core")] async fn wait_until_flow_suspends( flow: Uuid, mut queue: impl Stream + Unpin, db: &Pool, ) { loop { queue.by_ref().find(&flow).await.unwrap(); if sqlx::query_scalar!( "SELECT suspend > 0 AS \"r!\" FROM v2_job_queue WHERE id = $1", flow ) .fetch_one(db) .await .unwrap() { break; } } } #[cfg(feature = "deno_core")] fn flow() -> FlowValue { serde_json::from_value(serde_json::json!({ "modules": [{ "id": "a", "value": { "input_transforms": { "n": { "type": "javascript", "expr": "flow_input.n", }, "port": { "type": "javascript", "expr": "flow_input.port", }, "op": { "type": "javascript", "expr": "flow_input.op ?? 'resume'", }, }, "type": "rawscript", "language": "deno", "content": "\ export async function main(n, port, op) {\ const job = Deno.env.get('WM_JOB_ID'); const token = Deno.env.get('WM_TOKEN'); const r = await fetch( `http://localhost:${port}/api/w/test-workspace/jobs/job_signature/${job}/0?token=${token}&approver=ruben`,\ {\ method: 'GET',\ headers: { 'Authorization': `Bearer ${token}` }\ }\ );\ console.log(r);\ const secret = await r.text();\ console.log('Secret: ' + secret + ' ' + job + ' ' + token);\ const r2 = await fetch( `http://localhost:${port}/api/w/test-workspace/jobs_u/${op}/${job}/0/${secret}?approver=ruben`,\ {\ method: 'POST',\ body: JSON.stringify('from job'),\ headers: { 'content-type': 'application/json' }\ }\ );\ console.log(await r2.text());\ return n + 1;\ }", }, "suspend": { "required_events": 1 }, }, { "id": "b", "value": { "input_transforms": { "n": { "type": "javascript", "expr": "results.a", }, "resume": { "type": "javascript", "expr": "resume", }, "resumes": { "type": "javascript", "expr": "resumes", }, }, "type": "rawscript", "language": "deno", "content": "export function main(n, resume, resumes) { return { n: n + 1, resume, resumes } }" }, "suspend": { "required_events": 1 }, }, { "value": { "input_transforms": { "last": { "type": "javascript", "expr": "results.b", }, "resume": { "type": "javascript", "expr": "resume", }, "resumes": { "type": "javascript", "expr": "resumes", }, }, "type": "rawscript", "language": "deno", "content": "export function main(last, resume, resumes) { return { last, resume, resumes } }" }, }], })) .unwrap() } #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) .arg("n", json!(1)) .arg("port", json!(port)) .push(&db) .await; let mut completed = listen_for_completed_jobs(&db).await; let queue = listen_for_queue(&db).await; let db_ = db.clone(); in_test_worker(&db, async move { let db = db_; wait_until_flow_suspends(flow, queue, &db).await; // print_job(flow, &db).await; /* The first job resumes itself. */ let _first = completed.next().await.unwrap(); // print_job(_first, &db).await; /* ... and send a request resume it. */ let second = completed.next().await.unwrap(); // print_job(second, &db).await; let token = windmill_common::auth::create_token_for_owner(&db, "test-workspace", "u/test-user", "", 100, "", &Uuid::nil(), None, None).await.unwrap(); let secret = reqwest::get(format!( "http://localhost:{port}/api/w/test-workspace/jobs/job_signature/{second}/0?token={token}&approver=ruben" )) .await .unwrap() .error_for_status() .unwrap() .text().await.unwrap(); println!("{}", secret); /* ImZyb20gdGVzdCIK = base64 "from test" */ reqwest::get(format!( "http://localhost:{port}/api/w/test-workspace/jobs_u/resume/{second}/0/{secret}?payload=ImZyb20gdGVzdCIK&approver=ruben" )) .await .unwrap() .error_for_status() .unwrap(); completed.find(&flow).await.unwrap(); }, port) .await; server.close().await.unwrap(); let result = completed_job(flow, &db).await.json_result().unwrap(); assert_eq!( json!({ "last": { "resume": "from job", "resumes": ["from job"], "n": 3, }, "resume": "from test", "resumes": ["from test"], }), result ); // ensure resumes are cleaned up through CASCADE when the flow is finished assert_eq!( 0, sqlx::query_scalar!("SELECT count(*) AS \"count!\" FROM resume_job") .fetch_one(&db) .await .unwrap() ); Ok(()) } #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn cancel_from_job(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); let result = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) .arg("n", json!(1)) .arg("op", json!("cancel")) .arg("port", json!(port)) .run_until_complete(&db, false, port) .await .json_result() .unwrap(); server.close().await.unwrap(); assert_eq!( json!( {"error": {"name": "SuspendedDisapproved", "message": "Disapproved by ruben"}}), result ); Ok(()) } /// A suspend gate that ends without approval leaves the worker that ran the approval step /// alive, so the error handler it routes to must stay pinned to that worker rather than /// being unpinned and routed by tag — which would break the `./shared` contract of a /// `same_worker` flow. #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn disapproved_suspend_keeps_same_worker_pin(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); let value: FlowValue = serde_json::from_value(json!({ "same_worker": true, "modules": [{ "id": "a", "value": { "input_transforms": { "port": { "type": "javascript", "expr": "flow_input.port" }, }, "type": "rawscript", "language": "deno", "content": "\ export async function main(port) {\ const job = Deno.env.get('WM_JOB_ID');\ const token = Deno.env.get('WM_TOKEN');\ const secret = await (await fetch(\ `http://localhost:${port}/api/w/test-workspace/jobs/job_signature/${job}/0?token=${token}&approver=ruben`,\ { headers: { 'Authorization': `Bearer ${token}` } }\ )).text();\ await fetch(\ `http://localhost:${port}/api/w/test-workspace/jobs_u/cancel/${job}/0/${secret}?approver=ruben`,\ { method: 'POST', body: JSON.stringify('from job'), headers: { 'content-type': 'application/json' } }\ );\ return 'a ran';\ }", }, "suspend": { "required_events": 1 }, }, { "id": "b", "value": { "input_transforms": {}, "type": "rawscript", "language": "deno", "content": "export function main() { return 'b ran' }", }, // The gate holds `b` back, so `b` never runs and its error policy describes // nothing: honouring it here would skip `b` instead of reaching the handler. "continue_on_error": true, }], "failure_module": { "id": "failure", "value": { "input_transforms": {}, "type": "rawscript", "language": "deno", "content": "export function main() { return 'handled' }", }, }, }))?; let completed = RunJob::from(JobPayload::RawFlow { value, path: None, restarted_from: None }) .arg("port", json!(port)) .run_until_complete(&db, false, port) .await; server.close().await.unwrap(); assert_eq!(json!("handled"), completed.json_result().unwrap()); let same_worker: Option = sqlx::query_scalar( "SELECT same_worker FROM v2_job WHERE parent_job = $1 AND flow_step_id = 'failure'", ) .bind(completed.id) .fetch_one(&db) .await?; assert_eq!(Some(true), same_worker); Ok(()) } /// Test that self-approval is blocked when self_approval_disabled is true. /// /// This test verifies that when a flow has an approval step with self_approval_disabled=true, /// the user who triggered the flow cannot approve it themselves via the owner endpoint /// (POST /jobs/flow/resume/:id). /// /// Bug context: The owner endpoint was missing the approval condition check, allowing /// users to bypass self-approval restrictions by using the UI resume button instead /// of the HMAC-signed approval link. #[cfg(feature = "enterprise")] #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_self_approval_disabled_blocks_owner_resume( db: Pool, ) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); // Flow with self_approval_disabled=true on the approval step let flow_with_self_approval_disabled: FlowValue = serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step1'; }" }, "suspend": { "required_events": 1, "user_auth_required": true, "self_approval_disabled": true } }, { "id": "b", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step2 - after approval'; }" } }] })) .unwrap(); // Push flow as NON-ADMIN user (test-user-2) - admins bypass self-approval check // Use a path owned by test-user-2 so require_owner_of_path succeeds let flow = RunJob::from(JobPayload::RawFlow { value: flow_with_self_approval_disabled, path: Some("u/test-user-2/test_approval".to_string()), restarted_from: None, }) .push_as(&db, "test-user-2", "test2@windmill.dev") .await; let queue = listen_for_queue(&db).await; let db_ = db.clone(); in_test_worker( &db, async move { let db = db_; // Wait for flow to suspend at approval step wait_until_flow_suspends(flow, queue, &db).await; // Create a token for the same non-admin user who triggered the flow // This simulates clicking "Resume" in the UI as the flow owner // Args: db, w_id, owner, label, expires_in, email, job_id, perms, audit_span let token = windmill_common::auth::create_token_for_owner( &db, "test-workspace", "u/test-user-2", "test-token", 100, "test2@windmill.dev", &Uuid::nil(), None, None, ) .await .unwrap(); // Try to resume via the owner endpoint (POST /jobs/flow/resume/:id) // This should FAIL because self_approval_disabled=true and the user // is the same as the one who triggered the flow let response = reqwest::Client::new() .post(format!( "http://localhost:{port}/api/w/test-workspace/jobs/flow/resume/{flow}" )) .header("Authorization", format!("Bearer {token}")) .header("Content-Type", "application/json") .body("{}") .send() .await .unwrap(); let status = response.status(); // The request should be rejected with 403 Forbidden // (currently this test FAILS because the bug allows self-approval) assert!( status == reqwest::StatusCode::FORBIDDEN, "Self-approval should be blocked when self_approval_disabled=true. \ Expected 403 Forbidden, got {}. Response: {}", status, response.text().await.unwrap_or_default() ); }, port, ) .await; server.close().await.unwrap(); Ok(()) } /// The UI "Resume" button (POST /jobs_u/flow/resume_suspended/:job_id) must reject the /// triggerer approving their own self_approval_disabled step even when they own the flow /// path: only admins are exempt from the self-approval restriction, so owning the runnable /// does not grant the right to self-approve. #[cfg(feature = "enterprise")] #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_self_approval_disabled_blocks_ui_resume_for_owner( db: Pool, ) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); let flow_with_self_approval_disabled: FlowValue = serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step1'; }" }, "suspend": { "required_events": 1, "user_auth_required": true, "self_approval_disabled": true } }, { "id": "b", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step2 - after approval'; }" } }] })) .unwrap(); // Push as a non-admin who owns the flow path, so the owner shortcut is exercised. let flow = RunJob::from(JobPayload::RawFlow { value: flow_with_self_approval_disabled, path: Some("u/test-user-2/test_ui_resume".to_string()), restarted_from: None, }) .push_as(&db, "test-user-2", "test2@windmill.dev") .await; let queue = listen_for_queue(&db).await; let db_ = db.clone(); in_test_worker( &db, async move { let db = db_; wait_until_flow_suspends(flow, queue, &db).await; let token = windmill_common::auth::create_token_for_owner( &db, "test-workspace", "u/test-user-2", "test-token", 100, "test2@windmill.dev", &Uuid::nil(), None, None, ) .await .unwrap(); // Resume via the UI endpoint as the owner who triggered the flow. let response = reqwest::Client::new() .post(format!( "http://localhost:{port}/api/w/test-workspace/jobs_u/flow/resume_suspended/{flow}" )) .header("Authorization", format!("Bearer {token}")) .header("Content-Type", "application/json") .body("{}") .send() .await .unwrap(); let status = response.status(); assert!( status == reqwest::StatusCode::FORBIDDEN, "Self-approval via the UI resume endpoint should be blocked for the owner when \ self_approval_disabled=true. Expected 403 Forbidden, got {}. Response: {}", status, response.text().await.unwrap_or_default() ); }, port, ) .await; server.close().await.unwrap(); Ok(()) } /// self_approval_disabled must hold even when user_auth_required is not set: the worker must /// persist the condition and the resume boundary must enforce it for the authenticated /// triggerer. The triggerer here is a non-owner (folder path they don't own), so the owner /// shortcut is not involved and this exercises the persistence + authenticated-check path. #[cfg(feature = "enterprise")] #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_self_approval_disabled_without_user_auth_required( db: Pool, ) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); // self_approval_disabled without user_auth_required (as a raw-flow/CLI author could set). let flow_value: FlowValue = serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step1'; }" }, "suspend": { "required_events": 1, "self_approval_disabled": true } }, { "id": "b", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step2 - after approval'; }" } }] })) .unwrap(); // Folder path test-user-2 does not own -> non-owner triggerer (no folders in the base // fixture), so the owner shortcut is bypassed and only persistence matters here. let flow = RunJob::from(JobPayload::RawFlow { value: flow_value, path: Some("f/system/test_persist".to_string()), restarted_from: None, }) .push_as(&db, "test-user-2", "test2@windmill.dev") .await; let queue = listen_for_queue(&db).await; let db_ = db.clone(); in_test_worker( &db, async move { let db = db_; wait_until_flow_suspends(flow, queue, &db).await; let token = windmill_common::auth::create_token_for_owner( &db, "test-workspace", "u/test-user-2", "test-token", 100, "test2@windmill.dev", &Uuid::nil(), None, None, ) .await .unwrap(); let response = reqwest::Client::new() .post(format!( "http://localhost:{port}/api/w/test-workspace/jobs_u/flow/resume_suspended/{flow}" )) .header("Authorization", format!("Bearer {token}")) .header("Content-Type", "application/json") .body("{}") .send() .await .unwrap(); let status = response.status(); assert!( status == reqwest::StatusCode::FORBIDDEN, "Self-approval should be blocked when self_approval_disabled=true even without \ user_auth_required. Expected 403 Forbidden, got {}. Response: {}", status, response.text().await.unwrap_or_default() ); }, port, ) .await; server.close().await.unwrap(); Ok(()) } /// Test that self-approval WORKS when self_approval_disabled is false (default behavior). /// /// This is the complementary test to test_self_approval_disabled_blocks_owner_resume. /// When self_approval_disabled is NOT set, the flow owner should be able to approve /// their own flow via the owner endpoint. #[cfg(feature = "enterprise")] #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_self_approval_allowed_when_not_disabled( db: Pool, ) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); // Flow with user_auth_required but WITHOUT self_approval_disabled let flow_without_self_approval_disabled: FlowValue = serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step1'; }" }, "suspend": { "required_events": 1, "user_auth_required": true // self_approval_disabled is NOT set (defaults to false) } }, { "id": "b", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step2 - after approval'; }" } }] })) .unwrap(); // Push flow as non-admin user let flow = RunJob::from(JobPayload::RawFlow { value: flow_without_self_approval_disabled, path: Some("u/test-user-2/test_approval_allowed".to_string()), restarted_from: None, }) .push_as(&db, "test-user-2", "test2@windmill.dev") .await; let queue = listen_for_queue(&db).await; let db_ = db.clone(); in_test_worker( &db, async move { let db = db_; // Wait for flow to suspend at approval step wait_until_flow_suspends(flow, queue, &db).await; // Create a token for the same user who triggered the flow let token = windmill_common::auth::create_token_for_owner( &db, "test-workspace", "u/test-user-2", "test-token", 100, "test2@windmill.dev", &Uuid::nil(), None, None, ) .await .unwrap(); // Try to resume via the owner endpoint - this SHOULD succeed let response = reqwest::Client::new() .post(format!( "http://localhost:{port}/api/w/test-workspace/jobs/flow/resume/{flow}" )) .header("Authorization", format!("Bearer {token}")) .header("Content-Type", "application/json") .body("{}") .send() .await .unwrap(); let status = response.status(); // Self-approval should be allowed when self_approval_disabled is not set assert!( status.is_success(), "Self-approval should be allowed when self_approval_disabled is not set. \ Expected 2xx, got {}. Response: {}", status, response.text().await.unwrap_or_default() ); }, port, ) .await; server.close().await.unwrap(); Ok(()) } /// Test that a DIFFERENT user can approve a flow even when self_approval_disabled is true. /// /// This verifies that the self_approval_disabled setting only blocks the flow trigger, /// not other users. A different user should always be able to approve. #[cfg(feature = "enterprise")] #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn test_different_user_can_approve_when_self_approval_disabled( db: Pool, ) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); // Flow with self_approval_disabled=true let flow_with_self_approval_disabled: FlowValue = serde_json::from_value(json!({ "modules": [{ "id": "a", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step1'; }" }, "suspend": { "required_events": 1, "user_auth_required": true, "self_approval_disabled": true } }, { "id": "b", "value": { "type": "rawscript", "language": "deno", "content": "export function main() { return 'step2 - after approval'; }" } }] })) .unwrap(); // Push flow as test-user-2 let flow = RunJob::from(JobPayload::RawFlow { value: flow_with_self_approval_disabled, path: Some("u/test-user-2/test_approval_by_other".to_string()), restarted_from: None, }) .push_as(&db, "test-user-2", "test2@windmill.dev") .await; let queue = listen_for_queue(&db).await; let db_ = db.clone(); in_test_worker( &db, async move { let db = db_; // Wait for flow to suspend at approval step wait_until_flow_suspends(flow, queue, &db).await; // Create a token for a DIFFERENT user (test-user, who is admin) // This simulates a different person approving the flow let token = windmill_common::auth::create_token_for_owner( &db, "test-workspace", "u/test-user", "test-token", 100, "test@windmill.dev", &Uuid::nil(), None, None, ) .await .unwrap(); // Try to resume via the owner endpoint as a different user - this SHOULD succeed let response = reqwest::Client::new() .post(format!( "http://localhost:{port}/api/w/test-workspace/jobs/flow/resume/{flow}" )) .header("Authorization", format!("Bearer {token}")) .header("Content-Type", "application/json") .body("{}") .send() .await .unwrap(); let status = response.status(); // A different user should be able to approve even with self_approval_disabled assert!( status.is_success(), "Different user should be able to approve even with self_approval_disabled=true. \ Expected 2xx, got {}. Response: {}", status, response.text().await.unwrap_or_default() ); }, port, ) .await; server.close().await.unwrap(); Ok(()) } #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn cancel_after_suspend(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let port = server.addr.port(); let flow = RunJob::from(JobPayload::RawFlow { value: flow(), path: None, restarted_from: None }) .arg("n", json!(1)) .arg("port", json!(port)) .push(&db) .await; let mut completed = listen_for_completed_jobs(&db).await; let queue = listen_for_queue(&db).await; let db_ = db.clone(); in_test_worker(&db, async move { let db = db_; wait_until_flow_suspends(flow, queue, &db).await; /* The first job resumes itself. */ let _first = completed.next().await.unwrap(); /* ... and send a request resume it. */ let second = completed.next().await.unwrap(); let token = windmill_common::auth::create_token_for_owner(&db, "test-workspace", "u/test-user", "", 100, "", &Uuid::nil(), None, None).await.unwrap(); let secret = reqwest::get(format!( "http://localhost:{port}/api/w/test-workspace/jobs/job_signature/{second}/0?token={token}" )) .await .unwrap() .error_for_status() .unwrap() .text().await.unwrap(); println!("{}", secret); /* ImZyb20gdGVzdCIK = base64 "from test" */ reqwest::get(format!( "http://localhost:{port}/api/w/test-workspace/jobs_u/cancel/{second}/0/{secret}?payload=ImZyb20gdGVzdCIK" )) .await .unwrap() .error_for_status() .unwrap(); completed.find(&flow).await.unwrap(); }, port) .await; server.close().await.unwrap(); let result = completed_job(flow, &db).await.json_result().unwrap(); assert_eq!( json!( {"error": {"name": "SuspendedDisapproved", "message": "Disapproved by unknown"}}), result ); Ok(()) } /// A step that declares a `suspend` but is skipped via `skip_if` never arms /// its approval, so it must not gate the following step. If it did, the flow /// would park forever waiting for a resume event that is never dispatched. #[cfg(feature = "deno_core")] #[sqlx::test(fixtures("base"))] async fn skipped_suspend_step_does_not_block_next_step( db: Pool, ) -> anyhow::Result<()> { initialize_tracing().await; let server = ApiServer::start(db.clone()).await?; let flow: FlowValue = serde_json::from_value(json!({ "modules": [ { "id": "a", "skip_if": { "type": "javascript", "expr": "true" }, "suspend": { "required_events": 1, "timeout": 86400 }, "value": { "type": "rawscript", "language": "deno", "content": "export async function main() { return 1 }", "input_transforms": {}, }, }, { "id": "b", "value": { "type": "rawscript", "language": "deno", "content": "export async function main() { return 42 }", "input_transforms": {}, }, }, ], })) .unwrap(); let result = RunJob::from(JobPayload::RawFlow { value: flow, path: None, restarted_from: None }) .run_until_complete(&db, false, server.addr.port()) .await .json_result() .unwrap(); assert_eq!(result, json!(42)); Ok(()) } }