diff --git a/backend/tests/worker.rs b/backend/tests/worker.rs index 3328296faf..0a4cb580ae 100644 --- a/backend/tests/worker.rs +++ b/backend/tests/worker.rs @@ -2968,3 +2968,76 @@ async fn test_duckdb_ffi(db: Pool) -> anyhow::Result<()> { assert_eq!(result, serde_json::json!("Hello world!")); Ok(()) } + +/// Test that flow substeps with tags that are not available for the workspace fail. +/// This validates that `check_tag_available_for_workspace_internal` is properly called +/// when pushing jobs from worker_flow. +#[cfg(feature = "deno_core")] +#[sqlx::test(fixtures("base"))] +async fn test_flow_substep_tag_availability_check(db: Pool) -> anyhow::Result<()> { + use windmill_common::worker::{CustomTags, SpecificTagData, SpecificTagType, CUSTOM_TAGS_PER_WORKSPACE}; + + initialize_tracing().await; + + let server = ApiServer::start(db.clone()).await?; + + // Set up a restricted tag that is only available to "other-workspace" (not "test-workspace") + { + let mut custom_tags = CUSTOM_TAGS_PER_WORKSPACE.write().await; + *custom_tags = CustomTags { + global: vec![], + specific: std::collections::HashMap::from([( + "restricted-tag".to_string(), + SpecificTagData { + tag_type: SpecificTagType::NoneExcept, + workspaces: vec!["other-workspace".to_string()], + }, + )]), + }; + } + + // Create a flow with a substep that uses the restricted tag + let flow: FlowValue = serde_json::from_value(serde_json::json!({ + "modules": [{ + "id": "a", + "value": { + "type": "rawscript", + "language": "deno", + "content": "export function main() { return 42; }", + "tag": "restricted-tag", + }, + }], + })) + .unwrap(); + + let result = + RunJob::from(JobPayload::RawFlow { value: flow.clone(), path: None, restarted_from: None }) + .run_until_complete(&db, false, server.addr.port()) + .await; + + // The flow should fail because the tag is not available for test-workspace + assert!(!result.success, "Flow should have failed due to unavailable tag"); + + let result_json = result.json_result(); + assert!(result_json.is_some(), "Result should have error details"); + + let error_result = result_json.unwrap(); + let error_message = error_result["error"]["message"] + .as_str() + .unwrap_or(""); + + // Verify the error is about tag availability + assert!( + error_message.contains("restricted-tag") || error_message.contains("tag"), + "Error message should mention the tag issue: {}", + error_message + ); + + // Clean up: reset custom tags + { + let mut custom_tags = CUSTOM_TAGS_PER_WORKSPACE.write().await; + *custom_tags = CustomTags::default(); + } + + Ok(()) +} diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 0a4cb01339..07b9cc3315 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -43,7 +43,8 @@ use windmill_common::flow_status::{ }; use windmill_common::flows::{add_virtual_items_if_necessary, Branch, FlowNodeId, StopAfterIf}; use windmill_common::jobs::{ - script_path_to_payload, JobKind, JobPayload, OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE, + check_tag_available_for_workspace_internal, script_path_to_payload, JobKind, JobPayload, + OnBehalfOf, RawCode, ENTRYPOINT_OVERRIDE, }; use windmill_common::runnable_settings::{ConcurrencySettingsWithCustom, DebouncingSettings}; use windmill_common::scripts::{ScriptHash, ScriptRunnableSettingsInline}; @@ -3353,6 +3354,22 @@ async fn push_next_flow_job( ) }; + // Check tag availability for flow substeps to prevent abuse + if let Some(tag_str) = tag.as_deref().filter(|t| !t.is_empty()) { + check_tag_available_for_workspace_internal( + &db, + &flow_job.workspace_id, + tag_str, + email, + None, // no token for flow substeps so no scopes so no scope_tags + ) + .warn_after_seconds_with_sql( + 1, + "check_tag_available_for_workspace_internal".to_string(), + ) + .await?; + } + let evaluated_timeout = if let Some(timeout_transform) = &module.timeout { let ctx = get_transform_context(&flow_job, &previous_id, &status) .warn_after_seconds(3)