fix: check tag availability for flow substeps (#7468)

backend/tests/worker.rs

@@ -2968,3 +2968,76 @@ async fn test_duckdb_ffi(db: Pool<Postgres>) -> 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<Postgres>) -> 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(())
+}

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)