From afbbc0b1b2da1e468b10d4cd11e9f3ac7478a34d Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 12 Oct 2026 16:37:05 +0200 Subject: [PATCH] feat: audit declarative instance config writes (#11684) * feat: audit declarative instance config writes Co-Authored-By: Claude Opus 5.5 * fix: keep early logs, base url and license plan for declarative writers Co-Authored-By: Claude Opus 5.5 * refactor: require an observer on every config apply, drop license loading Co-Authored-By: Claude Opus 5.5 * fix: export only audit events from the declarative subcommands Co-Authored-By: Claude Opus 5.5 * chore: update ee-repo-ref to d6a6cdd343e7744ea49c610f9534c8013c8ffc2a This commit updates the EE repository reference after PR #872 was merged in windmill-ee-private. Previous ee-repo-ref: fc51c31bafb5975694fffe1da51f7b41fbf127a6 New ee-repo-ref: d6a6cdd343e7744ea49c610f9534c8013c8ffc2a Automated by sync-ee-ref workflow. --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: windmill-internal-app[bot] --- backend/Cargo.lock | 2 + backend/Cargo.toml | 1 + backend/ee-repo-ref.txt | 2 +- backend/src/main.rs | 71 ++++++-- backend/tests/instance_config.rs | 166 +++++++++++++++--- backend/windmill-api-schedule/src/lib.rs | 3 +- backend/windmill-api-settings/src/lib.rs | 147 ++-------------- backend/windmill-audit/src/instance_config.rs | 148 ++++++++++++++++ backend/windmill-audit/src/lib.rs | 1 + .../src/external_instance_pg.rs | 13 +- .../windmill-common/src/instance_config.rs | 76 +++++++- backend/windmill-common/src/otel_oss.rs | 13 ++ backend/windmill-common/src/tracing_init.rs | 22 +++ backend/windmill-operator/Cargo.toml | 1 + 14 files changed, 483 insertions(+), 183 deletions(-) create mode 100644 backend/windmill-audit/src/instance_config.rs diff --git a/backend/Cargo.lock b/backend/Cargo.lock index aa3d601492..90f40da8ca 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -15536,6 +15536,7 @@ dependencies = [ "windmill-api-scripts", "windmill-api-settings", "windmill-api-users", + "windmill-audit", "windmill-autoscaling", "windmill-common", "windmill-dep-map", @@ -16630,6 +16631,7 @@ dependencies = [ "thiserror 2.0.21", "tokio", "tracing", + "windmill-audit", "windmill-common", ] diff --git a/backend/Cargo.toml b/backend/Cargo.toml index d627328e27..15e9ec8e4d 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -281,6 +281,7 @@ windmill-api = { workspace = true, default-features = false } windmill-api-agent-workers = { workspace = true, optional = true } windmill-api-scripts.workspace = true windmill-api-settings.workspace = true +windmill-audit.workspace = true windmill-worker.workspace = true windmill-indexer = { workspace = true, optional = true } windmill-autoscaling = { workspace = true, optional = true } diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 012f205d9c..fbb75726df 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -06610ecc950ed64a2709d7f1aa8c7b49a95fdbae +d6a6cdd343e7744ea49c610f9534c8013c8ffc2a diff --git a/backend/src/main.rs b/backend/src/main.rs index b8fdd9353f..32e183491b 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -471,6 +471,38 @@ fn print_help() { println!("- At startup, Windmill logs currently set configuration keys for visibility."); } +async fn otel_environment(conn: &Connection) -> String { + if let Ok(environment) = std::env::var("OTEL_ENVIRONMENT") { + environment + } else { + load_base_url(conn) + .await + .unwrap_or_else(|_| "local".to_string()) + .trim_start_matches("https://") + .trim_start_matches("http://") + .split(".") + .next() + .unwrap_or_else(|| "local") + .to_string() + } +} + +/// Tracing for a subcommand that writes instance config. It never runs the server startup +/// that loads the OTel settings, and its audit events are exported under them. +/// +/// The subscriber it installs depends on those settings, so `early_logs` is what prints +/// the events of connecting and loading them; it gives way once the real one is in. +async fn init_declarative_writer_tracing( + db: &windmill_common::DB, + service: &str, + early_logs: tracing::subscriber::DefaultGuard, +) { + load_otel(db).await; + let environment = otel_environment(&Connection::Sql(db.clone())).await; + drop(early_logs); + windmill_common::tracing_init::initialize_subcommand_tracing(service, &HOSTNAME, &environment); +} + async fn windmill_main() -> anyhow::Result<()> { let (killpill_tx, mut killpill_rx) = KillpillSender::new(2); let mut monitor_killpill_rx = killpill_tx.subscribe(); @@ -574,7 +606,6 @@ async fn windmill_main() -> anyhow::Result<()> { return Ok(()); } "sync-config" => { - tracing_subscriber::fmt::init(); let path = std::env::args().nth(2).unwrap_or_else(|| { eprintln!("Usage: windmill sync-config "); std::process::exit(1); @@ -588,15 +619,34 @@ async fn windmill_main() -> anyhow::Result<()> { windmill_common::instance_config::resolve_env_refs(&mut config.global_settings) .map_err(|var| anyhow::anyhow!("environment variable '{var}' not found"))?; + let early_logs = tracing::subscriber::set_default(tracing_subscriber::fmt().finish()); tracing::info!("Connecting to database..."); let db = crate::db_connect::initial_connection().await?; - config.sync_to_db(&db).await?; + init_declarative_writer_tracing(&db, "windmill-sync-config", early_logs).await; + + let author = windmill_audit::instance_config::declarative_sync_author(); + let synced = config + .sync_to_db( + &db, + &windmill_audit::instance_config::InstanceConfigAuditor { + db: &db, + author: &author, + source: Some( + windmill_audit::instance_config::DeclarativeSource::SyncConfig, + ), + }, + ) + .await; + // Also on failure: the settings written before it are already audited, and the + // process exits before the exporter's next scheduled batch. + windmill_common::otel_oss::flush_otel_logs().await; + synced?; tracing::info!("Synced instance config from {path}"); return Ok(()); } #[cfg(feature = "operator")] "operator" => { - tracing_subscriber::fmt::init(); + let early_logs = tracing::subscriber::set_default(tracing_subscriber::fmt().finish()); tracing::info!("Starting Windmill Kubernetes operator..."); tracing::info!("Connecting to database..."); @@ -609,6 +659,7 @@ async fn windmill_main() -> anyhow::Result<()> { operator_killpill_rx, ) .await?; + init_declarative_writer_tracing(&db, "windmill-operator", early_logs).await; #[cfg(all(feature = "enterprise", feature = "private"))] tokio::spawn(async move { @@ -735,19 +786,7 @@ async fn windmill_main() -> anyhow::Result<()> { (Connection::Sql(db), None, None) }; - let environment = if let Ok(environment) = std::env::var("OTEL_ENVIRONMENT") { - environment - } else { - load_base_url(&conn) - .await - .unwrap_or_else(|_| "local".to_string()) - .trim_start_matches("https://") - .trim_start_matches("http://") - .split(".") - .next() - .unwrap_or_else(|| "local") - .to_string() - }; + let environment = otel_environment(&conn).await; let _guard = windmill_common::tracing_init::initialize_tracing(&hostname, &mode, &environment); diff --git a/backend/tests/instance_config.rs b/backend/tests/instance_config.rs index 52ebbee7e7..24937a3ce5 100644 --- a/backend/tests/instance_config.rs +++ b/backend/tests/instance_config.rs @@ -237,7 +237,7 @@ async fn test_diff_global_settings_empty_string_routes_to_delete(db: Pool) { ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); assert_eq!( get_global_setting(&db, "key_a").await, @@ -468,7 +468,7 @@ async fn test_apply_settings_diff_deletes_only(db: Pool) { ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); assert!(get_global_setting(&db, "to_delete_1").await.is_none()); assert!(get_global_setting(&db, "to_delete_2").await.is_none()); @@ -489,7 +489,7 @@ async fn test_apply_settings_diff_upserts_and_deletes(db: Pool) { ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); assert!(get_global_setting(&db, "old_key").await.is_none()); assert_eq!( @@ -503,7 +503,7 @@ async fn test_apply_settings_diff_empty_noop(db: Pool) { insert_global_setting(&db, "preexisting", serde_json::json!("value")).await; let diff = SettingsDiff::default(); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); assert_eq!( get_global_setting(&db, "preexisting").await, @@ -525,7 +525,7 @@ async fn test_apply_settings_diff_upsert_overwrites(db: Pool) { ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); assert_eq!( get_global_setting(&db, "overwrite_me").await, @@ -553,7 +553,7 @@ async fn test_apply_settings_diff_complex_json(db: Pool) { ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); let stored = get_global_setting(&db, "complex_setting").await.unwrap(); assert_eq!(stored, complex_value); @@ -568,7 +568,7 @@ async fn test_apply_settings_diff_delete_nonexistent_is_noop(db: Pool) }; // Should not error - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); } // ======================================================================== @@ -589,7 +589,7 @@ async fn test_apply_configs_diff_upserts_with_prefix(db: Pool) { deletes: vec![], }; - apply_configs_diff(&db, &diff).await.unwrap(); + apply_configs_diff(&db, &diff, &()).await.unwrap(); let stored = get_config(&db, "worker__mygroup").await.unwrap(); assert_eq!(stored["init_bash"], "echo hi"); @@ -601,7 +601,7 @@ async fn test_apply_configs_diff_deletes_with_prefix(db: Pool) { let diff = ConfigsDiff { upserts: BTreeMap::new(), deletes: vec!["to_remove".to_string()] }; - apply_configs_diff(&db, &diff).await.unwrap(); + apply_configs_diff(&db, &diff, &()).await.unwrap(); assert!(get_config(&db, "worker__to_remove").await.is_none()); } @@ -619,7 +619,7 @@ async fn test_apply_configs_diff_upsert_overwrites(db: Pool) { deletes: vec![], }; - apply_configs_diff(&db, &diff).await.unwrap(); + apply_configs_diff(&db, &diff, &()).await.unwrap(); let stored = get_config(&db, "worker__grp").await.unwrap(); assert_eq!(stored, serde_json::json!({"new": true})); @@ -630,7 +630,7 @@ async fn test_apply_configs_diff_empty_noop(db: Pool) { insert_config(&db, "worker__keep", serde_json::json!({"keep": true})).await; let diff = ConfigsDiff::default(); - apply_configs_diff(&db, &diff).await.unwrap(); + apply_configs_diff(&db, &diff, &()).await.unwrap(); assert!(get_config(&db, "worker__keep").await.is_some()); } @@ -641,7 +641,7 @@ async fn test_apply_configs_diff_does_not_touch_non_worker_configs(db: Pool) { assert!(diff.upserts.contains_key("expose_metrics")); assert!(diff.deletes.is_empty()); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); // Step 5: Read back and verify let config2 = InstanceConfig::from_db(&db).await.unwrap(); @@ -729,7 +729,7 @@ async fn test_roundtrip_replace_mode_deletes(db: Pool) { assert!(diff.deletes.contains(&"retention_period_secs".to_string())); assert!(diff.deletes.contains(&"expose_metrics".to_string())); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); assert!(get_global_setting(&db, "retention_period_secs") .await @@ -782,7 +782,7 @@ async fn test_roundtrip_worker_configs(db: Pool) { assert!(diff.upserts.contains_key("gpu")); // new assert_eq!(diff.deletes, vec!["legacy".to_string()]); - apply_configs_diff(&db, &diff).await.unwrap(); + apply_configs_diff(&db, &diff, &()).await.unwrap(); // Verify let config2 = InstanceConfig::from_db(&db).await.unwrap(); @@ -820,7 +820,7 @@ async fn test_roundtrip_to_settings_map_from_db_consistency(db: Pool) let diff = SettingsDiff { upserts: map.into_iter().collect(), deletes: vec![], ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); let config = InstanceConfig::from_db(&db).await.unwrap(); assert_eq!(config.global_settings.base_url, original.base_url); @@ -868,8 +868,8 @@ async fn test_idempotent_apply(db: Pool) { }; // Apply twice - apply_settings_diff(&db, &diff).await.unwrap(); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); assert_eq!( get_global_setting(&db, "idem_key").await, @@ -1069,7 +1069,7 @@ async fn test_diff_apply_only_touches_changed_rows(db: Pool) { assert!(!diff.upserts.contains_key("unchanged_1")); assert!(!diff.upserts.contains_key("unchanged_2")); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); // Verify all 4 settings present assert_eq!(count_global_settings(&db).await, 4); @@ -1148,7 +1148,7 @@ async fn test_replace_mode_protects_settings_in_integration(db: Pool) // But normal_setting should be deleted assert!(diff.deletes.contains(&"normal_setting".to_string())); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); // Verify protected settings survived assert!(get_global_setting(&db, "ducklake_settings").await.is_some()); @@ -1196,7 +1196,7 @@ async fn test_jwt_secret_roundtrip_as_string_or_secret_ref(db: Pool) { let diff = SettingsDiff { upserts: map.into_iter().collect(), deletes: vec![], ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); // Read back from DB — jwt_secret should survive the roundtrip let config = InstanceConfig::from_db(&db).await.unwrap(); @@ -1239,7 +1239,7 @@ async fn test_rsa_keys_roundtrip_via_extra(db: Pool) { let map = settings.to_settings_map(); let diff = SettingsDiff { upserts: map.into_iter().collect(), deletes: vec![], ..Default::default() }; - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); // Read back from DB let config = InstanceConfig::from_db(&db).await.unwrap(); @@ -1296,7 +1296,7 @@ async fn test_replace_mode_protects_jwt_secret_and_rsa_keys(db: Pool) "normal_setting should be deleted" ); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); // Both protected settings survive assert_eq!( @@ -1371,7 +1371,7 @@ async fn test_alert_config_in_global_settings_roundtrip(db: Pool) { "alert_config change should be detected in diff" ); - apply_settings_diff(&db, &diff).await.unwrap(); + apply_settings_diff(&db, &diff, &()).await.unwrap(); // Re-read and verify the update let config2 = InstanceConfig::from_db(&db).await.unwrap(); @@ -1455,6 +1455,7 @@ async fn declarative_sync_rejects_an_unusable_webhook_base_url(db: Pool &db, &BTreeMap::new(), &desired, + &(), ) .await .expect_err("a javascript: banner link must fail the sync"); @@ -1545,6 +1548,7 @@ async fn declarative_sync_rejects_a_malformed_max_token_expiration(db: Pool) { &db, &BTreeMap::new(), &desired, + &(), ) .await .expect_err("a non-hex accent color must fail the sync"); @@ -1681,7 +1691,109 @@ async fn declarative_sync_rejects_a_non_hex_accent_color(db: Pool) { &db, &BTreeMap::new(), &desired, + &(), ) .await .expect("a hex accent color must sync"); } + +/// A declarative sync has no user request behind it, so its observer is the only thing +/// that audits it: one entry per setting or worker config it actually writes, values +/// recorded under the API path's rule, and none for what it leaves as is. +#[cfg(all(feature = "enterprise", feature = "private"))] +#[sqlx::test(fixtures("base"))] +async fn test_declarative_sync_audits_only_what_it_writes(db: Pool) { + use windmill_audit::instance_config::{ + declarative_sync_author, DeclarativeSource, InstanceConfigAuditor, + }; + + insert_global_setting(&db, "retention_period_secs", serde_json::json!(100)).await; + insert_global_setting(&db, "expose_metrics", serde_json::json!(true)).await; + insert_global_setting(&db, "scim_token", serde_json::json!("old-scim-secret")).await; + insert_global_setting( + &db, + "npm_config_registry", + serde_json::json!("https://u:p@npm"), + ) + .await; + insert_config( + &db, + "worker__kept", + serde_json::json!({"worker_tags": ["a"]}), + ) + .await; + insert_config( + &db, + "worker__dropped", + serde_json::json!({"worker_tags": ["b"]}), + ) + .await; + + let config: windmill_common::instance_config::InstanceConfig = + serde_json::from_value(serde_json::json!({ + "global_settings": { + "retention_period_secs": 200, + "expose_metrics": true, + "scim_token": "new-scim-secret", + }, + "worker_configs": { + "kept": {"worker_tags": ["a"]}, + "added": {"init_bash": "export TOKEN=init-secret"}, + }, + })) + .unwrap(); + let author = declarative_sync_author(); + config + .sync_to_db( + &db, + &InstanceConfigAuditor { + db: &db, + author: &author, + source: Some(DeclarativeSource::SyncConfig), + }, + ) + .await + .unwrap(); + + let entries: Vec<(String, String, serde_json::Value)> = sqlx::query_as( + "SELECT operation, resource, parameters FROM audit_partitioned \ + WHERE workspace_id = 'global' AND username = $1 AND resource = ANY($2) \ + ORDER BY resource", + ) + .bind(&author.username) + .bind(vec![ + "retention_period_secs", + "expose_metrics", + "scim_token", + "npm_config_registry", + "worker__kept", + "worker__dropped", + "worker__added", + ]) + .fetch_all(&db) + .await + .unwrap(); + + let source = serde_json::json!({"source": "sync-config"}); + let entry = |operation: &str, resource: &str, parameters: &serde_json::Value| { + ( + operation.to_string(), + resource.to_string(), + parameters.clone(), + ) + }; + assert_eq!( + entries, + vec![ + entry("global_settings.delete", "npm_config_registry", &source), + entry( + "global_settings.update", + "retention_period_secs", + &serde_json::json!({"source": "sync-config", "value": "200"}), + ), + entry("global_settings.update", "scim_token", &source), + entry("worker_config.update", "worker__added", &source), + entry("worker_config.delete", "worker__dropped", &source), + ] + ); +} diff --git a/backend/windmill-api-schedule/src/lib.rs b/backend/windmill-api-schedule/src/lib.rs index 855f687431..b8ee9ded91 100644 --- a/backend/windmill-api-schedule/src/lib.rs +++ b/backend/windmill-api-schedule/src/lib.rs @@ -1492,7 +1492,8 @@ async fn set_default_error_handler( } else { windmill_api_settings::delete_global_setting(&db, key.as_str()).await?; } - windmill_api_settings::audit_setting_write(&db, &authed, &key, value.as_ref()).await?; + windmill_audit::instance_config::audit_setting_write(&db, &authed, &key, value.as_ref(), None) + .await?; if payload.override_existing { // The rewrite and its history rows go in one transaction: on separate diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index 116ff73457..1359e8c142 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -29,7 +29,11 @@ mod log_cleanup; mod storage_usage; use windmill_api_auth::{require_devops_role, require_super_admin, ApiAuthed}; -use windmill_audit::{audit_oss::audit_log, ActionKind}; +use windmill_audit::{ + audit_oss::audit_log, + instance_config::{audit_setting_write, InstanceConfigAuditor}, + ActionKind, +}; use windmill_common::hub_resource_types::{self, SyncSource}; use windmill_common::utils::HTTP_CLIENT_PERMISSIVE as HTTP_CLIENT; use windmill_common::DB; @@ -833,50 +837,7 @@ pub async fn set_global_setting( require_super_admin(&db, &authed).await?; let value = value.value.unwrap_or(serde_json::Value::Null); set_global_setting_internal(&db, key.clone(), value.clone()).await?; - audit_setting_write(&db, &authed, &key, Some(&value)).await -} - -/// A setting is recorded by name, with its value only when that is a boolean or a number. -/// Any other shape can carry a credential (a registry URL, a webhook, a header map), and no -/// list of secret-bearing settings stays complete as settings are added. -/// -/// AUTHORIZATION: records a write, checks nothing. The caller must have established that -/// `authed` may change instance settings before performing the write this records. -pub async fn audit_setting_write( - db: &DB, - authed: &ApiAuthed, - key: &str, - value: Option<&serde_json::Value>, -) -> error::Result<()> { - let deleted = match value { - None | Some(serde_json::Value::Null) => true, - Some(serde_json::Value::String(s)) => s.trim().is_empty(), - Some(_) => false, - }; - let recorded = match value { - Some(v @ (serde_json::Value::Bool(_) | serde_json::Value::Number(_))) => { - Some(v.to_string()) - } - _ => None, - }; - audit_log( - db, - authed, - if deleted { - "global_settings.delete" - } else { - "global_settings.update" - }, - if deleted { - ActionKind::Delete - } else { - ActionKind::Update - }, - "global", - Some(key), - recorded.as_deref().map(|v| [("value", v)].into()), - ) - .await + audit_setting_write(&db, &authed, &key, Some(&value), None).await } async fn audit_instance_action( @@ -1367,65 +1328,17 @@ async fn set_instance_config( run_setting_pre_write_hook(&db, key, value).await?; } } + let auditor = InstanceConfigAuditor { db: &db, author: &authed, source: None }; // The external-instance write removes its own key from the diff. - let external = settings_diff - .upserts - .get(EXTERNAL_INSTANCE_PG_SETTING) - .cloned() - .map(Some) - .or_else(|| { - settings_diff - .deletes - .iter() - .any(|k| k == EXTERNAL_INSTANCE_PG_SETTING) - .then_some(None) - }); windmill_common::external_instance_pg::write_external_instance_pg_from_diff( &db, &mut settings_diff, + &auditor, ) .await?; - if let Some(value) = external { - audit_setting_write(&db, &authed, EXTERNAL_INSTANCE_PG_SETTING, value.as_ref()).await?; - } - - // Applied and audited one setting at a time: the writes are not transactional, so a - // batch that fails midway must not leave the settings it did write unrecorded. - let instance_config::SettingsDiff { - upserts, - deletes, - mut previous_values, - unchanged_count, - } = settings_diff; - let steps = upserts - .into_iter() - .map(|(k, v)| (k, Some(v))) - .chain(deletes.into_iter().map(|k| (k, None))) - .map(Some) - .chain(std::iter::once(None)); - for step in steps { - let mut diff = instance_config::SettingsDiff::default(); - match &step { - Some((key, value)) => { - if let Some(previous) = previous_values.remove(key) { - diff.previous_values.insert(key.clone(), previous); - } - match value { - Some(value) => { - diff.upserts.insert(key.clone(), value.clone()); - } - None => diff.deletes.push(key.clone()), - } - } - None => diff.unchanged_count = unchanged_count, - } - instance_config::apply_settings_diff(&db, &diff) - .await - .map_err(|e| error::Error::internal_err(e.to_string()))?; - if let Some((key, value)) = &step { - audit_setting_write(&db, &authed, key, value.as_ref()).await?; - } - } + instance_config::apply_settings_diff(&db, &settings_diff, &auditor) + .await + .map_err(|e| error::Error::internal_err(e.to_string()))?; if ai_config_changed { bump_instance_ai_config_revision(); @@ -1455,37 +1368,13 @@ async fn set_instance_config( .collect(); let configs_diff = instance_config::diff_worker_configs(¤t_wc, &desired_wc, ApplyMode::Merge); - let steps = configs_diff - .upserts - .into_iter() - .map(|(g, c)| (g, Some(c))) - .chain(configs_diff.deletes.into_iter().map(|g| (g, None))); - for (group, config) in steps { - let mut diff = instance_config::ConfigsDiff::default(); - let (operation, action_kind) = match config { - Some(config) => { - diff.upserts.insert(group.clone(), config); - ("worker_config.update", ActionKind::Update) - } - None => { - diff.deletes.push(group.clone()); - ("worker_config.delete", ActionKind::Delete) - } - }; - instance_config::apply_configs_diff(&db, &diff) - .await - .map_err(|e| error::Error::internal_err(e.to_string()))?; - audit_log( - &db, - &authed, - operation, - action_kind, - "global", - Some(&format!("worker__{group}")), - None, - ) - .await?; - } + instance_config::apply_configs_diff( + &db, + &configs_diff, + &InstanceConfigAuditor { db: &db, author: &authed, source: None }, + ) + .await + .map_err(|e| error::Error::internal_err(e.to_string()))?; } Ok(()) diff --git a/backend/windmill-audit/src/instance_config.rs b/backend/windmill-audit/src/instance_config.rs new file mode 100644 index 0000000000..977e0fe6c6 --- /dev/null +++ b/backend/windmill-audit/src/instance_config.rs @@ -0,0 +1,148 @@ +//! Audit entries for writes to instance settings and worker group configs, whichever way +//! they arrive: an API request, or a declarative sync that no user request stands behind. + +use std::collections::HashMap; + +use windmill_common::{ + audit::AuditAuthor, error::Result, instance_config::ConfigWriteObserver, + users::SUPERADMIN_SYNC_EMAIL, DB, +}; + +use crate::{ + audit_oss::{audit_log, AuditAuthorable}, + ActionKind, +}; + +/// The declarative writer behind an entry, recorded as its `source` parameter. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DeclarativeSource { + /// `windmill sync-config ` + SyncConfig, + /// The Kubernetes operator's ConfigMap sync. + Operator, +} + +impl DeclarativeSource { + pub fn as_str(&self) -> &'static str { + match self { + DeclarativeSource::SyncConfig => "sync-config", + DeclarativeSource::Operator => "operator", + } + } +} + +/// Who a declarative sync's writes are attributed to. It runs on direct database +/// credentials, so there is no user to name. +pub fn declarative_sync_author() -> AuditAuthor { + AuditAuthor { + username: "instance_config_sync".to_string(), + email: SUPERADMIN_SYNC_EMAIL.to_string(), + username_override: None, + token_prefix: None, + } +} + +fn parameters<'a>( + value: Option<&'a str>, + source: Option, +) -> Option> { + let parameters: HashMap<&str, &str> = value + .map(|v| ("value", v)) + .into_iter() + .chain(source.map(|s| ("source", s.as_str()))) + .collect(); + (!parameters.is_empty()).then_some(parameters) +} + +/// A setting is recorded by name, with its value only when that is a boolean or a number. +/// Any other shape can carry a credential (a registry URL, a webhook, a header map), and no +/// list of secret-bearing settings stays complete as settings are added. +/// +/// AUTHORIZATION: records a write, checks nothing. The caller must have established that +/// `author` may change instance settings before performing the write this records. +pub async fn audit_setting_write( + db: &DB, + author: &impl AuditAuthorable, + key: &str, + value: Option<&serde_json::Value>, + source: Option, +) -> Result<()> { + let deleted = match value { + None | Some(serde_json::Value::Null) => true, + Some(serde_json::Value::String(s)) => s.trim().is_empty(), + Some(_) => false, + }; + let recorded = match value { + Some(v @ (serde_json::Value::Bool(_) | serde_json::Value::Number(_))) => { + Some(v.to_string()) + } + _ => None, + }; + audit_log( + db, + author, + if deleted { + "global_settings.delete" + } else { + "global_settings.update" + }, + if deleted { + ActionKind::Delete + } else { + ActionKind::Update + }, + "global", + Some(key), + parameters(recorded.as_deref(), source), + ) + .await +} + +/// A worker group config is recorded by name only: its env vars and init script are free +/// text that can carry a credential. +/// +/// AUTHORIZATION: records a write, checks nothing. The caller must have established that +/// `author` may change worker group configs before performing the write this records. +pub async fn audit_worker_config_write( + db: &DB, + author: &impl AuditAuthorable, + group: &str, + deleted: bool, + source: Option, +) -> Result<()> { + let (operation, action_kind) = if deleted { + ("worker_config.delete", ActionKind::Delete) + } else { + ("worker_config.update", ActionKind::Update) + }; + audit_log( + db, + author, + operation, + action_kind, + "global", + Some(&format!("worker__{group}")), + parameters(None, source), + ) + .await +} + +/// Audits each write of an instance config apply as `author`'s. +/// +/// AUTHORIZATION: as [`audit_setting_write`], for every write it is told of. +pub struct InstanceConfigAuditor<'a, A> { + pub db: &'a DB, + pub author: &'a A, + /// `None` for a write made through the API. + pub source: Option, +} + +impl ConfigWriteObserver for InstanceConfigAuditor<'_, A> { + async fn setting_written(&self, key: &str, value: Option<&serde_json::Value>) -> Result<()> { + audit_setting_write(self.db, self.author, key, value, self.source).await + } + + async fn worker_config_written(&self, group: &str, deleted: bool) -> Result<()> { + audit_worker_config_write(self.db, self.author, group, deleted, self.source).await + } +} diff --git a/backend/windmill-audit/src/lib.rs b/backend/windmill-audit/src/lib.rs index f1443fdb18..e9c71f1336 100644 --- a/backend/windmill-audit/src/lib.rs +++ b/backend/windmill-audit/src/lib.rs @@ -4,6 +4,7 @@ use sqlx::FromRow; #[cfg(feature = "private")] pub mod audit_ee; pub mod audit_oss; +pub mod instance_config; #[derive(sqlx::Type, Serialize, Deserialize, Debug)] #[sqlx(type_name = "ACTION_KIND", rename_all = "lowercase")] diff --git a/backend/windmill-common/src/external_instance_pg.rs b/backend/windmill-common/src/external_instance_pg.rs index c0b6203e40..72920a1874 100644 --- a/backend/windmill-common/src/external_instance_pg.rs +++ b/backend/windmill-common/src/external_instance_pg.rs @@ -198,7 +198,8 @@ pub async fn external_instance_database_usages<'c>( /// would be a login nothing can drop any more. Allowed on every edition, so a /// downgraded instance can still clear a setting it no longer uses. async fn ensure_external_instance_pg_removable(conn: &mut sqlx::PgConnection) -> Result<()> { - ensure_external_instance_pg_unused(conn, &format!("removing {EXTERNAL_INSTANCE_PG_SETTING}")).await + ensure_external_instance_pg_unused(conn, &format!("removing {EXTERNAL_INSTANCE_PG_SETTING}")) + .await } /// Refuse while Windmill has databases or data table roles on the cluster, or a workspace points @@ -394,16 +395,21 @@ pub async fn write_external_instance_pg_setting( } /// [`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. +/// it, and takes it out of the diff so the generic apply does not write it again. `observer` is +/// told of the write here, since the generic apply no longer sees the key. /// /// Authorization: checks nothing. Callers MUST be superadmin, or the declarative instance config /// sync, which applies what the operator deployed. pub async fn write_external_instance_pg_from_diff( db: &DB, diff: &mut crate::instance_config::SettingsDiff, + observer: &impl crate::instance_config::ConfigWriteObserver, ) -> Result<()> { if let Some(value) = diff.upserts.remove(EXTERNAL_INSTANCE_PG_SETTING) { write_external_instance_pg_setting(db, Some(&value)).await?; + observer + .setting_written(EXTERNAL_INSTANCE_PG_SETTING, Some(&value)) + .await?; } if let Some(i) = diff .deletes @@ -412,6 +418,9 @@ pub async fn write_external_instance_pg_from_diff( { diff.deletes.remove(i); write_external_instance_pg_setting(db, None).await?; + observer + .setting_written(EXTERNAL_INSTANCE_PG_SETTING, None) + .await?; } Ok(()) } diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index 3d50558c7f..0ae8152c1c 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -1015,6 +1015,43 @@ pub struct ConfigsDiff { pub deletes: Vec, } +/// Told of each write an instance config apply makes, as soon as it has landed: the writes +/// are not transactional, so an apply that fails midway must not leave the ones it did make +/// unrecorded. The audit entry is the implementor's to write, as this crate sits below +/// `windmill-audit`. +pub trait ConfigWriteObserver: Sync { + /// `value` is `None` for a deleted setting. + fn setting_written( + &self, + key: &str, + value: Option<&serde_json::Value>, + ) -> impl std::future::Future> + Send; + + fn worker_config_written( + &self, + group: &str, + deleted: bool, + ) -> impl std::future::Future> + Send; +} + +impl ConfigWriteObserver for () { + async fn setting_written( + &self, + _key: &str, + _value: Option<&serde_json::Value>, + ) -> crate::error::Result<()> { + Ok(()) + } + + async fn worker_config_written( + &self, + _group: &str, + _deleted: bool, + ) -> crate::error::Result<()> { + Ok(()) + } +} + /// Settings that must never be deleted by the operator or bulk API. pub const PROTECTED_SETTINGS: &[&str] = &[ "ducklake_user_pg_pwd", @@ -1398,10 +1435,14 @@ pub fn diff_worker_configs( /// AUTHORIZATION: replaces instance-wide settings and takes no authed context, so /// callers MUST have established superadmin or equivalent system authority (the CLI /// and the operator both run with direct instance credentials). +/// +/// `observer` is what audits the sync: no user request stands behind these writes, so +/// nothing else records them. pub async fn sync_global_settings_declarative( db: &sqlx::Pool, current: &BTreeMap, desired: &BTreeMap, + observer: &impl ConfigWriteObserver, ) -> anyhow::Result<()> { for url_key in [ crate::global_settings::GITHUB_APP_WEBHOOK_BASE_URL_SETTING, @@ -1465,16 +1506,22 @@ pub async fn sync_global_settings_declarative( .map_err(|e| anyhow::anyhow!("{max_expiration_key}: {e}"))?; 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?; + crate::external_instance_pg::write_external_instance_pg_from_diff(db, &mut diff, observer) + .await?; + apply_settings_diff(db, &diff, observer).await?; Ok(()) } -/// Apply a settings diff to the global_settings table. +/// Apply a settings diff to the global_settings table, reporting each write to `observer`. +/// +/// AUTHORIZATION: writes instance-wide settings, credentials included, and checks +/// nothing; `observer` records and authorizes nothing. Callers MUST have established +/// superadmin or equivalent system authority. pub async fn apply_settings_diff( db: &sqlx::Pool, diff: &SettingsDiff, + observer: &impl ConfigWriteObserver, ) -> anyhow::Result<()> { for (key, value) in &diff.upserts { sqlx::query( @@ -1497,6 +1544,7 @@ pub async fn apply_settings_diff( format_setting_value(key, value) ); } + observer.setting_written(key, Some(value)).await?; } for key in &diff.deletes { sqlx::query("DELETE FROM global_settings WHERE name = $1") @@ -1511,6 +1559,7 @@ pub async fn apply_settings_diff( } else { tracing::info!("Deleted global setting: {key}"); } + observer.setting_written(key, None).await?; } if diff.unchanged_count > 0 { tracing::info!("{} global setting(s) unchanged", diff.unchanged_count); @@ -1518,10 +1567,16 @@ pub async fn apply_settings_diff( Ok(()) } -/// Apply a configs diff to the config table (worker__ prefix). +/// Apply a configs diff to the config table (worker__ prefix), reporting each write to +/// `observer`. +/// +/// AUTHORIZATION: writes worker group configs, whose init scripts run on the workers, and +/// checks nothing; `observer` records and authorizes nothing. Callers MUST have +/// established superadmin or equivalent system authority. pub async fn apply_configs_diff( db: &sqlx::Pool, diff: &ConfigsDiff, + observer: &impl ConfigWriteObserver, ) -> anyhow::Result<()> { for (group_name, config_value) in &diff.upserts { let db_key = format!("worker__{group_name}"); @@ -1534,6 +1589,7 @@ pub async fn apply_configs_diff( .execute(db) .await?; tracing::info!("Synced worker config: {db_key}"); + observer.worker_config_written(group_name, false).await?; } for group_name in &diff.deletes { let db_key = format!("worker__{group_name}"); @@ -1542,6 +1598,7 @@ pub async fn apply_configs_diff( .execute(db) .await?; tracing::info!("Deleted worker config: {db_key}"); + observer.worker_config_written(group_name, true).await?; } Ok(()) } @@ -1663,7 +1720,11 @@ impl InstanceConfig { /// /// Reads the current state, computes a diff, then applies upserts and /// deletes for both global settings and worker configs. - pub async fn sync_to_db(&self, db: &sqlx::Pool) -> anyhow::Result<()> { + pub async fn sync_to_db( + &self, + db: &sqlx::Pool, + observer: &impl ConfigWriteObserver, + ) -> anyhow::Result<()> { let desired_settings = self.global_settings.to_settings_map(); // Current global settings @@ -1673,7 +1734,8 @@ impl InstanceConfig { .await?; let current_settings: BTreeMap = rows.into_iter().collect(); - sync_global_settings_declarative(db, ¤t_settings, &desired_settings).await?; + sync_global_settings_declarative(db, ¤t_settings, &desired_settings, observer) + .await?; // Worker configs let desired_configs: BTreeMap = self @@ -1701,7 +1763,7 @@ impl InstanceConfig { let configs_diff = diff_worker_configs(¤t_configs, &desired_configs, ApplyMode::Replace); - apply_configs_diff(db, &configs_diff).await?; + apply_configs_diff(db, &configs_diff, observer).await?; Ok(()) } diff --git a/backend/windmill-common/src/otel_oss.rs b/backend/windmill-common/src/otel_oss.rs index 95e635dd8f..9dac8a07db 100644 --- a/backend/windmill-common/src/otel_oss.rs +++ b/backend/windmill-common/src/otel_oss.rs @@ -54,6 +54,19 @@ pub(crate) fn init_logs_bridge(_mode: &Mode, _hostname: &str, _env: &str) -> Opt None } +#[cfg(not(feature = "private"))] +pub(crate) fn init_logs_bridge_for_service( + _service: &str, + _hostname: &str, + _env: &str, +) -> Option { + None +} + +/// Export the log records still batched, for a process about to exit. +#[cfg(not(feature = "private"))] +pub async fn flush_otel_logs() {} + #[cfg(all(feature = "otel", feature = "enterprise", not(feature = "private")))] pub(crate) fn init_otlp_tracer( _mode: &Mode, diff --git a/backend/windmill-common/src/tracing_init.rs b/backend/windmill-common/src/tracing_init.rs index 01e75ac489..855e077004 100644 --- a/backend/windmill-common/src/tracing_init.rs +++ b/backend/windmill-common/src/tracing_init.rs @@ -220,6 +220,28 @@ pub fn initialize_tracing( (_guard, meter_provider) } +/// Tracing for a subcommand that writes to the instance without running a server +/// (`sync-config`, the operator): stdout, plus the OTel logs bridge when OTel logs are +/// enabled, which is how the audit events of its writes reach a collector. It writes no +/// service log files, as nothing lists such a process to read them back. +/// +/// Only audit events are exported. These processes log each setting they write with its +/// value, masked by a list of known secret fields rather than by shape, and that is fit +/// for the operator's own stdout but not for a collector. +pub fn initialize_subcommand_tracing(service: &str, hostname: &str, environment: &str) { + let logs_bridge = crate::otel_oss::init_logs_bridge_for_service(service, hostname, environment); + tracing_subscriber::registry() + .with(logs_bridge.with_filter(Targets::new().with_target(AUDIT_TARGET, LevelFilter::INFO))) + .with( + tracing_subscriber::fmt::layer().with_filter( + EnvFilter::builder() + .with_default_directive(LevelFilter::INFO.into()) + .from_env_lossy(), + ), + ) + .init(); +} + lazy_static::lazy_static! { pub static ref LOG_COUNTING_BY_MIN: Arc>> = Arc::new(RwLock::new(HashMap::new())); } diff --git a/backend/windmill-operator/Cargo.toml b/backend/windmill-operator/Cargo.toml index 30a7c0bb7a..f847e4ab25 100644 --- a/backend/windmill-operator/Cargo.toml +++ b/backend/windmill-operator/Cargo.toml @@ -19,6 +19,7 @@ serde_json.workspace = true sqlx.workspace = true tracing.workspace = true windmill-common = { workspace = true, default-features = false } +windmill-audit.workspace = true anyhow.workspace = true thiserror.workspace = true kube.workspace = true