diff --git a/backend/.sqlx/query-9f663180166f53d117e794f3f3a5723a0a43db163ecca7d5a63d4e74ab1d3be1.json b/backend/.sqlx/query-9f663180166f53d117e794f3f3a5723a0a43db163ecca7d5a63d4e74ab1d3be1.json new file mode 100644 index 0000000000..4dd7ea9bd8 --- /dev/null +++ b/backend/.sqlx/query-9f663180166f53d117e794f3f3a5723a0a43db163ecca7d5a63d4e74ab1d3be1.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT pg_advisory_xact_lock(hashtext('datatable_role_catalog'))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "pg_advisory_xact_lock", + "type_info": "Void" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "9f663180166f53d117e794f3f3a5723a0a43db163ecca7d5a63d4e74ab1d3be1" +} diff --git a/backend/windmill-api-integration-tests/tests/datatable_roles.rs b/backend/windmill-api-integration-tests/tests/datatable_roles.rs index 7ec93dd159..d3e9255934 100644 --- a/backend/windmill-api-integration-tests/tests/datatable_roles.rs +++ b/backend/windmill-api-integration-tests/tests/datatable_roles.rs @@ -329,3 +329,51 @@ async fn a_caller_with_no_identity_reaches_a_permissioned_data_table_not_at_all( assert_eq!(resolved["dbname"], "dt_main", "{resolved}"); Ok(()) } + +#[sqlx::test(migrations = "../migrations", fixtures("base", "datatable_roles"))] +async fn concurrent_role_catalog_writes_do_not_lose_an_entry(db: Pool) -> anyhow::Result<()> { + use windmill_common::datatable_roles::{ + lock_role_catalog, read_role_catalog_tx, InstanceDatatableRole, + }; + + initialize_tracing().await; + // The catalog is one JSON document, so every mutation is read-modify-write. Without the lock + // two concurrent creates read the same snapshot and the second write drops the first — leaving + // the role it dropped as a live cluster login nobody recorded. No DDL here: the losable step is + // the catalog write, and that is what this pins. + let insert = |id: &'static str| { + let db = db.clone(); + async move { + let mut tx = db.begin().await?; + lock_role_catalog(&mut tx).await?; + let mut catalog = read_role_catalog_tx(&mut tx).await?; + // Widen the window the lock has to cover, so an unlocked version fails reliably rather + // than occasionally. + tokio::time::sleep(std::time::Duration::from_millis(150)).await; + catalog.insert( + id.to_string(), + InstanceDatatableRole { name: id.to_string(), enabled: true, pwd: None }, + ); + let value = serde_json::to_value(&catalog)?; + sqlx::query( + "UPDATE global_settings + SET value = jsonb_set(COALESCE(value, '{}'::jsonb), '{roles}', $1) + WHERE name = 'custom_instance_pg_databases'", + ) + .bind(value) + .execute(&mut *tx) + .await?; + tx.commit().await?; + Ok::<_, anyhow::Error>(()) + } + }; + + let (a, b) = tokio::join!(insert("first"), insert("second")); + a?; + b?; + + let catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?; + assert!(catalog.contains_key("first"), "lost 'first': {catalog:?}"); + assert!(catalog.contains_key("second"), "lost 'second': {catalog:?}"); + Ok(()) +} diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index ccc169d0f7..6d0168776d 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -2642,7 +2642,7 @@ fn datatable_role_infos( /// normally planted by the boot converge, but that swallows its own failures, so this is a check /// rather than an assumption. async fn write_role_catalog( - db: &DB, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, catalog: &windmill_common::datatable_roles::DatatableRoleCatalog, ) -> error::Result<()> { let value = serde_json::to_value(catalog).map_err(to_anyhow)?; @@ -2651,7 +2651,7 @@ async fn write_role_catalog( WHERE name = 'custom_instance_pg_databases'", value ) - .execute(db) + .execute(&mut **tx) .await? .rows_affected(); if written == 0 { @@ -2686,7 +2686,12 @@ async fn create_datatable_role( require_super_admin(&db, &authed).await?; windmill_common::datatable_roles::validate_role_name(&req.name)?; - let mut catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?; + // The catalog is one JSON document, so read, DDL and write are one critical section: two + // concurrent creates would otherwise both land in the cluster and the second write would drop + // the first, leaving a live login nobody recorded. + let mut tx = db.begin().await?; + windmill_common::datatable_roles::lock_role_catalog(&mut tx).await?; + let mut catalog = windmill_common::datatable_roles::read_role_catalog_tx(&mut tx).await?; if catalog.values().any(|r| r.name == req.name) { return Err(error::Error::BadRequest(format!( "A data table role named '{}' already exists", @@ -2696,7 +2701,7 @@ async fn create_datatable_role( let id = windmill_common::utils::rd_string(12); let pwd = uuid::Uuid::new_v4().to_string(); - windmill_common::datatable_roles::create_instance_role(&db, &req.name, &pwd).await?; + windmill_common::datatable_roles::create_instance_role(&mut tx, &req.name, &pwd).await?; catalog.insert( id.clone(), @@ -2706,7 +2711,8 @@ async fn create_datatable_role( pwd: Some(pwd), }, ); - write_role_catalog(&db, &catalog).await?; + write_role_catalog(&mut tx, &catalog).await?; + tx.commit().await?; converge_connect_grants_everywhere(&db, &catalog).await; windmill_common::feature_usage::log_feature_usage("datatable", "role_created", ""); @@ -2731,7 +2737,9 @@ async fn update_datatable_role( Json(req): Json, ) -> JsonResult { require_super_admin(&db, &authed).await?; - let mut catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?; + let mut tx = db.begin().await?; + windmill_common::datatable_roles::lock_role_catalog(&mut tx).await?; + let mut catalog = windmill_common::datatable_roles::read_role_catalog_tx(&mut tx).await?; let role = catalog .get(&id) .cloned() @@ -2748,19 +2756,20 @@ async fn update_datatable_role( // RENAME discards an md5-hashed password, so the role gets a fresh one in the same // statement and the catalog records it. Tenants name the id, so nothing else moves. let pwd = uuid::Uuid::new_v4().to_string(); - windmill_common::datatable_roles::rename_instance_role(&db, &role.name, &name, &pwd) + windmill_common::datatable_roles::rename_instance_role(&mut tx, &role.name, &name, &pwd) .await?; updated.name = name; updated.pwd = Some(pwd); } if let Some(enabled) = req.enabled.filter(|e| *e != role.enabled) { - windmill_common::datatable_roles::set_instance_role_login(&db, &updated.name, enabled) + windmill_common::datatable_roles::set_instance_role_login(&mut tx, &updated.name, enabled) .await?; updated.enabled = enabled; } catalog.insert(id.clone(), updated.clone()); - write_role_catalog(&db, &catalog).await?; + write_role_catalog(&mut tx, &catalog).await?; + tx.commit().await?; converge_connect_grants_everywhere(&db, &catalog).await; audit_log( @@ -2792,15 +2801,20 @@ async fn delete_datatable_role( Path(id): Path, ) -> JsonResult<()> { require_super_admin(&db, &authed).await?; - let mut catalog = windmill_common::datatable_roles::read_role_catalog(&db).await?; + let mut tx = db.begin().await?; + windmill_common::datatable_roles::lock_role_catalog(&mut tx).await?; + let mut catalog = windmill_common::datatable_roles::read_role_catalog_tx(&mut tx).await?; let role = catalog .get(&id) .cloned() .ok_or_else(|| error::Error::NotFound(format!("No data table role with id '{id}'")))?; - windmill_common::datatable_roles::drop_instance_role(&db, &role.name).await?; + windmill_common::datatable_roles::drop_instance_role(&db, &mut tx, &role.name).await?; catalog.remove(&id); - write_role_catalog(&db, &catalog).await?; + write_role_catalog(&mut tx, &catalog).await?; + tx.commit().await?; + // After the drop commits: a tenant naming a role that still exists is harmless, one naming a + // role that is gone is not, so this only ever runs once the cluster agrees it is gone. windmill_common::workspaces::forget_datatable_role_everywhere(&db, &id).await?; audit_log( diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 1ea9564abc..e9f0580162 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -2512,13 +2512,14 @@ async fn resolve_datatable_pg_as_caller( authed: &ApiAuthed, w_id: &str, datatable_name: &str, - role: Option<&str>, ) -> Result { let db_resource = get_datatable_resource_from_db( db, w_id, datatable_name, - role, + // The data table's default role. Browsing has no way to name another one yet; when the + // database manager grows a role picker it passes the pick through here. + None, DatatableAccess::Authed(authed.to_authed_ref()), ) .await?; @@ -2532,7 +2533,7 @@ async fn get_datatable_schema( w_id: &str, datatable_name: &str, ) -> Result { - let pg_db = resolve_datatable_pg_as_caller(db, authed, w_id, datatable_name, None).await?; + let pg_db = resolve_datatable_pg_as_caller(db, authed, w_id, datatable_name).await?; // Connect to the datatable database let (client, connection) = pg_db.connect(Some(db)).await?; @@ -2626,7 +2627,7 @@ async fn get_datatable_tables( w_id: &str, datatable_name: &str, ) -> Result { - let pg_db = resolve_datatable_pg_as_caller(db, authed, w_id, datatable_name, None).await?; + let pg_db = resolve_datatable_pg_as_caller(db, authed, w_id, datatable_name).await?; let (client, connection) = pg_db.connect(Some(db)).await?; tokio::spawn(async move { @@ -2704,7 +2705,7 @@ async fn get_datatable_table_columns( ))); } - let pg_db = resolve_datatable_pg_as_caller(db, authed, w_id, datatable_name, None).await?; + let pg_db = resolve_datatable_pg_as_caller(db, authed, w_id, datatable_name).await?; let (client, connection) = pg_db.connect(Some(db)).await?; tokio::spawn(async move { diff --git a/backend/windmill-common/src/datatable_roles.rs b/backend/windmill-common/src/datatable_roles.rs index 1a2f0df96b..19d396e763 100644 --- a/backend/windmill-common/src/datatable_roles.rs +++ b/backend/windmill-common/src/datatable_roles.rs @@ -45,6 +45,12 @@ pub struct InstanceDatatableRole { pub enabled: bool, /// Absent only for a role whose provisioning did not finish; resolving as it then errors /// rather than falling back to admin. + /// + /// A plain string rather than a `StringOrSecretRef` like the instance user's password beside + /// it: that one is a secret ref because an operator supplies it and may want it to come from + /// their own backend, while this one is minted here and never entered by anyone, so there is + /// nothing for a ref to point at. Encrypting generated secrets at rest is a separate change + /// that would take the replication password with it. #[serde(default, skip_serializing_if = "Option::is_none")] pub pwd: Option, } @@ -109,6 +115,23 @@ fn quote_literal(value: &str) -> String { format!("'{}'", value.replace('\'', "''")) } +/// Serialize every mutation of the catalog, from the read through the cluster DDL to the write. +/// +/// The catalog is one JSON document, so create/rename/enable/delete are all read-modify-write. +/// Without this, two concurrent creates both read the same snapshot, both succeed in the cluster, +/// and the second write drops the first — leaving a live Postgres login with a password nobody +/// recorded, which is exactly the state the whole delete path exists to avoid. Held for the +/// transaction, so the DDL has to run on that same transaction to be covered. +pub async fn lock_role_catalog(tx: &mut sqlx::Transaction<'_, sqlx::Postgres>) -> Result<()> { + sqlx::query!("SELECT pg_advisory_xact_lock(hashtext('datatable_role_catalog'))") + .execute(&mut **tx) + .await?; + Ok(()) +} + +/// Authorization: returns every role's stored Postgres password in plaintext. Callers MUST +/// restrict this to superadmin or internal server paths, and MUST NOT put what it returns into a +/// response, a log line or an audit record. pub async fn read_role_catalog(db: &DB) -> Result { let value = sqlx::query_scalar!( "SELECT value->'roles' FROM global_settings WHERE name = 'custom_instance_pg_databases'" @@ -116,10 +139,30 @@ pub async fn read_role_catalog(db: &DB) -> Result { .fetch_optional(db) .await? .flatten(); - match value { - Some(v) => Ok(serde_json::from_value(v).unwrap_or_default()), - None => Ok(Default::default()), - } + Ok(parse_role_catalog(value)) +} + +/// As [`read_role_catalog`], reading inside the caller's transaction so the value is the one +/// [`lock_role_catalog`] is protecting. Same authorization contract. +pub async fn read_role_catalog_tx( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, +) -> Result { + let value = sqlx::query_scalar!( + "SELECT value->'roles' FROM global_settings WHERE name = 'custom_instance_pg_databases'" + ) + .fetch_optional(&mut **tx) + .await? + .flatten(); + Ok(parse_role_catalog(value)) +} + +/// A catalog that will not deserialize is an empty one, which fails closed: tenants are keyed by +/// id independently of it, so every role then resolves to "no longer exists on this instance" +/// rather than to admin. +fn parse_role_catalog(value: Option) -> DatatableRoleCatalog { + value + .map(|v| serde_json::from_value(v).unwrap_or_default()) + .unwrap_or_default() } /// Resolve the role a caller named to its catalog id. A disabled role is an error rather than a @@ -161,6 +204,9 @@ pub async fn registered_instance_databases(db: &DB) -> Result> { /// `CONNECT` on `dbname` for every enabled role, and none for `PUBLIC`. Run at role creation, at /// database creation, and lazily whenever an instance data table is administered, so a database /// provisioned before a role existed is repaired rather than left silently unreachable. +/// +/// Authorization: rewrites a database's ACL with the server's own credentials and checks nothing. +/// Callers MUST restrict this to superadmin or internal server paths. pub async fn converge_connect_grants(db: &DB, dbname: &str) -> Result<()> { let catalog = read_role_catalog(db).await?; converge_connect_grants_with(db, dbname, &catalog).await @@ -188,13 +234,20 @@ pub async fn converge_connect_grants_with( /// `CREATE ROLE LOGIN PASSWORD ...; GRANT TO custom_instance_user`, and `CONNECT` on /// every registered database. No privileges beyond that — an admin grants them through SQL or the /// ACL editor. -pub async fn create_instance_role(db: &DB, name: &str, password: &str) -> Result<()> { +/// +/// Authorization: creates a cluster-wide Postgres login. Callers MUST restrict this to superadmin +/// paths, and MUST hold [`lock_role_catalog`] on the same transaction. +pub async fn create_instance_role( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + name: &str, + password: &str, +) -> Result<()> { validate_role_name(name)?; let exists = sqlx::query_scalar!( "SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)", name ) - .fetch_one(db) + .fetch_one(&mut **tx) .await? .unwrap_or(false); if exists { @@ -203,37 +256,59 @@ pub async fn create_instance_role(db: &DB, name: &str, password: &str) -> Result ))); } let quoted = quote_ident(name); - sqlx::raw_sql(&format!( - "CREATE ROLE {quoted} LOGIN PASSWORD {};\nGRANT {quoted} TO {};", - quote_literal(password), - quote_ident(CUSTOM_INSTANCE_USER), + // One statement per call rather than a batch: `raw_sql` takes the simple protocol, which is + // only needed for genuinely multi-statement SQL, and its future is not `Send` — which an axum + // handler holding this transaction requires. + sqlx::query(&format!( + "CREATE ROLE {quoted} LOGIN PASSWORD {}", + quote_literal(password) )) - .execute(db) + .execute(&mut **tx) + .await?; + sqlx::query(&format!( + "GRANT {quoted} TO {}", + quote_ident(CUSTOM_INSTANCE_USER) + )) + .execute(&mut **tx) .await?; Ok(()) } -pub async fn set_instance_role_login(db: &DB, name: &str, enabled: bool) -> Result<()> { +/// Authorization: alters a cluster-wide Postgres login. Callers MUST restrict this to superadmin +/// paths, and MUST hold [`lock_role_catalog`] on the same transaction. +pub async fn set_instance_role_login( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + name: &str, + enabled: bool, +) -> Result<()> { validate_role_name(name)?; sqlx::query(&format!( "ALTER ROLE {} {}", quote_ident(name), if enabled { "LOGIN" } else { "NOLOGIN" } )) - .execute(db) + .execute(&mut **tx) .await?; Ok(()) } /// A rename discards an md5-hashed password, so the caller has to hand over a fresh one. -pub async fn rename_instance_role(db: &DB, from: &str, to: &str, password: &str) -> Result<()> { +/// +/// Authorization: renames a cluster-wide Postgres login. Callers MUST restrict this to superadmin +/// paths, and MUST hold [`lock_role_catalog`] on the same transaction. +pub async fn rename_instance_role( + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + from: &str, + to: &str, + password: &str, +) -> Result<()> { validate_role_name(from)?; validate_role_name(to)?; let taken = sqlx::query_scalar!( "SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)", to ) - .fetch_one(db) + .fetch_one(&mut **tx) .await? .unwrap_or(false); if taken { @@ -241,14 +316,19 @@ pub async fn rename_instance_role(db: &DB, from: &str, to: &str, password: &str) "A Postgres role named '{to}' already exists on this cluster" ))); } - sqlx::raw_sql(&format!( - "ALTER ROLE {} RENAME TO {};\nALTER ROLE {} PASSWORD {};", + sqlx::query(&format!( + "ALTER ROLE {} RENAME TO {}", quote_ident(from), - quote_ident(to), - quote_ident(to), - quote_literal(password), + quote_ident(to) )) - .execute(db) + .execute(&mut **tx) + .await?; + sqlx::query(&format!( + "ALTER ROLE {} PASSWORD {}", + quote_ident(to), + quote_literal(password) + )) + .execute(&mut **tx) .await?; Ok(()) } @@ -262,7 +342,18 @@ pub async fn rename_instance_role(db: &DB, from: &str, to: &str, password: &str) /// owns the databases and can therefore revoke a grant whoever made it. `custom_instance_user` /// could only undo what it granted itself, so a privilege planted by an operator in psql — the /// ordinary way privileges reach a role — would survive and block the drop. -pub async fn drop_instance_role(db: &DB, name: &str) -> Result<()> { +/// +/// Authorization: drops a cluster-wide Postgres login and reassigns everything it owns. Callers +/// MUST restrict this to superadmin paths, and MUST hold [`lock_role_catalog`] on `tx`. +/// +/// The per-database passes open their own connections and cannot join `tx`; the lock is what keeps +/// a concurrent mutation out while they run. Only the final `DROP ROLE` is on `tx`, so it commits +/// or rolls back with the catalog write that forgets the role. +pub async fn drop_instance_role( + db: &DB, + tx: &mut sqlx::Transaction<'_, sqlx::Postgres>, + name: &str, +) -> Result<()> { validate_role_name(name)?; let quoted = quote_ident(name); let reassign = format!( @@ -291,8 +382,17 @@ pub async fn drop_instance_role(db: &DB, name: &str) -> Result<()> { })?; } - sqlx::raw_sql(&format!("{reassign}\nDROP ROLE {quoted};")) - .execute(db) + sqlx::query(&format!( + "REASSIGN OWNED BY {quoted} TO {}", + quote_ident(CUSTOM_INSTANCE_USER) + )) + .execute(&mut **tx) + .await?; + sqlx::query(&format!("DROP OWNED BY {quoted}")) + .execute(&mut **tx) + .await?; + sqlx::query(&format!("DROP ROLE {quoted}")) + .execute(&mut **tx) .await?; Ok(()) } diff --git a/frontend/src/lib/components/workspaceSettings/DataTableRolesSection.svelte b/frontend/src/lib/components/workspaceSettings/DataTableRolesSection.svelte index 59dbf813d5..e46b578597 100644 --- a/frontend/src/lib/components/workspaceSettings/DataTableRolesSection.svelte +++ b/frontend/src/lib/components/workspaceSettings/DataTableRolesSection.svelte @@ -10,7 +10,7 @@ import DataTable from '../table/DataTable.svelte' import Head from '../table/Head.svelte' import Row from '../table/Row.svelte' - import { Plus } from 'lucide-svelte' + import { Pencil, Plus } from 'lucide-svelte' import { SettingService, type InstanceDatatableRole } from '$lib/gen' import { sendUserToast } from '$lib/toast' @@ -42,10 +42,12 @@ try { await fn() sendUserToast(success) - await load() } catch (e) { sendUserToast(e?.body ?? e?.message ?? String(e), true) } finally { + // Reloaded whether or not it worked: the login toggle is driven by what the server + // holds, so a failed flip has to snap back rather than sit there claiming it landed. + await load() busy = false } } @@ -145,12 +147,17 @@ (renaming = undefined)} /> {:else} - +
+ {role.name} +
{/if}