diff --git a/rustfs/src/admin/handlers/audit.rs b/rustfs/src/admin/handlers/audit.rs index d13fa9a26..b7b5b2bab 100644 --- a/rustfs/src/admin/handlers/audit.rs +++ b/rustfs/src/admin/handlers/audit.rs @@ -17,14 +17,16 @@ use crate::admin::{ handlers::target_descriptor::{ AdminTargetSpec, AdminTargetValidator, EndpointKey, TargetDomain, TargetEndpointSource, allowed_target_keys, collect_validated_key_values as shared_collect_validated_key_values, - merge_target_endpoints as shared_merge_target_endpoints, + merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, target_service_name, target_spec, validate_target_request, }, router::{AdminOperation, Operation, S3Router}, }; use crate::auth::{check_key_valid, get_session_token}; -use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use crate::server::{ + ADMIN_PREFIX, RemoteAddr, is_audit_module_enabled, refresh_audit_module_enabled, refresh_persisted_module_switches_from_store, +}; use futures::stream::{FuturesUnordered, StreamExt}; use http::{HeaderMap, StatusCode}; use hyper::Method; @@ -168,6 +170,17 @@ fn audit_target_mutation_block_reason(config: &Config, target_type: &str, target ) } +async fn audit_target_operation_block_reason(action: &str) -> Option { + if let Err(err) = refresh_persisted_module_switches_from_store().await { + warn!( + error = %err, + "failed to reload persisted module switches before checking audit target operation gating" + ); + } + refresh_audit_module_enabled(); + target_module_disabled_reason("audit", rustfs_config::ENV_AUDIT_ENABLE, is_audit_module_enabled(), action) +} + fn merge_audit_endpoints(config: &Config, runtime_statuses: HashMap) -> Vec { shared_merge_target_endpoints(&audit_target_specs(), AUDIT_ROUTE_PREFIX, config, runtime_statuses) .into_iter() @@ -272,6 +285,9 @@ impl Operation for AuditTargetConfig { let (target_type, target_name) = extract_target_params(¶ms)?; authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?; + if let Some(reason) = audit_target_operation_block_reason("managing audit targets from the console").await { + return Err(s3_error!(InvalidRequest, "{reason}")); + } let config_snapshot = load_server_config_from_store().await?; if let Some(reason) = audit_target_mutation_block_reason(&config_snapshot, target_type, target_name) { return Err(s3_error!(InvalidRequest, "{reason}")); @@ -373,6 +389,9 @@ impl Operation for RemoveAuditTarget { let (target_type, target_name) = extract_target_params(¶ms)?; authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?; + if let Some(reason) = audit_target_operation_block_reason("managing audit targets from the console").await { + return Err(s3_error!(InvalidRequest, "{reason}")); + } let config_snapshot = load_server_config_from_store().await?; if let Some(reason) = audit_target_mutation_block_reason(&config_snapshot, target_type, target_name) { return Err(s3_error!(InvalidRequest, "{reason}")); @@ -534,6 +553,26 @@ mod tests { ); } + #[test] + fn audit_target_operation_block_reason_requires_audit_module_enable() { + with_var(rustfs_config::ENV_AUDIT_ENABLE, Some("false"), || { + let reason = + futures::executor::block_on(audit_target_operation_block_reason("managing audit targets from the console")); + assert!(reason.is_some()); + assert!(reason.unwrap().contains("set RUSTFS_AUDIT_ENABLE=true")); + }); + } + + #[test] + fn audit_target_operation_block_reason_allows_when_audit_module_enabled() { + with_var(rustfs_config::ENV_AUDIT_ENABLE, Some("true"), || { + assert!( + futures::executor::block_on(audit_target_operation_block_reason("managing audit targets from the console")) + .is_none() + ); + }); + } + #[test] fn audit_target_mutation_block_reason_rejects_mixed_target() { with_var("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_PRIMARY", Some("https://example.com/hook"), || { @@ -727,6 +766,10 @@ mod tests { put_block.contains("authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"), "audit target writes should require SetBucketTargetAction" ); + assert!( + put_block.contains("audit_target_operation_block_reason(\"managing audit targets from the console\")"), + "audit target writes should reject requests when the audit module is disabled" + ); assert!( list_block.contains("authorize_audit_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"), "audit target list should require GetBucketTargetAction" @@ -735,6 +778,10 @@ mod tests { delete_block.contains("authorize_audit_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"), "audit target deletion should require SetBucketTargetAction" ); + assert!( + delete_block.contains("audit_target_operation_block_reason(\"managing audit targets from the console\")"), + "audit target deletion should reject requests when the audit module is disabled" + ); } fn extract_block_between_markers<'a>(src: &'a str, start_marker: &str, end_marker: &str) -> &'a str { diff --git a/rustfs/src/admin/handlers/event.rs b/rustfs/src/admin/handlers/event.rs index dda500cf3..c363b2af4 100644 --- a/rustfs/src/admin/handlers/event.rs +++ b/rustfs/src/admin/handlers/event.rs @@ -17,14 +17,17 @@ use crate::admin::{ handlers::target_descriptor::{ AdminTargetSpec, AdminTargetValidator, EndpointKey, TargetDomain, TargetEndpointSource, allowed_target_keys, collect_validated_key_values as shared_collect_validated_key_values, - merge_target_endpoints as shared_merge_target_endpoints, + merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason, target_service_name, target_spec, validate_target_request, }, router::{AdminOperation, Operation, S3Router}, }; use crate::auth::{check_key_valid, get_session_token}; -use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use crate::server::{ + ADMIN_PREFIX, RemoteAddr, is_notify_module_enabled, refresh_notify_module_enabled, + refresh_persisted_module_switches_from_store, +}; use futures::stream::{FuturesUnordered, StreamExt}; use http::{HeaderMap, StatusCode}; use hyper::Method; @@ -167,6 +170,17 @@ fn target_mutation_block_reason(config: &Config, target_type: &str, target_name: ) } +async fn notification_target_operation_block_reason(action: &str) -> Option { + if let Err(err) = refresh_persisted_module_switches_from_store().await { + warn!( + error = %err, + "failed to reload persisted module switches before checking notification target operation gating" + ); + } + refresh_notify_module_enabled(); + target_module_disabled_reason("notify", rustfs_config::ENV_NOTIFY_ENABLE, is_notify_module_enabled(), action) +} + fn merge_notification_endpoints(config: &Config, runtime_statuses: HashMap) -> Vec { shared_merge_target_endpoints(¬ification_target_specs(), NOTIFY_ROUTE_PREFIX, config, runtime_statuses) .into_iter() @@ -197,6 +211,9 @@ impl Operation for NotificationTarget { let (target_type, target_name) = extract_target_params(¶ms)?; authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?; + if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await { + return Err(s3_error!(InvalidRequest, "{reason}")); + } let ns = get_notification_system()?; let config_snapshot = ns.config.read().await.clone(); if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name) { @@ -291,6 +308,13 @@ impl Operation for ListTargetsArns { let span = Span::current(); let _enter = span.enter(); authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?; + if let Some(reason) = notification_target_operation_block_reason( + "querying notification target ARNs for bucket associations from the console", + ) + .await + { + return Err(s3_error!(InvalidRequest, "{reason}")); + } let ns = get_notification_system()?; let targets = ns.get_target_values().await; @@ -336,6 +360,9 @@ impl Operation for RemoveNotificationTarget { let (target_type, target_name) = extract_target_params(¶ms)?; authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?; + if let Some(reason) = notification_target_operation_block_reason("managing notification targets from the console").await { + return Err(s3_error!(InvalidRequest, "{reason}")); + } let ns = get_notification_system()?; let config_snapshot = ns.config.read().await.clone(); if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name) { @@ -543,6 +570,29 @@ mod tests { ); } + #[test] + fn notification_target_operation_block_reason_requires_notify_module_enable() { + with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("false"), || { + let reason = futures::executor::block_on(notification_target_operation_block_reason( + "managing notification targets from the console", + )); + assert!(reason.is_some()); + assert!(reason.unwrap().contains("set RUSTFS_NOTIFY_ENABLE=true")); + }); + } + + #[test] + fn notification_target_operation_block_reason_allows_when_notify_module_enabled() { + with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"), || { + assert!( + futures::executor::block_on(notification_target_operation_block_reason( + "managing notification targets from the console" + )) + .is_none() + ); + }); + } + #[test] fn target_mutation_block_reason_rejects_mixed_target() { with_var("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_PRIMARY", Some("https://example.com/hook"), || { @@ -705,6 +755,11 @@ mod tests { put_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"), "notification target writes should require SetBucketTargetAction" ); + assert!( + put_block.contains("notification_target_operation_block_reason(") + && put_block.contains("\"managing notification targets from the console\""), + "notification target writes should reject requests when the notify module is disabled" + ); assert!( list_block.contains("authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"), "notification target list should require GetBucketTargetAction" @@ -713,10 +768,20 @@ mod tests { arns_block.contains("authorize_notification_admin_request(&req, AdminAction::GetBucketTargetAction).await?;"), "notification target arn listing should require GetBucketTargetAction" ); + assert!( + arns_block.contains("notification_target_operation_block_reason(") + && arns_block.contains("\"querying notification target ARNs for bucket associations from the console\""), + "notification target arn listing should reject requests when the notify module is disabled" + ); assert!( delete_block.contains("authorize_notification_admin_request(&req, AdminAction::SetBucketTargetAction).await?;"), "notification target deletion should require SetBucketTargetAction" ); + assert!( + delete_block.contains("notification_target_operation_block_reason(") + && delete_block.contains("\"managing notification targets from the console\""), + "notification target deletion should reject requests when the notify module is disabled" + ); } #[test] diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 5013a061d..93372f4f1 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -26,6 +26,7 @@ pub mod kms_dynamic; pub mod kms_keys; pub mod kms_management; pub mod metrics; +pub mod module_switch; pub mod oidc; pub mod policies; pub mod pools; @@ -55,6 +56,8 @@ mod tests { // Test that handler structs can be created let _account_handler = account_info::AccountInfoHandler {}; let _list_audit_targets = audit::ListAuditTargets {}; + let _get_module_switches = module_switch::GetModuleSwitchesHandler {}; + let _update_module_switches = module_switch::UpdateModuleSwitchesHandler {}; let _service_handler = system::ServiceHandle {}; let _server_info_handler = system::ServerInfoHandler {}; let _inspect_data_handler = system::InspectDataHandler {}; diff --git a/rustfs/src/admin/handlers/module_switch.rs b/rustfs/src/admin/handlers/module_switch.rs new file mode 100644 index 000000000..03717974d --- /dev/null +++ b/rustfs/src/admin/handlers/module_switch.rs @@ -0,0 +1,228 @@ +// 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, + router::{AdminOperation, Operation, S3Router}, +}; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ + ADMIN_PREFIX, ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches, RemoteAddr, current_module_switch_snapshot, + init_event_notifier, refresh_audit_module_enabled, refresh_notify_module_enabled, + refresh_persisted_module_switches_from_store, save_persisted_module_switches_to_store, shutdown_event_notifier, + start_audit_system, stop_audit_system, validate_module_switch_update, +}; +use http::{HeaderMap, StatusCode}; +use hyper::Method; +use matchit::Params; +use rustfs_audit::AuditError; +use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; +use rustfs_policy::policy::action::{Action, AdminAction}; +use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error}; +use serde::{Deserialize, Serialize}; + +pub fn register_module_switch_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/module-switches").as_str(), + AdminOperation(&GetModuleSwitchesHandler {}), + )?; + + r.insert( + Method::PUT, + format!("{}{}", ADMIN_PREFIX, "/v3/module-switches").as_str(), + AdminOperation(&UpdateModuleSwitchesHandler {}), + )?; + + Ok(()) +} + +#[derive(Debug, Deserialize)] +struct UpdateModuleSwitchesRequest { + notify_enabled: bool, + audit_enabled: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +struct ModuleSwitchesResponse { + notify_enabled: bool, + audit_enabled: bool, + persisted_notify_enabled: bool, + persisted_audit_enabled: bool, + notify_source: ModuleSwitchSource, + audit_source: ModuleSwitchSource, +} + +impl From for ModuleSwitchesResponse { + fn from(value: ModuleSwitchSnapshot) -> Self { + Self { + notify_enabled: value.notify_enabled, + audit_enabled: value.audit_enabled, + persisted_notify_enabled: value.persisted_notify_enabled, + persisted_audit_enabled: value.persisted_audit_enabled, + notify_source: value.notify_source, + audit_source: value.audit_source, + } + } +} + +fn build_response( + status: StatusCode, + body: &T, + request_id: Option<&http::HeaderValue>, +) -> S3Result> { + let data = serde_json::to_vec(body).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?; + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, "application/json".parse().unwrap()); + if let Some(v) = request_id { + header.insert("x-request-id", v.clone()); + } + Ok(S3Response::with_headers((status, Body::from(data)), header)) +} + +async fn authorize_module_switch_request(req: &S3Request, action: AdminAction) -> S3Result<()> { + let Some(input_cred) = &req.credentials else { + return Err(s3_error!(InvalidRequest, "authentication required")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(action)], + req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)), + ) + .await +} + +async fn refresh_module_switch_snapshot() -> S3Result { + // Re-read persisted values before every console read/write so the current + // node reflects the latest cluster-wide state instead of stale atomics. + refresh_persisted_module_switches_from_store() + .await + .map_err(|e| s3_error!(InternalError, "failed to reload persisted module switches: {}", e))?; + refresh_notify_module_enabled(); + refresh_audit_module_enabled(); + Ok(current_module_switch_snapshot()) +} + +pub struct GetModuleSwitchesHandler {} + +#[async_trait::async_trait] +impl Operation for GetModuleSwitchesHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_module_switch_request(&req, AdminAction::ServerInfoAdminAction).await?; + let snapshot = refresh_module_switch_snapshot().await?; + build_response(StatusCode::OK, &ModuleSwitchesResponse::from(snapshot), req.headers.get("x-request-id")) + } +} + +pub struct UpdateModuleSwitchesHandler {} + +#[async_trait::async_trait] +impl Operation for UpdateModuleSwitchesHandler { + async fn call(&self, mut req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_module_switch_request(&req, AdminAction::ConfigUpdateAdminAction).await?; + refresh_persisted_module_switches_from_store() + .await + .map_err(|e| s3_error!(InternalError, "failed to reload persisted module switches: {}", e))?; + + let body = req + .input + .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE) + .await + .map_err(|e| s3_error!(InvalidRequest, "failed to read request body: {}", e))?; + if body.is_empty() { + return Err(s3_error!(InvalidRequest, "request body is required")); + } + + let request: UpdateModuleSwitchesRequest = + serde_json::from_slice(&body).map_err(|e| s3_error!(InvalidRequest, "invalid JSON: {}", e))?; + let switches = PersistedModuleSwitches { + notify_enabled: request.notify_enabled, + audit_enabled: request.audit_enabled, + }; + + // Reject conflicting writes early so operators do not persist a console + // value that still cannot win over an explicit env override. + if let Err(err) = validate_module_switch_update(switches) { + let _ = refresh_module_switch_snapshot().await; + return Err(s3_error!(InvalidRequest, "{err}")); + } + + save_persisted_module_switches_to_store(switches) + .await + .map_err(|e| s3_error!(InternalError, "failed to save module switches: {}", e))?; + + // Apply the new effective values immediately on this node so the console + // response reflects the runtime state after to write completes. + if refresh_notify_module_enabled() { + init_event_notifier().await; + } else { + shutdown_event_notifier().await; + } + + if refresh_audit_module_enabled() { + match start_audit_system().await { + Ok(()) | Err(AuditError::AlreadyInitialized) => {} + Err(e) => return Err(s3_error!(InternalError, "failed to apply audit module switch: {}", e)), + } + } else { + stop_audit_system() + .await + .map_err(|e| s3_error!(InternalError, "failed to stop audit module after switch update: {}", e))?; + } + + let snapshot = current_module_switch_snapshot(); + build_response(StatusCode::OK, &ModuleSwitchesResponse::from(snapshot), req.headers.get("x-request-id")) + } +} + +#[cfg(test)] +mod tests { + #[test] + fn module_switch_handlers_require_admin_authorization_contract() { + let src = include_str!("module_switch.rs"); + let get_block = extract_block_between_markers( + src, + "impl Operation for GetModuleSwitchesHandler", + "pub struct UpdateModuleSwitchesHandler", + ); + let put_block = extract_block_between_markers(src, "impl Operation for UpdateModuleSwitchesHandler", "#[cfg(test)]"); + + assert!( + get_block.contains("authorize_module_switch_request(&req, AdminAction::ServerInfoAdminAction).await?;"), + "module switch GET should require ServerInfoAdminAction" + ); + assert!( + put_block.contains("authorize_module_switch_request(&req, AdminAction::ConfigUpdateAdminAction).await?;"), + "module switch PUT should require ConfigUpdateAdminAction" + ); + } + + fn extract_block_between_markers<'a>(src: &'a str, start_marker: &str, end_marker: &str) -> &'a str { + let start = src + .find(start_marker) + .unwrap_or_else(|| panic!("Expected marker `{start_marker}` in source")); + let after_start = &src[start..]; + let end = after_start + .find(end_marker) + .unwrap_or_else(|| panic!("Expected end marker `{end_marker}` in source")); + &after_start[..end] + } +} diff --git a/rustfs/src/admin/handlers/target_descriptor.rs b/rustfs/src/admin/handlers/target_descriptor.rs index 115cb83d3..686953fad 100644 --- a/rustfs/src/admin/handlers/target_descriptor.rs +++ b/rustfs/src/admin/handlers/target_descriptor.rs @@ -193,6 +193,14 @@ pub(crate) fn target_mutation_block_reason( } } +pub(crate) fn target_module_disabled_reason(module_name: &str, env_key: &str, enabled: bool, action: &str) -> Option { + (!enabled).then(|| { + format!( + "{module_name} module is disabled; enable the {module_name} module first in the console or set {env_key}=true before {action}" + ) + }) +} + pub(crate) fn merge_target_endpoints( specs: &[AdminTargetSpec], route_prefix: &str, diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 588781ecc..69f382cfa 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -25,8 +25,8 @@ mod console_test; mod route_registration_test; use handlers::{ - audit, bucket_meta, heal, health, kms, oidc, pools, profile_admin, quota, rebalance, replication, site_replication, sts, - system, tier, user, + audit, bucket_meta, heal, health, kms, module_switch, oidc, pools, profile_admin, quota, rebalance, replication, + site_replication, sts, system, tier, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -56,6 +56,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)?; audit::register_audit_target_route(&mut r)?; + module_switch::register_module_switch_route(&mut r)?; replication::register_replication_route(&mut r)?; site_replication::register_site_replication_route(&mut r)?; diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 9dbfc1517..9bdc7966c 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -14,8 +14,8 @@ use crate::admin::{ handlers::{ - audit, bucket_meta, heal, health, kms, oidc, pools, profile_admin, quota, rebalance, replication, site_replication, sts, - system, tier, user, + audit, bucket_meta, heal, health, kms, module_switch, oidc, pools, profile_admin, quota, rebalance, replication, + site_replication, sts, system, tier, user, }, router::{AdminOperation, S3Router}, }; @@ -52,6 +52,7 @@ fn register_admin_routes(router: &mut S3Router) { quota::register_quota_route(router).expect("register quota route"); bucket_meta::register_bucket_meta_route(router).expect("register bucket meta route"); audit::register_audit_target_route(router).expect("register audit target route"); + module_switch::register_module_switch_route(router).expect("register module switch route"); replication::register_replication_route(router).expect("register replication route"); site_replication::register_site_replication_route(router).expect("register site replication route"); profile_admin::register_profiling_route(router).expect("register profile route"); @@ -93,6 +94,8 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::GET, &admin_path("/v3/idp/builtin/policy-entities")); assert_route(&router, Method::GET, &admin_path("/v3/target/list")); assert_route(&router, Method::GET, &admin_path("/v3/audit/target/list")); + assert_route(&router, Method::GET, &admin_path("/v3/module-switches")); + assert_route(&router, Method::PUT, &admin_path("/v3/module-switches")); assert_route(&router, Method::PUT, &admin_path("/v3/audit/target/audit_webhook/test-audit")); assert_route(&router, Method::DELETE, &admin_path("/v3/audit/target/audit_webhook/test-audit/reset")); assert_route(&router, Method::GET, &admin_path("/v3/accountinfo")); @@ -185,6 +188,8 @@ fn test_admin_alias_paths_match_existing_admin_routes() { (Method::PUT, compat_admin_alias_path("/v3/set-bucket-quota")), (Method::GET, compat_admin_alias_path("/v3/get-bucket-quota")), (Method::GET, compat_admin_alias_path("/v3/audit/target/list")), + (Method::GET, compat_admin_alias_path("/v3/module-switches")), + (Method::PUT, compat_admin_alias_path("/v3/module-switches")), (Method::PUT, compat_admin_alias_path("/v3/audit/target/audit_webhook/test-audit")), (Method::DELETE, compat_admin_alias_path("/v3/audit/target/audit_webhook/test-audit/reset")), (Method::POST, compat_admin_alias_path("/v3/heal/")), diff --git a/rustfs/src/server/audit.rs b/rustfs/src/server/audit.rs index 66e161186..2ebe93e80 100644 --- a/rustfs/src/server/audit.rs +++ b/rustfs/src/server/audit.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use super::{module_switch::resolve_audit_module_state, refresh_persisted_module_switches_from_store}; use crate::app::context::resolve_server_config; use rustfs_audit::{AuditError, AuditResult, audit_system, init_audit_system, system::AuditSystemState}; use std::sync::atomic::{AtomicBool, Ordering}; @@ -24,7 +25,7 @@ fn server_config_from_context() -> Option { } pub fn refresh_audit_module_enabled() -> bool { - let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_AUDIT_ENABLE, rustfs_config::DEFAULT_AUDIT_ENABLE); + let enabled = resolve_audit_module_state().enabled; AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed); enabled } @@ -56,11 +57,15 @@ fn has_any_audit_targets(config: &rustfs_ecstore::config::Config) -> bool { /// If not configured, it skips the initialization. /// It also handles cases where the audit system is already running or if the global configuration is not loaded. pub async fn start_audit_system() -> AuditResult<()> { + if let Err(err) = refresh_persisted_module_switches_from_store().await { + warn!("Failed to refresh persisted audit module switch from store: {}", err); + } + let enabled = refresh_audit_module_enabled(); if !enabled { info!( target: "rustfs::main::start_audit_system", - "Audit module is disabled by RUSTFS_AUDIT_ENABLE=false, audit system initialization is skipped." + "Audit module is disabled, audit system initialization is skipped. Enable the audit module first." ); return Ok(()); } @@ -106,34 +111,69 @@ pub async fn start_audit_system() -> AuditResult<()> { target: "rustfs::main::start_audit_system", "Audit subsystem configuration detected and started initializing the audit system." ); - // 3. Initialize and start the audit system - let system = init_audit_system(); - // Check if the audit system is already running - let state = system.get_state().await; - if state == AuditSystemState::Running { - warn!( - target: "rustfs::main::start_audit_system", - "The audit system is running, skip repeated initialization." - ); - return Err(AuditError::AlreadyInitialized); - } - // Preparation before starting - match system.start(server_config).await { - Ok(_) => { - info!( - target: "rustfs::main::start_audit_system", - "Audit system started successfully with time: {}.", - jiff::Zoned::now() - ); - Ok(()) + + if let Some(system) = audit_system() { + match system.get_state().await { + AuditSystemState::Running | AuditSystemState::Paused | AuditSystemState::Starting => { + // Match notify behavior: prefer reloading the existing singleton + // instead of constructing a second lifecycle path on re-enable. + match system.reload_config(server_config).await { + Ok(()) => { + info!( + target: "rustfs::main::start_audit_system", + "Audit system reloaded successfully with time: {}.", + jiff::Zoned::now() + ); + Ok(()) + } + Err(e) => { + warn!( + target: "rustfs::main::start_audit_system", + "Audit system reload failed: {:?}", + e + ); + Err(e) + } + } + } + AuditSystemState::Stopped | AuditSystemState::Stopping => match system.start(server_config).await { + Ok(()) => { + info!( + target: "rustfs::main::start_audit_system", + "Audit system started successfully with time: {}.", + jiff::Zoned::now() + ); + Ok(()) + } + Err(e) => { + warn!( + target: "rustfs::main::start_audit_system", + "Audit system startup failed: {:?}", + e + ); + Err(e) + } + }, } - Err(e) => { - warn!( - target: "rustfs::main::start_audit_system", - "Audit system startup failed: {:?}", - e - ); - Err(e) + } else { + let system = init_audit_system(); + match system.start(server_config).await { + Ok(()) => { + info!( + target: "rustfs::main::start_audit_system", + "Audit system started successfully with time: {}.", + jiff::Zoned::now() + ); + Ok(()) + } + Err(e) => { + warn!( + target: "rustfs::main::start_audit_system", + "Audit system startup failed: {:?}", + e + ); + Err(e) + } } } } diff --git a/rustfs/src/server/event.rs b/rustfs/src/server/event.rs index 17fca2f3d..ca0e084d9 100644 --- a/rustfs/src/server/event.rs +++ b/rustfs/src/server/event.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use super::{module_switch::resolve_notify_module_state, refresh_persisted_module_switches_from_store}; use crate::app::context::resolve_server_config; use rustfs_ecstore::event_notification::{EventArgs as EcstoreEventArgs, register_event_dispatch_hook}; use rustfs_notify::EventArgs as NotifyEventArgs; @@ -27,7 +28,7 @@ fn server_config_from_context() -> Option { } pub fn refresh_notify_module_enabled() -> bool { - let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_NOTIFY_ENABLE, rustfs_config::DEFAULT_NOTIFY_ENABLE); + let enabled = resolve_notify_module_state().enabled; NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed); enabled } @@ -98,11 +99,15 @@ pub async fn shutdown_event_notifier() { #[instrument] pub async fn init_event_notifier() { + if let Err(err) = refresh_persisted_module_switches_from_store().await { + warn!("Failed to refresh persisted notify module switch from store: {}", err); + } + let enabled = refresh_notify_module_enabled(); if !enabled { info!( target: "rustfs::main::init_event_notifier", - "Notify module is disabled by RUSTFS_NOTIFY_ENABLE=false, event notifier initialization is skipped." + "Notify module is disabled, event notifier initialization is skipped. Enable the notify module first." ); return; } @@ -126,9 +131,18 @@ pub async fn init_event_notifier() { "Event notifier configuration found, proceeding with initialization." ); - // 2. Initialize the notification system asynchronously with a global configuration - // Use direct await for better error handling and faster initialization - if let Err(e) = rustfs_notify::initialize(server_config).await { + if let Some(system) = rustfs_notify::notification_system() { + // Reuse the existing global system on re-enable so bucket rules, metrics, + // and stream lifecycle stay aligned with the current process singleton. + if let Err(e) = system.reload_config(server_config).await { + error!("Failed to reload event notifier system: {}", e); + } else { + info!( + target: "rustfs::main::init_event_notifier", + "Event notifier system reloaded successfully." + ); + } + } else if let Err(e) = rustfs_notify::initialize(server_config).await { error!("Failed to initialize event notifier system: {}", e); } else { install_ecstore_event_dispatch_hook(); diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 41fc3967e..680bd2cae 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -19,6 +19,7 @@ mod event; mod http; mod hybrid; mod layer; +mod module_switch; mod prefix; mod readiness; mod runtime; @@ -38,6 +39,10 @@ pub use service_state::ShutdownSignal; pub use service_state::wait_for_shutdown; // Items only used within the library crate (admin handlers, server/http.rs, etc.). +pub(crate) use module_switch::{ + ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches, current_module_switch_snapshot, + refresh_persisted_module_switches_from_store, save_persisted_module_switches_to_store, validate_module_switch_update, +}; pub(crate) use prefix::{ ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TONIC_PREFIX, VERSION, diff --git a/rustfs/src/server/module_switch.rs b/rustfs/src/server/module_switch.rs new file mode 100644 index 000000000..d69832c7c --- /dev/null +++ b/rustfs/src/server/module_switch.rs @@ -0,0 +1,307 @@ +// 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_ecstore::{ + config::com::{read_config, save_config}, + error::Error as StorageError, + new_object_layer_fn, +}; +use serde::{Deserialize, Serialize}; +use std::sync::atomic::{AtomicBool, Ordering}; + +const MODULE_SWITCH_CONFIG_PATH: &str = "config/module_switches.json"; + +// Keep a cheap in-process snapshot so hot-path checks do not need to read +// cluster metadata after startup or console-triggered refresh. +static PERSISTED_NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE); +static PERSISTED_AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE); +static PERSISTED_MODULE_SWITCH_CONFIGURED: AtomicBool = AtomicBool::new(false); + +#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq, Serialize)] +pub(crate) struct PersistedModuleSwitches { + pub(crate) notify_enabled: bool, + pub(crate) audit_enabled: bool, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "lowercase")] +pub(crate) enum ModuleSwitchSource { + Env, + Console, + Default, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct ModuleSwitchResolution { + pub(crate) enabled: bool, + pub(crate) source: ModuleSwitchSource, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +pub(crate) struct ModuleSwitchSnapshot { + pub(crate) notify_enabled: bool, + pub(crate) audit_enabled: bool, + pub(crate) persisted_notify_enabled: bool, + pub(crate) persisted_audit_enabled: bool, + pub(crate) notify_source: ModuleSwitchSource, + pub(crate) audit_source: ModuleSwitchSource, +} + +pub(crate) fn current_persisted_module_switches() -> PersistedModuleSwitches { + PersistedModuleSwitches { + notify_enabled: PERSISTED_NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed), + audit_enabled: PERSISTED_AUDIT_MODULE_ENABLED.load(Ordering::Relaxed), + } +} + +fn persisted_module_switches_configured() -> bool { + PERSISTED_MODULE_SWITCH_CONFIGURED.load(Ordering::Relaxed) +} + +pub(crate) fn set_persisted_module_switches(config: PersistedModuleSwitches, configured: bool) { + PERSISTED_NOTIFY_MODULE_ENABLED.store(config.notify_enabled, Ordering::Relaxed); + PERSISTED_AUDIT_MODULE_ENABLED.store(config.audit_enabled, Ordering::Relaxed); + PERSISTED_MODULE_SWITCH_CONFIGURED.store(configured, Ordering::Relaxed); +} + +fn env_override_exists(key: &str) -> bool { + std::env::var_os(key).is_some() +} + +fn env_override_value(key: &str) -> Option { + rustfs_utils::get_env_opt_bool(key) +} + +fn effective_module_switch_state(env_key: &str, persisted_enabled: bool, default_enabled: bool) -> ModuleSwitchResolution { + // Explicit env remains the highest-priority source so process-level bootstrap + // cannot be silently overridden by a later console write. + if let Some(env_enabled) = env_override_value(env_key) { + return ModuleSwitchResolution { + enabled: env_enabled, + source: ModuleSwitchSource::Env, + }; + } + + if persisted_module_switches_configured() { + return ModuleSwitchResolution { + enabled: persisted_enabled, + source: ModuleSwitchSource::Console, + }; + } + + ModuleSwitchResolution { + enabled: default_enabled, + source: ModuleSwitchSource::Default, + } +} + +pub(crate) fn resolve_notify_module_state() -> ModuleSwitchResolution { + effective_module_switch_state( + rustfs_config::ENV_NOTIFY_ENABLE, + PERSISTED_NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed), + rustfs_config::DEFAULT_NOTIFY_ENABLE, + ) +} + +pub(crate) fn resolve_audit_module_state() -> ModuleSwitchResolution { + effective_module_switch_state( + rustfs_config::ENV_AUDIT_ENABLE, + PERSISTED_AUDIT_MODULE_ENABLED.load(Ordering::Relaxed), + rustfs_config::DEFAULT_AUDIT_ENABLE, + ) +} + +pub(crate) fn current_module_switch_snapshot() -> ModuleSwitchSnapshot { + let persisted = current_persisted_module_switches(); + let notify = resolve_notify_module_state(); + let audit = resolve_audit_module_state(); + + ModuleSwitchSnapshot { + notify_enabled: notify.enabled, + audit_enabled: audit.enabled, + persisted_notify_enabled: persisted.notify_enabled, + persisted_audit_enabled: persisted.audit_enabled, + notify_source: notify.source, + audit_source: audit.source, + } +} + +fn validate_env_override_for_request(env_key: &str, requested: bool, label: &str) -> Result<(), String> { + if !env_override_exists(env_key) { + return Ok(()); + } + + match env_override_value(env_key) { + // Matching values are safe: we still persist the console value, but the + // effective runtime source remains env until the operator changes it. + Some(value) if value == requested => Ok(()), + Some(value) => Err(format!( + "{label} is managed by environment variable {env_key}={value}; update the environment value first, then use the console to refresh the module switch state" + )), + None => Err(format!( + "{label} is managed by environment variable {env_key}, but its value is not a valid boolean; fix the environment value first, then use the console to refresh the module switch state" + )), + } +} + +pub(crate) fn validate_module_switch_update(requested: PersistedModuleSwitches) -> Result<(), String> { + validate_env_override_for_request(rustfs_config::ENV_NOTIFY_ENABLE, requested.notify_enabled, "notify module")?; + validate_env_override_for_request(rustfs_config::ENV_AUDIT_ENABLE, requested.audit_enabled, "audit module")?; + Ok(()) +} + +pub(crate) async fn refresh_persisted_module_switches_from_store() -> Result { + let Some(store) = new_object_layer_fn() else { + return Err("storage layer not initialized".to_string()); + }; + + let (config, configured) = match read_config(store, MODULE_SWITCH_CONFIG_PATH).await { + Ok(data) => ( + serde_json::from_slice::(&data) + .map_err(|e| format!("failed to deserialize module switch config: {e}"))?, + true, + ), + Err(StorageError::ConfigNotFound) => (PersistedModuleSwitches::default(), false), + Err(err) => return Err(format!("failed to load module switch config: {err}")), + }; + + // Track whether the persisted file exists so the effective state can + // distinguish "console configured false" from "never configured, use default". + set_persisted_module_switches(config, configured); + Ok(config) +} + +pub(crate) async fn save_persisted_module_switches_to_store(config: PersistedModuleSwitches) -> Result<(), String> { + let Some(store) = new_object_layer_fn() else { + return Err("storage layer not initialized".to_string()); + }; + + let data = serde_json::to_vec(&config).map_err(|e| format!("failed to serialize module switch config: {e}"))?; + save_config(store, MODULE_SWITCH_CONFIG_PATH, data) + .await + .map_err(|e| format!("failed to save module switch config: {e}"))?; + + set_persisted_module_switches(config, true); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use serial_test::serial; + use temp_env::{with_var, with_vars}; + + #[test] + #[serial] + fn resolve_module_switch_state_prefers_env_override() { + set_persisted_module_switches( + PersistedModuleSwitches { + notify_enabled: false, + audit_enabled: false, + }, + true, + ); + + with_vars( + [ + (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")), + (rustfs_config::ENV_AUDIT_ENABLE, Some("false")), + ], + || { + let notify = resolve_notify_module_state(); + let audit = resolve_audit_module_state(); + + assert!(notify.enabled); + assert_eq!(notify.source, ModuleSwitchSource::Env); + assert!(!audit.enabled); + assert_eq!(audit.source, ModuleSwitchSource::Env); + }, + ); + } + + #[test] + #[serial] + fn resolve_module_switch_state_falls_back_to_console_value() { + set_persisted_module_switches( + PersistedModuleSwitches { + notify_enabled: true, + audit_enabled: false, + }, + true, + ); + + with_vars( + [ + (rustfs_config::ENV_NOTIFY_ENABLE, None::<&str>), + (rustfs_config::ENV_AUDIT_ENABLE, None::<&str>), + ], + || { + let notify = resolve_notify_module_state(); + let audit = resolve_audit_module_state(); + + assert!(notify.enabled); + assert_eq!(notify.source, ModuleSwitchSource::Console); + assert!(!audit.enabled); + assert_eq!(audit.source, ModuleSwitchSource::Console); + }, + ); + } + + #[test] + #[serial] + fn validate_module_switch_update_rejects_env_conflict() { + with_var(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"), || { + let err = validate_module_switch_update(PersistedModuleSwitches { + notify_enabled: false, + audit_enabled: false, + }) + .unwrap_err(); + + assert!(err.contains(rustfs_config::ENV_NOTIFY_ENABLE)); + assert!(err.contains("update the environment value first")); + }); + } + + #[test] + #[serial] + fn validate_module_switch_update_allows_matching_env_override() { + with_vars( + [ + (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")), + (rustfs_config::ENV_AUDIT_ENABLE, Some("false")), + ], + || { + validate_module_switch_update(PersistedModuleSwitches { + notify_enabled: true, + audit_enabled: false, + }) + .expect("matching env override should be accepted"); + }, + ); + } + + #[test] + #[serial] + fn validate_module_switch_update_rejects_invalid_env_override() { + with_var(rustfs_config::ENV_AUDIT_ENABLE, Some("invalid"), || { + let err = validate_module_switch_update(PersistedModuleSwitches { + notify_enabled: false, + audit_enabled: true, + }) + .unwrap_err(); + + assert!(err.contains("not a valid boolean")); + }); + } +}