From 8577bd825eb550effdc0854354e664bc267e14c9 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 31 May 2026 19:50:13 +0800 Subject: [PATCH] feat(admin): restore config admin compatibility (#3133) * feat(admin): restore config admin compatibility Co-authored-by: weisd * fix(admin): align config admin clean rebuild Co-authored-by: weisd * fix(admin): align config history and peer signals * fix(admin): harden config admin mutations * fix(admin): tighten config review follow-ups * perf(admin): reuse env snapshot in config render * fix(ecstore): clean up config admin and listing error handling Remove redundant is_all_volume_not_found check in list_merged, add storage class encode/decode roundtrip tests, fresh boot integration test, and config admin clean rebuild improvements. Co-authored-by: hehutu * fix(admin): sync global server config on mutation and reload Change GLOBAL_SERVER_CONFIG from OnceLock to RwLock so config mutations (set/del/restore/reload) are visible to readers without restart. Call set_global_server_config after every store save and on snapshot reload. Register storage_class as a dynamic config subsystem. Co-authored-by: hehutu * style: apply rustfmt to config and admin tests Co-authored-by: hehutu * fix(test): update signal_service test for storage_class dynamic subsystem storage_class is now a valid dynamic config subsystem, so the "requires object layer" test should expect "storage layer not initialized" instead of "unsupported dynamic config subsystem". Co-authored-by: hehutu * fix(config): publish storage_class runtime config on dynamic reload Change GLOBAL_STORAGE_CLASS from OnceLock to RwLock so runtime updates are possible. apply_storage_class_runtime_config now actually publishes the parsed config via set_global_storage_class instead of dropping it. Addresses review feedback: storage_class was marked as dynamically applied but the parsed result was discarded, so mc admin config set returned config_applied=true while the runtime kept using stale parity settings until restart. Co-authored-by: hehutu * fix(admin): harden config init, history ordering, and env redaction - Change GLOBAL_SERVER_CONFIG from RwLock to RwLock> initialized with None, preserving "not initialized" detection via None - Move save_server_config_history before save_server_config_to_store in SetConfigKVHandler, DelConfigKVHandler, and SetConfigHandler so a restore point exists before mutations are persisted - Redact sensitive env override values with *redacted* instead of silently omitting the line, improving admin visibility - Add code comment explaining VolumeNotFound removal rationale in list_merged for listing paths Co-authored-by: hehutu * fix(config): keep in-memory config in sync after set/restore/reload GLOBAL_SERVER_CONFIG was a OnceLock set once at startup and never updated. After mc admin config set writes to the store, any fallback to get_global_server_config() returned stale init-time data. Similarly, reload_runtime_config_snapshot read from the store but discarded the result. - Replace OnceLock with RwLock for GLOBAL_SERVER_CONFIG and GLOBAL_STORAGE_CLASS so they can be updated at runtime - Add set_global_server_config / set_global_storage_class setters - Call set_global_server_config after every config save (set-kv, del-kv, set-config, restore-history) - Re-apply dynamic subsystems (storage_class, audit_webhook, audit_mqtt) and signal peers in reload_runtime_config_snapshot and full-config operations - Fix render_selected_config scope boundary check: track per-scope line count instead of checking global lines.is_empty() - Include STORAGE_CLASS_SUB_SYS in is_dynamic_config_subsystem so apply_storage_class_runtime_config is reachable Co-authored-by: hehutu * fix(storageclass): use CLASS_RRS key in lookup_config for RRS parity lookup_config used kvs.get(RRS) where RRS="REDUCED_REDUNDANCY", but the admin config path writes the key as CLASS_RRS="rrs". This caused RRS values to never be read back, always falling back to default parity. - Changed kvs.get(RRS) to kvs.get(CLASS_RRS) in lookup_config - Added regression tests verifying RRS read/write consistency Co-authored-by: hehutu * fix(config): add peer-side logging and don't swallow apply errors - Add tracing::warn! in reload_dynamic_config_runtime_state and reload_runtime_config_snapshot when config read or subsystem apply fails, so on-host diagnostics show which signal failed and why - Change `let _ = apply_dynamic_config_for_subsystem(...)` to `if let Err(err) = ... { warn!(...) }` in reload_runtime_config_snapshot so per-subsystem failures are logged instead of silently swallowed - Remove weak test global_server_config_returns_none_before_init that had no meaningful assertion due to shared global state Co-authored-by: hehutu * style: apply rustfmt to config and storageclass tests Co-authored-by: hehutu --------- Co-authored-by: weisd Co-authored-by: hehutu --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 45 + crates/ecstore/src/config/com.rs | 92 +- crates/ecstore/src/config/mod.rs | 49 +- crates/ecstore/src/config/storageclass.rs | 39 +- crates/ecstore/src/notification_sys.rs | 59 + crates/ecstore/src/rpc/mod.rs | 5 +- crates/ecstore/src/rpc/peer_rest_client.rs | 2 + crates/ecstore/src/set_disk.rs | 11 +- crates/ecstore/src/set_disk/heal.rs | 2 +- crates/ecstore/src/store.rs | 1 - crates/ecstore/src/store/rebalance.rs | 3 +- crates/ecstore/src/store_list_objects.rs | 10 +- rustfs/src/admin/handlers/config_admin.rs | 2121 +++++++++++++++++ rustfs/src/admin/handlers/mod.rs | 7 + rustfs/src/admin/mod.rs | 5 +- rustfs/src/admin/service/config.rs | 421 ++++ rustfs/src/admin/service/mod.rs | 1 + rustfs/src/server/layer.rs | 13 + rustfs/src/storage/rpc/node_service.rs | 180 +- 19 files changed, 3027 insertions(+), 39 deletions(-) create mode 100644 rustfs/src/admin/handlers/config_admin.rs create mode 100644 rustfs/src/admin/service/config.rs diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 58cb5c467..99896ed36 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -3043,6 +3043,51 @@ mod tests { ); } + #[tokio::test] + #[serial] + async fn ecstore_new_succeeds_on_fresh_local_volumes() { + let test_base_dir = format!("/tmp/rustfs_ecstore_empty_boot_{}", Uuid::new_v4()); + let temp_dir = PathBuf::from(&test_base_dir); + if temp_dir.exists() { + fs::remove_dir_all(&temp_dir).await.ok(); + } + fs::create_dir_all(&temp_dir).await.unwrap(); + + let disk_paths = vec![ + temp_dir.join("disk1"), + temp_dir.join("disk2"), + temp_dir.join("disk3"), + temp_dir.join("disk4"), + ]; + + for disk_path in &disk_paths { + fs::create_dir_all(disk_path).await.unwrap(); + } + + let mut endpoints = Vec::new(); + for (i, disk_path) in disk_paths.iter().enumerate() { + let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + endpoints.push(endpoint); + } + + let endpoint_pools = EndpointServerPools(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "fresh-boot-test".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }]); + + crate::store::init_local_disks(endpoint_pools.clone()).await.unwrap(); + + let ecstore = ECStore::new("127.0.0.1:0".parse().unwrap(), endpoint_pools, CancellationToken::new()).await; + assert!(ecstore.is_ok(), "fresh local ECStore boot should succeed, got {ecstore:?}"); + } + #[tokio::test] #[serial] async fn stale_multipart_cleanup_uses_default_expiry_without_lifecycle() { diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index bbef691e6..9a9c0cb86 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::config::{Config, GLOBAL_STORAGE_CLASS, KVS, audit, notify, oidc, storageclass}; +use crate::config::{Config, KVS, audit, notify, oidc, set_global_storage_class, storageclass}; use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::global::is_first_cluster_node_local; @@ -1164,11 +1164,8 @@ async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: for (i, count) in set_drive_counts.iter().enumerate() { match storageclass::lookup_config(&kvs, *count) { Ok(res) => { - if i == 0 - && GLOBAL_STORAGE_CLASS.get().is_none() - && let Err(r) = GLOBAL_STORAGE_CLASS.set(res) - { - error!("GLOBAL_STORAGE_CLASS.set failed {:?}", r); + if i == 0 { + set_global_storage_class(res); } } Err(err) => { @@ -2638,4 +2635,87 @@ mod tests { assert_eq!(object_info.bucket, crate::disk::RUSTFS_META_BUCKET); assert_eq!(object_info.name, "config/test.json"); } + + #[test] + fn test_encode_decode_roundtrip_preserves_storage_class() { + use crate::config::STORAGE_CLASS_SUB_SYS; + + // Create a config with custom storage class values + let mut cfg = Config::new(); + let kvs = storage_class_kvs_mut(&mut cfg); + kvs.insert("standard".to_string(), "EC:4".to_string()); + kvs.insert("rrs".to_string(), "EC:2".to_string()); + kvs.insert("optimize".to_string(), "availability".to_string()); + + // Encode to external format (what save_server_config produces) + let encoded = encode_server_config_blob(&cfg, None).expect("encode should succeed"); + + // Verify the encoded data uses "storageclass" (no underscore) + let v: serde_json::Value = serde_json::from_slice(&encoded).expect("should be valid json"); + assert!(v.get("storageclass").is_some(), "encoded should have 'storageclass' key"); + assert!(v.get("storage_class").is_none(), "encoded should NOT have 'storage_class' key"); + + // Decode back (what load_server_config_from_store produces) + let decoded = decode_server_config_blob(&encoded).expect("decode should succeed"); + + // Verify the decoded config has "storage_class" (with underscore) subsystem + let kvs = decoded + .get_value(STORAGE_CLASS_SUB_SYS, crate::config::DEFAULT_DELIMITER) + .expect("decoded config should have storage_class subsystem"); + assert_eq!(kvs.get("standard"), "EC:4", "standard should be EC:4"); + assert_eq!(kvs.get("rrs"), "EC:2", "rrs should be EC:2"); + assert_eq!(kvs.get("optimize"), "availability", "optimize should be availability"); + } + + #[test] + fn test_set_then_load_preserves_storage_class_values() { + use crate::config::STORAGE_CLASS_SUB_SYS; + + // Step 1: Start with a fresh config (simulates initial server state) + let mut cfg = Config::new(); + + // Step 2: Apply set directives (simulates "mc admin config set storage_class standard=EC:4 rrs=EC:2") + let kvs = cfg + .0 + .entry(STORAGE_CLASS_SUB_SYS.to_string()) + .or_default() + .entry(DEFAULT_DELIMITER.to_string()) + .or_default(); + kvs.insert("standard".to_string(), "EC:4".to_string()); + kvs.insert("rrs".to_string(), "EC:2".to_string()); + cfg.set_defaults(); + + // Step 3: Save (encode to external format) + let encoded = encode_server_config_blob(&cfg, None).expect("encode should succeed"); + + // Step 4: Load (decode from external format) + let loaded = decode_server_config_blob(&encoded).expect("decode should succeed"); + + // Step 5: Verify the values are preserved + let loaded_kvs = loaded + .get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER) + .expect("should have storage_class"); + assert_eq!(loaded_kvs.get("standard"), "EC:4"); + assert_eq!(loaded_kvs.get("rrs"), "EC:2"); + + // Step 6: Verify the subsystem is accessible by name (what render_selected_config does) + let targets = loaded + .0 + .get(STORAGE_CLASS_SUB_SYS) + .expect("storage_class subsystem should exist"); + let target_kvs = targets.get(DEFAULT_DELIMITER).expect("default target should exist"); + assert_eq!(target_kvs.get("standard"), "EC:4"); + assert_eq!(target_kvs.get("rrs"), "EC:2"); + } + + #[test] + fn test_config_unmarshal_fails_on_external_format() { + // This verifies that the external "storageclass" format cannot be + // mistakenly parsed by Config::unmarshal (which expects internal format). + // If unmarshal succeeds, the config would have "storageclass" instead of + // "storage_class", causing render_selected_config to fail. + let external_json = r#"{"version":"33","storageclass":{"standard":"EC:4","rrs":"EC:2"}}"#; + let result = Config::unmarshal(external_json.as_bytes()); + assert!(result.is_err(), "Config::unmarshal should fail on external format, but got: {:?}", result); + } } diff --git a/crates/ecstore/src/config/mod.rs b/crates/ecstore/src/config/mod.rs index 0b2935381..7b2cb6a51 100644 --- a/crates/ecstore/src/config/mod.rs +++ b/crates/ecstore/src/config/mod.rs @@ -37,11 +37,12 @@ use rustfs_config::oidc::IDENTITY_OPENID_SUB_SYS; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::LazyLock; -use std::sync::{Arc, OnceLock}; +use std::sync::{Arc, OnceLock, RwLock}; -pub static GLOBAL_STORAGE_CLASS: LazyLock> = LazyLock::new(OnceLock::new); +pub static GLOBAL_STORAGE_CLASS: LazyLock> = + LazyLock::new(|| RwLock::new(storageclass::Config::default())); pub static DEFAULT_KVS: LazyLock>> = LazyLock::new(OnceLock::new); -pub static GLOBAL_SERVER_CONFIG: LazyLock> = LazyLock::new(OnceLock::new); +pub static GLOBAL_SERVER_CONFIG: LazyLock>> = LazyLock::new(|| RwLock::new(None)); pub static GLOBAL_CONFIG_SYS: LazyLock = LazyLock::new(ConfigSys::new); pub static RUSTFS_CONFIG_PREFIX: &str = "config"; @@ -63,14 +64,30 @@ impl ConfigSys { lookup_configs(&mut cfg, api).await; - let _ = GLOBAL_SERVER_CONFIG.set(cfg); + set_global_server_config(cfg); Ok(()) } } pub fn get_global_server_config() -> Option { - GLOBAL_SERVER_CONFIG.get().cloned() + GLOBAL_SERVER_CONFIG.read().ok().and_then(|guard| (*guard).clone()) +} + +pub fn set_global_server_config(cfg: Config) { + if let Ok(mut guard) = GLOBAL_SERVER_CONFIG.write() { + *guard = Some(cfg); + } +} + +pub fn get_global_storage_class() -> Option { + GLOBAL_STORAGE_CLASS.read().ok().map(|guard| (*guard).clone()) +} + +pub fn set_global_storage_class(cfg: storageclass::Config) { + if let Ok(mut guard) = GLOBAL_STORAGE_CLASS.write() { + *guard = cfg; + } } pub async fn init_global_config_sys(api: Arc) -> Result<()> { @@ -262,3 +279,25 @@ pub fn init() { // Register all default configurations register_default_kvs(kvs) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn global_server_config_set_and_get_roundtrip() { + init(); + let mut cfg = Config::new(); + let mut kvs = KVS::new(); + kvs.insert("standard".to_string(), "EC:4".to_string()); + cfg.0 + .insert(STORAGE_CLASS_SUB_SYS.to_string(), HashMap::from([("_".to_string(), kvs)])); + + set_global_server_config(cfg.clone()); + let loaded = get_global_server_config().expect("global config should be set"); + let sc_kvs = loaded + .get_value(STORAGE_CLASS_SUB_SYS, "_") + .expect("storage_class should exist"); + assert_eq!(sc_kvs.get("standard"), "EC:4"); + } +} diff --git a/crates/ecstore/src/config/storageclass.rs b/crates/ecstore/src/config/storageclass.rs index 5fe70f4a0..faeb2ea09 100644 --- a/crates/ecstore/src/config/storageclass.rs +++ b/crates/ecstore/src/config/storageclass.rs @@ -101,13 +101,13 @@ pub static DEFAULT_KVS: LazyLock = LazyLock::new(|| { }); // StorageClass - holds storage class information -#[derive(Serialize, Deserialize, Debug, Default)] +#[derive(Serialize, Deserialize, Debug, Default, Clone)] pub struct StorageClass { parity: usize, } // Config storage class configuration -#[derive(Serialize, Deserialize, Debug, Default)] +#[derive(Serialize, Deserialize, Debug, Default, Clone)] pub struct Config { standard: StorageClass, rrs: StorageClass, @@ -205,7 +205,7 @@ pub fn lookup_config(kvs: &KVS, set_drive_count: usize) -> Result { if let Ok(ssc_str) = env::var(RRS_ENV) { ssc_str } else { - kvs.get(RRS) + kvs.get(CLASS_RRS) } }; @@ -336,3 +336,36 @@ pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_coun } Ok(()) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn lookup_config_reads_rrs_from_class_rrs_key() { + // Regression: kvs.get(RRS) used RRS="REDUCED_REDUNDANCY" instead of + // CLASS_RRS="rrs", so admin-written RRS values were never read back. + let mut kvs = KVS::new(); + kvs.insert(CLASS_STANDARD.to_string(), "EC:4".to_string()); + kvs.insert(CLASS_RRS.to_string(), "EC:2".to_string()); + + let cfg = lookup_config(&kvs, 8).expect("lookup should succeed"); + assert_eq!(cfg.standard.parity, 4, "standard parity should be 4"); + assert_eq!(cfg.rrs.parity, 2, "rrs parity should be 2"); + } + + #[test] + fn lookup_config_ignores_redundancy_key_name() { + // Ensure the old key name "REDUCED_REDUNDANCY" is NOT read. + let mut kvs = KVS::new(); + kvs.insert(CLASS_STANDARD.to_string(), "EC:4".to_string()); + kvs.insert(RRS.to_string(), "EC:2".to_string()); + + let cfg = lookup_config(&kvs, 8).expect("lookup should succeed"); + assert_eq!(cfg.standard.parity, 4); + assert_eq!( + cfg.rrs.parity, DEFAULT_RRS_PARITY, + "rrs should fall back to default when CLASS_RRS key is absent" + ); + } +} diff --git a/crates/ecstore/src/notification_sys.rs b/crates/ecstore/src/notification_sys.rs index 7edf4e7d3..452508e3e 100644 --- a/crates/ecstore/src/notification_sys.rs +++ b/crates/ecstore/src/notification_sys.rs @@ -189,6 +189,65 @@ impl NotificationSys { join_all(futures).await } + pub async fn reload_dynamic_config(&self, sub_sys: &str) -> Vec { + let mut futures = Vec::with_capacity(self.peer_clients.len()); + for client in self.peer_clients.iter() { + let sub_sys = sub_sys.to_string(); + futures.push(async move { + if let Some(client) = client { + match client + .signal_service(crate::rpc::SERVICE_SIGNAL_RELOAD_DYNAMIC, &sub_sys, false, SystemTime::UNIX_EPOCH) + .await + { + Ok(_) => NotificationPeerErr { + host: client.host.to_string(), + err: None, + }, + Err(e) => NotificationPeerErr { + host: client.host.to_string(), + err: Some(e), + }, + } + } else { + NotificationPeerErr { + host: "".to_string(), + err: Some(Error::other("peer is not reachable")), + } + } + }); + } + join_all(futures).await + } + + pub async fn refresh_config_snapshot(&self) -> Vec { + let mut futures = Vec::with_capacity(self.peer_clients.len()); + for client in self.peer_clients.iter() { + futures.push(async move { + if let Some(client) = client { + match client + .signal_service(crate::rpc::SERVICE_SIGNAL_REFRESH_CONFIG, "", false, SystemTime::UNIX_EPOCH) + .await + { + Ok(_) => NotificationPeerErr { + host: client.host.to_string(), + err: None, + }, + Err(e) => NotificationPeerErr { + host: client.host.to_string(), + err: Some(e), + }, + } + } else { + NotificationPeerErr { + host: "".to_string(), + err: Some(Error::other("peer is not reachable")), + } + } + }); + } + join_all(futures).await + } + pub async fn storage_info(&self, api: &S) -> rustfs_madmin::StorageInfo { let mut futures = Vec::with_capacity(self.peer_clients.len()); let endpoints = get_global_endpoints(); diff --git a/crates/ecstore/src/rpc/mod.rs b/crates/ecstore/src/rpc/mod.rs index 459052e40..a0ac57af5 100644 --- a/crates/ecstore/src/rpc/mod.rs +++ b/crates/ecstore/src/rpc/mod.rs @@ -29,7 +29,10 @@ pub use internode_data_transport::{ InternodeDataTransport, InternodeDataTransportCapabilities, ReadStreamRequest, TcpHttpInternodeDataTransport, WalkDirStreamRequest, WriteStreamRequest, build_internode_data_transport, build_internode_data_transport_from_env, }; -pub use peer_rest_client::PeerRestClient; +pub use peer_rest_client::{ + PEER_RESTDRY_RUN, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerRestClient, SERVICE_SIGNAL_REFRESH_CONFIG, + SERVICE_SIGNAL_RELOAD_DYNAMIC, +}; pub use peer_s3_client::{LocalPeerS3Client, PeerS3Client, RemotePeerS3Client, S3PeerSys}; pub use remote_disk::RemoteDisk; pub use remote_locker::RemoteClient; diff --git a/crates/ecstore/src/rpc/peer_rest_client.rs b/crates/ecstore/src/rpc/peer_rest_client.rs index aab0b0322..3dcd4afac 100644 --- a/crates/ecstore/src/rpc/peer_rest_client.rs +++ b/crates/ecstore/src/rpc/peer_rest_client.rs @@ -57,6 +57,8 @@ use tracing::warn; pub const PEER_RESTSIGNAL: &str = "signal"; pub const PEER_RESTSUB_SYS: &str = "sub-sys"; pub const PEER_RESTDRY_RUN: &str = "dry-run"; +pub const SERVICE_SIGNAL_REFRESH_CONFIG: u64 = 1; +pub const SERVICE_SIGNAL_RELOAD_DYNAMIC: u64 = 2; const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60; const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30); diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 3ea052286..678b5598f 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -41,7 +41,7 @@ use crate::{ LifecycleOps, gen_transition_objname, get_transitioned_object_reader, put_restore_opts, }, cache_value::metacache_set::{ListPathRawOptions, list_path_raw}, - config::{GLOBAL_STORAGE_CLASS, storageclass}, + config::{get_global_storage_class, storageclass}, disk::{ CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, FileInfoVersions, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, @@ -282,8 +282,7 @@ fn build_tiered_decommission_file_info( default_parity_count: usize, storage_class: Option<&str>, ) -> (FileInfo, usize) { - let parity_drives = GLOBAL_STORAGE_CLASS - .get() + let parity_drives = get_global_storage_class() .and_then(|sc| sc.get_parity_for_sc(storage_class.unwrap_or_default())) .unwrap_or(default_parity_count); let data_drives = disk_count - parity_drives; @@ -783,7 +782,7 @@ impl ObjectIO for SetDisks { } } let sc_parity_drives = { - if let Some(sc) = GLOBAL_STORAGE_CLASS.get() { + if let Some(sc) = get_global_storage_class() { sc.get_parity_for_sc(user_defined.get(AMZ_STORAGE_CLASS).cloned().unwrap_or_default().as_str()) } else { None @@ -837,7 +836,7 @@ impl ObjectIO for SetDisks { let erasure = erasure_coding::Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size); let is_inline_buffer = { - if let Some(sc) = GLOBAL_STORAGE_CLASS.get() { + if let Some(sc) = get_global_storage_class() { sc.should_inline(erasure.shard_file_size(data.size()), opts.versioned) } else { false @@ -3108,7 +3107,7 @@ impl MultipartOperations for SetDisks { } let sc_parity_drives = { - if let Some(sc) = GLOBAL_STORAGE_CLASS.get() { + if let Some(sc) = get_global_storage_class() { sc.get_parity_for_sc(user_defined.get(AMZ_STORAGE_CLASS).cloned().unwrap_or_default().as_str()) } else { None diff --git a/crates/ecstore/src/set_disk/heal.rs b/crates/ecstore/src/set_disk/heal.rs index 90d13ab7d..785c61313 100644 --- a/crates/ecstore/src/set_disk/heal.rs +++ b/crates/ecstore/src/set_disk/heal.rs @@ -417,7 +417,7 @@ impl SetDisks { } let is_inline_buffer = { - if let Some(sc) = GLOBAL_STORAGE_CLASS.get() { + if let Some(sc) = get_global_storage_class() { sc.should_inline(erasure.shard_file_size(latest_meta.size), false) } else { false diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index 21ccb78f2..f82283c94 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -31,7 +31,6 @@ use crate::bucket::utils::check_object_args; use crate::bucket::utils::check_put_object_args; use crate::bucket::utils::check_put_object_part_args; use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname}; -use crate::config::GLOBAL_STORAGE_CLASS; use crate::config::storageclass; use crate::disk::endpoint::{Endpoint, EndpointType}; use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions}; diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs index d5de6bcab..de788cfd5 100644 --- a/crates/ecstore/src/store/rebalance.rs +++ b/crates/ecstore/src/store/rebalance.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::config::get_global_storage_class; struct LatestObjectInfoCandidate { info: Option, @@ -605,7 +606,7 @@ impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_backend_info(&self) -> rustfs_madmin::BackendInfo { let (standard_sc_parity, rr_sc_parity) = { - if let Some(sc) = GLOBAL_STORAGE_CLASS.get() { + if let Some(sc) = get_global_storage_class() { let sc_parity = sc .get_parity_for_sc(storageclass::CLASS_STANDARD) .or(Some(self.pools[0].default_parity_count)); diff --git a/crates/ecstore/src/store_list_objects.rs b/crates/ecstore/src/store_list_objects.rs index 20332366c..828f4b777 100644 --- a/crates/ecstore/src/store_list_objects.rs +++ b/crates/ecstore/src/store_list_objects.rs @@ -841,11 +841,13 @@ impl ECStore { } } + // All sets returned not-found — the listing is simply empty. + // We intentionally do NOT distinguish VolumeNotFound here: during + // listing, a missing volume on a set is equivalent to "no entries" + // and must not surface as an error to the caller. The original + // VolumeNotFound check caused spurious errors when a pool had not + // yet created the bucket prefix on every set. if is_all_not_found(&errs) { - if is_all_volume_not_found(&errs) { - return Err(StorageError::VolumeNotFound); - } - return Ok(Vec::new()); } diff --git a/rustfs/src/admin/handlers/config_admin.rs b/rustfs/src/admin/handlers/config_admin.rs new file mode 100644 index 000000000..726cdfb43 --- /dev/null +++ b/rustfs/src/admin/handlers/config_admin.rs @@ -0,0 +1,2121 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::admin::auth::validate_admin_request; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::admin::service::config::{ + apply_dynamic_config_for_subsystem, is_dynamic_config_subsystem, signal_config_snapshot_reload, signal_dynamic_config_reload, + validate_server_config, +}; +use crate::admin::utils::{encode_compatible_admin_payload, is_compat_admin_request, read_compatible_admin_body}; +use crate::auth::{check_key_valid, get_session_token}; +use crate::error::ApiError; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use http::{HeaderMap, HeaderValue, Uri}; +use hyper::{Method, StatusCode}; +use matchit::Params; +use rustfs_config::audit::{ + AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, ENV_AUDIT_MQTT_BROKER, ENV_AUDIT_MQTT_ENABLE, ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL, + ENV_AUDIT_MQTT_PASSWORD, ENV_AUDIT_MQTT_QOS, ENV_AUDIT_MQTT_QUEUE_DIR, ENV_AUDIT_MQTT_QUEUE_LIMIT, + ENV_AUDIT_MQTT_RECONNECT_INTERVAL, ENV_AUDIT_MQTT_TOPIC, ENV_AUDIT_MQTT_USERNAME, ENV_AUDIT_WEBHOOK_AUTH_TOKEN, + ENV_AUDIT_WEBHOOK_CLIENT_CERT, ENV_AUDIT_WEBHOOK_CLIENT_KEY, ENV_AUDIT_WEBHOOK_ENABLE, ENV_AUDIT_WEBHOOK_ENDPOINT, + ENV_AUDIT_WEBHOOK_QUEUE_DIR, ENV_AUDIT_WEBHOOK_QUEUE_LIMIT, +}; +use rustfs_config::notify::{ + ENV_NOTIFY_MQTT_BROKER, ENV_NOTIFY_MQTT_ENABLE, ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL, ENV_NOTIFY_MQTT_PASSWORD, + ENV_NOTIFY_MQTT_QOS, ENV_NOTIFY_MQTT_QUEUE_DIR, ENV_NOTIFY_MQTT_QUEUE_LIMIT, ENV_NOTIFY_MQTT_RECONNECT_INTERVAL, + ENV_NOTIFY_MQTT_TOPIC, ENV_NOTIFY_MQTT_USERNAME, ENV_NOTIFY_WEBHOOK_AUTH_TOKEN, ENV_NOTIFY_WEBHOOK_CLIENT_CERT, + ENV_NOTIFY_WEBHOOK_CLIENT_KEY, ENV_NOTIFY_WEBHOOK_ENABLE, ENV_NOTIFY_WEBHOOK_ENDPOINT, ENV_NOTIFY_WEBHOOK_QUEUE_DIR, + ENV_NOTIFY_WEBHOOK_QUEUE_LIMIT, NOTIFY_MQTT_SUB_SYS, NOTIFY_WEBHOOK_SUB_SYS, +}; +use rustfs_config::oidc::{ + ENV_IDENTITY_OPENID_CLAIM_NAME, ENV_IDENTITY_OPENID_CLAIM_PREFIX, ENV_IDENTITY_OPENID_CLIENT_ID, + ENV_IDENTITY_OPENID_CLIENT_SECRET, ENV_IDENTITY_OPENID_CONFIG_URL, ENV_IDENTITY_OPENID_DISPLAY_NAME, + ENV_IDENTITY_OPENID_EMAIL_CLAIM, ENV_IDENTITY_OPENID_ENABLE, ENV_IDENTITY_OPENID_GROUPS_CLAIM, + ENV_IDENTITY_OPENID_REDIRECT_URI, ENV_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC, ENV_IDENTITY_OPENID_ROLE_POLICY, + ENV_IDENTITY_OPENID_SCOPES, ENV_IDENTITY_OPENID_USERNAME_CLAIM, IDENTITY_OPENID_SUB_SYS, OIDC_CLAIM_NAME, OIDC_CLAIM_PREFIX, + OIDC_CLIENT_ID, OIDC_CLIENT_SECRET, OIDC_CONFIG_URL, OIDC_DISPLAY_NAME, OIDC_EMAIL_CLAIM, OIDC_GROUPS_CLAIM, + OIDC_REDIRECT_URI, OIDC_REDIRECT_URI_DYNAMIC, OIDC_ROLE_POLICY, OIDC_SCOPES, OIDC_USERNAME_CLAIM, +}; +use rustfs_config::{ + COMMENT_KEY, DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, MAX_ADMIN_REQUEST_BODY_SIZE, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, + MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, + WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, + WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL, +}; +use rustfs_credentials::Credentials; +use rustfs_ecstore::config::com::STORAGE_CLASS_SUB_SYS; +use rustfs_ecstore::config::com::{delete_config, read_config, read_config_without_migrate, save_config, save_server_config}; +use rustfs_ecstore::config::storageclass::{INLINE_BLOCK_ENV, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV}; +use rustfs_ecstore::config::{Config as ServerConfig, DEFAULT_KVS, KV, KVS, get_global_server_config, set_global_server_config}; +use rustfs_ecstore::disk::RUSTFS_META_BUCKET; +use rustfs_ecstore::new_object_layer_fn; +use rustfs_ecstore::store_api::ListOperations; +use rustfs_policy::policy::action::{Action, AdminAction}; +use s3s::header::CONTENT_TYPE; +use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use serde::Serialize; +use std::collections::{BTreeSet, HashMap}; +use std::env; +use time::OffsetDateTime; +use uuid::Uuid; + +const REDACTED_VALUE: &str = "*redacted*"; +const OCTET_STREAM_CONTENT_TYPE: &str = "application/octet-stream"; +const JSON_CONTENT_TYPE: &str = "application/json"; +const TEXT_CONTENT_TYPE: &str = "text/plain; charset=utf-8"; +const CONFIG_HISTORY_PREFIX: &str = "config/history"; +const CONFIG_HISTORY_SUFFIX: &str = ".kv"; +const CONFIG_APPLIED_HEADER: &str = "x-rustfs-config-applied"; +const CONFIG_APPLIED_COMPAT_HEADER: &str = "x-minio-config-applied"; +const CONFIG_APPLIED_TRUE: &str = "true"; +const DEFAULT_COMMENT_DESCRIPTION: &str = "optionally add a comment to this setting"; + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ConfigEntry { + key: String, + value: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ConfigDirective { + sub_system: String, + target: String, + entries: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +struct ConfigSelector { + sub_system: String, + target: Option, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct HelpSubSystemMetadata { + key: &'static str, + description: &'static str, + multiple_targets: bool, + keys: &'static [HelpKeyMetadata], +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +struct HelpKeyMetadata { + key: &'static str, + type_name: &'static str, + description: &'static str, + optional: bool, +} + +#[derive(Debug, Serialize, PartialEq, Eq)] +struct ConfigHelpResponse { + #[serde(rename = "subSys")] + sub_sys: String, + description: String, + #[serde(rename = "multipleTargets")] + multiple_targets: bool, + #[serde(rename = "keysHelp")] + keys_help: Vec, +} + +#[derive(Debug, Serialize, PartialEq, Eq)] +struct ConfigHelpEntry { + key: String, + #[serde(rename = "type")] + type_name: String, + description: String, + optional: bool, + #[serde(rename = "multipleTargets")] + multiple_targets: bool, +} + +#[derive(Debug, Clone, Serialize, PartialEq, Eq)] +struct ConfigHistoryEntry { + #[serde(rename = "RestoreID")] + restore_id: String, + #[serde(rename = "CreateTime", with = "time::serde::rfc3339")] + create_time: OffsetDateTime, + #[serde(rename = "Data", skip_serializing_if = "Option::is_none")] + data: Option, +} + +const STORAGE_CLASS_HELP_KEYS: &[HelpKeyMetadata] = &[ + HelpKeyMetadata { + key: "standard", + type_name: "string", + description: "set the parity count for default standard storage class", + optional: true, + }, + HelpKeyMetadata { + key: "rrs", + type_name: "string", + description: "set the parity count for reduced redundancy storage class", + optional: true, + }, + HelpKeyMetadata { + key: "optimize", + type_name: "string", + description: "optimize parity calculation for standard storage class, set 'capacity' for capacity optimized", + optional: true, + }, + HelpKeyMetadata { + key: "inline_block", + type_name: "string", + description: "set the shard size threshold considered for inline blocks", + optional: true, + }, +]; + +const OIDC_HELP_KEYS: &[HelpKeyMetadata] = &[ + HelpKeyMetadata { + key: OIDC_CONFIG_URL, + type_name: "url", + description: "openid discovery document URL e.g. \"https://accounts.google.com/.well-known/openid-configuration\"", + optional: false, + }, + HelpKeyMetadata { + key: OIDC_CLIENT_ID, + type_name: "string", + description: "unique public identifier for the client application", + optional: false, + }, + HelpKeyMetadata { + key: OIDC_CLIENT_SECRET, + type_name: "string", + description: "secret for the client application identifier", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_SCOPES, + type_name: "csv", + description: "comma-separated list of OpenID scopes for the server", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_REDIRECT_URI, + type_name: "url", + description: "static redirect URI used when dynamic redirects are disabled", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_REDIRECT_URI_DYNAMIC, + type_name: "on|off", + description: "enable Host header based dynamic redirect URI", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_CLAIM_NAME, + type_name: "string", + description: "JWT canned policy claim name", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_CLAIM_PREFIX, + type_name: "string", + description: "prefix added to claims before policy mapping", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_ROLE_POLICY, + type_name: "string", + description: "IAM access policies mapped to this client application and identity provider", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_DISPLAY_NAME, + type_name: "string", + description: "friendly display name for this provider", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_GROUPS_CLAIM, + type_name: "string", + description: "claim name containing group memberships", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_EMAIL_CLAIM, + type_name: "string", + description: "claim name containing the user email", + optional: true, + }, + HelpKeyMetadata { + key: OIDC_USERNAME_CLAIM, + type_name: "string", + description: "claim name containing the username", + optional: true, + }, +]; + +const WEBHOOK_HELP_KEYS: &[HelpKeyMetadata] = &[ + HelpKeyMetadata { + key: WEBHOOK_ENDPOINT, + type_name: "url", + description: "webhook server endpoint e.g. \"http://localhost:8080/rustfs/events\"", + optional: false, + }, + HelpKeyMetadata { + key: WEBHOOK_AUTH_TOKEN, + type_name: "string", + description: "opaque string or JWT authorization token", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_QUEUE_DIR, + type_name: "path", + description: "staging dir for undelivered messages e.g. '/home/events'", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_QUEUE_LIMIT, + type_name: "number", + description: "maximum limit for undelivered messages", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_CLIENT_CERT, + type_name: "string", + description: "client cert for webhook mTLS authentication", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_CLIENT_KEY, + type_name: "string", + description: "client cert key for webhook mTLS authentication", + optional: true, + }, +]; + +const AUDIT_WEBHOOK_HELP_KEYS: &[HelpKeyMetadata] = &[ + HelpKeyMetadata { + key: WEBHOOK_ENDPOINT, + type_name: "url", + description: "HTTP(s) endpoint e.g. \"http://localhost:8080/rustfs/logs/audit\"", + optional: false, + }, + HelpKeyMetadata { + key: WEBHOOK_AUTH_TOKEN, + type_name: "string", + description: "opaque string or JWT authorization token", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_CLIENT_CERT, + type_name: "string", + description: "mTLS certificate for webhook authentication", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_CLIENT_KEY, + type_name: "string", + description: "mTLS certificate key for webhook authentication", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_BATCH_SIZE, + type_name: "number", + description: "number of events per HTTP send to the webhook target", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_QUEUE_LIMIT, + type_name: "number", + description: "channel queue size for webhook targets", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_QUEUE_DIR, + type_name: "path", + description: "staging dir for undelivered audit messages e.g. '/home/audit-events'", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_MAX_RETRY, + type_name: "number", + description: "maximum retry count before audit events are dropped", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_RETRY_INTERVAL, + type_name: "duration", + description: "sleep between retries e.g. '10s'", + optional: true, + }, + HelpKeyMetadata { + key: WEBHOOK_HTTP_TIMEOUT, + type_name: "duration", + description: "maximum duration for each HTTP request", + optional: true, + }, +]; + +const MQTT_HELP_KEYS: &[HelpKeyMetadata] = &[ + HelpKeyMetadata { + key: MQTT_BROKER, + type_name: "uri", + description: "MQTT server endpoint e.g. `tcp://localhost:1883`", + optional: false, + }, + HelpKeyMetadata { + key: MQTT_TOPIC, + type_name: "string", + description: "name of the MQTT topic to publish", + optional: false, + }, + HelpKeyMetadata { + key: MQTT_USERNAME, + type_name: "string", + description: "MQTT username", + optional: true, + }, + HelpKeyMetadata { + key: MQTT_PASSWORD, + type_name: "string", + description: "MQTT password", + optional: true, + }, + HelpKeyMetadata { + key: MQTT_QOS, + type_name: "number", + description: "quality of service priority for MQTT delivery", + optional: true, + }, + HelpKeyMetadata { + key: MQTT_KEEP_ALIVE_INTERVAL, + type_name: "duration", + description: "keep-alive interval for MQTT connections in s,m,h,d", + optional: true, + }, + HelpKeyMetadata { + key: MQTT_RECONNECT_INTERVAL, + type_name: "duration", + description: "reconnect interval for MQTT connections in s,m,h,d", + optional: true, + }, + HelpKeyMetadata { + key: MQTT_QUEUE_DIR, + type_name: "path", + description: "staging dir for undelivered messages e.g. '/home/events'", + optional: true, + }, + HelpKeyMetadata { + key: MQTT_QUEUE_LIMIT, + type_name: "number", + description: "maximum limit for undelivered messages", + optional: true, + }, +]; + +const HELP_SUBSYSTEMS: &[HelpSubSystemMetadata] = &[ + HelpSubSystemMetadata { + key: STORAGE_CLASS_SUB_SYS, + description: "define object level redundancy", + multiple_targets: false, + keys: STORAGE_CLASS_HELP_KEYS, + }, + HelpSubSystemMetadata { + key: IDENTITY_OPENID_SUB_SYS, + description: "enable OpenID SSO support", + multiple_targets: true, + keys: OIDC_HELP_KEYS, + }, + HelpSubSystemMetadata { + key: AUDIT_WEBHOOK_SUB_SYS, + description: "send audit logs to webhook endpoints", + multiple_targets: true, + keys: AUDIT_WEBHOOK_HELP_KEYS, + }, + HelpSubSystemMetadata { + key: AUDIT_MQTT_SUB_SYS, + description: "send audit logs to MQTT endpoints", + multiple_targets: true, + keys: MQTT_HELP_KEYS, + }, + HelpSubSystemMetadata { + key: NOTIFY_WEBHOOK_SUB_SYS, + description: "publish bucket notifications to webhook endpoints", + multiple_targets: true, + keys: WEBHOOK_HELP_KEYS, + }, + HelpSubSystemMetadata { + key: NOTIFY_MQTT_SUB_SYS, + description: "publish bucket notifications to MQTT endpoints", + multiple_targets: true, + keys: MQTT_HELP_KEYS, + }, +]; + +pub fn register_config_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/get-config-kv").as_str(), + AdminOperation(&GetConfigKVHandler {}), + )?; + r.insert( + Method::PUT, + format!("{ADMIN_PREFIX}/v3/set-config-kv").as_str(), + AdminOperation(&SetConfigKVHandler {}), + )?; + r.insert( + Method::DELETE, + format!("{ADMIN_PREFIX}/v3/del-config-kv").as_str(), + AdminOperation(&DelConfigKVHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/help-config-kv").as_str(), + AdminOperation(&HelpConfigKVHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/list-config-history-kv").as_str(), + AdminOperation(&ListConfigHistoryKVHandler {}), + )?; + r.insert( + Method::DELETE, + format!("{ADMIN_PREFIX}/v3/clear-config-history-kv").as_str(), + AdminOperation(&ClearConfigHistoryKVHandler {}), + )?; + r.insert( + Method::PUT, + format!("{ADMIN_PREFIX}/v3/restore-config-history-kv").as_str(), + AdminOperation(&RestoreConfigHistoryKVHandler {}), + )?; + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/config").as_str(), + AdminOperation(&GetConfigHandler {}), + )?; + r.insert( + Method::PUT, + format!("{ADMIN_PREFIX}/v3/config").as_str(), + AdminOperation(&SetConfigHandler {}), + )?; + + Ok(()) +} + +fn extract_query_params(uri: &Uri) -> HashMap { + let mut params = HashMap::new(); + + if let Some(query) = uri.query() { + for (key, value) in url::form_urlencoded::parse(query.as_bytes()) { + params.insert(key.into_owned(), value.into_owned()); + } + } + + params +} + +async fn validate_config_admin_request(req: &S3Request) -> S3Result { + let Some(input_cred) = req.credentials.as_ref() else { + return Err(s3_error!(InvalidRequest, "missing credentials")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + + let remote_addr = req + .extensions + .get::>() + .and_then(|opt| opt.map(|addr| addr.0)); + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(AdminAction::ConfigUpdateAdminAction)], + remote_addr, + ) + .await?; + + Ok(cred) +} + +fn header_value(content_type: &str) -> S3Result { + HeaderValue::from_str(content_type) + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid content type: {err}"))) +} + +fn response_with_content_type(status: StatusCode, body: Vec, content_type: &str) -> S3Result> { + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, header_value(content_type)?); + Ok(S3Response::with_headers((status, Body::from(body)), headers)) +} + +fn encode_config_payload(path: &str, secret_key: &str, data: Vec, plain_content_type: &str) -> S3Result<(Vec, String)> { + if is_compat_admin_request(path) { + let (encoded, _) = encode_compatible_admin_payload(path, secret_key, data)?; + Ok((encoded, OCTET_STREAM_CONTENT_TYPE.to_string())) + } else { + Ok((data, plain_content_type.to_string())) + } +} + +fn success_response(config_applied: bool) -> S3Result> { + if !config_applied { + return Ok(S3Response::new((StatusCode::OK, Body::default()))); + } + + let mut headers = HeaderMap::new(); + headers.insert(CONFIG_APPLIED_HEADER, header_value(CONFIG_APPLIED_TRUE)?); + headers.insert(CONFIG_APPLIED_COMPAT_HEADER, header_value(CONFIG_APPLIED_TRUE)?); + Ok(S3Response::with_headers((StatusCode::OK, Body::default()), headers)) +} + +fn object_store() -> S3Result> { + new_object_layer_fn().ok_or_else(|| s3_error!(InternalError, "server storage not initialized")) +} + +async fn load_server_config_from_store() -> S3Result { + let store = object_store()?; + read_config_without_migrate(store) + .await + .map_err(ApiError::from) + .map_err(Into::into) +} + +async fn load_active_server_config() -> S3Result { + if let Ok(config) = load_server_config_from_store().await { + return Ok(config); + } + + get_global_server_config().ok_or_else(|| s3_error!(InternalError, "server config is not initialized")) +} + +async fn save_server_config_to_store(config: &ServerConfig) -> S3Result<()> { + let store = object_store()?; + save_server_config(store, config) + .await + .map_err(ApiError::from) + .map_err(Into::into) +} + +fn history_object_name(restore_id: &str) -> String { + format!("{CONFIG_HISTORY_PREFIX}/{restore_id}{CONFIG_HISTORY_SUFFIX}") +} + +fn config_update_sub_system(directives: &[ConfigDirective]) -> S3Result> { + let sub_systems = directives + .iter() + .map(|directive| directive.sub_system.as_str()) + .collect::>(); + if sub_systems.len() > 1 { + return Err(s3_error!(InvalidRequest, "config update must target a single subsystem")); + } + Ok(sub_systems.iter().next().copied()) +} + +fn validate_config_directives(directives: &[ConfigDirective]) -> S3Result<()> { + if DEFAULT_KVS.get().is_none() { + rustfs_ecstore::config::init(); + } + let Some(defaults) = DEFAULT_KVS.get() else { + return Err(s3_error!(InternalError, "config defaults are not initialized")); + }; + + for directive in directives { + let Some(default_kvs) = defaults.get(&directive.sub_system) else { + return Err(s3_error!(InvalidRequest, "unsupported config subsystem '{}'", directive.sub_system)); + }; + + let valid_keys = default_kvs.keys().into_iter().collect::>(); + for entry in &directive.entries { + if !valid_keys.contains(&entry.key) { + return Err(s3_error!( + InvalidRequest, + "unsupported config key '{}' for subsystem '{}'", + entry.key, + directive.sub_system + )); + } + } + } + + Ok(()) +} + +fn history_restore_id_from_name(name: &str) -> Option { + name.strip_prefix(&format!("{CONFIG_HISTORY_PREFIX}/"))? + .strip_suffix(CONFIG_HISTORY_SUFFIX) + .map(ToString::to_string) +} + +fn trim_history_entries(mut entries: Vec, count: Option) -> Vec { + entries.sort_by_key(|lhs| lhs.create_time); + + if let Some(count) = count + && entries.len() > count + { + entries.drain(0..entries.len() - count); + } + + entries +} + +async fn save_server_config_history(data: &[u8]) -> S3Result { + let restore_id = Uuid::new_v4().to_string(); + let store = object_store()?; + save_config(store, &history_object_name(&restore_id), data.to_vec()) + .await + .map_err(ApiError::from) + .map_err(S3Error::from)?; + Ok(restore_id) +} + +async fn read_server_config_history(restore_id: &str) -> S3Result> { + let store = object_store()?; + read_config(store, &history_object_name(restore_id)) + .await + .map_err(ApiError::from) + .map_err(Into::into) +} + +async fn delete_server_config_history(restore_id: &str) -> S3Result<()> { + let store = object_store()?; + delete_config(store, &history_object_name(restore_id)) + .await + .map_err(ApiError::from) + .map_err(Into::into) +} + +async fn list_server_config_history(with_data: bool, count: Option) -> S3Result> { + let store = object_store()?; + let mut continuation_token = None; + let mut entries = Vec::new(); + + loop { + let page = store + .clone() + .list_objects_v2( + RUSTFS_META_BUCKET, + CONFIG_HISTORY_PREFIX, + continuation_token.clone(), + None, + 1000, + false, + None, + false, + ) + .await + .map_err(ApiError::from) + .map_err(S3Error::from)?; + + for object in page.objects { + if object.is_dir { + continue; + } + + let Some(restore_id) = history_restore_id_from_name(&object.name) else { + continue; + }; + + let data = if with_data { + Some( + String::from_utf8( + read_config(store.clone(), &object.name) + .await + .map_err(ApiError::from) + .map_err(S3Error::from)?, + ) + .map_err(ApiError::other) + .map_err(S3Error::from)?, + ) + } else { + None + }; + + entries.push(ConfigHistoryEntry { + restore_id, + create_time: object.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH), + data, + }); + } + + if !page.is_truncated { + break; + } + + continuation_token = page.next_continuation_token; + if continuation_token.is_none() { + break; + } + } + + Ok(trim_history_entries(entries, count)) +} + +fn normalize_target(target: &str) -> String { + if target.trim().is_empty() || target == "default" || target == DEFAULT_DELIMITER { + DEFAULT_DELIMITER.to_string() + } else { + target.trim().to_string() + } +} + +fn format_scope(sub_system: &str, target: &str) -> String { + if target == DEFAULT_DELIMITER { + sub_system.to_string() + } else { + format!("{sub_system}:{target}") + } +} + +fn escape_config_value(value: &str) -> String { + value.replace('\\', "\\\\").replace('"', "\\\"") +} + +fn format_kv_pair(key: &str, value: &str) -> String { + format!(r#"{key}="{}""#, escape_config_value(value)) +} + +fn tokenize_config_line(line: &str) -> S3Result> { + let mut tokens = Vec::new(); + let mut current = String::new(); + let mut quote: Option = None; + let mut escaped = false; + + for ch in line.chars() { + if escaped { + current.push(ch); + escaped = false; + continue; + } + + if ch == '\\' { + escaped = true; + continue; + } + + if let Some(active_quote) = quote { + if ch == active_quote { + quote = None; + } else { + current.push(ch); + } + continue; + } + + match ch { + '"' | '\'' => quote = Some(ch), + c if c.is_whitespace() => { + if !current.is_empty() { + tokens.push(std::mem::take(&mut current)); + } + } + _ => current.push(ch), + } + } + + if escaped { + current.push('\\'); + } + + if quote.is_some() { + return Err(s3_error!(InvalidRequest, "unterminated quoted config value")); + } + + if !current.is_empty() { + tokens.push(current); + } + + Ok(tokens) +} + +fn parse_directive_scope(scope: &str) -> S3Result<(String, String)> { + let (sub_system, target) = match scope.split_once(':') { + Some((sub_system, target)) => (sub_system.trim(), normalize_target(target)), + None => (scope.trim(), DEFAULT_DELIMITER.to_string()), + }; + + if sub_system.is_empty() { + return Err(s3_error!(InvalidRequest, "missing config subsystem")); + } + + Ok((sub_system.to_string(), target)) +} + +fn parse_config_directives(input: &str, allow_bare_keys: bool) -> S3Result> { + let mut directives = Vec::new(); + + for raw_line in input.lines() { + let line = raw_line.trim(); + if line.is_empty() || line.starts_with('#') { + continue; + } + + let tokens = tokenize_config_line(line)?; + if tokens.is_empty() { + continue; + } + + let (sub_system, target) = parse_directive_scope(&tokens[0])?; + let mut entries = Vec::new(); + + for token in tokens.iter().skip(1) { + if let Some((key, value)) = token.split_once('=') { + let key = key.trim(); + if key.is_empty() { + return Err(s3_error!(InvalidRequest, "config key cannot be empty")); + } + entries.push(ConfigEntry { + key: key.to_string(), + value: Some(value.to_string()), + }); + } else if allow_bare_keys { + entries.push(ConfigEntry { + key: token.trim().to_string(), + value: None, + }); + } else { + return Err(s3_error!(InvalidRequest, "config assignment must use key=value syntax")); + } + } + + directives.push(ConfigDirective { + sub_system, + target, + entries, + }); + } + + Ok(directives) +} + +fn parse_config_selector(input: &str) -> S3Result { + let raw = input.trim(); + if raw.is_empty() { + return Err(s3_error!(InvalidRequest, "missing config key selector")); + } + + let (sub_system, target) = match raw.split_once(':') { + Some((sub_system, target)) => (sub_system.trim(), Some(normalize_target(target))), + None => (raw, None), + }; + + if sub_system.is_empty() { + return Err(s3_error!(InvalidRequest, "missing config subsystem")); + } + + Ok(ConfigSelector { + sub_system: sub_system.to_string(), + target, + }) +} + +fn set_kvs_value(kvs: &mut KVS, key: &str, value: String) { + if let Some(existing) = kvs.0.iter_mut().find(|entry| entry.key == key) { + existing.value = value; + return; + } + + kvs.0.push(KV { + key: key.to_string(), + value, + hidden_if_empty: false, + }); +} + +fn is_sensitive_key_name(key: &str) -> bool { + let normalized = key.trim().to_ascii_lowercase(); + normalized.contains("secret") + || normalized.contains("password") + || normalized == "token" + || normalized.ends_with("_token") + || normalized == "client_key" + || normalized.ends_with("_client_key") + || normalized == "tls_client_key" + || normalized.ends_with("_tls_client_key") + || normalized == "private_key" + || normalized.ends_with("_private_key") +} + +fn apply_set_directives(config: &mut ServerConfig, directives: &[ConfigDirective]) -> S3Result<()> { + for directive in directives { + let targets = config.0.entry(directive.sub_system.clone()).or_default(); + let kvs = targets.entry(directive.target.clone()).or_default(); + + for entry in &directive.entries { + let value = entry.value.clone().ok_or_else(|| { + s3_error!( + InvalidRequest, + "config key '{}' in subsystem '{}' is missing a value", + entry.key, + directive.sub_system + ) + })?; + set_kvs_value(kvs, &entry.key, value); + } + } + + config.set_defaults(); + Ok(()) +} + +fn apply_delete_directives(config: &mut ServerConfig, directives: &[ConfigDirective]) { + for directive in directives { + let mut remove_subsystem = false; + + if let Some(targets) = config.0.get_mut(&directive.sub_system) { + if directive.entries.is_empty() { + targets.remove(&directive.target); + } else if let Some(kvs) = targets.get_mut(&directive.target) { + let keys = directive + .entries + .iter() + .map(|entry| entry.key.as_str()) + .collect::>(); + kvs.0.retain(|entry| !keys.contains(entry.key.as_str())); + if kvs.0.is_empty() { + targets.remove(&directive.target); + } + } + + remove_subsystem = targets.is_empty(); + } + + if remove_subsystem { + config.0.remove(&directive.sub_system); + } + } + + config.set_defaults(); +} + +fn render_entry_value(entry: &KV, redact_secrets: bool) -> String { + if redact_secrets && (entry.hidden_if_empty || is_sensitive_key_name(&entry.key)) && !entry.value.trim().is_empty() { + REDACTED_VALUE.to_string() + } else { + entry.value.clone() + } +} + +fn should_render_config_entry(entry: &KV) -> bool { + if entry.hidden_if_empty && entry.value.trim().is_empty() { + return false; + } + if entry.key == ENABLE_KEY + && entry + .value + .parse::() + .map(|state| state.is_enabled()) + .unwrap_or(false) + { + return false; + } + true +} + +fn sorted_kv_entries(kvs: &KVS) -> Vec<&KV> { + let mut entries = kvs.0.iter().filter(|entry| entry.key != COMMENT_KEY).collect::>(); + entries.sort_by(|lhs, rhs| lhs.key.cmp(&rhs.key)); + entries +} + +fn render_scope_line(sub_system: &str, target: &str, kvs: &KVS, redact_secrets: bool) -> Option { + let entries = sorted_kv_entries(kvs) + .into_iter() + .filter(|entry| should_render_config_entry(entry)) + .collect::>(); + if entries.is_empty() { + return None; + } + + let pairs = entries + .into_iter() + .map(|entry| format_kv_pair(&entry.key, &render_entry_value(entry, redact_secrets))) + .collect::>(); + + Some(format!("{} {}", format_scope(sub_system, target), pairs.join(" "))) +} + +fn env_var_name_for_target(sub_system: &str, target: &str, key: &str) -> String { + let base = env_help_key(sub_system, key); + if lookup_help_subsystem(sub_system).is_some_and(|metadata| metadata.multiple_targets) && target != DEFAULT_DELIMITER { + format!("{base}_{}", target.to_ascii_uppercase()) + } else { + base + } +} + +fn config_target_keys(kvs: &KVS) -> Vec<&str> { + let mut keys = sorted_kv_entries(kvs) + .into_iter() + .map(|entry| entry.key.as_str()) + .collect::>(); + if !keys.contains(&COMMENT_KEY) { + keys.push(COMMENT_KEY); + } + keys +} + +fn read_non_empty_env_vars() -> Vec<(String, String)> { + env::vars().filter(|(_, value)| !value.is_empty()).collect() +} + +fn discover_env_targets(sub_system: &str, kvs: &KVS, env_vars: &[(String, String)]) -> Vec { + if !lookup_help_subsystem(sub_system).is_some_and(|metadata| metadata.multiple_targets) { + return Vec::new(); + } + + let mut targets = BTreeSet::new(); + for key in config_target_keys(kvs) { + let prefix = format!("{}_", env_var_name_for_target(sub_system, DEFAULT_DELIMITER, key)); + for (name, _) in env_vars { + if let Some(target) = name.strip_prefix(&prefix) + && !target.is_empty() + { + targets.insert(target.to_ascii_lowercase()); + } + } + } + + targets.into_iter().collect() +} + +fn render_env_override_lines( + sub_system: &str, + target: &str, + kvs: &KVS, + redact_secrets: bool, + env_vars: &[(String, String)], +) -> Vec { + let mut lines = Vec::new(); + for key in config_target_keys(kvs) { + let env_name = env_var_name_for_target(sub_system, target, key); + let Some((_, value)) = env_vars.iter().find(|(name, _)| name == &env_name) else { + continue; + }; + if redact_secrets && is_sensitive_key_name(key) { + lines.push(format!("# {env_name}={REDACTED_VALUE}")); + continue; + } + lines.push(format!("# {env_name}={value}")); + } + + lines +} + +fn render_selected_config(config: &ServerConfig, selector: &ConfigSelector, redact_secrets: bool) -> S3Result> { + let Some(targets) = config.0.get(&selector.sub_system) else { + return Err(s3_error!(InvalidRequest, "config subsystem '{}' not found", selector.sub_system)); + }; + + let env_vars = read_non_empty_env_vars(); + let mut lines = Vec::new(); + let mut sorted_targets = targets.iter().collect::>(); + sorted_targets.sort_by_key(|(lhs, _)| *lhs); + + if let Some(target) = selector.target.as_ref() { + let default_kvs = targets.get(DEFAULT_DELIMITER).cloned().unwrap_or_else(KVS::new); + let discovered_targets = discover_env_targets(&selector.sub_system, &default_kvs, &env_vars); + let kvs = if let Some(kvs) = targets.get(target) { + kvs + } else if discovered_targets.iter().any(|name| name == target) { + &default_kvs + } else { + return Err(s3_error!( + InvalidRequest, + "config target '{}' not found for subsystem '{}'", + target, + selector.sub_system + )); + }; + + let scope_start = lines.len(); + lines.extend(render_env_override_lines(&selector.sub_system, target, kvs, redact_secrets, &env_vars)); + if let Some(line) = render_scope_line(&selector.sub_system, target, kvs, redact_secrets) { + lines.push(line); + } else if lines.len() > scope_start { + lines.push(format_scope(&selector.sub_system, target)); + } + } else { + let default_kvs = targets.get(DEFAULT_DELIMITER).cloned().unwrap_or_else(KVS::new); + let mut target_names = sorted_targets + .iter() + .map(|(target, _)| (*target).clone()) + .collect::>(); + for target in discover_env_targets(&selector.sub_system, &default_kvs, &env_vars) { + target_names.insert(target); + } + + for target in target_names { + let kvs = targets.get(&target).unwrap_or(&default_kvs); + let scope_start = lines.len(); + lines.extend(render_env_override_lines(&selector.sub_system, &target, kvs, redact_secrets, &env_vars)); + if let Some(line) = render_scope_line(&selector.sub_system, &target, kvs, redact_secrets) { + lines.push(line); + } else if lines.len() > scope_start { + lines.push(format_scope(&selector.sub_system, &target)); + } + } + } + + Ok(lines.join("\n").into_bytes()) +} + +fn render_full_config(config: &ServerConfig) -> Vec { + let mut subsystems = config.0.iter().collect::>(); + subsystems.sort_by_key(|(lhs, _)| *lhs); + + let mut lines = Vec::new(); + for (sub_system, targets) in subsystems { + let mut sorted_targets = targets.iter().collect::>(); + sorted_targets.sort_by_key(|(lhs, _)| *lhs); + + for (target, kvs) in sorted_targets { + if let Some(line) = render_scope_line(sub_system, target, kvs, false) { + lines.push(line); + } + } + } + + lines.join("\n").into_bytes() +} + +fn lookup_help_subsystem(sub_system: &str) -> Option<&'static HelpSubSystemMetadata> { + HELP_SUBSYSTEMS.iter().find(|metadata| metadata.key == sub_system) +} + +fn normalize_help_subsystem(sub_system: &str) -> &str { + sub_system + .split_once(':') + .map(|(base, _)| base.trim()) + .unwrap_or_else(|| sub_system.trim()) +} + +fn env_help_key(sub_system: &str, key: &str) -> String { + match (sub_system, key) { + (STORAGE_CLASS_SUB_SYS, "standard") => STANDARD_ENV.to_string(), + (STORAGE_CLASS_SUB_SYS, "rrs") => RRS_ENV.to_string(), + (STORAGE_CLASS_SUB_SYS, "optimize") => OPTIMIZE_ENV.to_string(), + (STORAGE_CLASS_SUB_SYS, "inline_block") => INLINE_BLOCK_ENV.to_string(), + (IDENTITY_OPENID_SUB_SYS, ENABLE_KEY) => ENV_IDENTITY_OPENID_ENABLE.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_CONFIG_URL) => ENV_IDENTITY_OPENID_CONFIG_URL.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_CLIENT_ID) => ENV_IDENTITY_OPENID_CLIENT_ID.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_CLIENT_SECRET) => ENV_IDENTITY_OPENID_CLIENT_SECRET.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_SCOPES) => ENV_IDENTITY_OPENID_SCOPES.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_REDIRECT_URI) => ENV_IDENTITY_OPENID_REDIRECT_URI.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_REDIRECT_URI_DYNAMIC) => ENV_IDENTITY_OPENID_REDIRECT_URI_DYNAMIC.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_CLAIM_NAME) => ENV_IDENTITY_OPENID_CLAIM_NAME.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_CLAIM_PREFIX) => ENV_IDENTITY_OPENID_CLAIM_PREFIX.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_ROLE_POLICY) => ENV_IDENTITY_OPENID_ROLE_POLICY.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_DISPLAY_NAME) => ENV_IDENTITY_OPENID_DISPLAY_NAME.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_GROUPS_CLAIM) => ENV_IDENTITY_OPENID_GROUPS_CLAIM.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_EMAIL_CLAIM) => ENV_IDENTITY_OPENID_EMAIL_CLAIM.to_string(), + (IDENTITY_OPENID_SUB_SYS, OIDC_USERNAME_CLAIM) => ENV_IDENTITY_OPENID_USERNAME_CLAIM.to_string(), + (NOTIFY_WEBHOOK_SUB_SYS, ENABLE_KEY) => ENV_NOTIFY_WEBHOOK_ENABLE.to_string(), + (NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_ENDPOINT) => ENV_NOTIFY_WEBHOOK_ENDPOINT.to_string(), + (NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_AUTH_TOKEN) => ENV_NOTIFY_WEBHOOK_AUTH_TOKEN.to_string(), + (NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_LIMIT) => ENV_NOTIFY_WEBHOOK_QUEUE_LIMIT.to_string(), + (NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_DIR) => ENV_NOTIFY_WEBHOOK_QUEUE_DIR.to_string(), + (NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_CERT) => ENV_NOTIFY_WEBHOOK_CLIENT_CERT.to_string(), + (NOTIFY_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_KEY) => ENV_NOTIFY_WEBHOOK_CLIENT_KEY.to_string(), + (NOTIFY_MQTT_SUB_SYS, ENABLE_KEY) => ENV_NOTIFY_MQTT_ENABLE.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_BROKER) => ENV_NOTIFY_MQTT_BROKER.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_TOPIC) => ENV_NOTIFY_MQTT_TOPIC.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_QOS) => ENV_NOTIFY_MQTT_QOS.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_USERNAME) => ENV_NOTIFY_MQTT_USERNAME.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_PASSWORD) => ENV_NOTIFY_MQTT_PASSWORD.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_RECONNECT_INTERVAL) => ENV_NOTIFY_MQTT_RECONNECT_INTERVAL.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_KEEP_ALIVE_INTERVAL) => ENV_NOTIFY_MQTT_KEEP_ALIVE_INTERVAL.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_QUEUE_DIR) => ENV_NOTIFY_MQTT_QUEUE_DIR.to_string(), + (NOTIFY_MQTT_SUB_SYS, MQTT_QUEUE_LIMIT) => ENV_NOTIFY_MQTT_QUEUE_LIMIT.to_string(), + (AUDIT_WEBHOOK_SUB_SYS, ENABLE_KEY) => ENV_AUDIT_WEBHOOK_ENABLE.to_string(), + (AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_ENDPOINT) => ENV_AUDIT_WEBHOOK_ENDPOINT.to_string(), + (AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_AUTH_TOKEN) => ENV_AUDIT_WEBHOOK_AUTH_TOKEN.to_string(), + (AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_LIMIT) => ENV_AUDIT_WEBHOOK_QUEUE_LIMIT.to_string(), + (AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_QUEUE_DIR) => ENV_AUDIT_WEBHOOK_QUEUE_DIR.to_string(), + (AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_CERT) => ENV_AUDIT_WEBHOOK_CLIENT_CERT.to_string(), + (AUDIT_WEBHOOK_SUB_SYS, WEBHOOK_CLIENT_KEY) => ENV_AUDIT_WEBHOOK_CLIENT_KEY.to_string(), + (AUDIT_MQTT_SUB_SYS, ENABLE_KEY) => ENV_AUDIT_MQTT_ENABLE.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_BROKER) => ENV_AUDIT_MQTT_BROKER.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_TOPIC) => ENV_AUDIT_MQTT_TOPIC.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_QOS) => ENV_AUDIT_MQTT_QOS.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_USERNAME) => ENV_AUDIT_MQTT_USERNAME.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_PASSWORD) => ENV_AUDIT_MQTT_PASSWORD.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_RECONNECT_INTERVAL) => ENV_AUDIT_MQTT_RECONNECT_INTERVAL.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_KEEP_ALIVE_INTERVAL) => ENV_AUDIT_MQTT_KEEP_ALIVE_INTERVAL.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_QUEUE_DIR) => ENV_AUDIT_MQTT_QUEUE_DIR.to_string(), + (AUDIT_MQTT_SUB_SYS, MQTT_QUEUE_LIMIT) => ENV_AUDIT_MQTT_QUEUE_LIMIT.to_string(), + _ => format!("{ENV_PREFIX}{}_{}", sub_system.to_ascii_uppercase(), key.to_ascii_uppercase()), + } +} + +fn default_help_postfix(sub_system: &str, key: &str) -> String { + if DEFAULT_KVS.get().is_none() { + rustfs_ecstore::config::init(); + } + + DEFAULT_KVS + .get() + .and_then(|defaults| defaults.get(sub_system)) + .and_then(|kvs| kvs.lookup(key)) + .filter(|value| !value.trim().is_empty()) + .map(|value| format!(" (default: '{}')", value)) + .unwrap_or_default() +} + +fn help_description(sub_system: &str, key: &str, description: &str) -> String { + format!("{description}{}", default_help_postfix(sub_system, key)) +} + +fn build_top_level_help_response() -> ConfigHelpResponse { + ConfigHelpResponse { + sub_sys: String::new(), + description: String::new(), + multiple_targets: false, + keys_help: HELP_SUBSYSTEMS + .iter() + .map(|metadata| ConfigHelpEntry { + key: metadata.key.to_string(), + type_name: String::new(), + description: metadata.description.to_string(), + optional: false, + multiple_targets: metadata.multiple_targets, + }) + .collect(), + } +} + +fn build_help_entries( + metadata: &HelpSubSystemMetadata, + key_filter: Option<&str>, + env_only: bool, +) -> S3Result> { + let enable_entry = metadata.multiple_targets.then(|| ConfigHelpEntry { + key: if env_only { + env_help_key(metadata.key, ENABLE_KEY) + } else { + ENABLE_KEY.to_string() + }, + type_name: "on|off".to_string(), + description: format!("enable {} target, default is 'off'", metadata.key), + optional: false, + multiple_targets: false, + }); + + let comment_entry = ConfigHelpEntry { + key: if env_only { + env_help_key(metadata.key, COMMENT_KEY) + } else { + COMMENT_KEY.to_string() + }, + type_name: "sentence".to_string(), + description: DEFAULT_COMMENT_DESCRIPTION.to_string(), + optional: true, + multiple_targets: false, + }; + + let mut entries = if let Some(key_filter) = key_filter.filter(|value| !value.trim().is_empty()) { + if key_filter == COMMENT_KEY { + vec![comment_entry] + } else { + let entry = metadata + .keys + .iter() + .find(|entry| entry.key == key_filter) + .ok_or_else(|| s3_error!(InvalidRequest, "unknown key {} for sub-system {}", key_filter, metadata.key))?; + vec![ConfigHelpEntry { + key: if env_only { + env_help_key(metadata.key, entry.key) + } else { + entry.key.to_string() + }, + type_name: entry.type_name.to_string(), + description: help_description(metadata.key, entry.key, entry.description), + optional: entry.optional, + multiple_targets: false, + }] + } + } else { + let mut entries = metadata + .keys + .iter() + .map(|entry| ConfigHelpEntry { + key: if env_only { + env_help_key(metadata.key, entry.key) + } else { + entry.key.to_string() + }, + type_name: entry.type_name.to_string(), + description: help_description(metadata.key, entry.key, entry.description), + optional: entry.optional, + multiple_targets: false, + }) + .collect::>(); + entries.push(comment_entry); + entries + }; + + if let Some(enable_entry) = enable_entry { + entries.insert(0, enable_entry); + } + + Ok(entries) +} + +fn build_help_response(sub_system: Option<&str>, key: Option<&str>, env_only: bool) -> S3Result { + let Some(sub_system) = sub_system.filter(|value| !value.trim().is_empty()) else { + return Ok(build_top_level_help_response()); + }; + let sub_system = normalize_help_subsystem(sub_system); + + let metadata = + lookup_help_subsystem(sub_system).ok_or_else(|| s3_error!(InvalidRequest, "unknown sub-system {}", sub_system))?; + let entries = build_help_entries(metadata, key, env_only)?; + + Ok(ConfigHelpResponse { + sub_sys: metadata.key.to_string(), + description: metadata.description.to_string(), + multiple_targets: metadata.multiple_targets, + keys_help: entries, + }) +} + +/// Re-apply all dynamic subsystems from the given config and signal peers to reload. +/// Used after full-config operations (restore, set-config) where the leader replaces +/// the entire config and must ensure runtime state (e.g. GLOBAL_STORAGE_CLASS) is +/// refreshed on both the leader and all peers. +async fn apply_and_signal_dynamic_subsystems(config: &ServerConfig) { + for sub_system in [STORAGE_CLASS_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, AUDIT_MQTT_SUB_SYS] { + if apply_dynamic_config_for_subsystem(config, sub_system).await.unwrap_or(false) { + signal_dynamic_config_reload(sub_system).await; + } + } +} + +pub struct GetConfigKVHandler {} + +#[async_trait::async_trait] +impl Operation for GetConfigKVHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_config_admin_request(&req).await?; + let queries = extract_query_params(&req.uri); + let selector = parse_config_selector( + queries + .get("key") + .ok_or_else(|| s3_error!(InvalidRequest, "missing config key selector"))?, + )?; + let config = load_active_server_config().await?; + let payload = render_selected_config(&config, &selector, true)?; + let (body, content_type) = encode_config_payload(req.uri.path(), &cred.secret_key, payload, TEXT_CONTENT_TYPE)?; + response_with_content_type(StatusCode::OK, body, &content_type) + } +} + +pub struct SetConfigKVHandler {} + +#[async_trait::async_trait] +impl Operation for SetConfigKVHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_config_admin_request(&req).await?; + let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?; + let directives = parse_config_directives(std::str::from_utf8(&body).map_err(ApiError::other)?, false)?; + if directives.is_empty() { + return Err(s3_error!(InvalidRequest, "config update body is empty")); + } + validate_config_directives(&directives)?; + + let sub_system = config_update_sub_system(&directives)?; + let mut config = load_server_config_from_store().await?; + apply_set_directives(&mut config, &directives)?; + validate_server_config(&config, sub_system).await?; + save_server_config_history(&body).await?; + save_server_config_to_store(&config).await?; + set_global_server_config(config.clone()); + let mut config_applied = false; + if let Some(sub_system) = sub_system + && is_dynamic_config_subsystem(sub_system) + { + config_applied = apply_dynamic_config_for_subsystem(&config, sub_system).await?; + if config_applied { + signal_dynamic_config_reload(sub_system).await; + } + } else { + signal_config_snapshot_reload().await; + } + + success_response(config_applied) + } +} + +pub struct DelConfigKVHandler {} + +#[async_trait::async_trait] +impl Operation for DelConfigKVHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_config_admin_request(&req).await?; + let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?; + let directives = parse_config_directives(std::str::from_utf8(&body).map_err(ApiError::other)?, true)?; + if directives.is_empty() { + return Err(s3_error!(InvalidRequest, "config delete body is empty")); + } + validate_config_directives(&directives)?; + + let sub_system = config_update_sub_system(&directives)?; + let mut config = load_server_config_from_store().await?; + apply_delete_directives(&mut config, &directives); + validate_server_config(&config, sub_system).await?; + save_server_config_history(&body).await?; + save_server_config_to_store(&config).await?; + set_global_server_config(config.clone()); + let mut config_applied = false; + if let Some(sub_system) = sub_system + && is_dynamic_config_subsystem(sub_system) + { + config_applied = apply_dynamic_config_for_subsystem(&config, sub_system).await?; + if config_applied { + signal_dynamic_config_reload(sub_system).await; + } + } else { + signal_config_snapshot_reload().await; + } + + success_response(config_applied) + } +} + +pub struct HelpConfigKVHandler {} + +#[async_trait::async_trait] +impl Operation for HelpConfigKVHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_config_admin_request(&req).await?; + let queries = extract_query_params(&req.uri); + let response = build_help_response( + queries.get("subSys").map(String::as_str), + queries.get("key").map(String::as_str), + queries.contains_key("env"), + )?; + let body = serde_json::to_vec(&response).map_err(ApiError::other)?; + response_with_content_type(StatusCode::OK, body, JSON_CONTENT_TYPE) + } +} + +pub struct ListConfigHistoryKVHandler {} + +#[async_trait::async_trait] +impl Operation for ListConfigHistoryKVHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_config_admin_request(&req).await?; + let queries = extract_query_params(&req.uri); + let count = queries + .get("count") + .ok_or_else(|| s3_error!(InvalidRequest, "missing count query parameter"))? + .parse::() + .map_err(ApiError::other) + .map_err(S3Error::from)?; + let entries = list_server_config_history(true, Some(count)).await?; + let payload = serde_json::to_vec(&entries).map_err(ApiError::other).map_err(S3Error::from)?; + let (body, content_type) = encode_config_payload(req.uri.path(), &cred.secret_key, payload, JSON_CONTENT_TYPE)?; + response_with_content_type(StatusCode::OK, body, &content_type) + } +} + +pub struct ClearConfigHistoryKVHandler {} + +#[async_trait::async_trait] +impl Operation for ClearConfigHistoryKVHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_config_admin_request(&req).await?; + let queries = extract_query_params(&req.uri); + let restore_id = queries + .get("restoreId") + .ok_or_else(|| s3_error!(InvalidRequest, "missing restoreId query parameter"))?; + + if restore_id == "all" { + for entry in list_server_config_history(false, None).await? { + delete_server_config_history(&entry.restore_id).await?; + } + } else { + delete_server_config_history(restore_id).await?; + } + + success_response(false) + } +} + +pub struct RestoreConfigHistoryKVHandler {} + +#[async_trait::async_trait] +impl Operation for RestoreConfigHistoryKVHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + validate_config_admin_request(&req).await?; + let queries = extract_query_params(&req.uri); + let restore_id = queries + .get("restoreId") + .ok_or_else(|| s3_error!(InvalidRequest, "missing restoreId query parameter"))?; + let history = read_server_config_history(restore_id).await?; + let directives = parse_config_directives(std::str::from_utf8(&history).map_err(ApiError::other)?, false)?; + if directives.is_empty() { + return Err(s3_error!(InvalidRequest, "history entry is empty")); + } + validate_config_directives(&directives)?; + + let mut config = ServerConfig::new(); + apply_set_directives(&mut config, &directives)?; + validate_server_config(&config, None).await?; + save_server_config_to_store(&config).await?; + set_global_server_config(config.clone()); + apply_and_signal_dynamic_subsystems(&config).await; + signal_config_snapshot_reload().await; + + success_response(false) + } +} + +pub struct GetConfigHandler {} + +#[async_trait::async_trait] +impl Operation for GetConfigHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_config_admin_request(&req).await?; + let config = load_active_server_config().await?; + let payload = render_full_config(&config); + let (body, content_type) = encode_config_payload(req.uri.path(), &cred.secret_key, payload, TEXT_CONTENT_TYPE)?; + response_with_content_type(StatusCode::OK, body, &content_type) + } +} + +pub struct SetConfigHandler {} + +#[async_trait::async_trait] +impl Operation for SetConfigHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let cred = validate_config_admin_request(&req).await?; + let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?; + let directives = parse_config_directives(std::str::from_utf8(&body).map_err(ApiError::other)?, false)?; + if directives.is_empty() { + return Err(s3_error!(InvalidRequest, "full config body is empty")); + } + validate_config_directives(&directives)?; + + let mut config = ServerConfig::new(); + apply_set_directives(&mut config, &directives)?; + validate_server_config(&config, None).await?; + save_server_config_history(&body).await?; + save_server_config_to_store(&config).await?; + set_global_server_config(config.clone()); + apply_and_signal_dynamic_subsystems(&config).await; + signal_config_snapshot_reload().await; + + success_response(false) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn tokenize_config_line_handles_quotes_and_escapes() { + let tokens = tokenize_config_line(r#"identity_openid client_id="console app" client_secret="s3cr\"et" enable=on"#) + .expect("tokenize"); + + assert_eq!( + tokens, + vec![ + "identity_openid".to_string(), + "client_id=console app".to_string(), + r#"client_secret=s3cr"et"#.to_string(), + "enable=on".to_string(), + ] + ); + } + + #[test] + fn parse_selector_supports_all_targets_and_default_target() { + let all_targets = parse_config_selector("notify_webhook").expect("parse selector"); + assert_eq!( + all_targets, + ConfigSelector { + sub_system: "notify_webhook".to_string(), + target: None, + } + ); + + let default_target = parse_config_selector("notify_webhook:").expect("parse selector"); + assert_eq!( + default_target, + ConfigSelector { + sub_system: "notify_webhook".to_string(), + target: Some(DEFAULT_DELIMITER.to_string()), + } + ); + } + + #[test] + fn validate_config_directives_rejects_unknown_subsystem_and_key() { + let unsupported_subsystem = + parse_config_directives(r#"not_real key="value""#, false).expect("parse unsupported subsystem directive"); + let err = validate_config_directives(&unsupported_subsystem).expect_err("unknown subsystem should fail"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + + let unsupported_key = + parse_config_directives(r#"identity_openid not_real="value""#, false).expect("parse unsupported key directive"); + let err = validate_config_directives(&unsupported_key).expect_err("unknown key should fail"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn set_get_and_delete_config_kv_round_trip() { + let mut config = ServerConfig::new(); + let directives = parse_config_directives( + r#"identity_openid config_url="https://issuer.example" client_id="console" client_secret="secret-value""#, + false, + ) + .expect("parse directives"); + apply_set_directives(&mut config, &directives).expect("apply directives"); + + let rendered = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "identity_openid".to_string(), + target: Some(DEFAULT_DELIMITER.to_string()), + }, + true, + ) + .expect("render config"), + ) + .expect("utf8"); + assert!(rendered.contains(r#"client_id="console""#)); + assert!(rendered.contains(r#"client_secret="*redacted*""#)); + + let delete_directives = parse_config_directives("identity_openid client_secret", true).expect("parse delete directives"); + apply_delete_directives(&mut config, &delete_directives); + + let rendered_after_delete = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "identity_openid".to_string(), + target: Some(DEFAULT_DELIMITER.to_string()), + }, + false, + ) + .expect("render config"), + ) + .expect("utf8"); + assert!(!rendered_after_delete.contains("client_secret=")); + } + + #[test] + fn full_config_export_can_be_reapplied() { + rustfs_ecstore::config::init(); + let mut original = ServerConfig::new(); + apply_set_directives( + &mut original, + &parse_config_directives( + r#"storage_class standard="EC:2" rrs="EC:1" +identity_openid config_url="https://issuer.example" client_id="console""#, + false, + ) + .expect("parse directives"), + ) + .expect("apply original directives"); + + let exported = String::from_utf8(render_full_config(&original)).expect("utf8 export"); + let mut restored = ServerConfig::new(); + apply_set_directives(&mut restored, &parse_config_directives(&exported, false).expect("parse exported config")) + .expect("apply restored directives"); + + assert_eq!(render_full_config(&original), render_full_config(&restored)); + } + + #[test] + fn build_help_response_reports_known_keys() { + let response = build_help_response(Some("identity_openid"), Some("client_secret"), false).expect("help response"); + + assert_eq!(response.sub_sys, "identity_openid"); + assert_eq!(response.description, "enable OpenID SSO support"); + assert!(response.multiple_targets); + assert_eq!(response.keys_help.len(), 2); + assert_eq!(response.keys_help[0].key, "enable"); + assert_eq!(response.keys_help[1].key, "client_secret"); + assert_eq!(response.keys_help[1].type_name, "string"); + } + + #[test] + fn build_help_response_supports_env_only_keys() { + let response = build_help_response(Some("notify_webhook"), Some("endpoint"), true).expect("env help response"); + + assert_eq!(response.sub_sys, "notify_webhook"); + assert_eq!(response.keys_help.len(), 2); + assert_eq!(response.keys_help[0].key, "RUSTFS_NOTIFY_WEBHOOK_ENABLE"); + assert_eq!(response.keys_help[0].description, "enable notify_webhook target, default is 'off'"); + assert_eq!(response.keys_help[1].key, "RUSTFS_NOTIFY_WEBHOOK_ENDPOINT"); + } + + #[test] + fn build_help_response_appends_default_value_postfix() { + rustfs_ecstore::config::init(); + let response = build_help_response(Some("identity_openid"), Some("scopes"), false).expect("help response"); + + assert_eq!(response.keys_help.len(), 2); + assert_eq!(response.keys_help[1].type_name, "csv"); + assert!( + response.keys_help[1] + .description + .contains("(default: 'openid,profile,email')") + ); + } + + #[test] + fn build_help_response_exposes_comment_key() { + let response = build_help_response(Some("notify_webhook"), Some("comment"), false).expect("comment help response"); + + assert_eq!(response.keys_help.len(), 2); + assert_eq!(response.keys_help[0].key, "enable"); + assert_eq!(response.keys_help[1].key, "comment"); + assert_eq!(response.keys_help[1].type_name, "sentence"); + assert_eq!(response.keys_help[1].description, DEFAULT_COMMENT_DESCRIPTION); + } + + #[test] + fn build_help_response_uses_target_specific_descriptions() { + let response = build_help_response(Some("notify_webhook"), Some("endpoint"), false).expect("webhook help response"); + + assert_eq!(response.keys_help.len(), 2); + assert_eq!( + response.keys_help[1].description, + "webhook server endpoint e.g. \"http://localhost:8080/rustfs/events\"" + ); + } + + #[test] + fn build_help_response_ignores_target_suffix_in_subsystem_query() { + let response = + build_help_response(Some("notify_webhook:primary"), Some("endpoint"), false).expect("targeted help response"); + + assert_eq!(response.sub_sys, "notify_webhook"); + assert_eq!(response.keys_help.len(), 2); + assert_eq!(response.keys_help[1].key, "endpoint"); + } + + #[test] + fn build_top_level_help_response_uses_empty_type_names() { + let response = build_help_response(None, None, false).expect("top level help response"); + + assert!(!response.keys_help.is_empty()); + assert!(response.keys_help.iter().all(|entry| entry.type_name.is_empty())); + } + + #[test] + fn render_selected_config_includes_env_override_lines() { + rustfs_ecstore::config::init(); + temp_env::with_vars( + [ + ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("http://env.example")), + ("RUSTFS_NOTIFY_WEBHOOK_AUTH_TOKEN_PRIMARY", Some("secret-token")), + ], + || { + let mut config = ServerConfig::new(); + apply_set_directives( + &mut config, + &parse_config_directives(r#"notify_webhook:primary endpoint="http://file.example""#, false) + .expect("parse directives"), + ) + .expect("apply directives"); + + let rendered = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "notify_webhook".to_string(), + target: None, + }, + true, + ) + .expect("render config"), + ) + .expect("utf8"); + + assert!(rendered.contains("# RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY=http://env.example")); + assert!(!rendered.contains("RUSTFS_NOTIFY_WEBHOOK_AUTH_TOKEN_PRIMARY")); + }, + ); + } + + #[test] + fn render_selected_config_lists_env_only_targets() { + rustfs_ecstore::config::init(); + temp_env::with_vars([("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("http://env.example"))], || { + let config = ServerConfig::new(); + let rendered = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "notify_webhook".to_string(), + target: None, + }, + true, + ) + .expect("render config"), + ) + .expect("utf8"); + + assert!(rendered.contains("# RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY=http://env.example")); + assert!(rendered.contains("notify_webhook:primary")); + }); + } + + #[test] + fn render_selected_config_supports_specific_env_only_target_queries() { + rustfs_ecstore::config::init(); + temp_env::with_vars([("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("http://env.example"))], || { + let config = ServerConfig::new(); + let rendered = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "notify_webhook".to_string(), + target: Some("primary".to_string()), + }, + true, + ) + .expect("render config"), + ) + .expect("utf8"); + + assert!(rendered.contains("# RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY=http://env.example")); + assert!(rendered.contains("notify_webhook:primary")); + }); + } + + #[test] + fn render_selected_config_orders_default_before_named_targets() { + rustfs_ecstore::config::init(); + temp_env::with_vars([("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_ALPHA", Some("http://alpha.example"))], || { + let mut config = ServerConfig::new(); + apply_set_directives( + &mut config, + &parse_config_directives( + r#"notify_webhook endpoint="http://default.example" +notify_webhook:beta endpoint="http://beta.example""#, + false, + ) + .expect("parse directives"), + ) + .expect("apply directives"); + + let rendered = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "notify_webhook".to_string(), + target: None, + }, + true, + ) + .expect("render config"), + ) + .expect("utf8"); + + let default_index = rendered.find("notify_webhook ").expect("default target"); + let alpha_index = rendered.find("notify_webhook:alpha").expect("alpha target"); + let beta_index = rendered.find("notify_webhook:beta").expect("beta target"); + assert!(default_index < alpha_index); + assert!(alpha_index < beta_index); + }); + } + + #[test] + fn render_scope_line_omits_enable_on_and_hidden_empty_values() { + let kvs = KVS(vec![ + KV { + key: ENABLE_KEY.to_string(), + value: "on".to_string(), + hidden_if_empty: false, + }, + KV { + key: WEBHOOK_ENDPOINT.to_string(), + value: "http://file.example".to_string(), + hidden_if_empty: false, + }, + KV { + key: WEBHOOK_AUTH_TOKEN.to_string(), + value: String::new(), + hidden_if_empty: true, + }, + ]); + + let rendered = render_scope_line("notify_webhook", "primary", &kvs, true).expect("render scope"); + assert!(rendered.contains(r#"endpoint="http://file.example""#)); + assert!(!rendered.contains("enable=")); + assert!(!rendered.contains("auth_token=")); + } + + #[test] + fn history_object_name_round_trips_restore_id() { + let name = history_object_name("restore-123"); + assert_eq!(name, "config/history/restore-123.kv"); + assert_eq!(history_restore_id_from_name(&name).as_deref(), Some("restore-123")); + assert!(history_restore_id_from_name("config/history/restore-123.txt").is_none()); + } + + #[test] + fn trim_history_entries_keeps_most_recent_count_in_time_order() { + let entries = vec![ + ConfigHistoryEntry { + restore_id: "oldest".to_string(), + create_time: OffsetDateTime::from_unix_timestamp(1).expect("timestamp"), + data: None, + }, + ConfigHistoryEntry { + restore_id: "middle".to_string(), + create_time: OffsetDateTime::from_unix_timestamp(2).expect("timestamp"), + data: None, + }, + ConfigHistoryEntry { + restore_id: "newest".to_string(), + create_time: OffsetDateTime::from_unix_timestamp(3).expect("timestamp"), + data: None, + }, + ]; + + let trimmed = trim_history_entries(entries, Some(2)); + assert_eq!(trimmed.len(), 2); + assert_eq!(trimmed[0].restore_id, "middle"); + assert_eq!(trimmed[1].restore_id, "newest"); + } + + #[test] + fn apply_set_directives_merge_into_existing_config() { + let mut config = ServerConfig::new(); + apply_set_directives( + &mut config, + &parse_config_directives( + r#"storage_class standard="EC:2" +identity_openid client_id="existing-client""#, + false, + ) + .expect("parse initial directives"), + ) + .expect("apply initial directives"); + + let history_directives = parse_config_directives( + r#"identity_openid config_url="https://issuer.example" client_secret="restored-secret""#, + false, + ) + .expect("parse history directives"); + apply_set_directives(&mut config, &history_directives).expect("apply history directives"); + + let mut webhook = ServerConfig::new(); + apply_set_directives( + &mut webhook, + &parse_config_directives(r#"notify_webhook client_key="s3cr3t-key""#, false).expect("parse client key directives"), + ) + .expect("apply client key directives"); + let webhook_rendered = String::from_utf8( + render_selected_config( + &webhook, + &ConfigSelector { + sub_system: "notify_webhook".to_string(), + target: Some(DEFAULT_DELIMITER.to_string()), + }, + true, + ) + .expect("render webhook config"), + ) + .expect("utf8"); + assert!(webhook_rendered.contains(r#"client_key="*redacted*""#)); + + let missing_value_directives = vec![ConfigDirective { + sub_system: "identity_openid".to_string(), + target: DEFAULT_DELIMITER.to_string(), + entries: vec![ConfigEntry { + key: "client_secret".to_string(), + value: None, + }], + }]; + let err = + apply_set_directives(&mut ServerConfig::new(), &missing_value_directives).expect_err("missing value should fail"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + + let storage_class = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "storage_class".to_string(), + target: Some(DEFAULT_DELIMITER.to_string()), + }, + false, + ) + .expect("render storage class"), + ) + .expect("utf8"); + assert!(storage_class.contains(r#"standard="EC:2""#)); + + let oidc = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "identity_openid".to_string(), + target: Some(DEFAULT_DELIMITER.to_string()), + }, + false, + ) + .expect("render oidc"), + ) + .expect("utf8"); + assert!(oidc.contains(r#"client_id="existing-client""#)); + assert!(oidc.contains(r#"config_url="https://issuer.example""#)); + assert!(oidc.contains(r#"client_secret="restored-secret""#)); + } + + #[test] + fn storage_class_get_target_none_matches_full_export() { + rustfs_ecstore::config::init(); + let mut config = ServerConfig::new(); + apply_set_directives( + &mut config, + &parse_config_directives(r#"storage_class standard="EC:4" rrs="EC:2""#, false).expect("parse"), + ) + .expect("apply"); + + // Simulate "mc admin config export" (render_full_config) + let full_output = String::from_utf8(render_full_config(&config)).expect("utf8"); + + // Simulate "mc admin config get storage_class" (target=None, redact=true) + let selected_output = String::from_utf8( + render_selected_config( + &config, + &ConfigSelector { + sub_system: "storage_class".to_string(), + target: None, + }, + true, + ) + .expect("render"), + ) + .expect("utf8"); + + assert!(full_output.contains("EC:4"), "full export should contain EC:4, got:\n{}", full_output); + assert!( + selected_output.contains("EC:4"), + "selected get should contain EC:4, got:\n{}", + selected_output + ); + assert!( + selected_output.contains("EC:2"), + "selected get should contain EC:2, got:\n{}", + selected_output + ); + + let full_sc_line = full_output.lines().find(|l| l.starts_with("storage_class")).unwrap(); + let selected_sc_line = selected_output.lines().find(|l| l.starts_with("storage_class")).unwrap(); + assert_eq!(full_sc_line, selected_sc_line); + } +} diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index ee95db6c0..245ecfbd9 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -16,6 +16,7 @@ pub mod account_info; pub mod audit; mod audit_runtime_config; pub mod bucket_meta; +pub mod config_admin; pub mod event; pub mod group; pub mod heal; @@ -60,6 +61,12 @@ mod tests { fn test_handler_struct_creation() { // Test that handler structs can be created let _account_handler = account_info::AccountInfoHandler {}; + let _get_config_kv_handler = config_admin::GetConfigKVHandler {}; + let _set_config_kv_handler = config_admin::SetConfigKVHandler {}; + let _del_config_kv_handler = config_admin::DelConfigKVHandler {}; + let _help_config_kv_handler = config_admin::HelpConfigKVHandler {}; + let _get_config_handler = config_admin::GetConfigHandler {}; + let _set_config_handler = config_admin::SetConfigHandler {}; let _list_audit_targets = audit::ListAuditTargets {}; let _get_module_switches = module_switch::GetModuleSwitchesHandler {}; let _get_plugin_catalog = plugins_catalog::GetPluginCatalogHandler {}; diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 4bc9cf63a..f8c3f09c4 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -27,8 +27,8 @@ mod console_test; mod route_registration_test; use handlers::{ - audit, bucket_meta, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools, profile_admin, quota, - rebalance, replication, site_replication, sts, system, tier, tls_debug, user, + audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools, + profile_admin, quota, rebalance, replication, site_replication, sts, system, tier, tls_debug, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -57,6 +57,7 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result quota::register_quota_route(&mut r)?; bucket_meta::register_bucket_meta_route(&mut r)?; + config_admin::register_config_route(&mut r)?; audit::register_audit_target_route(&mut r)?; module_switch::register_module_switch_route(&mut r)?; plugins_catalog::register_plugin_catalog_route(&mut r)?; diff --git a/rustfs/src/admin/service/config.rs b/rustfs/src/admin/service/config.rs new file mode 100644 index 000000000..25c14b17b --- /dev/null +++ b/rustfs/src/admin/service/config.rs @@ -0,0 +1,421 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use rustfs_audit::reload_audit_config; +use rustfs_config::audit::{AUDIT_MQTT_SUB_SYS, AUDIT_REDIS_DEFAULT_CHANNEL, AUDIT_WEBHOOK_SUB_SYS}; +use rustfs_config::notify::{NOTIFY_MQTT_SUB_SYS, NOTIFY_REDIS_DEFAULT_CHANNEL, NOTIFY_WEBHOOK_SUB_SYS}; +use rustfs_config::oidc::IDENTITY_OPENID_SUB_SYS; +use rustfs_config::{AUDIT_DEFAULT_DIR, EVENT_DEFAULT_DIR}; +use rustfs_config::{DEFAULT_DELIMITER, ENABLE_KEY, EnableState}; +use rustfs_ecstore::StorageAPI; +use rustfs_ecstore::config::com::{STORAGE_CLASS_SUB_SYS, read_config_without_migrate}; +use rustfs_ecstore::config::storageclass; +use rustfs_ecstore::config::{Config as ServerConfig, KVS, set_global_server_config, set_global_storage_class}; +use rustfs_ecstore::new_object_layer_fn; +use rustfs_ecstore::notification_sys::get_global_notification_sys; +use rustfs_iam::oidc::load_oidc_provider_configs_from_server_config; +use rustfs_targets::config::{ + validate_amqp_config, validate_kafka_config, validate_mqtt_config, validate_mysql_config, validate_nats_config, + validate_postgres_config, validate_pulsar_config, validate_redis_config, validate_webhook_config, +}; +use s3s::{S3Error, S3ErrorCode, S3Result}; +use tracing::warn; +use url::Url; + +pub fn is_dynamic_config_subsystem(sub_system: &str) -> bool { + matches!(sub_system, STORAGE_CLASS_SUB_SYS | AUDIT_WEBHOOK_SUB_SYS | AUDIT_MQTT_SUB_SYS) +} + +fn internal_error(message: impl Into) -> S3Error { + S3Error::with_message(S3ErrorCode::InternalError, message.into()) +} + +fn invalid_request(message: impl Into) -> S3Error { + S3Error::with_message(S3ErrorCode::InvalidRequest, message.into()) +} + +async fn apply_storage_class_runtime_config(config: &ServerConfig) -> S3Result<()> { + let Some(store) = new_object_layer_fn() else { + return Err(internal_error("storage layer not initialized")); + }; + + let kvs = config.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER).unwrap_or_default(); + let set_drive_count = store.set_drive_counts().into_iter().next().unwrap_or(1); + let parsed = storageclass::lookup_config(&kvs, set_drive_count) + .map_err(|err| internal_error(format!("failed to apply storage class config: {err}")))?; + set_global_storage_class(parsed); + Ok(()) +} + +fn validate_storage_class_kvs(kvs: &KVS, set_drive_counts: &[usize]) -> S3Result<()> { + for count in set_drive_counts { + storageclass::lookup_config(kvs, *count) + .map_err(|err| invalid_request(format!("invalid storage class config: {err}")))?; + } + + Ok(()) +} + +async fn validate_storage_class_config(config: &ServerConfig) -> S3Result<()> { + let Some(store) = new_object_layer_fn() else { + return Err(internal_error("storage layer not initialized")); + }; + + let kvs = config.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_DELIMITER).unwrap_or_default(); + let set_drive_counts = store.set_drive_counts(); + if set_drive_counts.is_empty() { + return validate_storage_class_kvs(&kvs, &[1]); + } + + validate_storage_class_kvs(&kvs, &set_drive_counts) +} + +fn target_enabled(kvs: &KVS) -> bool { + kvs.lookup(ENABLE_KEY) + .and_then(|value| value.parse::().ok()) + .is_some_and(|state| state.is_enabled()) +} + +fn target_kvs(config: &ServerConfig, sub_system: &str, target: &str) -> KVS { + let mut kvs = config.get_value(sub_system, DEFAULT_DELIMITER).unwrap_or_default(); + if target != DEFAULT_DELIMITER + && let Some(target_kvs) = config.0.get(sub_system).and_then(|targets| targets.get(target)) + { + kvs.extend(target_kvs.clone()); + } + + kvs +} + +fn validate_notify_subsystem_config(config: &ServerConfig, sub_system: &str) -> S3Result<()> { + let Some(targets) = config.0.get(sub_system) else { + return Ok(()); + }; + + for target in targets.keys() { + let kvs = target_kvs(config, sub_system, target); + if !target_enabled(&kvs) { + continue; + } + + let result = match sub_system { + "notify_webhook" => validate_webhook_config(&kvs, EVENT_DEFAULT_DIR), + "notify_amqp" => validate_amqp_config(&kvs, EVENT_DEFAULT_DIR), + "notify_kafka" => validate_kafka_config(&kvs, EVENT_DEFAULT_DIR), + "notify_mqtt" => validate_mqtt_config(&kvs), + "notify_mysql" => validate_mysql_config(&kvs, EVENT_DEFAULT_DIR), + "notify_nats" => validate_nats_config(&kvs, EVENT_DEFAULT_DIR), + "notify_postgres" => validate_postgres_config(&kvs, EVENT_DEFAULT_DIR), + "notify_pulsar" => validate_pulsar_config(&kvs, EVENT_DEFAULT_DIR), + "notify_redis" => validate_redis_config(&kvs, EVENT_DEFAULT_DIR, NOTIFY_REDIS_DEFAULT_CHANNEL), + _ => return Ok(()), + }; + + result.map_err(|err| invalid_request(format!("invalid {sub_system} config for target '{target}': {err}")))?; + } + + Ok(()) +} + +fn validate_audit_subsystem_config(config: &ServerConfig, sub_system: &str) -> S3Result<()> { + let Some(targets) = config.0.get(sub_system) else { + return Ok(()); + }; + + for target in targets.keys() { + let kvs = target_kvs(config, sub_system, target); + if !target_enabled(&kvs) { + continue; + } + + let result = match sub_system { + "audit_webhook" => validate_webhook_config(&kvs, AUDIT_DEFAULT_DIR), + "audit_amqp" => validate_amqp_config(&kvs, AUDIT_DEFAULT_DIR), + "audit_kafka" => validate_kafka_config(&kvs, AUDIT_DEFAULT_DIR), + "audit_mqtt" => validate_mqtt_config(&kvs), + "audit_mysql" => validate_mysql_config(&kvs, AUDIT_DEFAULT_DIR), + "audit_nats" => validate_nats_config(&kvs, AUDIT_DEFAULT_DIR), + "audit_postgres" => validate_postgres_config(&kvs, AUDIT_DEFAULT_DIR), + "audit_pulsar" => validate_pulsar_config(&kvs, AUDIT_DEFAULT_DIR), + "audit_redis" => validate_redis_config(&kvs, AUDIT_DEFAULT_DIR, AUDIT_REDIS_DEFAULT_CHANNEL), + _ => return Ok(()), + }; + + result.map_err(|err| invalid_request(format!("invalid {sub_system} config for target '{target}': {err}")))?; + } + + Ok(()) +} + +fn is_valid_provider_id(id: &str) -> bool { + !id.is_empty() && id.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-') +} + +fn is_valid_scheme(scheme: &str) -> bool { + scheme == "http" || scheme == "https" +} + +fn validate_absolute_http_url(value: &str, field_name: &str) -> S3Result<()> { + let parsed = Url::parse(value).map_err(|_| invalid_request(format!("{field_name} must be an absolute http/https URL")))?; + if !is_valid_scheme(parsed.scheme()) || parsed.host_str().is_none() { + return Err(invalid_request(format!("{field_name} must be an absolute http/https URL"))); + } + + Ok(()) +} + +fn validate_identity_openid_config(config: &ServerConfig) -> S3Result<()> { + if let Some(targets) = config.0.get(IDENTITY_OPENID_SUB_SYS) { + for target in targets.keys() { + if target == DEFAULT_DELIMITER { + continue; + } + if !is_valid_provider_id(target) { + return Err(invalid_request(format!("invalid provider_id '{target}'"))); + } + } + } + + for provider in load_oidc_provider_configs_from_server_config(config) { + if !is_valid_provider_id(&provider.id) { + return Err(invalid_request(format!("invalid provider_id '{}'", provider.id))); + } + if provider.config_url.trim().is_empty() { + return Err(invalid_request(format!("identity_openid provider '{}' requires config_url", provider.id))); + } + validate_absolute_http_url(&provider.config_url, "config_url")?; + + if provider.client_id.trim().is_empty() { + return Err(invalid_request(format!("identity_openid provider '{}' requires client_id", provider.id))); + } + + if !provider.redirect_uri_dynamic { + let Some(redirect_uri) = provider.redirect_uri.as_deref() else { + return Err(invalid_request(format!( + "identity_openid provider '{}' requires redirect_uri when redirect_uri_dynamic is off", + provider.id + ))); + }; + validate_absolute_http_url(redirect_uri, "redirect_uri")?; + } else if let Some(redirect_uri) = provider.redirect_uri.as_deref() { + validate_absolute_http_url(redirect_uri, "redirect_uri")?; + } + + if !provider.scopes.iter().any(|scope| scope == "openid") { + return Err(invalid_request(format!( + "identity_openid provider '{}' scopes must include openid", + provider.id + ))); + } + } + + Ok(()) +} + +pub async fn validate_server_config(config: &ServerConfig, sub_system: Option<&str>) -> S3Result<()> { + match sub_system { + Some(STORAGE_CLASS_SUB_SYS) => validate_storage_class_config(config).await, + Some(NOTIFY_WEBHOOK_SUB_SYS) => validate_notify_subsystem_config(config, NOTIFY_WEBHOOK_SUB_SYS), + Some(NOTIFY_MQTT_SUB_SYS) => validate_notify_subsystem_config(config, NOTIFY_MQTT_SUB_SYS), + Some(AUDIT_WEBHOOK_SUB_SYS) => validate_audit_subsystem_config(config, AUDIT_WEBHOOK_SUB_SYS), + Some(AUDIT_MQTT_SUB_SYS) => validate_audit_subsystem_config(config, AUDIT_MQTT_SUB_SYS), + Some(IDENTITY_OPENID_SUB_SYS) => validate_identity_openid_config(config), + Some(_) => Ok(()), + None => { + validate_storage_class_config(config).await?; + validate_notify_subsystem_config(config, NOTIFY_WEBHOOK_SUB_SYS)?; + validate_notify_subsystem_config(config, NOTIFY_MQTT_SUB_SYS)?; + validate_audit_subsystem_config(config, AUDIT_WEBHOOK_SUB_SYS)?; + validate_audit_subsystem_config(config, AUDIT_MQTT_SUB_SYS)?; + validate_identity_openid_config(config)?; + Ok(()) + } + } +} + +pub async fn apply_dynamic_config_for_subsystem(config: &ServerConfig, sub_system: &str) -> S3Result { + if !is_dynamic_config_subsystem(sub_system) { + return Ok(false); + } + + match sub_system { + STORAGE_CLASS_SUB_SYS => apply_storage_class_runtime_config(config).await?, + AUDIT_WEBHOOK_SUB_SYS | AUDIT_MQTT_SUB_SYS => reload_audit_config(config.clone()) + .await + .map_err(|err| internal_error(format!("failed to reload audit config: {err}")))?, + _ => return Ok(false), + } + + Ok(true) +} + +pub async fn reload_dynamic_config_runtime_state(sub_system: &str) -> S3Result<()> { + if !is_dynamic_config_subsystem(sub_system) { + return Err(internal_error(format!("unsupported dynamic config subsystem: {sub_system}"))); + } + + let Some(store) = new_object_layer_fn() else { + return Err(internal_error("storage layer not initialized")); + }; + + let config = read_config_without_migrate(store).await.map_err(|err| { + warn!("peer reload_dynamic_config: failed to load server config for {sub_system}: {err}"); + internal_error(format!("failed to load server config: {err}")) + })?; + apply_dynamic_config_for_subsystem(&config, sub_system).await.map_err(|err| { + warn!("peer reload_dynamic_config: failed to apply {sub_system}: {err}"); + err + })?; + Ok(()) +} + +pub async fn reload_runtime_config_snapshot() -> S3Result<()> { + let Some(store) = new_object_layer_fn() else { + return Err(internal_error("storage layer not initialized")); + }; + + let config = read_config_without_migrate(store).await.map_err(|err| { + warn!("peer reload_runtime_config_snapshot: failed to load server config: {err}"); + internal_error(format!("failed to load server config: {err}")) + })?; + + // Re-apply dynamic subsystems before publishing the snapshot, so that + // runtime state (e.g. GLOBAL_STORAGE_CLASS) is refreshed on this peer. + for sub_system in [STORAGE_CLASS_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, AUDIT_MQTT_SUB_SYS] { + if let Err(err) = apply_dynamic_config_for_subsystem(&config, sub_system).await { + warn!("peer reload_runtime_config_snapshot: failed to apply {sub_system}: {err}"); + } + } + + set_global_server_config(config); + Ok(()) +} + +pub async fn signal_dynamic_config_reload(sub_system: &str) { + if !is_dynamic_config_subsystem(sub_system) { + return; + } + + let Some(notification_sys) = get_global_notification_sys() else { + return; + }; + + for failure in notification_sys.reload_dynamic_config(sub_system).await { + if let Some(err) = failure.err { + tracing::warn!("peer {} dynamic config reload for {} failed: {}", failure.host, sub_system, err); + } + } +} + +pub async fn signal_config_snapshot_reload() { + let Some(notification_sys) = get_global_notification_sys() else { + return; + }; + + for failure in notification_sys.refresh_config_snapshot().await { + if let Some(err) = failure.err { + tracing::warn!("peer config snapshot refresh failed for {}: {}", failure.host, err); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_config::notify::NOTIFY_WEBHOOK_SUB_SYS; + use rustfs_config::oidc::{OIDC_CLIENT_ID, OIDC_CONFIG_URL, OIDC_SCOPES}; + use rustfs_config::{MQTT_BROKER, MQTT_QUEUE_DIR, MQTT_TOPIC, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR}; + + #[test] + fn dynamic_config_subsystems_match_runtime_apply_support() { + assert!(is_dynamic_config_subsystem(AUDIT_WEBHOOK_SUB_SYS)); + assert!(is_dynamic_config_subsystem(AUDIT_MQTT_SUB_SYS)); + assert!(is_dynamic_config_subsystem(STORAGE_CLASS_SUB_SYS)); + assert!(!is_dynamic_config_subsystem("identity_openid")); + assert!(!is_dynamic_config_subsystem("notify_webhook")); + } + + #[test] + fn validate_storage_class_kvs_rejects_invalid_parity() { + let mut kvs = KVS::new(); + kvs.insert("standard".to_string(), "EC:5".to_string()); + + let err = validate_storage_class_kvs(&kvs, &[4]).expect_err("invalid parity should fail"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn validate_notify_subsystem_config_rejects_invalid_webhook_endpoint() { + rustfs_ecstore::config::init(); + let mut config = ServerConfig::new(); + let targets = config.0.get_mut(NOTIFY_WEBHOOK_SUB_SYS).expect("notify webhook defaults"); + let kvs = targets.get_mut(DEFAULT_DELIMITER).expect("default target"); + kvs.insert(ENABLE_KEY.to_string(), EnableState::On.to_string()); + kvs.insert(WEBHOOK_ENDPOINT.to_string(), "not-a-url".to_string()); + kvs.insert(WEBHOOK_QUEUE_DIR.to_string(), "/tmp/rustfs-notify".to_string()); + + let err = validate_notify_subsystem_config(&config, NOTIFY_WEBHOOK_SUB_SYS).expect_err("invalid endpoint should fail"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn validate_audit_subsystem_config_rejects_relative_queue_dir() { + rustfs_ecstore::config::init(); + let mut config = ServerConfig::new(); + let targets = config.0.get_mut(AUDIT_MQTT_SUB_SYS).expect("audit mqtt defaults"); + let kvs = targets.get_mut(DEFAULT_DELIMITER).expect("default target"); + kvs.insert(ENABLE_KEY.to_string(), EnableState::On.to_string()); + kvs.insert(MQTT_BROKER.to_string(), "mqtt://localhost:1883".to_string()); + kvs.insert(MQTT_TOPIC.to_string(), "audit-events".to_string()); + kvs.insert(MQTT_QUEUE_DIR.to_string(), "relative/dir".to_string()); + + let err = validate_audit_subsystem_config(&config, AUDIT_MQTT_SUB_SYS).expect_err("relative queue dir should fail"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn validate_identity_openid_config_rejects_missing_openid_scope() { + rustfs_ecstore::config::init(); + let mut config = ServerConfig::new(); + let targets = config.0.get_mut(IDENTITY_OPENID_SUB_SYS).expect("openid defaults"); + let kvs = targets.get_mut(DEFAULT_DELIMITER).expect("default target"); + kvs.insert( + OIDC_CONFIG_URL.to_string(), + "https://issuer.example/.well-known/openid-configuration".to_string(), + ); + kvs.insert(OIDC_CLIENT_ID.to_string(), "console".to_string()); + kvs.insert(OIDC_SCOPES.to_string(), "profile,email".to_string()); + + let err = validate_identity_openid_config(&config).expect_err("openid scope should be required"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } + + #[test] + fn validate_identity_openid_config_rejects_invalid_named_provider_id() { + rustfs_ecstore::config::init(); + let mut config = ServerConfig::new(); + let targets = config.0.get_mut(IDENTITY_OPENID_SUB_SYS).expect("openid defaults"); + let default_kvs = targets.get(DEFAULT_DELIMITER).cloned().expect("default target"); + let mut named_kvs = default_kvs; + named_kvs.insert( + OIDC_CONFIG_URL.to_string(), + "https://issuer.example/.well-known/openid-configuration".to_string(), + ); + named_kvs.insert(OIDC_CLIENT_ID.to_string(), "console".to_string()); + targets.insert("bad$id".to_string(), named_kvs); + + let err = validate_identity_openid_config(&config).expect_err("provider id should be validated"); + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + } +} diff --git a/rustfs/src/admin/service/mod.rs b/rustfs/src/admin/service/mod.rs index 4fc81d80f..5af8b3f7e 100644 --- a/rustfs/src/admin/service/mod.rs +++ b/rustfs/src/admin/service/mod.rs @@ -12,4 +12,5 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub mod config; pub mod site_replication; diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 0c4cd1662..23d471e0c 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -293,8 +293,10 @@ fn is_empty_body_admin_path(method: &Method, uri: &http::Uri) -> bool { path, "/minio/admin/v3/set-user-status" | "/minio/admin/v3/set-group-status" + | "/minio/admin/v3/restore-config-history-kv" | "/rustfs/admin/v3/set-user-status" | "/rustfs/admin/v3/set-group-status" + | "/rustfs/admin/v3/restore-config-history-kv" ), Method::POST => { matches!( @@ -1571,6 +1573,17 @@ mod tests { assert!(!should_force_zero_content_length_for_empty_body_route(&request)); } + #[test] + fn admin_restore_config_history_without_content_length_is_normalized() { + let request = Request::builder() + .method(Method::PUT) + .uri("/minio/admin/v3/restore-config-history-kv?restoreId=test") + .body(()) + .expect("request"); + + assert!(should_force_zero_content_length_for_empty_body_route(&request)); + } + #[test] fn s3_put_object_is_not_normalized() { let request = Request::builder() diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 6cf1980e6..82d09a478 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -12,7 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::admin::service::site_replication::reload_site_replication_runtime_state; +use crate::admin::service::{ + config::{reload_dynamic_config_runtime_state, reload_runtime_config_snapshot}, + site_replication::reload_site_replication_runtime_state, +}; use bytes::Bytes; use futures::Stream; use futures_util::future::join_all; @@ -29,7 +32,10 @@ use rustfs_ecstore::{ global::GLOBAL_TierConfigMgr, metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics}, new_object_layer_fn, - rpc::{LocalPeerS3Client, PeerS3Client}, + rpc::{ + LocalPeerS3Client, PEER_RESTSIGNAL, PEER_RESTSUB_SYS, PeerS3Client, SERVICE_SIGNAL_REFRESH_CONFIG, + SERVICE_SIGNAL_RELOAD_DYNAMIC, + }, store::{all_local_disk_path, find_local_disk_by_ref}, store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI}, }; @@ -45,7 +51,7 @@ use rustfs_protos::{ proto_gen::node_service::{node_service_server::NodeService as Node, *}, }; use serde::Deserialize; -use std::{io::Cursor, pin::Pin, sync::Arc}; +use std::{collections::HashMap, io::Cursor, pin::Pin, sync::Arc}; use tokio::spawn; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; @@ -794,8 +800,49 @@ impl Node for NodeService { } async fn signal_service(&self, request: Request) -> Result, Status> { - let _request = request.into_inner(); - Err(unimplemented_rpc("signal_service")) + let request = request.into_inner(); + let vars = match request.vars { + Some(vars) => vars.value, + None => HashMap::new(), + }; + let raw_signal = vars.get(PEER_RESTSIGNAL).map(String::as_str); + let signal = raw_signal.and_then(|value| value.parse::().ok()); + let sub_system = vars.get(PEER_RESTSUB_SYS).map(String::as_str).unwrap_or_default(); + + match signal { + Some(SERVICE_SIGNAL_REFRESH_CONFIG) => match reload_runtime_config_snapshot().await { + Ok(()) => Ok(Response::new(SignalServiceResponse { + success: true, + error_info: None, + })), + Err(err) => Ok(Response::new(SignalServiceResponse { + success: false, + error_info: Some(err.to_string()), + })), + }, + Some(SERVICE_SIGNAL_RELOAD_DYNAMIC) => match reload_dynamic_config_runtime_state(sub_system).await { + Ok(()) => Ok(Response::new(SignalServiceResponse { + success: true, + error_info: None, + })), + Err(err) => Ok(Response::new(SignalServiceResponse { + success: false, + error_info: Some(err.to_string()), + })), + }, + Some(other) => Ok(Response::new(SignalServiceResponse { + success: false, + error_info: Some(format!("unsupported service signal: {other}")), + })), + None if raw_signal.is_some() => Ok(Response::new(SignalServiceResponse { + success: false, + error_info: Some(format!("invalid service signal value: {}", raw_signal.unwrap_or_default())), + })), + None => Ok(Response::new(SignalServiceResponse { + success: false, + error_info: Some("missing service signal".to_string()), + })), + } } async fn background_heal_status( @@ -2450,6 +2497,125 @@ mod tests { assert!(reload_response.error_info.is_some()); } + #[tokio::test] + async fn test_signal_service_rejects_missing_signal() { + let service = create_test_node_service(); + + let request = Request::new(SignalServiceRequest { + vars: Some(Mss { value: HashMap::new() }), + }); + + let response = service.signal_service(request).await; + assert!(response.is_ok()); + + let signal_response = response.unwrap().into_inner(); + assert!(!signal_response.success); + assert_eq!(signal_response.error_info.as_deref(), Some("missing service signal")); + } + + #[tokio::test] + async fn test_signal_service_rejects_invalid_signal_value() { + let service = create_test_node_service(); + + let mut vars = HashMap::new(); + vars.insert(PEER_RESTSIGNAL.to_string(), "abc".to_string()); + + let request = Request::new(SignalServiceRequest { + vars: Some(Mss { value: vars }), + }); + + let response = service.signal_service(request).await; + assert!(response.is_ok()); + + let signal_response = response.unwrap().into_inner(); + assert!(!signal_response.success); + assert_eq!(signal_response.error_info.as_deref(), Some("invalid service signal value: abc")); + } + + #[tokio::test] + async fn test_signal_service_rejects_unsupported_signal() { + let service = create_test_node_service(); + + let mut vars = HashMap::new(); + vars.insert(PEER_RESTSIGNAL.to_string(), "99".to_string()); + + let request = Request::new(SignalServiceRequest { + vars: Some(Mss { value: vars }), + }); + + let response = service.signal_service(request).await; + assert!(response.is_ok()); + + let signal_response = response.unwrap().into_inner(); + assert!(!signal_response.success); + assert_eq!(signal_response.error_info.as_deref(), Some("unsupported service signal: 99")); + } + + #[tokio::test] + async fn test_signal_service_rejects_non_dynamic_subsystem() { + let service = create_test_node_service(); + + let mut vars = HashMap::new(); + vars.insert(PEER_RESTSIGNAL.to_string(), SERVICE_SIGNAL_RELOAD_DYNAMIC.to_string()); + vars.insert(PEER_RESTSUB_SYS.to_string(), "identity_openid".to_string()); + + let request = Request::new(SignalServiceRequest { + vars: Some(Mss { value: vars }), + }); + + let response = service.signal_service(request).await; + assert!(response.is_ok()); + + let signal_response = response.unwrap().into_inner(); + assert!(!signal_response.success); + let error_info = signal_response.error_info.expect("expected error info"); + assert!(error_info.contains("unsupported dynamic config subsystem: identity_openid")); + } + + #[tokio::test] + async fn test_signal_service_refresh_config_requires_object_layer() { + let service = create_test_node_service(); + + let mut vars = HashMap::new(); + vars.insert(PEER_RESTSIGNAL.to_string(), SERVICE_SIGNAL_REFRESH_CONFIG.to_string()); + + let request = Request::new(SignalServiceRequest { + vars: Some(Mss { value: vars }), + }); + + let response = service.signal_service(request).await; + assert!(response.is_ok()); + + let signal_response = response.unwrap().into_inner(); + assert!(!signal_response.success); + let error_info = signal_response.error_info.expect("expected error info"); + assert!(error_info.contains("storage layer not initialized")); + } + + #[tokio::test] + async fn test_signal_service_reload_dynamic_requires_object_layer() { + let service = create_test_node_service(); + + let mut vars = HashMap::new(); + vars.insert(PEER_RESTSIGNAL.to_string(), SERVICE_SIGNAL_RELOAD_DYNAMIC.to_string()); + vars.insert( + PEER_RESTSUB_SYS.to_string(), + rustfs_ecstore::config::com::STORAGE_CLASS_SUB_SYS.to_string(), + ); + + let request = Request::new(SignalServiceRequest { + vars: Some(Mss { value: vars }), + }); + + let response = service.signal_service(request).await; + assert!(response.is_ok()); + + let signal_response = response.unwrap().into_inner(); + assert!(!signal_response.success); + let error_info = signal_response.error_info.expect("expected error info"); + assert!(error_info.contains("storage layer not initialized")); + } + fn assert_unimplemented_status(response: Result, Status>, method: &str) { let err = match response { Ok(_) => panic!("unimplemented RPC should return an error status"), @@ -2493,10 +2659,6 @@ mod tests { .await, "get_all_bucket_stats", ); - assert_unimplemented_status( - service.signal_service(Request::new(SignalServiceRequest::default())).await, - "signal_service", - ); assert_unimplemented_status( service .background_heal_status(Request::new(BackgroundHealStatusRequest::default()))