starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-10-14 00:02:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: audit declarative instance config writes (#11684)

Browse Source 
* feat: audit declarative instance config writes

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: keep early logs, base url and license plan for declarative writers

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* refactor: require an observer on every config apply, drop license loading

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: export only audit events from the declarative subcommands

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* 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 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-10-12 16:37:05 +02:00
committed by GitHub
co-authored by Claude Opus 5.5 windmill-internal-app[bot]
parent 1e3aef877c
commit afbbc0b1b2
14 changed files with 483 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+2
View File
@@ -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",
]
backend/Cargo.toml
+1
View File
@@ -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 }
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
06610ecc950ed64a2709d7f1aa8c7b49a95fdbae
d6a6cdd343e7744ea49c610f9534c8013c8ffc2a
backend/src/main.rs
+55 -16
View File
@@ -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 <file>");
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);
backend/tests/instance_config.rs
+139 -27
View File
@@ -237,7 +237,7 @@ async fn test_diff_global_settings_empty_string_routes_to_delete(db: Pool<Postgr
"empty-string value must route to deletes"
);
apply_settings_diff(&db, &diff).await.unwrap();
apply_settings_diff(&db, &diff, &()).await.unwrap();
assert!(
get_global_setting(&db, "npmrc").await.is_none(),
"npmrc row must be deleted after apply"
@@ -319,7 +319,7 @@ async fn test_diff_global_settings_rejects_worker_configs_upsert(db: Pool<Postgr
"worker_configs must never be written as a global setting"
);
apply_settings_diff(&db, &diff).await.unwrap();
apply_settings_diff(&db, &diff, &()).await.unwrap();
assert!(
get_global_setting(&db, "worker_configs").await.is_none(),
"worker_configs row must not exist in global_settings after apply"
@@ -444,7 +444,7 @@ async fn test_apply_settings_diff_upserts_only(db: Pool<Postgres>) {
..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<Postgres>) {
..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<Postgres>) {
..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<Postgres>) {
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<Postgres>) {
..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<Postgres>) {
..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<Postgres>)
};
// 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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Post
let diff = ConfigsDiff { upserts: BTreeMap::new(), deletes: vec!["server_config".to_string()] };
apply_configs_diff(&db, &diff).await.unwrap();
apply_configs_diff(&db, &diff, &()).await.unwrap();
// The delete targets "worker__server_config", not "server_config"
assert!(
@@ -692,7 +692,7 @@ async fn test_roundtrip_write_read_modify_apply(db: Pool<Postgres>) {
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<Postgres>) {
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<Postgres>) {
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<Postgres>)
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<Postgres>) {
};
// 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<Postgres>) {
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<Postgres>)
// 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<Postgres>) {
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<Postgres>) {
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<Postgres>)
"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<Postgres>) {
"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<Postgres
&db,
&BTreeMap::new(),
&desired,
&(),
)
.await
.expect_err("an invalid webhook base url must fail the sync");
@@ -1474,6 +1475,7 @@ async fn declarative_sync_rejects_an_unusable_webhook_base_url(db: Pool<Postgres
&db,
&BTreeMap::new(),
&wrong_type,
&(),
)
.await
.expect_err("a non-string webhook base url must fail the sync");
@@ -1511,6 +1513,7 @@ async fn declarative_sync_rejects_an_unusable_instance_banner(db: Pool<Postgres>
&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<Post
&db,
&BTreeMap::new(),
&desired,
&(),
)
.await
.expect_err("a ceiling the token routes cannot read must fail the sync");
@@ -1578,6 +1582,7 @@ async fn declarative_sync_rejects_an_unusable_default_allowed_origins(db: Pool<P
&db,
&BTreeMap::new(),
&desired,
&(),
)
.await
.expect_err(&format!("{bad} must fail the sync"));
@@ -1604,6 +1609,7 @@ async fn declarative_sync_rejects_an_unusable_default_allowed_origins(db: Pool<P
&db,
&BTreeMap::new(),
&desired,
&(),
)
.await
.expect("a valid origin list must sync");
@@ -1635,7 +1641,10 @@ async fn declarative_sync_keeps_the_external_cluster_while_it_holds_databases(db
let (db, current) = (db.clone(), current.clone());
async move {
windmill_common::instance_config::sync_global_settings_declarative(
&db, &current, &desired,
&db,
&current,
&desired,
&(),
)
.await
}
@@ -1670,6 +1679,7 @@ async fn declarative_sync_rejects_a_non_hex_accent_color(db: Pool<Postgres>) {
&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<Postgres>) {
&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<Postgres>) {
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),
]
);
}
backend/windmill-api-schedule/src/lib.rs
+2 -1
View File
@@ -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
backend/windmill-api-settings/src/lib.rs
+18 -129
View File
@@ -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(&current_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(())
backend/windmill-audit/src/instance_config.rs
+148
View File
@@ -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 <file>`
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<DeclarativeSource>,
) -> Option<HashMap<&'a str, &'a str>> {
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<DeclarativeSource>,
) -> 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<DeclarativeSource>,
) -> 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<DeclarativeSource>,
}
impl<A: AuditAuthorable + Sync> 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
}
}
backend/windmill-audit/src/lib.rs
+1
View File
@@ -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")]
backend/windmill-common/src/external_instance_pg.rs
+11 -2
View File
@@ -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(())
}
backend/windmill-common/src/instance_config.rs
+69 -7
View File
@@ -1015,6 +1015,43 @@ pub struct ConfigsDiff {
pub deletes: Vec<String>,
}
/// 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<Output = crate::error::Result<()>> + Send;
fn worker_config_written(
&self,
group: &str,
deleted: bool,
) -> impl std::future::Future<Output = crate::error::Result<()>> + 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<sqlx::Postgres>,
current: &BTreeMap<String, serde_json::Value>,
desired: &BTreeMap<String, serde_json::Value>,
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<sqlx::Postgres>,
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<sqlx::Postgres>,
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<sqlx::Postgres>) -> anyhow::Result<()> {
pub async fn sync_to_db(
&self,
db: &sqlx::Pool<sqlx::Postgres>,
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<String, serde_json::Value> = rows.into_iter().collect();
sync_global_settings_declarative(db, &current_settings, &desired_settings).await?;
sync_global_settings_declarative(db, &current_settings, &desired_settings, observer)
.await?;
// Worker configs
let desired_configs: BTreeMap<String, serde_json::Value> = self
@@ -1701,7 +1763,7 @@ impl InstanceConfig {
let configs_diff =
diff_worker_configs(&current_configs, &desired_configs, ApplyMode::Replace);
apply_configs_diff(db, &configs_diff).await?;
apply_configs_diff(db, &configs_diff, observer).await?;
Ok(())
}
backend/windmill-common/src/otel_oss.rs
+13
View File
@@ -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<EnvFilter> {
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,
backend/windmill-common/src/tracing_init.rs
+22
View File
@@ -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<RwLock<HashMap<String, LogCounter>>> = Arc::new(RwLock::new(HashMap::new()));
}
backend/windmill-operator/Cargo.toml
+1
View File
@@ -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