feat: option to preserve on_behalf_of and edited_by for admins and users in the new wm_deployers group (#8079)

This commit is contained in:
wendrul
2026-02-25 13:05:22 +01:00
committed by GitHub
parent 382b27bbbc
commit 96b5b3e8bb
58 changed files with 5276 additions and 1251 deletions
@@ -20,8 +20,7 @@
"resource",
"variable",
"ducklake",
"datatable",
"volume"
"datatable"
]
}
}
@@ -46,11 +46,11 @@
]
},
"nullable": [
false,
false,
false,
false,
false,
true,
true,
true,
true,
true,
true,
true
]
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT email, edited_by FROM websocket_trigger WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "edited_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "075d4749299af2cb81162bf396bec6aa89de43ec201c911196763e03e644ca7a"
}
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT email, edited_by FROM schedule WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "edited_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "17aafb72843659df9594d6d2466d2afaf26e666ffe52e0ea85792ea31b63410c"
}
@@ -15,7 +15,7 @@
]
},
"nullable": [
true
null
]
},
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
@@ -0,0 +1,29 @@
{
"db_name": "PostgreSQL",
"query": "SELECT email, edited_by FROM http_trigger WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "email",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "edited_by",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false,
false
]
},
"hash": "8311a553c44221751ffdbbe6a997d6feba8d43292daf6c5433b66bd8450e8854"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT on_behalf_of_email FROM flow WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "on_behalf_of_email",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "85a6a85fd126a8bfedd65d6b38d22c65911ab9cf0414c33a3321a1d43af49795"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE schedule SET\n schedule = $1,\n timezone = $2,\n args = $3,\n on_failure = $4,\n on_failure_times = $5,\n on_failure_exact = $6,\n on_failure_extra_args = $7,\n on_recovery = $8,\n on_recovery_times = $9,\n on_recovery_extra_args = $10,\n on_success = $11,\n on_success_extra_args = $12,\n ws_error_handler_muted = $13,\n retry = $14,\n summary = $15,\n no_flow_overlap = $16,\n tag = $17,\n paused_until = $18,\n path = $19,\n workspace_id = $20,\n cron_version = COALESCE($21, cron_version),\n description = $22,\n dynamic_skip = $23\n WHERE path = $19 AND workspace_id = $20\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version,\n dynamic_skip\n ",
"query": "\n UPDATE schedule SET\n schedule = $1,\n timezone = $2,\n args = $3,\n on_failure = $4,\n on_failure_times = $5,\n on_failure_exact = $6,\n on_failure_extra_args = $7,\n on_recovery = $8,\n on_recovery_times = $9,\n on_recovery_extra_args = $10,\n on_success = $11,\n on_success_extra_args = $12,\n ws_error_handler_muted = $13,\n retry = $14,\n summary = $15,\n no_flow_overlap = $16,\n tag = $17,\n paused_until = $18,\n path = $19,\n workspace_id = $20,\n cron_version = COALESCE($21, cron_version),\n description = $22,\n dynamic_skip = $23,\n email = COALESCE($24, email),\n edited_by = $25\n WHERE path = $19 AND workspace_id = $20\n RETURNING\n workspace_id,\n path,\n edited_by,\n edited_at,\n schedule,\n timezone,\n enabled,\n script_path,\n is_flow,\n args AS \"args: _\",\n extra_perms,\n email,\n error,\n on_failure,\n on_failure_times,\n on_failure_exact,\n on_failure_extra_args AS \"on_failure_extra_args: _\",\n on_recovery,\n on_recovery_times,\n on_recovery_extra_args AS \"on_recovery_extra_args: _\",\n on_success,\n on_success_extra_args AS \"on_success_extra_args: _\",\n ws_error_handler_muted,\n retry,\n no_flow_overlap,\n summary,\n description,\n tag,\n paused_until,\n cron_version,\n dynamic_skip\n ",
"describe": {
"columns": [
{
@@ -183,6 +183,8 @@
"Text",
"Text",
"Text",
"Varchar",
"Varchar",
"Varchar"
]
},
@@ -220,5 +222,5 @@
true
]
},
"hash": "4144c87c25a939aafb2f57da189d94d038bcad7a36fbf87e0403c89a979c5b3f"
"hash": "987d79f7c6d7bc148cc8aab67e47161cfca045966e995e28c7a7ad090cffeda0"
}
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow_version (workspace_id, path, value, schema, created_by) \n VALUES ($1, $2, $3, $4::text::json, $5)\n RETURNING id",
"query": "INSERT INTO flow_version (workspace_id, path, value, schema, created_by)\n VALUES ($1, $2, $3, $4::text::json, $5)\n RETURNING id",
"describe": {
"columns": [
{
@@ -22,5 +22,5 @@
false
]
},
"hash": "07f5290e90533eac50b890a0d7f4a5e73ac111c838f687fe8647636827aae8b5"
"hash": "a9c805423e700b0acceb7c3dc43d1d3f9d4f56da25f588d281638e449d99a0d9"
}
@@ -16,8 +16,7 @@
"resource",
"variable",
"ducklake",
"datatable",
"volume"
"datatable"
]
}
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT policy FROM app WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "policy",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
false
]
},
"hash": "b12fba75788e44daefd9b3540a3aebe9167431aaa0a902b4558bc141c85ed825"
}
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO group_\n VALUES ($1, 'wm_deployers', 'Members can preserve the original author when deploying to this workspace')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar"
]
},
"nullable": []
},
"hash": "dda45bcc53e94659838e98b6b9e7a55be0e31aee3008d5190f09c1f15e5b47dd"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT on_behalf_of_email FROM script WHERE path = $1 AND workspace_id = $2 ORDER BY created_at DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "on_behalf_of_email",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "e1f43cb65201b4f0965a4e18f0c918ae51fee667472d0cc2796ffdba4138d2ee"
}
@@ -0,0 +1,23 @@
{
"db_name": "PostgreSQL",
"query": "SELECT on_behalf_of_email FROM script WHERE path = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "on_behalf_of_email",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
true
]
},
"hash": "e8d948274840699c5f7485ee4bc00b72c11bd226f99eade7e9a0da4605539283"
}
+1 -1
View File
@@ -1 +1 @@
cb25312072c15c0e9cc375ebc824d41995a52898
a4546264d41ce7122dfd127f4b17f724eb63c40a
@@ -0,0 +1,5 @@
INSERT INTO group_ (workspace_id, name, summary, extra_perms)
SELECT id, 'wm_deployers', 'Members can preserve the original author when deploying to this workspace', '{}'::jsonb
FROM workspace
WHERE NOT deleted
ON CONFLICT (workspace_id, name) DO UPDATE SET summary = EXCLUDED.summary;
+2
View File
@@ -447,6 +447,7 @@ def main():
deployment_message: None,
visible_to_runner_only: None,
on_behalf_of_email: None,
preserve_on_behalf_of: None,
ws_error_handler_muted: None,
})
.send()
@@ -508,6 +509,7 @@ def main():
policy: None,
deployment_message: None,
custom_path: None,
preserve_on_behalf_of: None,
})
.send()
.await
+186
View File
@@ -0,0 +1,186 @@
-- Fixture for preserve_on_behalf_of integration tests
-- Extends base.sql with a deployer user in the wm_deployers group
-- Include all base setup (workspace, admin user, etc.)
INSERT INTO workspace
(id, name, owner)
VALUES ('test-workspace', 'test-workspace', 'test-user')
ON CONFLICT DO NOTHING;
INSERT INTO usr(workspace_id, email, username, is_admin, role) VALUES
('test-workspace', 'test@windmill.dev', 'test-user', true, 'Admin')
ON CONFLICT DO NOTHING;
INSERT INTO workspace_key(workspace_id, kind, key) VALUES
('test-workspace', 'cloud', 'test-key')
ON CONFLICT DO NOTHING;
INSERT INTO workspace_settings (workspace_id) VALUES
('test-workspace')
ON CONFLICT DO NOTHING;
INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES
('test-workspace', 'all', 'All users', '{}')
ON CONFLICT DO NOTHING;
-- Create the wm_deployers group
INSERT INTO group_ (workspace_id, name, summary, extra_perms) VALUES
('test-workspace', 'wm_deployers', 'Users allowed to deploy and preserve on_behalf_of', '{}')
ON CONFLICT DO NOTHING;
INSERT INTO password(email, password_hash, login_type, super_admin, verified, name, username)
VALUES ('test@windmill.dev', 'not-a-real-hash', 'password', true, true, 'Test User', 'test-user')
ON CONFLICT DO NOTHING;
INSERT INTO password(email, password_hash, login_type, super_admin, verified, name)
VALUES ('test2@windmill.dev', 'not-a-real-hash', 'password', false, true, 'Test User 2')
ON CONFLICT DO NOTHING;
-- Deployer user (non-admin but in wm_deployers group)
INSERT INTO password(email, password_hash, login_type, super_admin, verified, name)
VALUES ('deployer@windmill.dev', 'not-a-real-hash', 'password', false, true, 'Deployer User')
ON CONFLICT DO NOTHING;
-- Original user whose on_behalf_of should be preserved
INSERT INTO password(email, password_hash, login_type, super_admin, verified, name)
VALUES ('original@windmill.dev', 'not-a-real-hash', 'password', false, true, 'Original User')
ON CONFLICT DO NOTHING;
INSERT INTO usr(workspace_id, email, username, is_admin, role) VALUES
('test-workspace', 'test2@windmill.dev', 'test-user-2', false, 'User')
ON CONFLICT DO NOTHING;
-- Deployer user in workspace
INSERT INTO usr(workspace_id, email, username, is_admin, role) VALUES
('test-workspace', 'deployer@windmill.dev', 'deployer-user', false, 'User')
ON CONFLICT DO NOTHING;
-- Original user in workspace (whose on_behalf_of should be preserved)
INSERT INTO usr(workspace_id, email, username, is_admin, role) VALUES
('test-workspace', 'original@windmill.dev', 'original-user', false, 'User')
ON CONFLICT DO NOTHING;
-- Add deployer user to wm_deployers group
INSERT INTO usr_to_group(workspace_id, group_, usr) VALUES
('test-workspace', 'wm_deployers', 'deployer-user')
ON CONFLICT DO NOTHING;
-- Tokens for all users
INSERT INTO token(token, email, label, super_admin) VALUES ('SECRET_TOKEN', 'test@windmill.dev', 'test token', true)
ON CONFLICT DO NOTHING;
INSERT INTO token(token, email, label, super_admin) VALUES ('SECRET_TOKEN_2', 'test2@windmill.dev', 'test token 2', false)
ON CONFLICT DO NOTHING;
INSERT INTO token(token, email, label, super_admin) VALUES ('DEPLOYER_TOKEN', 'deployer@windmill.dev', 'deployer token', false)
ON CONFLICT DO NOTHING;
INSERT INTO token(token, email, label, super_admin) VALUES ('ORIGINAL_TOKEN', 'original@windmill.dev', 'original token', false)
ON CONFLICT DO NOTHING;
GRANT ALL PRIVILEGES ON TABLE workspace_key TO windmill_admin;
GRANT ALL PRIVILEGES ON TABLE workspace_key TO windmill_user;
CREATE OR REPLACE FUNCTION "notify_insert_on_completed_job" ()
RETURNS TRIGGER AS $$
BEGIN
PERFORM pg_notify('completed', NEW.id::text);
RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;
DROP TRIGGER IF EXISTS "notify_insert_on_completed_job" ON "v2_job_completed";
CREATE TRIGGER "notify_insert_on_completed_job"
AFTER INSERT ON "v2_job_completed"
FOR EACH ROW
EXECUTE FUNCTION "notify_insert_on_completed_job" ();
CREATE OR REPLACE FUNCTION "notify_queue" ()
RETURNS TRIGGER AS $$
BEGIN
PERFORM pg_notify('queued', NEW.id::text);
RETURN NEW;
END;
$$ LANGUAGE PLPGSQL;
DROP TRIGGER IF EXISTS "notify_queue_after_insert" ON "v2_job_queue";
CREATE TRIGGER "notify_queue_after_insert"
AFTER INSERT ON "v2_job_queue"
FOR EACH ROW
EXECUTE FUNCTION "notify_queue" ();
DROP TRIGGER IF EXISTS "notify_queue_after_flow_status_update" ON "v2_job_status";
CREATE TRIGGER "notify_queue_after_flow_status_update"
AFTER UPDATE ON "v2_job_status"
FOR EACH ROW
WHEN (NEW.flow_status IS DISTINCT FROM OLD.flow_status)
EXECUTE FUNCTION "notify_queue" ();
-- Apply phase 4:
DROP FUNCTION IF EXISTS v2_job_after_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_completed_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_completed_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_after_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_queue_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_runtime_before_update CASCADE;
DROP FUNCTION IF EXISTS v2_job_status_before_insert CASCADE;
DROP FUNCTION IF EXISTS v2_job_status_before_update CASCADE;
DROP VIEW IF EXISTS completed_job, completed_job_view, job, queue, queue_view CASCADE;
ALTER TABLE v2_job_queue
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __last_ping CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __flow_status CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __same_worker CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __pre_run_error CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __mem_peak CASCADE,
DROP COLUMN IF EXISTS __root_job CASCADE,
DROP COLUMN IF EXISTS __leaf_jobs CASCADE,
DROP COLUMN IF EXISTS __concurrent_limit CASCADE,
DROP COLUMN IF EXISTS __concurrency_time_window_s CASCADE,
DROP COLUMN IF EXISTS __timeout CASCADE,
DROP COLUMN IF EXISTS __flow_step_id CASCADE,
DROP COLUMN IF EXISTS __cache_ttl CASCADE;
LOCK TABLE v2_job_queue IN ACCESS EXCLUSIVE MODE;
ALTER TABLE v2_job_completed
DROP COLUMN IF EXISTS __parent_job CASCADE,
DROP COLUMN IF EXISTS __created_by CASCADE,
DROP COLUMN IF EXISTS __created_at CASCADE,
DROP COLUMN IF EXISTS __success CASCADE,
DROP COLUMN IF EXISTS __script_hash CASCADE,
DROP COLUMN IF EXISTS __script_path CASCADE,
DROP COLUMN IF EXISTS __args CASCADE,
DROP COLUMN IF EXISTS __logs CASCADE,
DROP COLUMN IF EXISTS __raw_code CASCADE,
DROP COLUMN IF EXISTS __canceled CASCADE,
DROP COLUMN IF EXISTS __job_kind CASCADE,
DROP COLUMN IF EXISTS __env_id CASCADE,
DROP COLUMN IF EXISTS __schedule_path CASCADE,
DROP COLUMN IF EXISTS __permissioned_as CASCADE,
DROP COLUMN IF EXISTS __raw_flow CASCADE,
DROP COLUMN IF EXISTS __is_flow_step CASCADE,
DROP COLUMN IF EXISTS __language CASCADE,
DROP COLUMN IF EXISTS __is_skipped CASCADE,
DROP COLUMN IF EXISTS __raw_lock CASCADE,
DROP COLUMN IF EXISTS __email CASCADE,
DROP COLUMN IF EXISTS __visible_to_owner CASCADE,
DROP COLUMN IF EXISTS __tag CASCADE,
DROP COLUMN IF EXISTS __priority CASCADE;
File diff suppressed because it is too large Load Diff
+306
View File
@@ -0,0 +1,306 @@
//! Integration tests for workspace protection rulesets.
//!
//! Tests verify that DisableDirectDeployment protection rules correctly
//! block/allow operations based on user permissions.
use serde_json::json;
use sqlx::{Pool, Postgres};
use windmill_common::workspaces::invalidate_protection_rules_cache;
use windmill_test_utils::*;
fn client() -> reqwest::Client {
reqwest::Client::new()
}
fn authed(builder: reqwest::RequestBuilder, token: &str) -> reqwest::RequestBuilder {
builder.header("Authorization", format!("Bearer {}", token))
}
fn new_script(path: &str, summary: &str) -> serde_json::Value {
json!({
"path": path,
"summary": summary,
"description": "",
"content": "export async function main() { return 42; }",
"language": "deno",
"schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {},
"required": []
}
})
}
fn new_flow(path: &str, summary: &str) -> serde_json::Value {
json!({
"path": path,
"summary": summary,
"description": "",
"value": { "modules": [] },
"schema": {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"type": "object",
"properties": {},
"required": []
}
})
}
/// Comprehensive test for protection rules functionality.
/// Tests all essential cases in a single test to avoid cache interference.
#[sqlx::test(fixtures("base"))]
async fn test_protection_rules(db: Pool<Postgres>) -> anyhow::Result<()> {
initialize_tracing().await;
invalidate_protection_rules_cache("test-workspace");
let server = ApiServer::start(db.clone()).await?;
let port = server.addr.port();
let base = format!("http://localhost:{port}/api/w/test-workspace");
// ========================================
// 1. Without protection rule, non-admin can create scripts and flows
// ========================================
let resp = authed(
client().post(format!("{base}/scripts/create")),
"SECRET_TOKEN_2",
)
.json(&new_script("u/test-user-2/script_no_rule", "No rule"))
.send()
.await?;
assert_eq!(
resp.status(),
201,
"Should create script without rule: {}",
resp.text().await?
);
let resp = authed(
client().post(format!("{base}/flows/create")),
"SECRET_TOKEN_2",
)
.json(&new_flow("u/test-user-2/flow_no_rule", "No rule"))
.send()
.await?;
assert_eq!(
resp.status(),
201,
"Should create flow without rule: {}",
resp.text().await?
);
// ========================================
// 2. Non-admin cannot create protection rules
// ========================================
let resp = authed(
client().post(format!("{base}/workspaces/protection_rules")),
"SECRET_TOKEN_2",
)
.json(&json!({
"name": "unauthorized-rule",
"rules": ["DisableDirectDeployment"],
"bypass_users": [],
"bypass_groups": []
}))
.send()
.await?;
assert!(
!resp.status().is_success(),
"Non-admin should not create rules: {}",
resp.status()
);
// ========================================
// 3. Admin creates protection rule
// ========================================
let resp = authed(
client().post(format!("{base}/workspaces/protection_rules")),
"SECRET_TOKEN",
)
.json(&json!({
"name": "test-rule",
"rules": ["DisableDirectDeployment"],
"bypass_users": [],
"bypass_groups": []
}))
.send()
.await?;
assert_eq!(
resp.status(),
200,
"Admin should create rule: {}",
resp.text().await?
);
// ========================================
// 4. With rule, non-admin is blocked from creating scripts/flows
// ========================================
let resp = authed(
client().post(format!("{base}/scripts/create")),
"SECRET_TOKEN_2",
)
.json(&new_script("u/test-user-2/blocked_script", "Blocked"))
.send()
.await?;
assert!(
!resp.status().is_success(),
"Non-admin should be blocked from scripts: {}",
resp.status()
);
let body = resp.text().await?;
assert!(
body.contains("blocked") || body.contains("Blocked"),
"Error should mention blocking: {}",
body
);
let resp = authed(
client().post(format!("{base}/flows/create")),
"SECRET_TOKEN_2",
)
.json(&new_flow("u/test-user-2/blocked_flow", "Blocked"))
.send()
.await?;
assert!(
!resp.status().is_success(),
"Non-admin should be blocked from flows: {}",
resp.status()
);
// ========================================
// 5. Admin bypasses protection rule
// ========================================
let resp = authed(
client().post(format!("{base}/scripts/create")),
"SECRET_TOKEN",
)
.json(&new_script("u/test-user/admin_script", "Admin"))
.send()
.await?;
assert_eq!(
resp.status(),
201,
"Admin should bypass rule: {}",
resp.text().await?
);
// ========================================
// 6. Update rule to bypass test-user-2
// ========================================
let resp = authed(
client().post(format!("{base}/workspaces/protection_rules/test-rule")),
"SECRET_TOKEN",
)
.json(&json!({
"rules": ["DisableDirectDeployment"],
"bypass_users": ["test-user-2"],
"bypass_groups": []
}))
.send()
.await?;
assert_eq!(
resp.status(),
200,
"Should update rule: {}",
resp.text().await?
);
// Invalidate cache to pick up the update
invalidate_protection_rules_cache("test-workspace");
// ========================================
// 7. Bypassed user can now create
// ========================================
let resp = authed(
client().post(format!("{base}/scripts/create")),
"SECRET_TOKEN_2",
)
.json(&new_script("u/test-user-2/bypassed_script", "Bypassed"))
.send()
.await?;
assert_eq!(
resp.status(),
201,
"Bypassed user should create: {}",
resp.text().await?
);
// ========================================
// 8. Non-bypassed user (test-user-3) is still blocked
// ========================================
let resp = authed(
client().post(format!("{base}/scripts/create")),
"SECRET_TOKEN_3",
)
.json(&new_script("u/test-user-3/still_blocked", "Blocked"))
.send()
.await?;
assert!(
!resp.status().is_success(),
"Non-bypassed user should be blocked: {}",
resp.status()
);
// ========================================
// 9. Delete rule
// ========================================
let resp = authed(
client().delete(format!("{base}/workspaces/protection_rules/test-rule")),
"SECRET_TOKEN",
)
.send()
.await?;
assert_eq!(
resp.status(),
200,
"Should delete rule: {}",
resp.text().await?
);
// Invalidate cache to pick up the deletion
invalidate_protection_rules_cache("test-workspace");
// ========================================
// 10. After deletion, non-admin can create again
// ========================================
let resp = authed(
client().post(format!("{base}/scripts/create")),
"SECRET_TOKEN_3",
)
.json(&new_script("u/test-user-3/after_delete", "After delete"))
.send()
.await?;
assert_eq!(
resp.status(),
201,
"Should create after rule deletion: {}",
resp.text().await?
);
// ========================================
// 11. Verify rule list is empty
// ========================================
let resp = authed(
client().get(format!("{base}/workspaces/protection_rules")),
"SECRET_TOKEN",
)
.send()
.await?;
assert_eq!(resp.status(), 200);
let rules: Vec<serde_json::Value> = resp.json().await?;
assert!(rules.is_empty(), "Should have no rules after deletion");
Ok(())
}
+57 -3
View File
@@ -503,7 +503,11 @@ async fn create_flow(
nf.tag,
nf.dedicated_worker,
nf.visible_to_runner_only.unwrap_or(false),
nf.on_behalf_of_email.and(Some(&authed.email)),
windmill_common::resolve_on_behalf_of_email(
nf.on_behalf_of_email.as_deref(),
nf.preserve_on_behalf_of.unwrap_or(false),
&authed,
),
nf.ws_error_handler_muted.unwrap_or(false),
sqlx::types::Json(&nf.value) as _,
schema_str,
@@ -513,7 +517,7 @@ async fn create_flow(
.await?;
let version = sqlx::query_scalar!(
"INSERT INTO flow_version (workspace_id, path, value, schema, created_by)
"INSERT INTO flow_version (workspace_id, path, value, schema, created_by)
VALUES ($1, $2, $3, $4::text::json, $5)
RETURNING id",
w_id,
@@ -555,6 +559,29 @@ async fn create_flow(
),
)
.await?;
if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
nf.on_behalf_of_email.as_deref(),
nf.preserve_on_behalf_of.unwrap_or(false),
&authed,
&authed.email,
) {
audit_log(
&mut *tx,
&authed,
"flows.on_behalf_of",
ActionKind::Create,
&w_id,
Some(&nf.path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "create"),
]
.into(),
),
)
.await?;
}
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = nf.deployment_message {
@@ -940,7 +967,11 @@ async fn update_flow(
nf.tag,
nf.dedicated_worker,
nf.visible_to_runner_only.unwrap_or(false),
nf.on_behalf_of_email.and(Some(&authed.email)),
windmill_common::resolve_on_behalf_of_email(
nf.on_behalf_of_email.as_deref(),
nf.preserve_on_behalf_of.unwrap_or(false),
&authed,
),
nf.ws_error_handler_muted.unwrap_or(false),
sqlx::types::Json(&nf.value) as _,
schema_str,
@@ -1103,6 +1134,29 @@ async fn update_flow(
),
)
.await?;
if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
nf.on_behalf_of_email.as_deref(),
nf.preserve_on_behalf_of.unwrap_or(false),
&authed,
&authed.email,
) {
audit_log(
&mut *tx,
&authed,
"flows.on_behalf_of",
ActionKind::Update,
&w_id,
Some(&nf.path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "update"),
]
.into(),
),
)
.await?;
}
webhook.send_message(
w_id.clone(),
+103 -4
View File
@@ -21,6 +21,7 @@ use windmill_audit::audit_oss::audit_log;
use windmill_audit::ActionKind;
use windmill_common::DB;
use windmill_common::{
can_preserve_on_behalf_of,
db::UserDB,
error::{Error, JsonResult, Result},
schedule::Schedule,
@@ -30,6 +31,45 @@ use windmill_common::{
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
use windmill_queue::schedule::push_scheduled_job;
/// Resolves the email to use for a schedule based on preservation settings.
/// When preserving, looks up the email from the provided username.
async fn resolve_email(
username: Option<&String>,
preserve_email: Option<bool>,
authed: &ApiAuthed,
db: &DB,
w_id: &str,
) -> Result<String> {
if let Some(username) = username {
if preserve_email.unwrap_or(false) && can_preserve_on_behalf_of(authed) {
let email = sqlx::query_scalar!(
"SELECT email FROM usr WHERE username = $1 AND workspace_id = $2",
username,
w_id
)
.fetch_optional(db)
.await?;
if let Some(email) = email {
return Ok(email);
}
}
}
Ok(authed.email.clone())
}
fn resolve_edited_by(
username: Option<&String>,
preserve_edited_by: Option<bool>,
authed: &ApiAuthed,
) -> String {
if let Some(username) = username {
if preserve_edited_by.unwrap_or(false) && can_preserve_on_behalf_of(authed) {
return username.clone();
}
}
authed.username.clone()
}
pub fn workspaced_service() -> Router {
Router::new()
.route("/list", get(list_schedule))
@@ -75,6 +115,8 @@ pub struct NewSchedule {
pub paused_until: Option<DateTime<Utc>>,
pub cron_version: Option<String>,
pub dynamic_skip: Option<String>,
pub email: Option<String>,
pub preserve_email: Option<bool>,
}
#[derive(Serialize, Deserialize)]
@@ -200,6 +242,8 @@ async fn create_schedule(
validate_dynamic_skip(&mut tx, &w_id, handler_path).await?;
}
let resolved_edited_by = resolve_edited_by(ns.email.as_ref(), ns.preserve_email, &authed);
let schedule = sqlx::query_as!(
Schedule,
r#"
@@ -257,13 +301,13 @@ async fn create_schedule(
ns.path,
ns.schedule,
ns.timezone,
authed.username,
resolved_edited_by,
ns.script_path,
ns.is_flow,
to_json_raw_opt(ns.args.as_ref())
as Option<sqlx::types::Json<Box<serde_json::value::RawValue>>>,
ns.enabled.unwrap_or(false),
authed.email,
resolve_email(ns.email.as_ref(), ns.preserve_email, &authed, &db, &w_id).await?,
ns.on_failure,
ns.on_failure_times,
ns.on_failure_exact,
@@ -308,6 +352,29 @@ async fn create_schedule(
),
)
.await?;
if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
ns.email.as_deref(),
ns.preserve_email.unwrap_or(false),
&authed,
&authed.username,
) {
audit_log(
&mut *tx,
&authed,
"schedule.on_behalf_of",
ActionKind::Create,
&w_id,
Some(&ns.path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "create"),
]
.into(),
),
)
.await?;
}
if ns.enabled.unwrap_or(true) {
tx = push_scheduled_job(&db, tx, &schedule, Some(&authed.clone().into()), None).await?
@@ -351,6 +418,9 @@ async fn edit_schedule(
}
clear_schedule(&mut tx, path, &w_id).await?;
let resolved_edited_by = resolve_edited_by(es.email.as_ref(), es.preserve_email, &authed);
let schedule = sqlx::query_as!(
Schedule,
r#"
@@ -377,7 +447,9 @@ async fn edit_schedule(
workspace_id = $20,
cron_version = COALESCE($21, cron_version),
description = $22,
dynamic_skip = $23
dynamic_skip = $23,
email = COALESCE($24, email),
edited_by = $25
WHERE path = $19 AND workspace_id = $20
RETURNING
workspace_id,
@@ -438,7 +510,9 @@ async fn edit_schedule(
w_id,
es.cron_version,
es.description,
es.dynamic_skip
es.dynamic_skip,
Some(resolve_email(es.email.as_ref(), es.preserve_email, &authed, &db, &w_id).await?),
resolved_edited_by
)
.fetch_one(&mut *tx)
.await
@@ -459,6 +533,29 @@ async fn edit_schedule(
),
)
.await?;
if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
es.email.as_deref(),
es.preserve_email.unwrap_or(false),
&authed,
&authed.username,
) {
audit_log(
&mut *tx,
&authed,
"schedule.on_behalf_of",
ActionKind::Update,
&w_id,
Some(path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "update"),
]
.into(),
),
)
.await?;
}
if schedule.enabled {
tx = push_scheduled_job(&db, tx, &schedule, None, None).await?;
@@ -1073,6 +1170,8 @@ pub struct EditSchedule {
pub paused_until: Option<DateTime<Utc>>,
pub cron_version: Option<String>,
pub dynamic_skip: Option<String>,
pub email: Option<String>,
pub preserve_email: Option<bool>,
}
pub use windmill_queue::schedule::clear_schedule;
+51 -5
View File
@@ -927,11 +927,11 @@ async fn create_script_internal<'c>(
no_main_func.filter(|x: &bool| *x), // should be Some(true) or None
codebase,
has_preprocessor.filter(|x: &bool| *x), // should be Some(true) or None
if ns.on_behalf_of_email.is_some() {
Some(&authed.email)
} else {
None
},
windmill_common::resolve_on_behalf_of_email(
ns.on_behalf_of_email.as_deref(),
ns.preserve_on_behalf_of.unwrap_or(false),
&authed,
),
validate_schema,
ns.assets.as_ref().and_then(|a| serde_json::to_value(a).ok()),
guarded_debounce_key,
@@ -1027,6 +1027,29 @@ async fn create_script_internal<'c>(
Some([("hash", hash.to_string().as_str())].into()),
)
.await?;
if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
ns.on_behalf_of_email.as_deref(),
ns.preserve_on_behalf_of.unwrap_or(false),
&authed,
&authed.email,
) {
audit_log(
&mut *tx,
&authed,
"scripts.on_behalf_of",
ActionKind::Update,
&w_id,
Some(&ns.path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "update"),
]
.into(),
),
)
.await?;
}
webhook.send_message(
w_id.clone(),
WebhookMessage::UpdateScript {
@@ -1052,6 +1075,29 @@ async fn create_script_internal<'c>(
),
)
.await?;
if let Some(on_behalf_of) = windmill_common::check_on_behalf_of_preservation(
ns.on_behalf_of_email.as_deref(),
ns.preserve_on_behalf_of.unwrap_or(false),
&authed,
&authed.email,
) {
audit_log(
&mut *tx,
&authed,
"scripts.on_behalf_of",
ActionKind::Create,
&w_id,
Some(&ns.path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "create"),
]
.into(),
),
)
.await?;
}
webhook.send_message(
w_id.clone(),
WebhookMessage::CreateScript {
@@ -2788,6 +2788,14 @@ async fn create_workspace(
.execute(&mut *tx)
.await?;
sqlx::query!(
"INSERT INTO group_
VALUES ($1, 'wm_deployers', 'Members can preserve the original author when deploying to this workspace')",
nw.id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"INSERT INTO usr_to_group
VALUES ($1, 'all', $2)",
+132
View File
@@ -8215,6 +8215,9 @@ paths:
type: string
custom_path:
type: string
preserve_on_behalf_of:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original on_behalf_of value in the policy instead of overwriting it."
required:
- path
- value
@@ -8260,6 +8263,9 @@ paths:
type: string
custom_path:
type: string
preserve_on_behalf_of:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original on_behalf_of value in the policy instead of overwriting it."
required:
- path
- value
@@ -8571,6 +8577,9 @@ paths:
type: string
custom_path:
type: string
preserve_on_behalf_of:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original on_behalf_of value in the policy instead of overwriting it."
responses:
"200":
description: app updated
@@ -8610,6 +8619,9 @@ paths:
type: string
custom_path:
type: string
preserve_on_behalf_of:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original on_behalf_of value in the policy instead of overwriting it."
js:
type: string
css:
@@ -18497,6 +18509,9 @@ components:
type: boolean
on_behalf_of_email:
type: string
preserve_on_behalf_of:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original on_behalf_of_email value instead of overwriting it."
assets:
type: array
items:
@@ -20151,6 +20166,12 @@ components:
type: string
nullable: true
description: Path to a script that validates scheduled datetimes. Receives scheduled_for datetime and returns boolean to skip (true) or run (false)
email:
type: string
description: Email of the user who the scheduled jobs run as. Used during deployment to preserve the original schedule owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- schedule
@@ -20238,6 +20259,12 @@ components:
type: string
nullable: true
description: Path to a script that validates scheduled datetimes. Receives scheduled_for datetime and returns boolean to skip (true) or run (false)
email:
type: string
description: Email of the user who the scheduled jobs run as. Used during deployment to preserve the original schedule owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- schedule
- timezone
@@ -20594,6 +20621,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
@@ -20680,6 +20713,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -20842,6 +20881,12 @@ components:
retry:
description: Retry configuration for failed executions
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
@@ -20904,6 +20949,12 @@ components:
retry:
description: Retry configuration for failed executions
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
@@ -21073,6 +21124,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -21127,6 +21184,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -21263,6 +21326,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: "Retry configuration for failed executions."
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -21425,6 +21494,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- queue_url
- aws_resource_path
@@ -21471,6 +21546,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- queue_url
- aws_resource_path
@@ -21629,6 +21710,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -21671,6 +21758,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -21778,6 +21871,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
@@ -21831,6 +21930,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
@@ -21932,6 +22037,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
@@ -21981,6 +22092,12 @@ components:
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -22029,6 +22146,12 @@ components:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
mode:
$ref: "#/components/schemas/TriggerMode"
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
@@ -22055,6 +22178,12 @@ components:
$ref: "#/components/schemas/ScriptArgs"
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
email:
type: string
description: Email of the user who triggered jobs run as. Used during deployment to preserve the original trigger owner.
preserve_email:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original email value instead of overwriting it."
required:
- path
- script_path
@@ -22474,6 +22603,9 @@ components:
type: boolean
on_behalf_of_email:
type: string
preserve_on_behalf_of:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original on_behalf_of_email value instead of overwriting it."
required:
- path
+69 -13
View File
@@ -39,8 +39,6 @@ use itertools::Itertools;
use lazy_static::lazy_static;
use magic_crypt::MagicCryptTrait;
#[cfg(feature = "parquet")]
use windmill_object_store::object_store_reexports::{Attribute, Attributes};
#[cfg(feature = "parquet")]
use regex::Regex;
use serde::{Deserialize, Serialize};
use serde_json::{json, value::RawValue};
@@ -67,6 +65,8 @@ use windmill_common::{
workspaces::{check_user_against_rule, ProtectionRuleKind, RuleCheckResult},
HUB_BASE_URL,
};
#[cfg(feature = "parquet")]
use windmill_object_store::object_store_reexports::{Attribute, Attributes};
use windmill_store::resources::get_resource_value_interpolated_internal;
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
@@ -75,11 +75,7 @@ use windmill_queue::{push, PushArgs, PushArgsOwned, PushIsolationLevel};
#[cfg(feature = "parquet")]
use hmac::Mac;
#[cfg(feature = "parquet")]
use windmill_common::{
jwt,
oauth2::HmacSha256,
variables::get_workspace_key,
};
use windmill_common::{jwt, oauth2::HmacSha256, variables::get_workspace_key};
#[cfg(feature = "parquet")]
use windmill_types::s3::{S3Object, S3Permission};
@@ -279,6 +275,7 @@ pub struct CreateApp {
pub draft_only: Option<bool>,
pub deployment_message: Option<String>,
pub custom_path: Option<String>,
pub preserve_on_behalf_of: Option<bool>,
}
#[derive(Serialize, Deserialize)]
@@ -289,6 +286,7 @@ pub struct EditApp {
pub policy: Option<Policy>,
pub deployment_message: Option<String>,
pub custom_path: Option<String>,
pub preserve_on_behalf_of: Option<bool>,
}
#[derive(Serialize, FromRow)]
@@ -443,7 +441,9 @@ async fn get_raw_app_data(
if let Some(os) = object_store {
let path = format!("/app_bundles/{}/{}.{}", w_id, id, file_type);
let stream = os
.get(&windmill_object_store::object_store_reexports::Path::from(path))
.get(&windmill_object_store::object_store_reexports::Path::from(
path,
))
.await
.map_err(windmill_object_store::object_store_error_to_error)?
.bytes()
@@ -958,7 +958,10 @@ async fn store_raw_app_file<'a>(
if let Some(os) = object_store {
if let Err(e) = os
.put(&windmill_object_store::object_store_reexports::Path::from(path.clone()), data.into())
.put(
&windmill_object_store::object_store_reexports::Path::from(path.clone()),
data.into(),
)
.await
{
tracing::error!("Failed to put snapshot to s3 at {path}: {:?}", e);
@@ -1178,8 +1181,14 @@ async fn create_app_internal<'a>(
}
}
let mut tx = user_db.clone().begin(&authed).await?;
app.policy.on_behalf_of = Some(username_to_permissioned_as(&authed.username));
app.policy.on_behalf_of_email = Some(authed.email.clone());
let should_preserve = app.preserve_on_behalf_of.unwrap_or(false)
&& windmill_common::can_preserve_on_behalf_of(&authed)
&& app.policy.on_behalf_of.is_some();
if !should_preserve {
app.policy.on_behalf_of = Some(username_to_permissioned_as(&authed.username));
app.policy.on_behalf_of_email = Some(authed.email.clone());
}
let path = app.path.clone();
if &app.path == "" {
return Err(Error::BadRequest("App path cannot be empty".to_string()));
@@ -1270,6 +1279,22 @@ async fn create_app_internal<'a>(
None,
)
.await?;
if should_preserve {
if let Some(ref obo_email) = app.policy.on_behalf_of_email {
if obo_email != &authed.email {
audit_log(
&mut *tx,
&authed,
"apps.on_behalf_of",
ActionKind::Create,
w_id,
Some(&app.path),
Some([("on_behalf_of", obo_email.as_str()), ("action", "create")].into()),
)
.await?;
}
}
}
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = &app.deployment_message {
args.insert("deployment_message".to_string(), to_raw_value(&dm));
@@ -1599,6 +1624,7 @@ async fn update_app_internal<'a>(
use sql_builder::prelude::*;
let mut tx = user_db.clone().begin(&authed).await?;
let mut preserved_on_behalf_of: Option<String> = None;
let npath = if ns.policy.is_some()
|| ns.path.is_some()
|| ns.summary.is_some()
@@ -1664,8 +1690,20 @@ async fn update_app_internal<'a>(
}
if let Some(mut npolicy) = ns.policy {
npolicy.on_behalf_of = Some(username_to_permissioned_as(&authed.username));
npolicy.on_behalf_of_email = Some(authed.email.clone());
let should_preserve = ns.preserve_on_behalf_of.unwrap_or(false)
&& windmill_common::can_preserve_on_behalf_of(&authed)
&& npolicy.on_behalf_of.is_some();
if should_preserve {
if let Some(ref obo_email) = npolicy.on_behalf_of_email {
if obo_email != &authed.email {
preserved_on_behalf_of = Some(obo_email.clone());
}
}
} else {
npolicy.on_behalf_of = Some(username_to_permissioned_as(&authed.username));
npolicy.on_behalf_of_email = Some(authed.email.clone());
}
sqlb.set(
"policy",
quote(serde_json::to_string(&json!(npolicy)).map_err(|e| {
@@ -1747,6 +1785,24 @@ async fn update_app_internal<'a>(
None,
)
.await?;
if let Some(on_behalf_of) = preserved_on_behalf_of {
audit_log(
&mut *tx,
&authed,
"apps.on_behalf_of",
ActionKind::Update,
w_id,
Some(&npath),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "update"),
]
.into(),
),
)
.await?;
}
let tx = PushIsolationLevel::Transaction(tx);
let mut args: HashMap<String, Box<serde_json::value::RawValue>> = HashMap::new();
if let Some(dm) = ns.deployment_message {
+62 -27
View File
@@ -179,13 +179,12 @@ pub async fn benchmark_verify(benchmark_jobs: i32, db: &DB) {
let canceled = row.canceled.unwrap_or(0);
let total = succeeded + failed + canceled;
let remaining_in_queue = sqlx::query_scalar!(
"SELECT COUNT(*) FROM v2_job_queue WHERE workspace_id = 'admins'",
)
.fetch_one(db)
.await
.expect("benchmark verify queue query failed")
.unwrap_or(0);
let remaining_in_queue =
sqlx::query_scalar!("SELECT COUNT(*) FROM v2_job_queue WHERE workspace_id = 'admins'",)
.fetch_one(db)
.await
.expect("benchmark verify queue query failed")
.unwrap_or(0);
println!("=== BENCHMARK VERIFICATION ===");
println!(" kind: {benchmark_kind}");
@@ -248,10 +247,12 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
.execute(db)
.await
.unwrap_or_else(|e| panic!("failed to clean up job_perms: {e:#}"));
sqlx::query!("DELETE FROM concurrency_key WHERE key LIKE 'bench_%' OR key LIKE 'u/admin/bench_%'")
.execute(db)
.await
.unwrap_or_else(|e| panic!("failed to clean up concurrency_key: {e:#}"));
sqlx::query!(
"DELETE FROM concurrency_key WHERE key LIKE 'bench_%' OR key LIKE 'u/admin/bench_%'"
)
.execute(db)
.await
.unwrap_or_else(|e| panic!("failed to clean up concurrency_key: {e:#}"));
sqlx::query!("DELETE FROM concurrency_counter WHERE concurrency_id LIKE 'bench_%' OR concurrency_id LIKE 'u/admin/bench_%'")
.execute(db)
.await
@@ -637,9 +638,13 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &noop_uuids, "admins", "deno")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed noop queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &noop_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&noop_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed noop runtime"));
.await
.unwrap_or_else(|_e| panic!("failed to insert mixed noop runtime"));
// 2) sequentialflow jobs
if portion > 0 {
@@ -661,9 +666,13 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sf_uuids, "admins", "flow")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed sequentialflow queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sf_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&sf_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed sequentialflow runtime"));
.await
.unwrap_or_else(|_e| panic!("failed to insert mixed sequentialflow runtime"));
sqlx::query!(
"INSERT INTO v2_job_status (id, flow_status) SELECT unnest($1::uuid[]), $2",
&sf_uuids,
@@ -693,9 +702,13 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sl_uuids, "admins", "deno")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed scriptlogs queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sl_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&sl_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed scriptlogs runtime"));
.await
.unwrap_or_else(|_e| panic!("failed to insert mixed scriptlogs runtime"));
}
// 4) concurrencylimit jobs
@@ -720,9 +733,13 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &cl_uuids, "admins", "deno")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &cl_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&cl_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit runtime"));
.await
.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencylimit runtime"));
let cl_concurrency_id = "u/admin/bench_conclimit";
sqlx::query!(
"INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb) ON CONFLICT (concurrency_id) DO NOTHING",
@@ -763,9 +780,13 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &ck_uuids, "admins", "deno")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &ck_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&ck_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey runtime"));
.await
.unwrap_or_else(|_e| panic!("failed to insert mixed concurrencykey runtime"));
let ck_concurrency_id = "bench_shared_concurrency_key";
sqlx::query!(
"INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, '{}'::jsonb) ON CONFLICT (concurrency_id) DO NOTHING",
@@ -807,9 +828,13 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &noop_uuids, "admins", "deno")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc noop queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &noop_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&noop_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc noop runtime"));
.await
.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc noop runtime"));
// 2) sequentialflow jobs
if portion > 0 {
@@ -831,9 +856,15 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sf_uuids, "admins", "flow")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc sequentialflow queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sf_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&sf_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc sequentialflow runtime"));
.await
.unwrap_or_else(|_e| {
panic!("failed to insert mixed_no_cc sequentialflow runtime")
});
sqlx::query!(
"INSERT INTO v2_job_status (id, flow_status) SELECT unnest($1::uuid[]), $2",
&sf_uuids,
@@ -863,9 +894,13 @@ pub async fn benchmark_init(benchmark_jobs: i32, db: &DB) {
sqlx::query!("INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, tag) SELECT unnest($1::uuid[]), $2, now(), $3", &sl_uuids, "admins", "deno")
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc scriptlogs queue"));
sqlx::query!("INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])", &sl_uuids)
sqlx::query!(
"INSERT INTO v2_job_runtime (id) SELECT unnest($1::uuid[])",
&sl_uuids
)
.execute(&mut *tx)
.await.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc scriptlogs runtime"));
.await
.unwrap_or_else(|_e| panic!("failed to insert mixed_no_cc scriptlogs runtime"));
}
}
"none" => {}
+44
View File
@@ -109,6 +109,50 @@ pub const DEFAULT_MAX_CONNECTIONS_INDEXER: u32 = 5;
pub const DEFAULT_HUB_BASE_URL: &str = "https://hub.windmill.dev";
pub const PRIVATE_HUB_MIN_VERSION: i32 = 10_000_000;
pub const SERVICE_LOG_RETENTION_SECS: i64 = 60 * 60 * 24 * 14; // 2 weeks retention period for logs
pub const WM_DEPLOYERS_GROUP: &str = "wm_deployers";
/// Checks if the user is allowed to preserve on_behalf_of values (admin or deployer).
pub fn can_preserve_on_behalf_of(authed: &impl db::Authable) -> bool {
authed.is_admin() || authed.groups().iter().any(|g| g == &WM_DEPLOYERS_GROUP)
}
/// Checks if on-behalf-of preservation actually happened (the target user differs from the acting user).
/// Returns Some(target_identifier) if preservation occurred, None otherwise.
pub fn check_on_behalf_of_preservation(
on_behalf_of_identifier: Option<&str>,
preserve: bool,
authed: &impl db::Authable,
authed_identifier: &str,
) -> Option<String> {
if preserve && can_preserve_on_behalf_of(authed) {
if let Some(id) = on_behalf_of_identifier {
if id != authed_identifier {
return Some(id.to_string());
}
}
}
None
}
/// Determines the on_behalf_of_email value to use when creating/updating a flow or script.
/// - If `on_behalf_of_email` is None, returns None
/// - If `preserve` is true and the user is admin or in the deployers group, returns the original value
/// - Otherwise, returns the authenticated user's email
pub fn resolve_on_behalf_of_email<'a>(
on_behalf_of_email: Option<&'a str>,
preserve: bool,
authed: &'a impl db::Authable,
) -> Option<&'a str> {
if on_behalf_of_email.is_some() {
if preserve && can_preserve_on_behalf_of(authed) {
on_behalf_of_email
} else {
Some(authed.email())
}
} else {
None
}
}
#[macro_export]
macro_rules! add_time {
@@ -52,7 +52,10 @@ fn extract_assets_from_raw_value(
if prefix {
let s = serde_json::from_str::<String>(value.get()).ok()?;
let (kind, path) = parse_asset_syntax(&s, false)?;
assets.push(RuntimeAsset { path: path.to_string(), kind: crate::assets::asset_kind_from_parser(kind) });
assets.push(RuntimeAsset {
path: path.to_string(),
kind: crate::assets::asset_kind_from_parser(kind),
});
}
}
None
+6 -13
View File
@@ -49,9 +49,7 @@ pub fn extract_workspace_dependencies_annotated_refs(
Some(&RE_PYTHON),
runnable_path,
),
Go => {
WorkspaceDependenciesAnnotatedRefs::parse("//", "go_mod", code, None, runnable_path)
}
Go => WorkspaceDependenciesAnnotatedRefs::parse("//", "go_mod", code, None, runnable_path),
Php => WorkspaceDependenciesAnnotatedRefs::parse(
"//",
"composer_json",
@@ -67,11 +65,8 @@ pub async fn prefetch_cached_script(
script: Script<ScriptRunnableSettingsHandle>,
db: &DB,
) -> crate::error::Result<Script<ScriptRunnableSettingsInline>> {
let rs = runnable_settings::from_handle(
script.runnable_settings.runnable_settings_handle,
db,
)
.await?;
let rs = runnable_settings::from_handle(script.runnable_settings.runnable_settings_handle, db)
.await?;
let (debouncing_settings, concurrency_settings) =
runnable_settings::prefetch_cached(&rs, db).await?;
@@ -379,11 +374,8 @@ pub async fn clone_script<'c>(
)));
};
let rs = runnable_settings::from_handle(
s.runnable_settings.runnable_settings_handle,
db,
)
.await?;
let rs =
runnable_settings::from_handle(s.runnable_settings.runnable_settings_handle, db).await?;
let (debouncing_settings, concurrency_settings) =
runnable_settings::prefetch_cached(&rs, db).await?;
@@ -424,6 +416,7 @@ pub async fn clone_script<'c>(
codebase: s.codebase,
has_preprocessor: s.has_preprocessor,
on_behalf_of_email: s.on_behalf_of_email,
preserve_on_behalf_of: None,
assets: s.assets,
};
@@ -384,8 +384,11 @@ impl WorkspaceDependenciesPrefetched {
Box::pin(async {
let r = if let Some(wdar) =
crate::scripts::extract_workspace_dependencies_annotated_refs(&language, code, runnable_path)
{
crate::scripts::extract_workspace_dependencies_annotated_refs(
&language,
code,
runnable_path,
) {
tracing::debug!(workspace_id, ?language, "found explicit annotations");
let expanded = wdar
+23 -10
View File
@@ -138,6 +138,7 @@ fn check_no_duplicates(
pub async fn insert_new_trigger_into_db(
authed: &ApiAuthed,
db: &DB,
tx: &mut PgConnection,
w_id: &str,
trigger: &TriggerData<HttpConfigRequest>,
@@ -146,6 +147,8 @@ pub async fn insert_new_trigger_into_db(
require_admin(authed.is_admin, &authed.username)?;
let request_type = trigger.config.request_type;
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, w_id).await?;
sqlx::query!(
r#"
@@ -196,8 +199,8 @@ pub async fn insert_new_trigger_into_db(
trigger.config.authentication_method as _,
trigger.config.http_method as _,
trigger.config.static_asset_config as _,
&authed.username,
&authed.email,
&resolved_edited_by,
resolved_email,
trigger.config.is_static_website,
trigger.error_handling.error_handler_path,
trigger.error_handling.error_handler_args as _,
@@ -247,9 +250,16 @@ pub async fn create_many_http_triggers(
let mut tx = user_db.begin(&authed).await?;
for (new_http_trigger, route_path_key) in new_http_triggers.iter().zip(route_path_keys.iter()) {
insert_new_trigger_into_db(&authed, &mut tx, &w_id, new_http_trigger, route_path_key)
.await
.map_err(|err| error_wrapper(&new_http_trigger.config.route_path, err))?;
insert_new_trigger_into_db(
&authed,
&db,
&mut tx,
&w_id,
new_http_trigger,
route_path_key,
)
.await
.map_err(|err| error_wrapper(&new_http_trigger.config.route_path, err))?;
audit_log(
&mut *tx,
@@ -403,7 +413,7 @@ impl TriggerCrud for HttpTrigger {
) -> Result<()> {
let route_path_key = check_if_route_exist(db, &trigger.config, &w_id, None).await?;
insert_new_trigger_into_db(authed, tx, w_id, &trigger, &route_path_key).await?;
insert_new_trigger_into_db(authed, db, tx, w_id, &trigger, &route_path_key).await?;
increase_trigger_version(tx).await?;
@@ -419,6 +429,9 @@ impl TriggerCrud for HttpTrigger {
path: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, workspace_id).await?;
if authed.is_admin {
if trigger.config.route_path.is_empty() {
return Err(Error::BadRequest("route_path is required".to_string()));
@@ -478,8 +491,8 @@ impl TriggerCrud for HttpTrigger {
trigger.base.mode() as _,
trigger.config.http_method as _,
trigger.config.static_asset_config as _,
&authed.username,
&authed.email,
&resolved_edited_by,
resolved_email,
request_type as _,
trigger.config.authentication_method as _,
trigger.config.summary,
@@ -534,8 +547,8 @@ impl TriggerCrud for HttpTrigger {
trigger.base.mode() as _,
trigger.config.http_method as _,
trigger.config.static_asset_config as _,
&authed.username,
&authed.email,
&resolved_edited_by,
resolved_email,
request_type as _,
trigger.config.authentication_method as _,
trigger.config.summary,
+10 -6
View File
@@ -66,12 +66,14 @@ impl TriggerCrud for MqttTrigger {
async fn create_trigger(
&self,
_db: &DB,
db: &DB,
tx: &mut PgConnection,
authed: &ApiAuthed,
w_id: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, w_id).await?;
let subscribe_topics = trigger
.config
.subscribe_topics
@@ -114,9 +116,9 @@ impl TriggerCrud for MqttTrigger {
trigger.base.path,
trigger.base.script_path,
trigger.base.is_flow,
authed.email,
resolved_email,
trigger.base.mode() as _,
authed.username,
&resolved_edited_by,
trigger.error_handling.error_handler_path,
trigger.error_handling.error_handler_args as _,
trigger.error_handling.retry as _
@@ -129,13 +131,15 @@ impl TriggerCrud for MqttTrigger {
async fn update_trigger(
&self,
_db: &DB,
db: &DB,
tx: &mut PgConnection,
authed: &ApiAuthed,
workspace_id: &str,
path: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, workspace_id).await?;
let subscribe_topics = trigger
.config
.subscribe_topics
@@ -179,8 +183,8 @@ impl TriggerCrud for MqttTrigger {
v3_config as Option<SqlxJson<MqttV3Config>>,
v5_config as Option<SqlxJson<MqttV5Config>>,
trigger.base.is_flow,
authed.username,
authed.email,
&resolved_edited_by,
resolved_email,
trigger.base.script_path,
trigger.base.path,
workspace_id,
@@ -70,6 +70,8 @@ impl TriggerCrud for PostgresTrigger {
w_id: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, w_id).await?;
let Self::TriggerConfigRequest {
postgres_resource_path,
publication_name,
@@ -137,8 +139,8 @@ impl TriggerCrud for PostgresTrigger {
trigger.base.script_path,
trigger.base.is_flow,
trigger.base.mode() as _,
authed.username,
authed.email,
&resolved_edited_by,
resolved_email,
trigger.error_handling.error_handler_path,
trigger.error_handling.error_handler_args as _,
trigger.error_handling.retry as _
@@ -157,6 +159,8 @@ impl TriggerCrud for PostgresTrigger {
path: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, w_id).await?;
let Self::TriggerConfigRequest {
replication_slot_name,
publication_name,
@@ -235,8 +239,8 @@ impl TriggerCrud for PostgresTrigger {
trigger.base.script_path,
trigger.base.path,
trigger.base.is_flow,
authed.username,
authed.email,
&resolved_edited_by,
resolved_email,
w_id,
path,
trigger.error_handling.error_handler_path,
@@ -72,12 +72,14 @@ impl TriggerCrud for WebsocketTrigger {
async fn create_trigger(
&self,
_db: &DB,
db: &DB,
tx: &mut PgConnection,
authed: &ApiAuthed,
w_id: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, w_id).await?;
let filters = trigger
.config
.filters
@@ -127,10 +129,10 @@ impl TriggerCrud for WebsocketTrigger {
.config
.url_runnable_args
.map(|v| SqlxJson(serde_json::value::to_raw_value(&v).unwrap())) as _,
authed.username,
&resolved_edited_by,
trigger.config.can_return_message,
trigger.config.can_return_error_result,
authed.email,
resolved_email,
trigger.error_handling.error_handler_path,
trigger.error_handling.error_handler_args as _,
trigger.error_handling.retry as _
@@ -142,13 +144,15 @@ impl TriggerCrud for WebsocketTrigger {
async fn update_trigger(
&self,
_db: &DB,
db: &DB,
tx: &mut PgConnection,
authed: &ApiAuthed,
w_id: &str,
path: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_email = trigger.base.resolve_email(authed, db, w_id).await?;
let filters = trigger
.config
.filters
@@ -200,8 +204,8 @@ impl TriggerCrud for WebsocketTrigger {
.url_runnable_args
.map(|v| SqlxJson(serde_json::value::to_raw_value(&v).unwrap()))
as Option<SqlxJson<Box<RawValue>>>,
&authed.username,
&authed.email,
&resolved_edited_by,
resolved_email,
trigger.config.can_return_message,
trigger.config.can_return_error_result,
w_id,
+48
View File
@@ -416,6 +416,12 @@ async fn create_trigger<T: TriggerCrud>(
let mut tx = user_db.begin(&authed).await?;
let new_path = new_trigger.base.path.clone();
let on_behalf_of_info = windmill_common::check_on_behalf_of_preservation(
new_trigger.base.email.as_deref(),
new_trigger.base.preserve_email.unwrap_or(false),
&authed,
&authed.username,
);
handler
.create_trigger(&db, &mut *tx, &authed, &workspace_id, new_trigger)
@@ -431,6 +437,24 @@ async fn create_trigger<T: TriggerCrud>(
None,
)
.await?;
if let Some(on_behalf_of) = on_behalf_of_info {
audit_log(
&mut *tx,
&authed,
&format!("{}_triggers.on_behalf_of", T::TRIGGER_TYPE),
ActionKind::Create,
&workspace_id,
Some(&new_path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "create"),
]
.into(),
),
)
.await?;
}
handle_deployment_metadata(
&authed.email,
@@ -510,6 +534,12 @@ async fn update_trigger<T: TriggerCrud>(
let mut tx = user_db.begin(&authed).await?;
let new_path = edit_trigger.base.path.to_string();
let on_behalf_of_info = windmill_common::check_on_behalf_of_preservation(
edit_trigger.base.email.as_deref(),
edit_trigger.base.preserve_email.unwrap_or(false),
&authed,
&authed.username,
);
handler
.update_trigger(&db, &mut *tx, &authed, &workspace_id, path, edit_trigger)
@@ -525,6 +555,24 @@ async fn update_trigger<T: TriggerCrud>(
None,
)
.await?;
if let Some(on_behalf_of) = on_behalf_of_info {
audit_log(
&mut *tx,
&authed,
&format!("{}_triggers.on_behalf_of", T::TRIGGER_TYPE),
ActionKind::Update,
&workspace_id,
Some(&new_path),
Some(
[
("on_behalf_of", on_behalf_of.as_str()),
("action", "update"),
]
.into(),
),
)
.await?;
}
let parent_path = if path != new_path {
Some(path.to_string())
+58 -11
View File
@@ -8,9 +8,9 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use sqlx::{types::Json as SqlxJson, FromRow};
use sqlx::{types::Json as SqlxJson, FromRow, Pool, Postgres};
use std::{collections::HashMap, fmt::Debug};
use windmill_common::jobs::JobTriggerKind;
use windmill_common::{db::Authable, error::Result, jobs::JobTriggerKind};
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
@@ -103,6 +103,12 @@ pub struct BaseTriggerData {
#[deprecated(note = "Use mode instead")]
enabled: Option<bool>, // Kept for backwards compatibility, use mode instead
mode: Option<TriggerMode>,
/// Optional email for deployment - when set, the trigger will run jobs as this user
#[serde(skip_serializing_if = "Option::is_none")]
pub email: Option<String>,
/// If true and user is admin/wm_deployers, preserve the provided email instead of using deploying user's email
#[serde(skip_serializing_if = "Option::is_none")]
pub preserve_email: Option<bool>,
}
impl BaseTriggerData {
@@ -116,6 +122,42 @@ impl BaseTriggerData {
},
)
}
pub async fn resolve_email(
&self,
authed: &impl Authable,
db: &Pool<Postgres>,
w_id: &str,
) -> Result<String> {
if let Some(ref username) = self.email {
if self.preserve_email.unwrap_or(false)
&& windmill_common::can_preserve_on_behalf_of(authed)
{
let email = sqlx::query_scalar!(
"SELECT email FROM usr WHERE username = $1 AND workspace_id = $2",
username,
w_id
)
.fetch_optional(db)
.await?;
if let Some(email) = email {
return Ok(email);
}
}
}
Ok(authed.email().to_string())
}
pub fn resolve_edited_by(&self, authed: &impl Authable) -> String {
if let Some(ref username) = self.email {
if self.preserve_email.unwrap_or(false)
&& windmill_common::can_preserve_on_behalf_of(authed)
{
return username.clone();
}
}
authed.username().to_string()
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -166,9 +208,18 @@ mod tests {
#[test]
fn test_trigger_mode_serialize() {
assert_eq!(serde_json::to_value(TriggerMode::Enabled).unwrap(), json!("enabled"));
assert_eq!(serde_json::to_value(TriggerMode::Disabled).unwrap(), json!("disabled"));
assert_eq!(serde_json::to_value(TriggerMode::Suspended).unwrap(), json!("suspended"));
assert_eq!(
serde_json::to_value(TriggerMode::Enabled).unwrap(),
json!("enabled")
);
assert_eq!(
serde_json::to_value(TriggerMode::Disabled).unwrap(),
json!("disabled")
);
assert_eq!(
serde_json::to_value(TriggerMode::Suspended).unwrap(),
json!("suspended")
);
}
#[test]
@@ -181,7 +232,7 @@ mod tests {
#[test]
fn test_trigger_mode_invalid() {
let result: Result<TriggerMode, _> = serde_json::from_value(json!("paused"));
let result: std::result::Result<TriggerMode, _> = serde_json::from_value(json!("paused"));
assert!(result.is_err());
}
@@ -314,11 +365,7 @@ mod tests {
#[test]
fn test_server_state_skip_none_fields() {
let state = ServerState {
server_id: None,
last_server_ping: None,
error: None,
};
let state = ServerState { server_id: None, last_server_ping: None, error: None };
let json = serde_json::to_value(&state).unwrap();
assert!(!json.as_object().unwrap().contains_key("server_id"));
assert!(!json.as_object().unwrap().contains_key("error"));
+1
View File
@@ -98,6 +98,7 @@ pub struct NewFlow {
pub deployment_message: Option<String>,
pub visible_to_runner_only: Option<bool>,
pub on_behalf_of_email: Option<String>,
pub preserve_on_behalf_of: Option<bool>,
pub ws_error_handler_muted: Option<bool>,
}
+1
View File
@@ -490,6 +490,7 @@ pub struct NewScript {
pub codebase: Option<String>,
pub has_preprocessor: Option<bool>,
pub on_behalf_of_email: Option<String>,
pub preserve_on_behalf_of: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
pub assets: Option<Vec<AssetWithAltAccessType>>,
}
File diff suppressed because it is too large Load Diff
+266 -457
View File
@@ -1,39 +1,43 @@
<script lang="ts">
import { createEventDispatcher, untrack } from 'svelte'
import { base } from '$lib/base'
import { enterpriseLicense, superadmin, workspaceStore } from '$lib/stores'
import { enterpriseLicense, superadmin, userStore, workspaceStore } from '$lib/stores'
import {
AppService,
FlowService,
FolderService,
ResourceService,
ScheduleService,
ScriptService,
UserService,
VariableService,
WorkspaceService
} from '$lib/gen'
import { getAllModules } from './flows/flowExplorer'
import Button from './common/button/Button.svelte'
import Tooltip from './Tooltip.svelte'
import Alert from './common/alert/Alert.svelte'
import Toggle from './Toggle.svelte'
import { Loader2 } from 'lucide-svelte'
import { DiffIcon, Loader2 } from 'lucide-svelte'
import Badge from './common/badge/Badge.svelte'
import DiffDrawer from './DiffDrawer.svelte'
import {
existsTrigger,
getTriggerDependency,
getTriggersDeployData,
getTriggerValue,
type AdditionalInformation,
type Kind
} from '$lib/utils_deployable'
import type { TriggerKind } from './triggers'
import {
checkItemExists,
deployItem,
getItemValue,
getOnBehalfOfEmail
} from '$lib/utils_workspace_deploy'
import type { App } from './apps/types'
import { getAllGridItems } from './apps/editor/appUtils'
import { isRunnableByPath } from './apps/inputType'
import type { Runnable } from './raw_apps/utils'
import WorkspaceDeployLayout from './WorkspaceDeployLayout.svelte'
import OnBehalfOfSelector, {
needsOnBehalfOfSelection,
type OnBehalfOfChoice
} from './OnBehalfOfSelector.svelte'
import ParentWorkspaceProtectionAlert from './ParentWorkspaceProtectionAlert.svelte'
const dispatch = createEventDispatcher()
@@ -52,13 +56,14 @@
additionalInformation = undefined,
workspaceToDeployTo = $bindable(undefined),
hideButton = false,
canDeployToWorkspace = $bindable(false)
canDeployToWorkspace = $bindable(true)
}: Props = $props()
let canSeeTarget: 'yes' | 'cant-deploy-to-workspace' | 'cant-see-all-deps' | undefined =
$state(undefined)
let dependencies: { kind: Kind; path: string; include: boolean }[] | undefined = $state(undefined)
type Dependency = { kind: Kind; path: string; include: boolean }
let dependencies: Dependency[] | undefined = $state<Dependency[] | undefined>(undefined)
const allAlreadyExists: { [key: string]: boolean } = $state({})
@@ -66,14 +71,49 @@
let notSet: boolean | undefined = $state(undefined)
let isFlow: boolean | undefined = $state(undefined)
// On-behalf-of tracking for flows and scripts
// Source workspace on_behalf_of emails (keyed by kind:path)
let sourceOnBehalfOfInfo = $state<Record<string, string | undefined>>({})
// Target workspace on_behalf_of emails (keyed by kind:path)
let targetOnBehalfOfInfo = $state<Record<string, string | undefined>>({})
let onBehalfOfChoice = $state<Record<string, OnBehalfOfChoice>>({})
let canPreserveOnBehalfOf = $state(false)
// Check if an item needs on_behalf_of selection (more than 1 unique option)
function itemNeedsOnBehalfOfSelection(statusPath: string, kind: string): boolean {
const myIdentity = kind === 'trigger' ? $userStore?.username : $userStore?.email
return needsOnBehalfOfSelection(
kind,
sourceOnBehalfOfInfo[statusPath],
targetOnBehalfOfInfo[statusPath],
myIdentity
)
}
// Get the email to use for deployment based on user's choice
function getOnBehalfOfEmailForDeploy(statusPath: string): string | undefined {
const choice = onBehalfOfChoice[statusPath]
if (choice === 'source') return sourceOnBehalfOfInfo[statusPath]
if (choice === 'target') return targetOnBehalfOfInfo[statusPath]
// 'me' or undefined = don't pass, backend will use deploying user's email
return undefined
}
async function reload(path: string) {
try {
if (!$superadmin) {
await UserService.whoami({ workspace: workspaceToDeployTo! })
const targetUser = await UserService.whoami({ workspace: workspaceToDeployTo! })
canPreserveOnBehalfOf =
targetUser.is_admin ||
targetUser.groups?.includes('wm_deployers') ||
false
} else {
canPreserveOnBehalfOf = true
}
canSeeTarget = 'yes'
} catch {
canSeeTarget = 'cant-deploy-to-workspace'
canPreserveOnBehalfOf = false
return
}
@@ -105,6 +145,33 @@
x.kind != 'resource_type' &&
(x.kind != 'folder' || !allAlreadyExists[computeStatusPath(x.kind, x.path)])
}))
// Fetch on_behalf_of_email for flows, scripts, apps, and triggers from both workspaces
for (const dep of sortedSet.filter((d) =>
['flow', 'script', 'app', 'trigger'].includes(d.kind)
)) {
const key = computeStatusPath(dep.kind, dep.path)
try {
sourceOnBehalfOfInfo[key] = await getOnBehalfOfEmail(
dep.kind,
dep.path,
$workspaceStore!,
additionalInformation
)
} catch {
sourceOnBehalfOfInfo[key] = undefined
}
try {
targetOnBehalfOfInfo[key] = await getOnBehalfOfEmail(
dep.kind,
dep.path,
workspaceToDeployTo!,
additionalInformation
)
} catch {
targetOnBehalfOfInfo[key] = undefined
}
}
}
async function getDependencies(
@@ -217,78 +284,7 @@
}
async function checkAlreadyExists(kind: Kind, path: string): Promise<boolean> {
let exists: boolean
if (kind == 'flow') {
exists = await FlowService.existsFlowByPath({
workspace: workspaceToDeployTo!,
path: path
})
} else if (kind == 'script') {
exists = await ScriptService.existsScriptByPath({
workspace: workspaceToDeployTo!,
path: path
})
} else if (kind == 'app') {
exists = await AppService.existsApp({
workspace: workspaceToDeployTo!,
path: path
})
} else if (kind == 'variable') {
exists = await VariableService.existsVariable({
workspace: workspaceToDeployTo!,
path: path
})
} else if (kind == 'resource') {
exists = await ResourceService.existsResource({
workspace: workspaceToDeployTo!,
path: path
})
} else if (kind == 'schedule') {
exists = await ScheduleService.existsSchedule({
workspace: workspaceToDeployTo!,
path: path
})
} else if (kind == 'resource_type') {
exists = await ResourceService.existsResourceType({
workspace: workspaceToDeployTo!,
path: path
})
} else if (kind == 'folder') {
exists = await FolderService.existsFolder({
workspace: workspaceToDeployTo!,
name: path
})
} else if (kind === 'trigger') {
const triggersKind: TriggerKind[] = [
'kafka',
'mqtt',
'nats',
'postgres',
'routes',
'schedules',
'sqs',
'websockets',
'gcp'
]
if (
additionalInformation?.triggers &&
triggersKind.includes(additionalInformation.triggers.kind)
) {
exists = await existsTrigger(
{ workspace: workspaceToDeployTo!, path },
additionalInformation.triggers.kind
)
} else {
throw new Error(
`Unexpected triggers kind, expected one of: '${triggersKind.join(', ')}' got: ${
additionalInformation?.triggers?.kind
}`
)
}
} else {
throw new Error(`Unknown kind ${kind}`)
}
return exists
return checkItemExists(kind, path, workspaceToDeployTo!, additionalInformation)
}
const deploymentStatus: Record<
@@ -297,240 +293,23 @@
> = $state({})
async function deploy(kind: Kind, path: string) {
const statusPath = `${kind}:${path}`
const statusPath = computeStatusPath(kind, path)
deploymentStatus[statusPath] = { status: 'loading' }
try {
let alreadyExists = await checkAlreadyExists(kind, path)
if (kind == 'flow') {
const flow = await FlowService.getFlowByPath({
workspace: $workspaceStore!,
path: path
})
getAllModules(flow.value.modules).forEach((x) => {
if (x.value.type == 'script' && x.value.hash != undefined) {
x.value.hash = undefined
}
})
if (alreadyExists) {
await FlowService.updateFlow({
workspace: workspaceToDeployTo!,
path: path,
requestBody: {
...flow
}
})
} else {
await FlowService.createFlow({
workspace: workspaceToDeployTo!,
requestBody: {
...flow
}
})
}
} else if (kind == 'script') {
const script = await ScriptService.getScriptByPath({
workspace: $workspaceStore!,
path: path
})
await ScriptService.createScript({
workspace: workspaceToDeployTo!,
requestBody: {
...script,
lock: script.lock,
parent_hash: alreadyExists
? (
await ScriptService.getScriptByPath({
workspace: workspaceToDeployTo!,
path: path
})
).hash
: undefined
}
})
} else if (kind == 'app') {
const app = await AppService.getAppByPath({
workspace: $workspaceStore!,
path: path
})
if (alreadyExists) {
if (app.raw_app) {
const secret = await AppService.getPublicSecretOfLatestVersionOfApp({
workspace: $workspaceStore!,
path: app.path
})
const js = await AppService.getRawAppData({
secretWithExtension: `${secret}.js`,
workspace: $workspaceStore!
})
const css = await AppService.getRawAppData({
secretWithExtension: `${secret}.css`,
workspace: $workspaceStore!
})
await AppService.updateAppRaw({
workspace: workspaceToDeployTo!,
path: path,
formData: {
app,
css,
js
}
})
} else {
await AppService.updateApp({
workspace: workspaceToDeployTo!,
path: path,
requestBody: {
...app
}
})
}
} else {
if (app.raw_app) {
const secret = await AppService.getPublicSecretOfLatestVersionOfApp({
workspace: $workspaceStore!,
path: app.path
})
const js = await AppService.getRawAppData({
secretWithExtension: `${secret}.js`,
workspace: $workspaceStore!
})
const css = await AppService.getRawAppData({
secretWithExtension: `${secret}.css`,
workspace: $workspaceStore!
})
await AppService.createAppRaw({
workspace: workspaceToDeployTo!,
formData: {
app,
css,
js
}
})
} else {
await AppService.createApp({
workspace: workspaceToDeployTo!,
requestBody: {
...app
}
})
}
}
} else if (kind == 'variable') {
const variable = await VariableService.getVariable({
workspace: $workspaceStore!,
path: path,
decryptSecret: true
})
if (alreadyExists) {
await VariableService.updateVariable({
workspace: workspaceToDeployTo!,
path: path,
requestBody: {
path: path,
value: variable.value ?? '',
is_secret: variable.is_secret,
description: variable.description ?? ''
},
alreadyEncrypted: false
})
} else {
await VariableService.createVariable({
workspace: workspaceToDeployTo!,
requestBody: {
path: path,
value: variable.value ?? '',
is_secret: variable.is_secret,
description: variable.description ?? ''
}
})
}
} else if (kind == 'resource') {
const resource = await ResourceService.getResource({
workspace: $workspaceStore!,
path: path
})
if (alreadyExists) {
await ResourceService.updateResource({
workspace: workspaceToDeployTo!,
path: path,
requestBody: {
path: path,
value: resource.value ?? '',
description: resource.description ?? ''
}
})
} else {
await ResourceService.createResource({
workspace: workspaceToDeployTo!,
requestBody: {
path: path,
value: resource.value ?? '',
resource_type: resource.resource_type,
description: resource.description ?? ''
}
})
}
} else if (kind == 'resource_type') {
const resource = await ResourceService.getResourceType({
workspace: $workspaceStore!,
path: path
})
if (alreadyExists) {
await ResourceService.updateResourceType({
workspace: workspaceToDeployTo!,
path: path,
requestBody: {
schema: resource.schema,
description: resource.description ?? ''
}
})
} else {
await ResourceService.createResourceType({
workspace: workspaceToDeployTo!,
requestBody: {
description: resource.description ?? '',
schema: resource.schema,
name: resource.name
}
})
}
} else if (kind == 'folder') {
await FolderService.createFolder({
workspace: workspaceToDeployTo!,
requestBody: {
name: path
}
})
} else if (kind === 'trigger') {
if (additionalInformation?.triggers) {
const { data, createFn, updateFn } = await getTriggersDeployData(
additionalInformation.triggers.kind,
path,
$workspaceStore!
)
if (alreadyExists) {
await updateFn({
path,
workspace: workspaceToDeployTo!,
requestBody: data
} as any)
} else {
await createFn({
workspace: workspaceToDeployTo!,
requestBody: data
} as any)
}
} else {
throw new Error('Missing triggers kind')
}
} else {
throw new Error(`Unknown kind ${kind}`)
}
const result = await deployItem({
kind,
path,
workspaceFrom: $workspaceStore!,
workspaceTo: workspaceToDeployTo!,
additionalInformation,
onBehalfOfEmail: getOnBehalfOfEmailForDeploy(statusPath)
})
if (result.success) {
allAlreadyExists[statusPath] = true
deploymentStatus[statusPath] = { status: 'deployed' }
} catch (e) {
deploymentStatus[statusPath] = { status: 'failed', error: e.body || e.message }
} else {
deploymentStatus[statusPath] = { status: 'failed', error: result.error }
}
}
@@ -547,83 +326,11 @@
return `${kind}:${path}`
}
async function getValue(kind: Kind, path: string, workspace: string) {
try {
if (kind == 'flow') {
const flow = await FlowService.getFlowByPath({
workspace: workspace,
path: path
})
getAllModules(flow.value.modules).forEach((x) => {
if (x.value.type == 'script' && x.value.hash != undefined) {
x.value.hash = undefined
}
})
return { summary: flow.summary, description: flow.description, value: flow.value }
} else if (kind == 'script') {
const script = await ScriptService.getScriptByPath({
workspace: workspace,
path: path
})
return {
content: script.content,
lock: script.lock,
schema: script.schema,
summary: script.summary,
language: script.language
}
} else if (kind == 'app') {
const app = await AppService.getAppByPath({
workspace: workspace,
path: path
})
return app
} else if (kind == 'variable') {
const variable = await VariableService.getVariable({
workspace: workspace,
path: path,
decryptSecret: true
})
return variable.value
} else if (kind == 'resource') {
const resource = await ResourceService.getResource({
workspace: workspace,
path: path
})
return resource.value
} else if (kind == 'resource_type') {
const resource = await ResourceService.getResourceType({
workspace: workspace,
path: path
})
return resource.schema
} else if (kind == 'folder') {
const folder = await FolderService.getFolder({
workspace: workspace,
name: path
})
return {
name: folder.name
}
} else if (kind == 'trigger') {
if (additionalInformation?.triggers) {
return await getTriggerValue(additionalInformation.triggers.kind, path, workspace)
} else {
throw new Error(`Missing trigger information`)
}
} else {
throw new Error(`Unknown kind ${kind}`)
}
} catch {
return {}
}
}
async function showDiff(kind: Kind, path: string) {
diffDrawer?.openDrawer()
let values = await Promise.all([
getValue(kind, path, workspaceToDeployTo!),
getValue(kind, path, $workspaceStore!)
getItemValue(kind, path, workspaceToDeployTo!, additionalInformation),
getItemValue(kind, path, $workspaceStore!, additionalInformation)
])
diffDrawer?.setDiff({
mode: 'simple',
@@ -632,6 +339,7 @@
title: 'Staging/prod <> Dev'
})
}
$effect(() => {
WorkspaceService.getDeployTo({ workspace: $workspaceStore! }).then((x) => {
workspaceToDeployTo = x.deploy_to
@@ -640,9 +348,66 @@
}
})
})
$effect(() => {
workspaceToDeployTo && initialPath && untrack(() => reload(initialPath))
})
// Transform dependencies to deployable item format for the shared layout
let deployableItems = $derived(
(dependencies ?? []).map((dep) => ({
key: computeStatusPath(dep.kind, dep.path),
path: dep.path,
kind: dep.kind,
include: dep.include,
triggerKind: dep.kind === 'trigger' ? additionalInformation?.triggers?.kind : undefined
}))
)
let selectedItems = $derived<string[]>(
(dependencies ?? [])
.filter((dep) => dep.include)
.map((dep) => computeStatusPath(dep.kind, dep.path))
)
let allSelected = $derived(
dependencies != null && dependencies.length > 0 && dependencies.every((dep) => dep.include)
)
// Check if all required on_behalf_of selections are made
let hasUnselectedOnBehalfOf = $derived(
selectedItems.some((statusPath) => {
const dep = dependencies?.find((d) => computeStatusPath(d.kind, d.path) === statusPath)
if (!dep) return false
return (
itemNeedsOnBehalfOfSelection(statusPath, dep.kind) &&
onBehalfOfChoice[statusPath] === undefined
)
})
)
function toggleItem(item: { key: string }) {
if (dependencies) {
const idx = dependencies.findIndex(
(dep) => computeStatusPath(dep.kind, dep.path) === item.key
)
if (idx !== -1) {
dependencies[idx].include = !dependencies[idx].include
}
}
}
function selectAll() {
if (dependencies) {
dependencies = dependencies.map((dep) => ({ ...dep, include: true }))
}
}
function deselectAll() {
if (dependencies) {
dependencies = dependencies.map((dep) => ({ ...dep, include: false }))
}
}
</script>
<div class="mt-6"></div>
@@ -668,6 +433,15 @@
>
<input class="max-w-xs" type="text" disabled value={workspaceToDeployTo} />
{#if workspaceToDeployTo}
<ParentWorkspaceProtectionAlert
parentWorkspaceId={workspaceToDeployTo}
onUpdateCanDeploy={(canDeploy) => {
canDeployToWorkspace = canDeploy
}}
/>
{/if}
{#if canSeeTarget == undefined}
<div class="mt-6"></div>
<Loader2 class="animate-spin" />
@@ -675,79 +449,114 @@
<h3 class="mb-6 mt-16">All related deployable items</h3>
<DiffDrawer bind:this={diffDrawer} {isFlow} />
<div class="grid grid-cols-9 justify-center max-w-3xl gap-2">
{#each dependencies ?? [] as { kind, path, include }, i}
{@const statusPath = computeStatusPath(kind, path)}
<div class="col-span-1 truncate text-secondary text-sm pt-0.5">{kind}</div><div
class="col-span-5 truncate font-semibold">{path}</div
><div class="col-span-1 pt-1.5">
<Toggle
size="xs"
checked={include}
on:change={(e) => {
if (dependencies?.[i]) {
dependencies[i].include = e.detail
}
}}
<WorkspaceDeployLayout
items={deployableItems}
{selectedItems}
{deploymentStatus}
selectablePredicate={() => true}
{allSelected}
onToggleItem={toggleItem}
onSelectAll={selectAll}
onDeselectAll={deselectAll}
emptyMessage="No deployable items found"
>
{#snippet itemActions(item)}
{@const statusPath = item.key}
{@const exists = allAlreadyExists[statusPath]}
{@const status = deploymentStatus[statusPath]}
{@const sourceEmail = sourceOnBehalfOfInfo[statusPath]}
{@const targetEmail = targetOnBehalfOfInfo[statusPath]}
<!-- On-behalf-of selector -->
{#if itemNeedsOnBehalfOfSelection(statusPath, item.kind)}
<OnBehalfOfSelector
{sourceEmail}
{targetEmail}
selected={onBehalfOfChoice[statusPath]}
onSelect={(choice) => (onBehalfOfChoice[statusPath] = choice)}
kind={item.kind}
canPreserve={canPreserveOnBehalfOf}
/>
</div>
<div class="col-span-1">
{#if allAlreadyExists[statusPath] == false}
{#if include}
<Badge
>New <Tooltip
>This {kind} doesn't exist yet on the target and will be created by the deployment</Tooltip
></Badge
>
{:else}
<Badge color="red">
Missing
<Tooltip
>{#if kind == 'resource_type'}
Resource types are not re-deployed by default. We strongly recommend to add
shared resource types in 'admin' workspace, which will have them be shared to
every workspace.
{:else}
This {kind} doesn't exist and is not included in the deployment. Variables and Resources
are considered to be workspace specific and are never included by default.
{/if}</Tooltip
>
</Badge>
{/if}
{:else if allAlreadyExists[statusPath] == true}
<button
class="text-blue-600 font-normal mt-1"
onclick={() => {
showDiff(kind, path)
isFlow = kind === 'flow'
}}>diff</button
{/if}
{#if exists === false}
{#if item.include}
<Badge
>New <Tooltip
>This {item.kind} doesn't exist yet on the target and will be created by the deployment</Tooltip
></Badge
>
{/if}</div
>
<div class="col-span-1 pr-1">
{#if deploymentStatus[statusPath]}
{#if deploymentStatus[statusPath].status == 'loading'}
<Loader2 class="animate-spin" />
{:else if deploymentStatus[statusPath].status == 'deployed'}
<Badge color="green">Deployed</Badge>
{:else if deploymentStatus[statusPath].status == 'failed'}
<div class="inline-flex gap-1">
<Badge color="red">Failed</Badge>
<Tooltip>{deploymentStatus[statusPath].error}</Tooltip></div
>
{/if}
{:else}
<Button color="light" size="xs" disabled={!canDeployToWorkspace} on:click={() => deploy(kind, path)}>Deploy</Button>
<Badge color="red">
Missing
<Tooltip
>{#if item.kind == 'resource_type'}
Resource types are not re-deployed by default. We strongly recommend to add shared
resource types in 'admin' workspace, which will have them be shared to every
workspace.
{:else}
This {item.kind} doesn't exist and is not included in the deployment. Variables and
Resources are considered to be workspace specific and are never included by default.
{/if}</Tooltip
>
</Badge>
{/if}
{:else if exists === true && !status}
<Button
size="xs"
variant="subtle"
onclick={() => {
showDiff(item.kind, item.path)
isFlow = item.kind === 'flow'
}}
>
<DiffIcon class="w-3 h-3" />
Show diff
</Button>
{/if}
{#if !status}
<Button
color="light"
size="xs"
disabled={!canDeployToWorkspace ||
(itemNeedsOnBehalfOfSelection(statusPath, item.kind) &&
onBehalfOfChoice[statusPath] === undefined)}
onclick={() => deploy(item.kind, item.path)}>Deploy</Button
>
{/if}
{/snippet}
{#snippet footer()}
<div class="flex flex-col items-end gap-2">
{#if !hideButton}
<Button on:click={deployAll} disabled={!canDeployToWorkspace || hasUnselectedOnBehalfOf}
>Deploy all toggled</Button
>
{/if}
{#if hasUnselectedOnBehalfOf}
<span class="text-xs text-yellow-600">
{#if kind === 'trigger'}
You must set the "edited by" user for all triggers before deploying
<Tooltip class="text-yellow-600">
The "edited by" field defines which user's permissions will be applied
when the trigger runs. Make sure this is set to an appropriate user
before deploying.
</Tooltip>
{:else}
You must set the "on behalf of" user for all items before deploying
<Tooltip class="text-yellow-600">
The "run on behalf of" field defines which user's permissions will be
applied during execution. Make sure this is set to an appropriate user
before deploying.
</Tooltip>
{/if}
</span>
{/if}
</div>
{/each}
</div>
{#if !hideButton}
<div class="mt-16 flex flex-row-reverse max-w-3xl"
><Button on:click={deployAll} disabled={!canDeployToWorkspace}>Deploy all toggled</Button></div
>
{/if}
{/snippet}
</WorkspaceDeployLayout>
{:else if canSeeTarget == 'cant-see-all-deps'}
<div class="my-2"></div>
<Alert type="error" title="User doesn't have visibility over all dependencies"
@@ -2,7 +2,6 @@
import { Button, Drawer } from './common'
import DrawerContent from './common/drawer/DrawerContent.svelte'
import DeployWorkspace from './DeployWorkspace.svelte'
import ParentWorkspaceProtectionAlert from './ParentWorkspaceProtectionAlert.svelte'
import { type AdditionalInformation, type Kind } from '$lib/utils_deployable'
let initialPath: string | undefined = $state(undefined)
@@ -27,17 +26,9 @@
<Drawer bind:this={drawer} size="900px">
<DrawerContent title="Deploy {initialPath}" on:close={drawer.closeDrawer}>
{#if workspaceToDeployTo}
<ParentWorkspaceProtectionAlert
parentWorkspaceId={workspaceToDeployTo}
onUpdateCanDeploy={(canDeploy) => {
canDeployToWorkspace = canDeploy
}}
/>
{/if}
{#if (kind != 'trigger' && kind != undefined && initialPath != undefined) || (kind === 'trigger' && initialPath != undefined && additionalInformation?.triggers != undefined)}
<DeployWorkspace
hideButton
hideButton={false}
{initialPath}
{kind}
{additionalInformation}
+4 -2
View File
@@ -225,7 +225,7 @@
lockChanges = true
replaceFlow(event.data)
timeout && clearTimeout(timeout)
timeout = setTimeout(() => {
timeout = window.setTimeout(() => {
lockChanges = false
}, 500)
} else if (event.data.type == 'error') {
@@ -539,7 +539,9 @@
}),
currentEditor: writable(undefined),
modulesTestStates,
outputPickerOpenFns
outputPickerOpenFns,
preserveOnBehalfOf: writable(false),
savedOnBehalfOfEmail: writable<string | undefined>(undefined)
})
setContext<PropPickerContext>('PropPickerContext', {
flowPropPickerConfig: writable<FlowPropPickerConfig | undefined>(undefined),
+10 -2
View File
@@ -131,6 +131,10 @@
let initialPathStore = writable(initialPath)
// For preserve_on_behalf_of feature
let preserveOnBehalfOf = writable(false)
let savedOnBehalfOfEmail = writable<string | undefined>(savedFlow?.on_behalf_of_email)
// used for new flows for captures
let fakeInitialPath =
'u/' +
@@ -481,6 +485,7 @@
dedicated_worker: flow.dedicated_worker,
visible_to_runner_only: flow.visible_to_runner_only,
on_behalf_of_email: flow.on_behalf_of_email,
preserve_on_behalf_of: $preserveOnBehalfOf || undefined,
deployment_message: deploymentMsg || undefined
}
})
@@ -533,6 +538,7 @@
ws_error_handler_muted: flow.ws_error_handler_muted,
visible_to_runner_only: flow.visible_to_runner_only,
on_behalf_of_email: flow.on_behalf_of_email,
preserve_on_behalf_of: $preserveOnBehalfOf || undefined,
deployment_message: deploymentMsg || undefined
}
})
@@ -560,7 +566,7 @@
function saveSessionDraft() {
timeout && clearTimeout(timeout)
timeout = setTimeout(() => {
timeout = window.setTimeout(() => {
try {
localStorage.setItem(
initialPath && initialPath != '' ? `flow-${initialPath}` : 'flow',
@@ -641,7 +647,9 @@
executionCount: writable(0),
flowInputEditorState: flowInputEditorStateStore,
modulesTestStates,
outputPickerOpenFns
outputPickerOpenFns,
preserveOnBehalfOf,
savedOnBehalfOfEmail
})
// Set up NoteEditor context for note editing capabilities
@@ -21,6 +21,7 @@
import TextInput from './text_input/TextInput.svelte'
import { Trash } from 'lucide-svelte'
import PermissionHistory from './PermissionHistory.svelte'
import Alert from './common/alert/Alert.svelte'
interface Props {
name: string
@@ -116,6 +117,13 @@
</script>
<div class="flex flex-col gap-6">
{#if name === 'wm_deployers'}
<Alert type="info" title="Deployer permissions">
Members of this group can preserve the original author (on_behalf_of / edited_by) when
deploying scripts, flows, apps, and triggers to this workspace. Without this permission,
deployed items will be reassigned to the deploying user.
</Alert>
{/if}
<Label label="Summary" for="summary">
<div class="flex flex-row gap-2">
<TextInput
@@ -0,0 +1,94 @@
<script lang="ts" module>
export type OnBehalfOfChoice = 'source' | 'target' | 'me' | undefined
/**
* Check if an item needs on_behalf_of selection (more than 1 unique option available)
*/
export function needsOnBehalfOfSelection(
kind: string,
sourceEmail: string | undefined,
targetEmail: string | undefined,
myEmail: string | undefined
): boolean {
if (kind !== 'flow' && kind !== 'script' && kind !== 'app' && kind !== 'trigger') return false
// Don't show if no on_behalf_of is set in source
if (!sourceEmail) return false
// Count unique options: source, target (even if undefined counts as different), me
const options = new Set([sourceEmail, myEmail])
// Target is a unique option if it differs from source (including undefined != defined)
if (targetEmail !== sourceEmail) {
options.add(targetEmail ?? '__not_set__')
}
// Show if more than 1 unique option
return options.size > 1
}
</script>
<script lang="ts">
import { Check, UserCog } from 'lucide-svelte'
import MeltPopover from './meltComponents/Popover.svelte'
import { userStore } from '$lib/stores'
interface Props {
sourceEmail: string | undefined
targetEmail: string | undefined
selected: OnBehalfOfChoice
onSelect: (choice: OnBehalfOfChoice) => void
kind: string
canPreserve: boolean
}
let { sourceEmail, targetEmail, selected, onSelect, kind, canPreserve }: Props = $props()
let label = $derived(
kind === 'trigger'
? 'Set the user this will be recorded as edited by:'
: 'Set the user this will be run on behalf of:'
)
</script>
<MeltPopover placement="bottom">
<svelte:fragment slot="trigger">
<UserCog class="w-4 h-4 {selected ? 'text-green-500' : 'text-yellow-500'}" />
</svelte:fragment>
<div slot="content" class="p-3 flex flex-col gap-2 min-w-48">
<div class="text-xs font-medium text-secondary mb-1">{label}</div>
<button
class="flex items-center gap-2 px-2 py-1.5 rounded text-left text-xs hover:bg-surface-hover {!canPreserve
? 'opacity-50 cursor-not-allowed'
: ''}"
disabled={!canPreserve}
onclick={() => onSelect('source')}
>
<Check class="w-3 h-3 {selected === 'source' ? 'opacity-100' : 'opacity-0'}" />
<span class="truncate max-w-40">{sourceEmail}</span>
<span class="text-xs text-tertiary">(source)</span>
</button>
<button
class="flex items-center gap-2 px-2 py-1.5 rounded text-left text-xs hover:bg-surface-hover {!canPreserve || !targetEmail
? 'opacity-50 cursor-not-allowed'
: ''}"
disabled={!canPreserve || !targetEmail}
onclick={() => onSelect('target')}
>
<Check class="w-3 h-3 {selected === 'target' ? 'opacity-100' : 'opacity-0'}" />
<span class="truncate max-w-40 {!targetEmail ? 'italic text-tertiary' : ''}"
>{targetEmail ?? 'unknown'}</span
>
<span class="text-xs text-tertiary">(target)</span>
</button>
<button
class="flex items-center gap-2 px-2 py-1.5 rounded text-left text-xs hover:bg-surface-hover"
onclick={() => onSelect('me')}
>
<Check class="w-3 h-3 {selected === 'me' ? 'opacity-100' : 'opacity-0'}" />
<span class="truncate max-w-40"
>{kind === 'trigger' ? $userStore?.username : $userStore?.email}</span
>
<span class="text-xs text-tertiary">(me)</span>
</button>
</div>
</MeltPopover>
@@ -157,6 +157,17 @@
let args: Record<string, any> = $state(initialArgs) // Test args input
let selectedInputTab: 'main' | 'preprocessor' = $state('main')
let hasPreprocessor = $state(false)
let preserveOnBehalfOf = $state(false)
const WM_DEPLOYERS_GROUP = 'wm_deployers'
let isDeployer = $derived($userStore?.groups?.includes(WM_DEPLOYERS_GROUP) ?? false)
let originalOnBehalfOfEmail = $derived(savedScript?.on_behalf_of_email)
let showPreserveToggle = $derived(
isDeployer &&
script.on_behalf_of_email &&
originalOnBehalfOfEmail &&
originalOnBehalfOfEmail !== $userStore?.email
)
let metadataOpen = $state(
!neverShowMeta &&
@@ -541,6 +552,7 @@
has_preprocessor: script.has_preprocessor,
deployment_message: deploymentMsg || undefined,
on_behalf_of_email: script.on_behalf_of_email,
preserve_on_behalf_of: preserveOnBehalfOf || undefined,
assets: script.assets
}
})
@@ -1591,6 +1603,7 @@
on:change={() => {
if (script.on_behalf_of_email) {
script.on_behalf_of_email = undefined
preserveOnBehalfOf = false
} else {
script.on_behalf_of_email = $userStore?.email
}
@@ -1599,6 +1612,15 @@
right: 'Run on behalf of last editor'
}}
/>
{#if showPreserveToggle}
<Toggle
size="sm"
bind:checked={preserveOnBehalfOf}
options={{
right: `Keep original author (${originalOnBehalfOfEmail})`
}}
/>
{/if}
</div>
</Section>
{#if !isCloudHosted()}
@@ -0,0 +1,156 @@
<script lang="ts">
import { Loader2 } from 'lucide-svelte'
import { Badge } from './common'
import Row from './common/table/Row.svelte'
import Tooltip from './Tooltip.svelte'
import type { Snippet } from 'svelte'
import type { Kind } from '$lib/utils_deployable'
interface DeployableItem {
key: string
path: string
kind: Kind
triggerKind?: string
[key: string]: unknown
}
interface Props {
items: DeployableItem[]
selectedItems: string[]
selectablePredicate?: (item: DeployableItem) => boolean
deploymentStatus: Record<string, { status: 'loading' | 'deployed' | 'failed'; error?: string }>
allSelected?: boolean
emptyMessage?: string
// Snippets for customization
header?: Snippet
alerts?: Snippet
itemSummary?: Snippet<[DeployableItem]>
itemActions?: Snippet<[DeployableItem]>
footer?: Snippet
// Callbacks
onToggleItem?: (item: DeployableItem) => void
onSelectAll?: () => void
onDeselectAll?: () => void
}
let {
items,
selectedItems,
selectablePredicate = () => true,
deploymentStatus,
allSelected = false,
emptyMessage = 'No items to deploy',
header,
alerts,
itemSummary,
itemActions,
footer,
onToggleItem,
onSelectAll,
onDeselectAll
}: Props = $props()
let selectableItems = $derived(items.filter(selectablePredicate))
let hasSelectableItems = $derived(selectableItems.length > 0)
</script>
<div class="flex flex-col h-full">
<!-- Header section -->
{#if header}
<div class="bg-surface">
{@render header()}
</div>
{/if}
<!-- Alerts section -->
{#if alerts}
{@render alerts()}
{/if}
{#if items.length > 0}
<!-- Select all row -->
<div class="px-4 py-2 flex items-center justify-between">
<div
class="flex items-center gap-2 text-secondary text-sm"
class:opacity-50={!hasSelectableItems}
>
<input
type="checkbox"
disabled={!hasSelectableItems}
checked={allSelected}
onchange={allSelected ? onDeselectAll : onSelectAll}
class="rounded max-w-4 w-full"
/> Select all
</div>
</div>
<!-- Items list -->
<div class="overflow-y-auto">
<div class="border rounded-md bg-surface-tertiary">
{#each items as item (item.key)}
{@const isSelectable = selectablePredicate(item)}
{@const isSelected = selectedItems.includes(item.key)}
{@const status = deploymentStatus[item.key]}
{@const isDeployed = status?.status === 'deployed'}
<Row
isSelectable={isSelectable && !isDeployed}
alignWithSelectable={true}
disabled={!isSelectable}
selected={isSelected && !isDeployed}
onSelect={() => onToggleItem?.(item)}
path={item.kind !== 'resource' &&
item.kind !== 'variable' &&
item.kind !== 'resource_type'
? item.path
: ''}
marked={undefined}
kind={item.kind}
triggerKind={item.triggerKind}
canFavorite={false}
workspaceId=""
>
{#snippet customSummary()}
{#if itemSummary}
{@render itemSummary(item)}
{:else}
{item.path}
{/if}
{/snippet}
{#snippet actions()}
{#if itemActions}
{@render itemActions(item)}
{/if}
<!-- Deployment status always shown -->
{#if status}
{#if status.status === 'loading'}
<Loader2 class="animate-spin" />
{:else if status.status === 'deployed'}
<Badge color="green">Deployed</Badge>
{:else if status.status === 'failed'}
<div class="inline-flex gap-1">
<Badge color="red">Failed</Badge>
<Tooltip>{status.error}</Tooltip>
</div>
{/if}
{/if}
{/snippet}
</Row>
{/each}
</div>
</div>
{:else}
<div class="flex items-center justify-center h-full">
<div class="text-gray-500">{emptyMessage}</div>
</div>
{/if}
<!-- Footer -->
{#if footer}
<div class="p-4 bg-surface">
{@render footer()}
</div>
{/if}
</div>
@@ -158,6 +158,7 @@
let debugAppDrawerOpen = $state(false)
let lazyDrawerOpen = $state(false)
let deploymentMsg = $state('')
let preserveOnBehalfOf = $state(false)
function closeSaveDrawer() {
saveDrawerOpen = false
@@ -178,7 +179,8 @@
summary: $summary,
policy,
deployment_message: deploymentMsg,
custom_path: customPath
custom_path: customPath,
preserve_on_behalf_of: preserveOnBehalfOf || undefined
}
})
savedApp = {
@@ -274,7 +276,8 @@
// custom_path requires admin so to accept update without it, we need to send as undefined when non-admin (when undefined, it will be ignored)
// it also means that customPath needs to be set to '' instead of undefined to unset it (when admin)
custom_path:
$userStore?.is_admin || $userStore?.is_super_admin ? (customPath ?? '') : undefined
$userStore?.is_admin || $userStore?.is_super_admin ? (customPath ?? '') : undefined,
preserve_on_behalf_of: preserveOnBehalfOf || undefined
}
})
savedApp = {
@@ -851,6 +854,7 @@
bind:customPathError
bind:pathError
bind:newEditedPath
bind:preserveOnBehalfOf
hideSecretUrl={false}
/>
</DrawerContent>
@@ -16,6 +16,8 @@
import EEOnly from '$lib/components/EEOnly.svelte'
import TextInput from '$lib/components/text_input/TextInput.svelte'
const WM_DEPLOYERS_GROUP = 'wm_deployers'
let {
policy,
setPublishState,
@@ -29,7 +31,8 @@
pathError = $bindable(),
newEditedPath = $bindable(),
newPath,
hideSecretUrl = false
hideSecretUrl = false,
preserveOnBehalfOf = $bindable(false)
}: {
policy: any
setPublishState: () => void
@@ -44,7 +47,17 @@
newEditedPath: string
newPath: string
hideSecretUrl?: boolean
preserveOnBehalfOf?: boolean
} = $props()
let isDeployer = $derived($userStore?.groups?.includes(WM_DEPLOYERS_GROUP) ?? false)
let savedOnBehalfOfEmail = $derived(savedApp?.policy?.on_behalf_of_email)
let showPreserveToggle = $derived(
isDeployer &&
policy?.on_behalf_of_email &&
savedOnBehalfOfEmail &&
savedOnBehalfOfEmail !== $userStore?.email
)
let dirtyCustomPath = $state(false)
let path: Path | undefined = $state(undefined)
@@ -182,6 +195,18 @@
</Tooltip>
</Alert>
{#if showPreserveToggle}
<div class="mt-4">
<Toggle
size="sm"
bind:checked={preserveOnBehalfOf}
options={{
right: `Keep original author (${savedOnBehalfOfEmail})`
}}
/>
</div>
{/if}
<div class="mt-10"></div>
{#if !hideSecretUrl}
@@ -16,7 +16,8 @@
errorHandlerMuted?: boolean
aiId?: string | undefined
aiDescription?: string | undefined
kind?: 'script' | 'flow' | 'app' | 'raw_app' | 'resource' | 'variable' | 'resource_type'
kind?: 'script' | 'flow' | 'app' | 'raw_app' | 'resource' | 'variable' | 'resource_type' | 'folder' | 'schedule' | 'trigger'
triggerKind?: string | undefined
summary?: string | undefined
path: string
href?: string
@@ -43,6 +44,7 @@
aiId = undefined,
aiDescription = undefined,
kind = 'script',
triggerKind = undefined,
summary = undefined,
path,
href = undefined,
@@ -105,7 +107,7 @@
</div>
{/if}
{#if canFavorite && kind !== 'resource' && kind !== 'variable' && kind !== 'resource_type'}
{#if canFavorite && (kind == 'app' || kind == 'raw_app' || kind == 'script' || kind == 'flow')}
<div class="center-center h-full text-xs font-semibold text-secondary w-9">
<Star {kind} {path} {workspaceId} {summary} />
</div>
@@ -120,7 +122,7 @@
{#snippet rowContent()}
<div class="shrink">
<RowIcon {kind} />
<RowIcon {kind} {triggerKind} />
</div>
<div class="grow">
<div class="text-emphasis flex-wrap text-left text-xs font-semibold">
@@ -1,25 +1,85 @@
<script lang="ts">
import BarsStaggered from '$lib/components/icons/BarsStaggered.svelte'
import { Boxes, Code2, DollarSign, Folder, LayoutDashboard } from 'lucide-svelte'
import KafkaIcon from '$lib/components/icons/KafkaIcon.svelte'
import NatsIcon from '$lib/components/icons/NatsIcon.svelte'
import MqttIcon from '$lib/components/icons/MqttIcon.svelte'
import AwsIcon from '$lib/components/icons/AwsIcon.svelte'
import GoogleCloudIcon from '$lib/components/icons/GoogleCloudIcon.svelte'
import {
Boxes,
Calendar,
Code2,
Database,
DollarSign,
Folder,
LayoutDashboard,
Route,
Unplug
} from 'lucide-svelte'
export let kind: 'script' | 'flow' | 'app' | 'raw_app' | 'resource' | 'variable' | 'resource_type' | 'folder'
export let kind:
| 'script'
| 'flow'
| 'app'
| 'raw_app'
| 'resource'
| 'variable'
| 'resource_type'
| 'folder'
| 'schedule'
| 'trigger'
| 'routes'
| 'schedules'
| 'websockets'
| 'postgres'
| 'kafka'
| 'nats'
| 'mqtt'
| 'sqs'
| 'gcp'
/** For 'trigger' kind, specifies the specific trigger type (routes, schedules, etc.) */
export let triggerKind: string | undefined = undefined
// Use triggerKind if kind is 'trigger' and triggerKind is provided
$: effectiveKind = kind === 'trigger' && triggerKind ? triggerKind : kind
</script>
<div class="flex justify-center items-center" title={kind}>
{#if kind === 'flow'}
<div class="flex justify-center items-center" title={effectiveKind}>
{#if effectiveKind === 'flow'}
<BarsStaggered size={16} class="text-teal-500" />
{:else if kind === 'app' || kind === 'raw_app'}
{:else if effectiveKind === 'app' || effectiveKind === 'raw_app'}
<LayoutDashboard size={16} class="text-orange-500" />
{:else if kind === 'script'}
{:else if effectiveKind === 'script'}
<Code2 size={16} class="text-blue-500" />
{:else if kind === 'variable'}
{:else if effectiveKind === 'variable'}
<DollarSign size={16} class="text-gray-400" />
{:else if kind === 'resource'}
{:else if effectiveKind === 'resource'}
<Boxes size={16} class="text-gray-400" />
{:else if kind === 'resource_type'}
<div style="width: 16px; height: 16px;" class="bg-gray-100 rounded-full" ></div>
{:else if kind === 'folder'}
{:else if effectiveKind === 'resource_type'}
<div style="width: 16px; height: 16px;" class="bg-gray-100 rounded-full"></div>
{:else if effectiveKind === 'folder'}
<Folder size={16} class="text-gray-400" />
{:else if effectiveKind === 'schedule' || effectiveKind === 'schedules'}
<Calendar size={16} class="text-gray-400" />
{:else if effectiveKind === 'routes'}
<Route size={16} class="text-gray-400" />
{:else if effectiveKind === 'websockets'}
<Unplug size={16} class="text-gray-400" />
{:else if effectiveKind === 'postgres'}
<Database size={16} class="text-gray-400" />
{:else if effectiveKind === 'kafka'}
<KafkaIcon size={16} class="text-gray-400" />
{:else if effectiveKind === 'nats'}
<NatsIcon size={16} class="text-gray-400" />
{:else if effectiveKind === 'mqtt'}
<MqttIcon size={16} class="text-gray-400" />
{:else if effectiveKind === 'sqs'}
<AwsIcon size={16} class="text-gray-400" />
{:else if effectiveKind === 'gcp'}
<GoogleCloudIcon size={16} />
{:else if effectiveKind === 'trigger'}
<Calendar size={16} class="text-gray-400" />
{:else}
<div class="w-[16px]"></div>
{/if}
@@ -34,8 +34,24 @@
let { noEditor, enableAi }: Props = $props()
const { flowStore, initialPathStore, previewArgs, pathStore, customUi } =
getContext<FlowEditorContext>('FlowEditorContext')
const {
flowStore,
initialPathStore,
previewArgs,
pathStore,
customUi,
preserveOnBehalfOf,
savedOnBehalfOfEmail
} = getContext<FlowEditorContext>('FlowEditorContext')
const WM_DEPLOYERS_GROUP = 'wm_deployers'
let isDeployer = $derived($userStore?.groups?.includes(WM_DEPLOYERS_GROUP) ?? false)
let showPreserveToggle = $derived(
isDeployer &&
flowStore.val.on_behalf_of_email &&
$savedOnBehalfOfEmail &&
$savedOnBehalfOfEmail !== $userStore?.email
)
function asSchema(x: any) {
return x as Schema
@@ -373,6 +389,7 @@
on:change={() => {
if (flowStore.val.on_behalf_of_email) {
flowStore.val.on_behalf_of_email = undefined
$preserveOnBehalfOf = false
} else {
flowStore.val.on_behalf_of_email = $userStore?.email
}
@@ -383,6 +400,16 @@
'When this option is enabled, the flow will be run with the permissions of the last editor.'
}}
/>
{#if showPreserveToggle}
<Toggle
textClass="font-medium"
size="xs"
bind:checked={$preserveOnBehalfOf}
options={{
right: `Keep original author (${$savedOnBehalfOfEmail})`
}}
/>
{/if}
<!-- Error Handler Section -->
<div class="flex flex-row items-center py-1">
@@ -93,6 +93,8 @@ export type FlowEditorContext = {
executionCount: Writable<number>
modulesTestStates: ModulesTestStates
outputPickerOpenFns: Record<string, () => void>
preserveOnBehalfOf: Writable<boolean>
savedOnBehalfOfEmail: Writable<string | undefined>
}
export type FlowGraphAssetContext = StateStore<{
+66 -10
View File
@@ -93,7 +93,18 @@ export async function existsTrigger(
)
}
export async function getTriggersDeployData(kind: TriggerKind, path: string, workspace: string) {
/**
* Get trigger deployment data with optional email preservation.
* @param onBehalfOfEmail - If set, the trigger will be deployed with this email and preserve_email=true.
*/
export async function getTriggersDeployData(
kind: TriggerKind,
path: string,
workspace: string,
onBehalfOfEmail?: string
) {
const preserveEmail = onBehalfOfEmail !== undefined
if (kind === 'sqs') {
const sqsTrigger = await SqsTriggerService.getSqsTrigger({
workspace: workspace!,
@@ -101,7 +112,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
})
return {
data: sqsTrigger,
data: { ...sqsTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: SqsTriggerService.createSqsTrigger,
updateFn: SqsTriggerService.updateSqsTrigger
}
@@ -112,7 +123,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
})
return {
data: kafkaTrigger,
data: { ...kafkaTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: KafkaTriggerService.createKafkaTrigger,
updateFn: KafkaTriggerService.updateKafkaTrigger
}
@@ -123,7 +134,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
})
return {
data: mqttTrigger,
data: { ...mqttTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: MqttTriggerService.createMqttTrigger,
updateFn: MqttTriggerService.updateMqttTrigger
}
@@ -134,7 +145,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
})
return {
data: natsTrigger,
data: { ...natsTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: NatsTriggerService.createNatsTrigger,
updateFn: NatsTriggerService.updateNatsTrigger
}
@@ -155,7 +166,9 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
...gcpTrigger,
delivery_config: gcpTrigger.delivery_config ?? undefined,
base_endpoint:
gcpTrigger.delivery_type === 'push' ? `${window.location.origin}${base}` : undefined
gcpTrigger.delivery_type === 'push' ? `${window.location.origin}${base}` : undefined,
email: onBehalfOfEmail,
preserve_email: preserveEmail
}
return {
@@ -170,7 +183,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
})
return {
data: postgresTrigger,
data: { ...postgresTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: PostgresTriggerService.createPostgresTrigger,
updateFn: PostgresTriggerService.updatePostgresTrigger
}
@@ -181,7 +194,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
})
return {
data: websocketTrigger,
data: { ...websocketTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: WebsocketTriggerService.createWebsocketTrigger,
updateFn: WebsocketTriggerService.updateWebsocketTrigger
}
@@ -192,7 +205,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
})
return {
data: httpTrigger,
data: { ...httpTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: HttpTriggerService.createHttpTrigger,
updateFn: HttpTriggerService.updateHttpTrigger
}
@@ -202,7 +215,7 @@ export async function getTriggersDeployData(kind: TriggerKind, path: string, wor
path: path
})
return {
data: schedulesTrigger,
data: { ...schedulesTrigger, email: onBehalfOfEmail, preserve_email: preserveEmail },
createFn: ScheduleService.createSchedule,
updateFn: ScheduleService.updateSchedule
}
@@ -435,6 +448,49 @@ export async function getTriggerValue(kind: TriggerKind, path: string, workspace
throw new Error(`Unexpected trigger kind got: ${kind}`)
}
/**
* Get the email for a trigger (used for on_behalf_of during deployment).
*/
export async function getTriggerEmail(
kind: TriggerKind,
path: string,
workspace: string
): Promise<string | undefined> {
try {
if (kind === 'sqs') {
const trigger = await SqsTriggerService.getSqsTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'kafka') {
const trigger = await KafkaTriggerService.getKafkaTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'mqtt') {
const trigger = await MqttTriggerService.getMqttTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'nats') {
const trigger = await NatsTriggerService.getNatsTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'gcp') {
const trigger = await GcpTriggerService.getGcpTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'postgres') {
const trigger = await PostgresTriggerService.getPostgresTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'websockets') {
const trigger = await WebsocketTriggerService.getWebsocketTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'routes') {
const trigger = await HttpTriggerService.getHttpTrigger({ workspace, path })
return trigger.edited_by
} else if (kind === 'schedules') {
const trigger = await ScheduleService.getSchedule({ workspace, path })
return trigger.edited_by
}
} catch {
// Trigger may not exist in the workspace
}
return undefined
}
function retrieveScriptOrFlowKind(path: string, is_flow: boolean): { kind: Kind; path: string } {
return {
kind: is_flow ? 'flow' : 'script',
+476
View File
@@ -0,0 +1,476 @@
import {
AppService,
FlowService,
FolderService,
ResourceService,
ScheduleService,
ScriptService,
VariableService
} from '$lib/gen'
import { getAllModules } from './components/flows/flowExplorer'
import {
existsTrigger,
getTriggersDeployData,
getTriggerEmail,
getTriggerValue,
type AdditionalInformation,
type Kind
} from '$lib/utils_deployable'
import type { TriggerKind } from './components/triggers'
export interface DeployItemParams {
kind: Kind
path: string
workspaceFrom: string
workspaceTo: string
additionalInformation?: AdditionalInformation
/** The email to use for on_behalf_of. If set, preserve_on_behalf_of will be true. If undefined, deploying user's email is used. */
onBehalfOfEmail?: string
}
export interface DeployResult {
success: boolean
error?: string
}
/**
* Deploy an item from one workspace to another.
* Handles all item kinds: flow, script, app, variable, resource, resource_type, folder, trigger.
*/
export async function deployItem(params: DeployItemParams): Promise<DeployResult> {
const { kind, path, workspaceFrom, workspaceTo, additionalInformation, onBehalfOfEmail } = params
// When onBehalfOfEmail is set, we preserve the on_behalf_of setting with the specified email
const preserveOnBehalfOf = onBehalfOfEmail !== undefined
try {
const alreadyExists = await checkItemExists(kind, path, workspaceTo, additionalInformation)
if (kind === 'flow') {
const flow = await FlowService.getFlowByPath({
workspace: workspaceFrom,
path: path
})
getAllModules(flow.value.modules).forEach((x) => {
if (x.value.type === 'script' && x.value.hash != undefined) {
x.value.hash = undefined
}
})
if (alreadyExists) {
await FlowService.updateFlow({
workspace: workspaceTo,
path: path,
requestBody: {
...flow,
preserve_on_behalf_of: preserveOnBehalfOf,
on_behalf_of_email: onBehalfOfEmail
}
})
} else {
await FlowService.createFlow({
workspace: workspaceTo,
requestBody: {
...flow,
preserve_on_behalf_of: preserveOnBehalfOf,
on_behalf_of_email: onBehalfOfEmail
}
})
}
} else if (kind === 'script') {
const script = await ScriptService.getScriptByPath({
workspace: workspaceFrom,
path: path
})
await ScriptService.createScript({
workspace: workspaceTo,
requestBody: {
...script,
lock: script.lock,
parent_hash: alreadyExists
? (
await ScriptService.getScriptByPath({
workspace: workspaceTo,
path: path
})
).hash
: undefined,
preserve_on_behalf_of: preserveOnBehalfOf,
on_behalf_of_email: onBehalfOfEmail
}
})
} else if (kind === 'app') {
const app = await AppService.getAppByPath({
workspace: workspaceFrom,
path: path
})
if (alreadyExists) {
if (app.raw_app) {
const secret = await AppService.getPublicSecretOfLatestVersionOfApp({
workspace: workspaceFrom,
path: app.path
})
const js = await AppService.getRawAppData({
secretWithExtension: `${secret}.js`,
workspace: workspaceFrom
})
const css = await AppService.getRawAppData({
secretWithExtension: `${secret}.css`,
workspace: workspaceFrom
})
await AppService.updateAppRaw({
workspace: workspaceTo,
path: path,
formData: {
app: { ...app, preserve_on_behalf_of: preserveOnBehalfOf },
css,
js
}
})
} else {
await AppService.updateApp({
workspace: workspaceTo,
path: path,
requestBody: {
...app,
preserve_on_behalf_of: preserveOnBehalfOf
}
})
}
} else {
if (app.raw_app) {
const secret = await AppService.getPublicSecretOfLatestVersionOfApp({
workspace: workspaceFrom,
path: app.path
})
const js = await AppService.getRawAppData({
secretWithExtension: `${secret}.js`,
workspace: workspaceFrom
})
const css = await AppService.getRawAppData({
secretWithExtension: `${secret}.css`,
workspace: workspaceFrom
})
await AppService.createAppRaw({
workspace: workspaceTo,
formData: {
app: { ...app, preserve_on_behalf_of: preserveOnBehalfOf },
css,
js
}
})
} else {
await AppService.createApp({
workspace: workspaceTo,
requestBody: {
...app,
preserve_on_behalf_of: preserveOnBehalfOf
}
})
}
}
} else if (kind === 'variable') {
const variable = await VariableService.getVariable({
workspace: workspaceFrom,
path: path,
decryptSecret: true
})
if (alreadyExists) {
await VariableService.updateVariable({
workspace: workspaceTo,
path: path,
requestBody: {
path: path,
value: variable.value ?? '',
is_secret: variable.is_secret,
description: variable.description ?? ''
},
alreadyEncrypted: false
})
} else {
await VariableService.createVariable({
workspace: workspaceTo,
requestBody: {
path: path,
value: variable.value ?? '',
is_secret: variable.is_secret,
description: variable.description ?? ''
}
})
}
} else if (kind === 'resource') {
const resource = await ResourceService.getResource({
workspace: workspaceFrom,
path: path
})
if (alreadyExists) {
await ResourceService.updateResource({
workspace: workspaceTo,
path: path,
requestBody: {
path: path,
value: resource.value ?? '',
description: resource.description ?? ''
}
})
} else {
await ResourceService.createResource({
workspace: workspaceTo,
requestBody: {
path: path,
value: resource.value ?? '',
resource_type: resource.resource_type,
description: resource.description ?? ''
}
})
}
} else if (kind === 'resource_type') {
const resource = await ResourceService.getResourceType({
workspace: workspaceFrom,
path: path
})
if (alreadyExists) {
await ResourceService.updateResourceType({
workspace: workspaceTo,
path: path,
requestBody: {
schema: resource.schema,
description: resource.description ?? ''
}
})
} else {
await ResourceService.createResourceType({
workspace: workspaceTo,
requestBody: {
description: resource.description ?? '',
schema: resource.schema,
name: resource.name
}
})
}
} else if (kind === 'folder') {
await FolderService.createFolder({
workspace: workspaceTo,
requestBody: {
name: path
}
})
} else if (kind === 'trigger') {
if (additionalInformation?.triggers) {
const { data, createFn, updateFn } = await getTriggersDeployData(
additionalInformation.triggers.kind,
path,
workspaceFrom,
onBehalfOfEmail
)
if (alreadyExists) {
await updateFn({
path,
workspace: workspaceTo,
requestBody: data
} as any)
} else {
await createFn({
workspace: workspaceTo,
requestBody: data
} as any)
}
} else {
throw new Error('Missing triggers kind')
}
} else {
throw new Error(`Unknown kind ${kind}`)
}
return { success: true }
} catch (e: any) {
return { success: false, error: e.body || e.message }
}
}
/**
* Check if an item already exists in the target workspace.
*/
export async function checkItemExists(
kind: Kind,
path: string,
workspace: string,
additionalInformation?: AdditionalInformation
): Promise<boolean> {
if (kind === 'flow') {
return await FlowService.existsFlowByPath({
workspace: workspace,
path: path
})
} else if (kind === 'script') {
return await ScriptService.existsScriptByPath({
workspace: workspace,
path: path
})
} else if (kind === 'app') {
return await AppService.existsApp({
workspace: workspace,
path: path
})
} else if (kind === 'variable') {
return await VariableService.existsVariable({
workspace: workspace,
path: path
})
} else if (kind === 'resource') {
return await ResourceService.existsResource({
workspace: workspace,
path: path
})
} else if (kind === 'schedule') {
return await ScheduleService.existsSchedule({
workspace: workspace,
path: path
})
} else if (kind === 'resource_type') {
return await ResourceService.existsResourceType({
workspace: workspace,
path: path
})
} else if (kind === 'folder') {
return await FolderService.existsFolder({
workspace: workspace,
name: path
})
} else if (kind === 'trigger') {
const triggersKind: TriggerKind[] = [
'kafka',
'mqtt',
'nats',
'postgres',
'routes',
'schedules',
'sqs',
'websockets',
'gcp'
]
if (
additionalInformation?.triggers &&
triggersKind.includes(additionalInformation.triggers.kind)
) {
return await existsTrigger(
{ workspace: workspace, path },
additionalInformation.triggers.kind
)
} else {
throw new Error(
`Unexpected triggers kind, expected one of: '${triggersKind.join(', ')}' got: ${
additionalInformation?.triggers?.kind
}`
)
}
} else {
throw new Error(`Unknown kind ${kind}`)
}
}
/**
* Get the value of an item for diff comparison.
*/
export async function getItemValue(
kind: Kind,
path: string,
workspace: string,
additionalInformation?: AdditionalInformation
): Promise<unknown> {
try {
if (kind === 'flow') {
const flow = await FlowService.getFlowByPath({
workspace: workspace,
path: path
})
getAllModules(flow.value.modules).forEach((x) => {
if (x.value.type === 'script' && x.value.hash != undefined) {
x.value.hash = undefined
}
})
return { summary: flow.summary, description: flow.description, value: flow.value }
} else if (kind === 'script') {
const script = await ScriptService.getScriptByPath({
workspace: workspace,
path: path
})
return {
content: script.content,
lock: script.lock,
schema: script.schema,
summary: script.summary,
language: script.language
}
} else if (kind === 'app') {
const app = await AppService.getAppByPath({
workspace: workspace,
path: path
})
return app
} else if (kind === 'variable') {
const variable = await VariableService.getVariable({
workspace: workspace,
path: path,
decryptSecret: true
})
return variable.value
} else if (kind === 'resource') {
const resource = await ResourceService.getResource({
workspace: workspace,
path: path
})
return resource.value
} else if (kind === 'resource_type') {
const resource = await ResourceService.getResourceType({
workspace: workspace,
path: path
})
return resource.schema
} else if (kind === 'folder') {
const folder = await FolderService.getFolder({
workspace: workspace,
name: path
})
return {
name: folder.name
}
} else if (kind === 'trigger') {
if (additionalInformation?.triggers) {
return await getTriggerValue(additionalInformation.triggers.kind, path, workspace)
} else {
throw new Error(`Missing trigger information`)
}
} else {
throw new Error(`Unknown kind ${kind}`)
}
} catch {
return {}
}
}
/**
* Get the on_behalf_of_email for a flow, script, app, or trigger (including schedule).
*/
export async function getOnBehalfOfEmail(
kind: Kind,
path: string,
workspace: string,
additionalInformation?: AdditionalInformation
): Promise<string | undefined> {
try {
if (kind === 'flow') {
const flow = await FlowService.getFlowByPath({ workspace, path })
return flow.on_behalf_of_email
} else if (kind === 'script') {
const script = await ScriptService.getScriptByPath({ workspace, path })
return script.on_behalf_of_email
} else if (kind === 'app') {
const app = await AppService.getAppByPath({ workspace, path })
return app.policy.on_behalf_of_email
} else if (kind === 'trigger' && additionalInformation?.triggers) {
return await getTriggerEmail(additionalInformation.triggers.kind, path, workspace)
}
} catch {
// Item may not exist in the workspace
}
return undefined
}
+3 -1
View File
@@ -113,7 +113,9 @@
}),
currentEditor: writable(undefined),
modulesTestStates: new ModulesTestStates(),
outputPickerOpenFns: {}
outputPickerOpenFns: {},
preserveOnBehalfOf: writable(false),
savedOnBehalfOfEmail: writable<string | undefined>(undefined)
})
setContext<PropPickerContext>('PropPickerContext', {
flowPropPickerConfig: writable<FlowPropPickerConfig | undefined>(undefined),