fran/rustfs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(rustfs): gate audit/notify by global env switches (#2669)

Browse Source
This commit is contained in:
houseme
2026-04-24 09:14:19 +00:00
committed by GitHub
parent fefb308b35
commit 92f812fc80
6 changed files with 311 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/main.rs
+1
View File
@@ -457,6 +457,7 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
// Initialize event notifier
init_event_notifier().await;
// Start the audit system
match start_audit_system().await {
Ok(_) => info!(target: "rustfs::main::run","Audit system started successfully."),
rustfs/src/server/audit.rs
+22
View File
@@ -14,12 +14,25 @@
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};
use tracing::{info, warn};
static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
fn server_config_from_context() -> Option<rustfs_ecstore::config::Config> {
resolve_server_config()
}
pub fn refresh_audit_module_enabled() -> bool {
let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_AUDIT_ENABLE, rustfs_config::DEFAULT_AUDIT_ENABLE);
AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
enabled
}
pub fn is_audit_module_enabled() -> bool {
AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
}
fn has_any_audit_targets(config: &rustfs_ecstore::config::Config) -> bool {
for subsystem in [
rustfs_config::audit::AUDIT_MQTT_SUB_SYS,
@@ -43,6 +56,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<()> {
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."
);
return Ok(());
}
info!(
target: "rustfs::main::start_audit_system",
"Initializing the audit system..."
rustfs/src/server/event.rs
+22
View File
@@ -16,13 +16,26 @@ 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;
use rustfs_s3_common::EventName;
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::spawn;
use tracing::{error, info, instrument, warn};
static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
fn server_config_from_context() -> Option<rustfs_ecstore::config::Config> {
resolve_server_config()
}
pub fn refresh_notify_module_enabled() -> bool {
let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_NOTIFY_ENABLE, rustfs_config::DEFAULT_NOTIFY_ENABLE);
NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
enabled
}
pub fn is_notify_module_enabled() -> bool {
NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed)
}
fn convert_ecstore_event_args(args: EcstoreEventArgs) -> NotifyEventArgs {
let version_id = args.object.version_id.map(|v| v.to_string()).unwrap_or_default();
let (host, port) = match args.host.rsplit_once(':') {
@@ -85,6 +98,15 @@ pub async fn shutdown_event_notifier() {
#[instrument]
pub async fn init_event_notifier() {
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."
);
return;
}
info!(
target: "rustfs::main::init_event_notifier",
"Initializing event notifier..."
rustfs/src/server/mod.rs
+2 -2
View File
@@ -26,8 +26,8 @@ mod service_state;
pub mod tls_material;
// Items used by main.rs (binary crate) and/or embedded.rs — must be fully pub.
pub use audit::{start_audit_system, stop_audit_system};
pub use event::{init_event_notifier, shutdown_event_notifier};
pub use audit::{is_audit_module_enabled, refresh_audit_module_enabled, start_audit_system, stop_audit_system};
pub use event::{init_event_notifier, is_notify_module_enabled, refresh_notify_module_enabled, shutdown_event_notifier};
pub use http::start_http_server;
pub use prefix::LOGO;
pub use runtime::build_tokio_runtime;
rustfs/src/storage/helper.rs
+249 -125
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::server::{is_audit_module_enabled, is_notify_module_enabled};
use crate::storage::access::{ReqInfo, request_context_from_req};
use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
use hashbrown::HashMap;
@@ -70,7 +71,14 @@ where
}
/// A unified helper structure for building and distributing audit logs and event notifications via RAII mode at the end of an S3 operation scope.
pub struct OperationHelper {
pub enum OperationHelper {
Disabled,
Enabled(Box<EnabledOperationHelper>),
}
pub struct EnabledOperationHelper {
audit_enabled: bool,
notify_enabled: bool,
audit_builder: Option<AuditEntryBuilder>,
api_builder: ApiDetailsBuilder,
event_builder: Option<EventArgsBuilder>,
@@ -81,7 +89,17 @@ pub struct OperationHelper {
impl OperationHelper {
/// Create a new OperationHelper for S3 requests.
pub fn new(req: &S3Request<impl Send + Sync>, event: EventName, op: S3Operation) -> Self {
counter!("rustfs.log.chain.audit.total").increment(1);
let audit_enabled = is_audit_module_enabled();
let notify_enabled = is_notify_module_enabled();
// Fast path: when both chains are disabled, avoid all request parsing/builder work.
if !audit_enabled && !notify_enabled {
return Self::Disabled;
}
if audit_enabled {
counter!("rustfs.log.chain.audit.total").increment(1);
}
// Parse path -> bucket/object
let path = req.uri.path().trim_start_matches('/');
let mut segs = path.splitn(2, '/');
@@ -131,13 +149,19 @@ impl OperationHelper {
.map(|ctx| ctx.request_id.clone())
.unwrap_or_else(|| extract_request_id_from_headers(&req.headers));
let audit_builder = AuditEntryBuilder::new("1.0", event, trigger, ApiDetails::default())
.remote_host(remote_host)
.user_agent(get_request_user_agent(&req.headers))
.req_host(get_request_host(&req.headers))
.req_path(req.uri.path().to_string())
.req_query(extract_req_params(req))
.request_id(&request_id);
let audit_builder = if audit_enabled {
Some(
AuditEntryBuilder::new("1.0", event, trigger, ApiDetails::default())
.remote_host(remote_host)
.user_agent(get_request_user_agent(&req.headers))
.req_host(get_request_host(&req.headers))
.req_path(req.uri.path().to_string())
.req_query(extract_req_params(req))
.request_id(&request_id),
)
} else {
None
};
let event_object = ObjectInfo {
bucket: bucket.clone(),
@@ -162,54 +186,67 @@ impl OperationHelper {
// initialize event builder
// object is a placeholder that must be set later using the `object()` method.
let mut event_builder = EventArgsBuilder::new(event, bucket, event_object)
.host(get_request_host(&req.headers))
.port(get_request_port(&req.headers))
.user_agent(get_request_user_agent(&req.headers))
.req_params(req_params);
if let Some(version_id) = req_info
.and_then(|info| info.version_id.clone())
.filter(|value| !value.is_empty())
{
event_builder = event_builder.version_id(version_id);
}
let event_builder = if notify_enabled {
let mut event_builder = EventArgsBuilder::new(event, bucket, event_object)
.host(get_request_host(&req.headers))
.port(get_request_port(&req.headers))
.user_agent(get_request_user_agent(&req.headers))
.req_params(req_params);
if let Some(version_id) = req_info
.and_then(|info| info.version_id.clone())
.filter(|value| !value.is_empty())
{
event_builder = event_builder.version_id(version_id);
}
Some(event_builder)
} else {
None
};
Self {
audit_builder: Some(audit_builder),
Self::Enabled(Box::new(EnabledOperationHelper {
audit_enabled,
notify_enabled,
audit_builder,
api_builder,
event_builder: Some(event_builder),
event_builder,
start_time: request_context
.as_ref()
.map(|ctx| ctx.start_time)
.unwrap_or_else(std::time::Instant::now),
request_context,
}
}))
}
/// Sets the ObjectInfo for event notification.
pub fn object(mut self, object_info: ObjectInfo) -> Self {
if let Some(builder) = self.event_builder.take() {
self.event_builder = Some(builder.object(object_info));
if let Self::Enabled(state) = &mut self
&& let Some(builder) = state.event_builder.take()
{
state.event_builder = Some(builder.object(object_info));
}
self
}
/// Set the version ID for event notifications.
pub fn version_id(mut self, version_id: impl Into<String>) -> Self {
if let Some(builder) = self.event_builder.take() {
self.event_builder = Some(builder.version_id(version_id));
if let Self::Enabled(state) = &mut self
&& let Some(builder) = state.event_builder.take()
{
state.event_builder = Some(builder.version_id(version_id));
}
self
}
/// Set the event name for event notifications.
pub fn event_name(mut self, event_name: EventName) -> Self {
if let Some(builder) = self.event_builder.take() {
self.event_builder = Some(builder.event_name(event_name));
}
if let Self::Enabled(state) = &mut self {
if let Some(builder) = state.event_builder.take() {
state.event_builder = Some(builder.event_name(event_name));
}
if let Some(builder) = self.audit_builder.take() {
self.audit_builder = Some(builder.event(event_name));
if let Some(builder) = state.audit_builder.take() {
state.audit_builder = Some(builder.event(event_name));
}
}
self
@@ -219,19 +256,26 @@ impl OperationHelper {
/// This method should be called immediately before the function returns.
/// It consumes and prepares auxiliary structures for use during `drop`.
pub fn complete<T>(mut self, result: &S3Result<S3Response<T>>) -> Self {
// Complete audit log
if let Some(builder) = self.audit_builder.take() {
let (status, status_code, error_msg) = match result {
Ok(res) => ("success".to_string(), res.status.unwrap_or(StatusCode::OK).as_u16() as i32, None),
Err(e) => (
"failure".to_string(),
e.status_code().unwrap_or(StatusCode::BAD_REQUEST).as_u16() as i32,
e.message().map(|s| s.to_string()),
),
};
let Self::Enabled(state) = &mut self else {
return self;
};
let ttr = self.start_time.elapsed();
let api_details = self
let (status, status_code, error_msg) = match result {
Ok(res) => ("success".to_string(), res.status.unwrap_or(StatusCode::OK).as_u16() as i32, None),
Err(e) => (
"failure".to_string(),
e.status_code().unwrap_or(StatusCode::BAD_REQUEST).as_u16() as i32,
e.message().map(|s| s.to_string()),
),
};
state.api_builder = state.api_builder.clone().status(status.clone()).status_code(status_code);
// Complete audit log
if state.audit_enabled
&& let Some(builder) = state.audit_builder.take()
{
let ttr = state.start_time.elapsed();
let api_details = state
.api_builder
.clone()
.status(status)
@@ -253,7 +297,7 @@ impl OperationHelper {
}
// Inject OpenTelemetry trace context into audit tags for distributed tracing correlation
if let Some(ref ctx) = self.request_context
if let Some(ref ctx) = state.request_context
&& (ctx.trace_id.is_some() || ctx.span_id.is_some())
{
let mut tags = HashMap::new();
@@ -266,13 +310,15 @@ impl OperationHelper {
final_builder = final_builder.tags(tags);
}
self.audit_builder = Some(final_builder);
self.api_builder = ApiDetailsBuilder(api_details); // Store final details for Drop use
state.audit_builder = Some(final_builder);
state.api_builder = ApiDetailsBuilder(api_details); // Store final details for Drop use
}
// Completion event notification (only on success)
if let (Some(builder), Ok(res)) = (self.event_builder.take(), result) {
self.event_builder = Some(builder.resp_elements(extract_resp_elements(res)));
if state.notify_enabled
&& let (Some(builder), Ok(res)) = (state.event_builder.take(), result)
{
state.event_builder = Some(builder.resp_elements(extract_resp_elements(res)));
}
self
@@ -280,29 +326,38 @@ impl OperationHelper {
/// Suppresses the automatic event notification on drop.
pub fn suppress_event(mut self) -> Self {
self.event_builder = None;
if let Self::Enabled(state) = &mut self {
state.event_builder = None;
}
self
}
}
impl Drop for OperationHelper {
fn drop(&mut self) {
let Self::Enabled(state) = self else {
return;
};
// Distribute audit logs
if let Some(builder) = self.audit_builder.take() {
let ctx = self.request_context.clone();
if state.audit_enabled
&& let Some(builder) = state.audit_builder.take()
{
let ctx = state.request_context.clone();
spawn_background_with_context(ctx, async move {
AuditLogger::log(builder.build()).await;
});
}
// Distribute event notification (only on success)
if self.api_builder.0.status.as_deref() == Some("success")
&& let Some(builder) = self.event_builder.take()
if state.notify_enabled
&& state.api_builder.0.status.as_deref() == Some("success")
&& let Some(builder) = state.event_builder.take()
{
let event_args = builder.build();
// Avoid generating notifications for copy requests
if !event_args.is_replication_request() {
let ctx = self.request_context.clone();
let ctx = state.request_context.clone();
spawn_background_with_context(ctx, async move {
notifier_global::notify(event_args).await;
});
@@ -314,9 +369,11 @@ impl Drop for OperationHelper {
#[cfg(test)]
mod tests {
use super::*;
use crate::server::{refresh_audit_module_enabled, refresh_notify_module_enabled};
use http::{Extensions, HeaderMap, HeaderValue, Method, Uri};
use rustfs_credentials::Credentials;
use s3s::dto::DeleteObjectTaggingInput;
use temp_env::with_vars;
fn build_request<T>(input: T, method: Method, uri: Uri) -> S3Request<T> {
S3Request {
@@ -334,91 +391,158 @@ mod tests {
#[test]
fn operation_helper_uses_req_info_for_notification_context() {
let input = DeleteObjectTaggingInput::builder()
.bucket("input-bucket".to_string())
.key("input-object".to_string())
.build()
.unwrap();
let mut req = build_request(input, Method::DELETE, Uri::from_static("/from-uri/ignored"));
req.headers.insert("host", HeaderValue::from_static("example.com"));
req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test"));
req.extensions.insert(ReqInfo {
cred: Some(Credentials {
access_key: "notifyTag".to_string(),
..Default::default()
}),
bucket: Some("issue-2292-bucket".to_string()),
object: Some("prefix/issue-2292.txt".to_string()),
version_id: Some("version-123".to_string()),
..Default::default()
});
with_vars(
[
(rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
(rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
],
|| {
refresh_notify_module_enabled();
refresh_audit_module_enabled();
let input = DeleteObjectTaggingInput::builder()
.bucket("input-bucket".to_string())
.key("input-object".to_string())
.build()
.unwrap();
let mut req = build_request(input, Method::DELETE, Uri::from_static("/from-uri/ignored"));
req.headers.insert("host", HeaderValue::from_static("example.com"));
req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test"));
req.extensions.insert(ReqInfo {
cred: Some(Credentials {
access_key: "notifyTag".to_string(),
..Default::default()
}),
bucket: Some("issue-2292-bucket".to_string()),
object: Some("prefix/issue-2292.txt".to_string()),
version_id: Some("version-123".to_string()),
..Default::default()
});
let helper = OperationHelper::new(&req, EventName::ObjectTaggingPut, S3Operation::PutObjectTagging);
let event_args = helper.event_builder.clone().expect("event builder should exist").build();
let helper = OperationHelper::new(&req, EventName::ObjectTaggingPut, S3Operation::PutObjectTagging);
let event_args = match &helper {
OperationHelper::Enabled(state) => state.event_builder.clone().expect("event builder should exist").build(),
OperationHelper::Disabled => panic!("helper should be enabled when notify/audit switches are on"),
};
assert_eq!(event_args.bucket_name, "issue-2292-bucket");
assert_eq!(event_args.object.bucket, "issue-2292-bucket");
assert_eq!(event_args.object.name, "prefix/issue-2292.txt");
assert_eq!(event_args.version_id, "version-123");
assert_eq!(event_args.req_params.get("principalId").map(String::as_str), Some("notifyTag"));
assert_eq!(event_args.bucket_name, "issue-2292-bucket");
assert_eq!(event_args.object.bucket, "issue-2292-bucket");
assert_eq!(event_args.object.name, "prefix/issue-2292.txt");
assert_eq!(event_args.version_id, "version-123");
assert_eq!(event_args.req_params.get("principalId").map(String::as_str), Some("notifyTag"));
},
);
}
#[test]
fn operation_helper_prioritizes_request_context_for_request_id() {
let input = DeleteObjectTaggingInput::builder()
.bucket("test-bucket".to_string())
.key("test-key".to_string())
.build()
.unwrap();
let mut req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key"));
req.headers.insert("host", HeaderValue::from_static("example.com"));
req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test"));
with_vars(
[
(rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
(rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
],
|| {
refresh_notify_module_enabled();
refresh_audit_module_enabled();
// Insert RequestContext (set by ingress layer) with a specific request_id
req.extensions.insert(RequestContext {
request_id: "ingress-canonical-uuid".to_string(),
x_amz_request_id: "ingress-canonical-uuid".to_string(),
trace_id: None,
span_id: None,
start_time: std::time::Instant::now(),
});
let input = DeleteObjectTaggingInput::builder()
.bucket("test-bucket".to_string())
.key("test-key".to_string())
.build()
.unwrap();
let mut req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key"));
req.headers.insert("host", HeaderValue::from_static("example.com"));
req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test"));
req.extensions.insert(ReqInfo {
bucket: Some("test-bucket".to_string()),
object: Some("test-key".to_string()),
..Default::default()
});
// Insert RequestContext (set by ingress layer) with a specific request_id
req.extensions.insert(RequestContext {
request_id: "ingress-canonical-uuid".to_string(),
x_amz_request_id: "ingress-canonical-uuid".to_string(),
trace_id: None,
span_id: None,
start_time: std::time::Instant::now(),
});
let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject);
req.extensions.insert(ReqInfo {
bucket: Some("test-bucket".to_string()),
object: Some("test-key".to_string()),
..Default::default()
});
// Verify the helper stored the RequestContext
assert!(helper.request_context.is_some());
assert_eq!(helper.request_context.as_ref().unwrap().request_id, "ingress-canonical-uuid");
let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject);
// Verify the helper stored the RequestContext
match &helper {
OperationHelper::Enabled(state) => {
assert!(state.request_context.is_some());
assert_eq!(state.request_context.as_ref().unwrap().request_id, "ingress-canonical-uuid");
}
OperationHelper::Disabled => panic!("helper should be enabled when notify/audit switches are on"),
}
},
);
}
#[test]
fn operation_helper_no_request_context_when_absent() {
let input = DeleteObjectTaggingInput::builder()
.bucket("test-bucket".to_string())
.key("test-key".to_string())
.build()
.unwrap();
let mut req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key"));
req.headers.insert("host", HeaderValue::from_static("example.com"));
req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test"));
req.headers
.insert("x-amz-request-id", HeaderValue::from_static("amz-header-uuid"));
with_vars(
[
(rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
(rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
],
|| {
refresh_notify_module_enabled();
refresh_audit_module_enabled();
// No RequestContext inserted
req.extensions.insert(ReqInfo {
bucket: Some("test-bucket".to_string()),
object: Some("test-key".to_string()),
..Default::default()
});
let input = DeleteObjectTaggingInput::builder()
.bucket("test-bucket".to_string())
.key("test-key".to_string())
.build()
.unwrap();
let mut req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key"));
req.headers.insert("host", HeaderValue::from_static("example.com"));
req.headers.insert("user-agent", HeaderValue::from_static("rustfs-test"));
req.headers
.insert("x-amz-request-id", HeaderValue::from_static("amz-header-uuid"));
let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject);
// No RequestContext inserted
req.extensions.insert(ReqInfo {
bucket: Some("test-bucket".to_string()),
object: Some("test-key".to_string()),
..Default::default()
});
// Verify the helper has no RequestContext
assert!(helper.request_context.is_none());
let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject);
// Verify the helper has no RequestContext
match &helper {
OperationHelper::Enabled(state) => assert!(state.request_context.is_none()),
OperationHelper::Disabled => panic!("helper should be enabled when notify/audit switches are on"),
}
},
);
}
#[test]
fn operation_helper_returns_disabled_when_both_switches_off() {
with_vars(
[
(rustfs_config::ENV_NOTIFY_ENABLE, Some("false")),
(rustfs_config::ENV_AUDIT_ENABLE, Some("false")),
],
|| {
refresh_notify_module_enabled();
refresh_audit_module_enabled();
let input = DeleteObjectTaggingInput::builder()
.bucket("test-bucket".to_string())
.key("test-key".to_string())
.build()
.unwrap();
let req = build_request(input, Method::DELETE, Uri::from_static("/test-bucket/test-key"));
let helper = OperationHelper::new(&req, EventName::ObjectAccessedGet, S3Operation::GetObject);
assert!(matches!(helper, OperationHelper::Disabled));
},
);
}
}