626 lines
23 KiB
Rust
626 lines
23 KiB
Rust
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<Item = Uuid> + Unpin,
|
|
db: &Pool<Postgres>,
|
|
) {
|
|
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<Postgres>) -> 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<Postgres>) -> 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(())
|
|
}
|
|
|
|
/// 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<Postgres>) -> 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(())
|
|
}
|
|
|
|
/// 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<Postgres>) -> 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<Postgres>) -> 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<Postgres>) -> 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(())
|
|
}
|
|
}
|