mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-25 08:00:59 +00:00
Ruben Fiszel
0dbd9c1231
* perf: eliminate dual-connection DB pool contention across worker, queue, and api Reuse the held transaction (or move pool reads before begin()) instead of checking out a second pool connection while a tx is open, extending the fix from #9789/#7861. Targets the per-worker pool (max 5) hot paths plus several server-pool API handlers. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix: pass owned pool to get_email_from_permissioned_as in http trigger handler The generified signature takes impl PgExecutor; the http trigger handler passed &db where db is already &DB, yielding &&Pool which does not impl PgExecutor (only surfaced under the full feature set in CI). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * fix: keep RLS-exposed reads on the non-RLS pool and isolate flow-eval reads in a savepoint Addresses review of the dual-connection sweep: - worker_flow: wrap the stop_after_all_iters_if reads in a SAVEPOINT. The caller swallows the error and keeps using tx, so a DB read failure must not leave the outer transaction aborted (it would fail the later commit). Matches the previous pool-read semantics. - Revert reads that were moved onto an RLS (user_db) transaction back to the non-RLS pool, since RLS row-visibility/role context can change results: push_scheduled_job (email/tag/settings lookups; reachable with a user_db tx from api-schedule/api-flows), push_inner native-retry dedicated_worker routing (RLS isolation variants), resources.rs app-namespace folder auto-create (non-admins must not be blocked), and the script archive/delete UPDATEs. Non-RLS db.begin() reuse and move-before-begin are kept. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * test: failpoint proving the stop_after_all_iters_if savepoint isolates an aborted read Adds a worker-crate failpoints feature and a data-driven hook: when the stop_after_all_iters_if expr is the magic sentinel, the in-evaluation read runs SELECT 1/0 to abort its (savepoint) transaction. The test asserts the flow still completes (iteration marked failed) — which only holds if the savepoint keeps the outer status-update transaction committable. Without the savepoint the abort would poison the outer tx and the job would never complete. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
575 lines
19 KiB
Rust
575 lines
19 KiB
Rust
/*
|
|
* Author: Ruben Fiszel
|
|
* Copyright: Windmill Labs, Inc 2022
|
|
* This file and its contents are licensed under the AGPLv3 License.
|
|
* Please see the included NOTICE for copyright information and
|
|
* LICENSE-AGPL for a copy of the license.
|
|
*/
|
|
|
|
use axum::{
|
|
extract::{Extension, Path},
|
|
routing::{get, post},
|
|
Json, Router,
|
|
};
|
|
use windmill_api_auth::require_owner_of_path;
|
|
use windmill_audit::audit_oss::audit_log;
|
|
use windmill_audit::ActionKind;
|
|
use windmill_common::scripts::ScriptHash;
|
|
use windmill_common::DB;
|
|
use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
|
|
|
|
use windmill_api_auth::ApiAuthed;
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
use windmill_common::{
|
|
db::UserDB,
|
|
error::{Error, JsonResult, Result},
|
|
utils::{not_found_if_none, StripPath},
|
|
};
|
|
|
|
/// Map a granular-ACL kind segment to the audit-log action prefix used by the
|
|
/// per-kind CRUD endpoints (e.g. `flows.update`, `scripts.update`). The
|
|
/// resulting action is suffixed with `.grant_acl` / `.revoke_acl` so the
|
|
/// audit log keeps a per-resource record of every `/acls/*` mutation —
|
|
/// folder/group already log via their dedicated permission-history tables.
|
|
fn audit_action_prefix_for_acl_kind(kind: &str) -> Option<&'static str> {
|
|
match kind {
|
|
"script" => Some("scripts"),
|
|
"flow" => Some("flows"),
|
|
"app" => Some("apps"),
|
|
// Distinct prefix so dashboards aggregating on `action` can separate
|
|
// raw_app ACL mutations from regular app ones without parsing the
|
|
// `kind` parameters field. (The granular_acls SQL routes raw_app
|
|
// writes to the same `app` table; audit log identity is separate.)
|
|
"raw_app" => Some("raw_apps"),
|
|
"resource" => Some("resources"),
|
|
"variable" => Some("variables"),
|
|
"schedule" => Some("schedules"),
|
|
"http_trigger" | "websocket_trigger" | "kafka_trigger" | "nats_trigger"
|
|
| "postgres_trigger" | "mqtt_trigger" | "gcp_trigger" | "azure_trigger" | "sqs_trigger"
|
|
| "email_trigger" => Some("triggers"),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
const KINDS: [&str; 20] = [
|
|
"script",
|
|
"group_",
|
|
"resource",
|
|
"schedule",
|
|
"variable",
|
|
"flow",
|
|
"folder",
|
|
"app",
|
|
"raw_app",
|
|
"http_trigger",
|
|
"websocket_trigger",
|
|
"kafka_trigger",
|
|
"nats_trigger",
|
|
"postgres_trigger",
|
|
"mqtt_trigger",
|
|
"gcp_trigger",
|
|
"azure_trigger",
|
|
"sqs_trigger",
|
|
"email_trigger",
|
|
"volume",
|
|
];
|
|
|
|
pub fn workspaced_service() -> Router {
|
|
Router::new()
|
|
.route("/get/{*path}", get(get_granular_acls))
|
|
.route("/add/{*path}", post(add_granular_acl))
|
|
.route("/remove/{*path}", post(remove_granular_acl))
|
|
}
|
|
|
|
#[derive(Serialize, Deserialize)]
|
|
pub struct GranularAcl {
|
|
pub owner: String,
|
|
pub write: Option<bool>,
|
|
}
|
|
|
|
async fn add_granular_acl(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
Json(GranularAcl { owner, write }): Json<GranularAcl>,
|
|
) -> Result<String> {
|
|
let path = path.to_path();
|
|
|
|
let (kind, path) = path
|
|
.split_once('/')
|
|
.ok_or_else(|| Error::BadRequest("Invalid path or kind".to_string()))?;
|
|
|
|
if !KINDS.contains(&kind) {
|
|
return Err(Error::BadRequest("Invalid kind".to_string()));
|
|
}
|
|
|
|
let identifier = if kind == "group_" || kind == "folder" || kind == "volume" {
|
|
"name"
|
|
} else {
|
|
"path"
|
|
};
|
|
|
|
if !authed.is_admin {
|
|
if kind == "folder" {
|
|
crate::folders::require_is_owner(&authed, path)?;
|
|
} else if kind == "group_" {
|
|
crate::groups::require_is_owner(path, &authed.username, &authed.groups, &w_id, &db)
|
|
.await?;
|
|
} else if kind == "volume" {
|
|
let created_by = sqlx::query_scalar!(
|
|
"SELECT created_by FROM volume WHERE name = $1 AND workspace_id = $2",
|
|
path,
|
|
&w_id
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.ok_or_else(|| Error::NotFound(format!("volume '{path}' not found")))?;
|
|
// created_by is stored with u/ prefix (from job.permissioned_as)
|
|
let owner_username = created_by.strip_prefix("u/").unwrap_or(&created_by);
|
|
if owner_username != authed.username {
|
|
return Err(Error::NotAuthorized(
|
|
"Only the volume owner or an admin can modify permissions".to_string(),
|
|
));
|
|
}
|
|
} else {
|
|
require_owner_of_path(&authed, path)?;
|
|
}
|
|
}
|
|
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
if kind == "folder" {
|
|
if let Some(obj) = sqlx::query_scalar!(
|
|
"SELECT owners FROM folder WHERE name = $1 AND workspace_id = $2",
|
|
path,
|
|
w_id
|
|
)
|
|
.fetch_optional(&mut *tx)
|
|
.await?
|
|
{
|
|
if obj.contains(&owner) {
|
|
if write != Some(true) {
|
|
return Err(Error::BadRequest(
|
|
"Cannot remove write permission for folder owner".to_string(),
|
|
));
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// v2 raw apps are stored in the `app` table (with `app_version.raw_app = true`
|
|
// distinguishing them from regular apps); the legacy `raw_app` table no longer
|
|
// backs the workspace export, so granting/revoking on `raw_app` must hit `app`
|
|
// for the change to be visible. Git-sync dispatch still uses
|
|
// DeployedObject::RawApp so the worker writes back `<path>.raw_app.json`.
|
|
let table = if kind == "raw_app" { "app" } else { kind };
|
|
// SAFETY: `kind` has been validated against the `KINDS` allowlist before reaching this function.
|
|
let obj_o = sqlx::query_scalar::<_, serde_json::Value>(&format!(
|
|
"UPDATE {table} SET extra_perms = jsonb_set(extra_perms, $1, to_jsonb($2), \
|
|
true) WHERE {identifier} = $3 AND workspace_id = $4 RETURNING extra_perms"
|
|
))
|
|
.bind(vec![owner.clone()])
|
|
.bind(write.unwrap_or(false))
|
|
.bind(path)
|
|
.bind(&w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
let _ = not_found_if_none(obj_o, &kind, &path)?;
|
|
|
|
if kind == "folder" {
|
|
let change_type = if write.unwrap_or(false) {
|
|
"grant_read"
|
|
} else {
|
|
"grant_write"
|
|
};
|
|
crate::folders::log_folder_permission_change(
|
|
&mut *tx,
|
|
&w_id,
|
|
path,
|
|
&authed.username,
|
|
change_type,
|
|
Some(&owner),
|
|
)
|
|
.await?;
|
|
} else if kind == "group_" {
|
|
let change_type = if write.unwrap_or(false) {
|
|
"grant_admin"
|
|
} else {
|
|
"grant_member_only"
|
|
};
|
|
crate::groups::log_group_permission_change(
|
|
&mut *tx,
|
|
&w_id,
|
|
path,
|
|
&authed.username,
|
|
change_type,
|
|
Some(&owner),
|
|
)
|
|
.await?;
|
|
} else if let Some(prefix) = audit_action_prefix_for_acl_kind(kind) {
|
|
// Mirror the folder/group permission-history coverage for every other
|
|
// ACLable kind. Folder/group already wrote a dedicated history row
|
|
// above; everything else (script/flow/app/raw_app/resource/...) lands
|
|
// in the general audit_log table here.
|
|
let access = if write.unwrap_or(false) {
|
|
"write"
|
|
} else {
|
|
"read"
|
|
};
|
|
let action = format!("{}.grant_acl", prefix);
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed,
|
|
action.as_str(),
|
|
ActionKind::Update,
|
|
&w_id,
|
|
Some(path),
|
|
Some(
|
|
[
|
|
("kind", kind),
|
|
("owner", owner.as_str()),
|
|
("access", access),
|
|
]
|
|
.into(),
|
|
),
|
|
)
|
|
.await?;
|
|
}
|
|
|
|
tx.commit().await?;
|
|
|
|
match kind {
|
|
"folder" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Folder { path: format!("f/{}", path) },
|
|
Some(format!("Folder '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"app" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::App { path: path.to_string(), parent_path: None, version: 0 },
|
|
Some(format!("App '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"raw_app" => {
|
|
// RawApp deliberately uses its own DeployedObject variant: the
|
|
// git-sync worker reads `path_type` ("app" vs "raw_app") to decide
|
|
// whether to write `<path>.app.json` or `<path>.raw_app.json`.
|
|
// Collapsing this into `App` would dispatch raw_app perm changes
|
|
// against the wrong file in the repo.
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::RawApp { path: path.to_string(), parent_path: None, version: 0 },
|
|
Some(format!("Raw App '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"script" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Script {
|
|
path: path.to_string(),
|
|
parent_path: None,
|
|
hash: ScriptHash(0),
|
|
},
|
|
Some(format!("Script '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"flow" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Flow { path: path.to_string(), parent_path: None, version: 0 },
|
|
Some(format!("Flow '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
_ => (),
|
|
}
|
|
|
|
Ok("Successfully modified granular acl".to_string())
|
|
}
|
|
|
|
async fn remove_granular_acl(
|
|
authed: ApiAuthed,
|
|
Extension(db): Extension<DB>,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
Json(GranularAcl { owner, .. }): Json<GranularAcl>,
|
|
) -> Result<String> {
|
|
let path = path.to_path();
|
|
|
|
let (kind, path) = path
|
|
.split_once('/')
|
|
.ok_or_else(|| Error::BadRequest("Invalid path or kind".to_string()))?;
|
|
|
|
if !KINDS.contains(&kind) {
|
|
return Err(Error::BadRequest("Invalid kind".to_string()));
|
|
}
|
|
|
|
if !authed.is_admin {
|
|
if kind == "folder" {
|
|
crate::folders::require_is_owner(&authed, path)?;
|
|
} else if kind == "group_" {
|
|
crate::groups::require_is_owner(path, &authed.username, &authed.groups, &w_id, &db)
|
|
.await?;
|
|
} else if kind == "volume" {
|
|
let created_by = sqlx::query_scalar!(
|
|
"SELECT created_by FROM volume WHERE name = $1 AND workspace_id = $2",
|
|
path,
|
|
&w_id
|
|
)
|
|
.fetch_optional(&db)
|
|
.await?
|
|
.ok_or_else(|| Error::NotFound(format!("volume '{path}' not found")))?;
|
|
// created_by is stored with u/ prefix (from job.permissioned_as)
|
|
let owner_username = created_by.strip_prefix("u/").unwrap_or(&created_by);
|
|
if owner_username != authed.username {
|
|
return Err(Error::NotAuthorized(
|
|
"Only the volume owner or an admin can modify permissions".to_string(),
|
|
));
|
|
}
|
|
} else {
|
|
require_owner_of_path(&authed, path)?;
|
|
}
|
|
}
|
|
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
let identifier = if kind == "group_" || kind == "folder" || kind == "volume" {
|
|
"name"
|
|
} else {
|
|
"path"
|
|
};
|
|
|
|
if identifier == "path" {
|
|
require_owner_of_path(&authed, path)?;
|
|
}
|
|
|
|
// See add_granular_acl: kind="raw_app" must hit the `app` table because v2
|
|
// raw apps live there and the legacy `raw_app` table no longer backs the
|
|
// workspace export.
|
|
let table = if kind == "raw_app" { "app" } else { kind };
|
|
// SAFETY: `kind` has been validated against the `KINDS` allowlist before reaching this function.
|
|
// LIMIT 1: `script` shares (workspace_id, path) across versions, so `old` can
|
|
// return >1 row, which would break the scalar subquery in RETURNING.
|
|
let obj_o = sqlx::query_scalar::<_, bool>(&format!(
|
|
"WITH old AS (
|
|
SELECT extra_perms->$1 as old_write FROM {table}
|
|
WHERE {identifier} = $2 AND workspace_id = $3 AND extra_perms ? $1
|
|
)
|
|
UPDATE {table} SET extra_perms = extra_perms - $1
|
|
WHERE {identifier} = $2 AND workspace_id = $3 AND extra_perms ? $1
|
|
RETURNING (SELECT old_write FROM old LIMIT 1)::bool"
|
|
))
|
|
.bind(&owner)
|
|
.bind(path)
|
|
.bind(&w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
// Only log if something was actually removed (obj_o is Some)
|
|
if let Some(write) = obj_o {
|
|
// Log permission changes for folders and groups
|
|
if kind == "folder" {
|
|
let change_type = if write { "revoke_write" } else { "revoke_read" };
|
|
crate::folders::log_folder_permission_change(
|
|
&mut *tx,
|
|
&w_id,
|
|
path,
|
|
&authed.username,
|
|
change_type,
|
|
Some(&owner),
|
|
)
|
|
.await?;
|
|
} else if kind == "group_" {
|
|
crate::groups::log_group_permission_change(
|
|
&mut *tx,
|
|
&w_id,
|
|
path,
|
|
&authed.username,
|
|
"revoke_admin",
|
|
Some(&owner),
|
|
)
|
|
.await?;
|
|
} else if let Some(prefix) = audit_action_prefix_for_acl_kind(kind) {
|
|
// Mirror the add path: standard audit_log row for every kind that
|
|
// doesn't have a dedicated permission-history table.
|
|
let access = if write { "write" } else { "read" };
|
|
let action = format!("{}.revoke_acl", prefix);
|
|
audit_log(
|
|
&mut *tx,
|
|
&authed,
|
|
action.as_str(),
|
|
ActionKind::Update,
|
|
&w_id,
|
|
Some(path),
|
|
Some(
|
|
[
|
|
("kind", kind),
|
|
("owner", owner.as_str()),
|
|
("access", access),
|
|
]
|
|
.into(),
|
|
),
|
|
)
|
|
.await?;
|
|
}
|
|
|
|
tx.commit().await?;
|
|
|
|
match kind {
|
|
"folder" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Folder { path: format!("f/{}", path) },
|
|
Some(format!("Folder '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"app" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::App { path: path.to_string(), parent_path: None, version: 0 },
|
|
Some(format!("App '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"raw_app" => {
|
|
// See add_granular_acl: raw_app must use its own DeployedObject
|
|
// variant so git-sync writes `<path>.raw_app.json`.
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::RawApp {
|
|
path: path.to_string(),
|
|
parent_path: None,
|
|
version: 0,
|
|
},
|
|
Some(format!("Raw App '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"script" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Script {
|
|
path: path.to_string(),
|
|
parent_path: None,
|
|
hash: ScriptHash(0),
|
|
},
|
|
Some(format!("Script '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
"flow" => {
|
|
handle_deployment_metadata(
|
|
&authed.email,
|
|
&authed.username,
|
|
&db,
|
|
&w_id,
|
|
DeployedObject::Flow { path: path.to_string(), parent_path: None, version: 0 },
|
|
Some(format!("Flow '{}' changed permissions", path)),
|
|
true,
|
|
None,
|
|
)
|
|
.await?
|
|
}
|
|
_ => (),
|
|
}
|
|
}
|
|
|
|
Ok("Successfully removed granular acl".to_string())
|
|
}
|
|
|
|
async fn get_granular_acls(
|
|
authed: ApiAuthed,
|
|
Extension(user_db): Extension<UserDB>,
|
|
Path((w_id, path)): Path<(String, StripPath)>,
|
|
) -> JsonResult<serde_json::Value> {
|
|
let path = path.to_path();
|
|
let (kind, path) = path
|
|
.split_once('/')
|
|
.ok_or_else(|| Error::BadRequest("Invalid path or kind".to_string()))?;
|
|
|
|
if !KINDS.contains(&kind) {
|
|
return Err(Error::BadRequest("Invalid kind".to_string()));
|
|
}
|
|
|
|
let mut tx = user_db.begin(&authed).await?;
|
|
|
|
let identifier = if kind == "group_" || kind == "folder" || kind == "volume" {
|
|
"name"
|
|
} else {
|
|
"path"
|
|
};
|
|
// See add_granular_acl: raw_app rows live in the `app` table now, so the
|
|
// read path must also target `app` — otherwise GET would return stale or
|
|
// 404 state while POST /acls/add and /acls/remove write to `app`.
|
|
let table = if kind == "raw_app" { "app" } else { kind };
|
|
// SAFETY: `kind` has been validated against the `KINDS` allowlist before reaching this function.
|
|
let obj_o = sqlx::query_scalar::<_, serde_json::Value>(&format!(
|
|
"SELECT extra_perms from {table} WHERE {identifier} = $1 AND workspace_id = $2"
|
|
))
|
|
.bind(path)
|
|
.bind(w_id)
|
|
.fetch_optional(&mut *tx)
|
|
.await?;
|
|
|
|
let obj = not_found_if_none(obj_o, &kind, &path)?;
|
|
tx.commit().await?;
|
|
|
|
Ok(Json(obj))
|
|
}
|