diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5a9ef2b01c..46b2d5b5c7 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -8efda07e42b4154e50c6340eaf7f1be7d2a0f827 \ No newline at end of file +438e5efda9674b37d25e6c02e286adbf2566d25c \ No newline at end of file diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index fc627f115f..3bf04d1ef7 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -893,6 +893,14 @@ pub async fn set_global_setting_internal( ))); } + if key == EXTERNAL_INSTANCE_PG_SETTING { + return windmill_common::external_instance_pg::write_external_instance_pg_setting( + db, + Some(&value), + ) + .await; + } + run_setting_pre_write_hook(db, &key, &value).await?; match value { @@ -954,13 +962,6 @@ async fn run_setting_pre_write_hook( value: &serde_json::Value, ) -> error::Result<()> { match key { - EXTERNAL_INSTANCE_PG_SETTING => { - windmill_common::external_instance_pg::check_external_instance_pg_write( - db, - Some(value), - ) - .await?; - } // The instance AI config is written as an untyped blob through this generic // endpoint, so it never passes the typed check the workspace handler applies. // Rates that reach a cost total unbounded would make it negative or infinite. @@ -1281,7 +1282,7 @@ async fn set_instance_config( let desired_map = desired.global_settings.to_settings_map(); if !desired_map.is_empty() { let current_map = current.global_settings.to_settings_map(); - let settings_diff = + let mut settings_diff = instance_config::diff_global_settings(¤t_map, &desired_map, ApplyMode::Merge); let ai_config_changed = settings_diff .upserts @@ -1310,16 +1311,15 @@ async fn set_instance_config( } for (key, value) in &settings_diff.upserts { - run_setting_pre_write_hook(&db, key, value).await?; - } - if settings_diff - .deletes - .iter() - .any(|k| k == EXTERNAL_INSTANCE_PG_SETTING) - { - windmill_common::external_instance_pg::check_external_instance_pg_write(&db, None) - .await?; + if key != EXTERNAL_INSTANCE_PG_SETTING { + run_setting_pre_write_hook(&db, key, value).await?; + } } + windmill_common::external_instance_pg::write_external_instance_pg_from_diff( + &db, + &mut settings_diff, + ) + .await?; instance_config::apply_settings_diff(&db, &settings_diff) .await diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index cabd362cac..d8c1d7c497 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -3373,7 +3373,7 @@ fn apply_pg_tls_env( cmd.env("PGSSLROOTCERT", "system"); } Some("verify-ca") => { - if let Some(bundle) = system_ca_bundle() { + if let Some(bundle) = windmill_common::system_ca_bundle() { cmd.env("PGSSLROOTCERT", bundle); } } @@ -3383,21 +3383,6 @@ fn apply_pg_tls_env( Ok(None) } -fn system_ca_bundle() -> Option { - std::env::var_os("SSL_CERT_FILE") - .map(std::path::PathBuf::from) - .into_iter() - .chain( - [ - "/etc/ssl/certs/ca-certificates.crt", - "/etc/pki/tls/certs/ca-bundle.crt", - "/etc/ssl/cert.pem", - "/etc/ssl/ca-bundle.pem", - ] - .map(std::path::PathBuf::from), - ) - .find(|path| path.is_file()) -} #[cfg(test)] mod pg_tls_env_tests { @@ -8352,6 +8337,13 @@ async fn apply_forked_datatable( })?, }; + if database.resource_type == DataTableCatalogResourceType::ExternalInstance { + windmill_common::external_instance_pg::ensure_external_instance_database_registered( + tx, + &fdt.new_dbname, + ) + .await?; + } if database.resource_type.is_windmill_managed() { // The whole `database` object, not just its `resource_path`: a pointer entry has none to // patch. `reference` goes with it — exactly one of the two may be set. The copy was created diff --git a/backend/windmill-common/src/external_instance_pg.rs b/backend/windmill-common/src/external_instance_pg.rs index cb5691b2a6..c9cb9bb1a2 100644 --- a/backend/windmill-common/src/external_instance_pg.rs +++ b/backend/windmill-common/src/external_instance_pg.rs @@ -302,23 +302,73 @@ pub async fn ensure_external_instance_database_registered( ))) } -/// Check a write to [`EXTERNAL_INSTANCE_PG_SETTING`] before it happens: `None`, null or an empty -/// string unsets it. Every writer of global settings calls this, the per-key and bulk endpoints -/// as well as the declarative sync. -pub async fn check_external_instance_pg_write( +/// Write [`EXTERNAL_INSTANCE_PG_SETTING`]: `None`, null or an empty string unsets it. Every writer +/// of global settings goes through this for that key — the per-key and bulk endpoints as well as +/// the declarative sync — instead of writing the row itself. +/// +/// The checks and the write share one transaction holding [`lock_external_instance_pg_state`]. A +/// check taken outside it could pass while a database create still reads the old cluster, which +/// would then register a database there after the setting names another one. +/// +/// Authorization: checks nothing. Callers MUST be superadmin. +pub async fn write_external_instance_pg_setting( db: &DB, value: Option<&serde_json::Value>, ) -> Result<()> { + let value = match value { + None | Some(serde_json::Value::Null) => None, + Some(serde_json::Value::String(s)) if s.trim().is_empty() => None, + Some(value) => Some(value), + }; + let mut tx = db.begin().await?; + lock_external_instance_pg_state(&mut tx).await?; match value { - None | Some(serde_json::Value::Null) => ensure_external_instance_pg_removable(db).await, - Some(serde_json::Value::String(s)) if s.trim().is_empty() => { - ensure_external_instance_pg_removable(db).await + None => { + ensure_external_instance_pg_removable(db).await?; + sqlx::query("DELETE FROM global_settings WHERE name = $1") + .bind(EXTERNAL_INSTANCE_PG_SETTING) + .execute(&mut *tx) + .await?; } Some(value) => { crate::external_instance_pg_oss::validate_external_instance_pg_setting(value)?; - ensure_external_instance_pg_not_repointed(db, value).await + ensure_external_instance_pg_not_repointed(db, value).await?; + sqlx::query( + "INSERT INTO global_settings (name, value) VALUES ($1, $2) + ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value, updated_at = now()", + ) + .bind(EXTERNAL_INSTANCE_PG_SETTING) + .bind(value) + .execute(&mut *tx) + .await?; } } + tx.commit().await?; + tracing::info!( + "{} global setting {EXTERNAL_INSTANCE_PG_SETTING}", + if value.is_some() { "Set" } else { "Unset" } + ); + Ok(()) +} + +/// [`write_external_instance_pg_setting`] for a settings diff: writes the key if the diff touches +/// it, and takes it out of the diff so the generic apply does not write it again. +pub async fn write_external_instance_pg_from_diff( + db: &DB, + diff: &mut crate::instance_config::SettingsDiff, +) -> Result<()> { + if let Some(value) = diff.upserts.remove(EXTERNAL_INSTANCE_PG_SETTING) { + write_external_instance_pg_setting(db, Some(&value)).await?; + } + if let Some(i) = diff + .deletes + .iter() + .position(|k| k == EXTERNAL_INSTANCE_PG_SETTING) + { + diff.deletes.remove(i); + write_external_instance_pg_setting(db, None).await?; + } + Ok(()) } /// Refuse pointing the setting at another host or port while databases or data table roles live on diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index d26b974717..f5683db687 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -1395,14 +1395,8 @@ pub async fn sync_global_settings_declarative( crate::global_settings::parse_allowed_origins_setting(desired.get(origins_key)) .map_err(|e| anyhow::anyhow!("{origins_key}: {e}"))?; - let diff = diff_global_settings(current, desired, ApplyMode::Replace); - let external_pg_key = crate::global_settings::EXTERNAL_INSTANCE_PG_SETTING; - if diff.deletes.iter().any(|k| k == external_pg_key) { - crate::external_instance_pg::check_external_instance_pg_write(db, None).await?; - } - if let Some(value) = diff.upserts.get(external_pg_key) { - crate::external_instance_pg::check_external_instance_pg_write(db, Some(value)).await?; - } + let mut diff = diff_global_settings(current, desired, ApplyMode::Replace); + crate::external_instance_pg::write_external_instance_pg_from_diff(db, &mut diff).await?; apply_settings_diff(db, &diff).await?; Ok(()) diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 1122c46ac8..6d84f136ff 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -1664,6 +1664,24 @@ pub async fn create_custom_instance_database( Ok(()) } +/// The system's CA bundle file, for libpq clients that cannot take `sslrootcert=system`: that value +/// needs libpq 16, and verify-full only. +pub fn system_ca_bundle() -> Option { + std::env::var_os("SSL_CERT_FILE") + .map(std::path::PathBuf::from) + .into_iter() + .chain( + [ + "/etc/ssl/certs/ca-certificates.crt", + "/etc/pki/tls/certs/ca-bundle.crt", + "/etc/ssl/cert.pem", + "/etc/ssl/ca-bundle.pem", + ] + .map(std::path::PathBuf::from), + ) + .find(|path| path.is_file()) +} + /// Connection options parsed from a database URL. /// /// The only place a database URL becomes `PgConnectOptions`. Providers that mint the password diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs index 87bbba4761..9780bb35f0 100644 --- a/backend/windmill-common/src/workspaces.rs +++ b/backend/windmill-common/src/workspaces.rs @@ -3018,6 +3018,14 @@ async fn register_fork_ducklake_namespace( { return Ok(()); } + let mut tx = db.begin().await?; + // A row naming an external database counts as a use of it. Written under the lock a drop takes, + // and only while the database is still registered, so a drop cannot slip in between the + // settings this attach resolved and the row that protects the database. + if let Some(dbname) = catalog.strip_prefix("external_instance:") { + crate::external_instance_pg::ensure_external_instance_database_registered(&mut tx, dbname) + .await?; + } sqlx::query!( "INSERT INTO fork_ducklake_namespace (workspace_id, ducklake_name, metadata_schema, catalog, storage, storage_ref, data_path) @@ -3032,9 +3040,10 @@ async fn register_fork_ducklake_namespace( &storage_ref, data_path, ) - .execute(db) + .execute(&mut *tx) .await .map_err(|e| Error::internal_err(format!("registering fork ducklake namespace: {e:#}")))?; + tx.commit().await?; let mut locations = FORK_DUCKLAKE_REGISTERED .get(w_id) .filter(|(_, exp)| *exp > now) diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 3ad0e1edba..d2e0400192 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -2240,11 +2240,64 @@ fn parse_attach_db_resource<'a>(query: &'a str) -> Option Result> { + let mode = match res.sslmode.as_deref() { + Some(mode @ ("verify-ca" | "verify-full")) if res.accept_invalid_certs == Some(false) => mode, + _ => return Ok(None), + }; + let bundle = windmill_common::system_ca_bundle() + .map(std::fs::read_to_string) + .transpose() + .map_err(|e| Error::ExecutionErr(format!("Failed to read the system CA bundle: {e}")))? + .unwrap_or_default(); + let pem = res.root_certificate_pem.as_deref().unwrap_or_default(); + if bundle.is_empty() && pem.is_empty() { + return Err(Error::ExecutionErr(format!( + "sslmode {mode} needs a root certificate, and this worker has no system CA bundle" + ))); + } + let roots = format!("{bundle}\n{pem}\n"); + use sha2::Digest; + let path = std::env::temp_dir().join(format!( + "windmill-pg-roots-{}.pem", + hex::encode(&sha2::Sha256::digest(roots.as_bytes())[..8]) + )); + if !path.is_file() { + // Renamed into place: a job attaching concurrently must never read a half-written file. + let partial = path.with_extension(format!("{}.partial", Uuid::new_v4())); + std::fs::write(&partial, &roots) + .and_then(|()| std::fs::rename(&partial, &path)) + .map_err(|e| Error::ExecutionErr(format!("Failed to write root certificates: {e}")))?; + } + Ok(Some((mode, path))) +} + +fn pg_attach_uri(res: &PgDatabase) -> Result { + let uri = res.to_uri(); + let Some((mode, roots)) = pg_attach_verification(res)? else { + return Ok(uri); + }; + let base = uri.strip_suffix("?sslmode=require").ok_or_else(|| { + Error::internal_err("unexpected sslmode in a postgres connection URI".to_string()) + })?; + Ok(format!( + "{base}?sslmode={mode}&sslrootcert={}", + urlencoding::encode(&roots.to_string_lossy()) + )) +} + fn format_attach_db_conn_str(db_resource: Value, db_type: &str) -> Result { let s = match db_type.to_lowercase().as_str() { "postgres" | "postgresql" => { let res: PgDatabase = serde_json::from_value(db_resource)?; - res.to_uri() + pg_attach_uri(&res)? } #[cfg(feature = "mysql")] "mysql" => { @@ -2703,10 +2756,21 @@ fn pg_secret_attach_statements(db_resource: Value, alias_name: &str) -> Result "disable", - Some("require") | Some("verify-ca") | Some("verify-full") => "require", - _ => "prefer", + let sslmode = match pg_attach_verification(&res)? { + // A libpq keyword/value string: the path is quoted for libpq, then for the DuckDB literal. + Some((mode, roots)) => format!( + "{mode} sslrootcert=''{}''", + roots + .to_string_lossy() + .replace('\\', "\\\\") + .replace('\'', "\\''") + ), + None => match res.sslmode.as_deref() { + Some("disable") => "disable", + Some("require") | Some("verify-ca") | Some("verify-full") => "require", + _ => "prefer", + } + .to_string(), }; let secret_name = datatable_secret_name(alias_name); Ok(vec![ @@ -2794,6 +2858,36 @@ pub struct Arg { mod tests { use super::*; + #[test] + fn pg_attach_keeps_verification_only_when_required() { + let pg = |sslmode: &str, accept_invalid_certs: Option| PgDatabase { + host: "db.internal".to_string(), + user: Some("custom_instance_user".to_string()), + password: Some("pw".to_string()), + port: None, + sslmode: Some(sslmode.to_string()), + dbname: "dt".to_string(), + root_certificate_pem: Some("-----BEGIN CERTIFICATE-----test".to_string()), + accept_invalid_certs, + use_iam_auth: None, + region: None, + }; + let uri = pg_attach_uri(&pg("verify-full", Some(false))).unwrap(); + assert!(uri.contains("?sslmode=verify-full&sslrootcert="), "{uri}"); + let root = urlencoding::decode(uri.split("sslrootcert=").nth(1).unwrap()).unwrap(); + let roots = std::fs::read_to_string(root.as_ref()).unwrap(); + assert!(roots.contains("-----BEGIN CERTIFICATE-----test")); + let external = serde_json::to_value(pg("verify-full", Some(false))).unwrap(); + let attach = &pg_secret_attach_statements(external, "dt").unwrap()[3]; + assert!( + attach.starts_with(&format!("ATTACH 'sslmode=verify-full sslrootcert=''{}''", root)), + "{attach}" + ); + // A resource that never opted in keeps the historical downgrade. + assert!(pg_attach_uri(&pg("verify-full", None)).unwrap().ends_with("?sslmode=require")); + assert!(pg_attach_uri(&pg("require", Some(false))).unwrap().ends_with("?sslmode=require")); + } + #[test] fn attach_datatable_parses_name_and_role() { let reference_of = |q: &str| parse_attach_datatable(q).unwrap().reference;