fix(audit): emit entries for multipart parts and bucket config writes (#9067)

* fix(audit): replay queued payloads for internal event names

Audit entries are persisted by the store targets and read back through
serde. Three internal events (CreateMultipartUpload, AbortMultipartUpload,
DeleteObjects) serialized to a string that EventName::deserialize rejected,
so every queued entry for them was dropped on replay. Read the persisted
event name through a lenient deserialize_with function; EventName's own
Deserialize and parse stay strict because they also back notification
configuration.

* fix(audit): emit entries for multipart parts and bucket config writes

UploadPart, UploadPartCopy, AbortMultipartUpload and the mutating bucket
configuration operations never produced an audit entry: the audit path
required a bucket notification event and these operations have none, or
(AbortMultipartUpload) never built a helper. Add an audit-only
OperationHelper constructor that builds no notification event and wrap
the S3 handlers for these operations with it. Operations without a
notification event use one coarse audit-only event name, with the
operation in api.name.

---------

Co-authored-by: Hauser Zhi <housemecn@gmail.com>

crates/notify/src/rules/config_test.rs

@@ -259,6 +259,34 @@ mod integration_tests {
         assert!(!targets.is_empty(), "All files should match");
     }
 
+    /// Internal events are never valid in a user-supplied notification configuration,
+    /// even though the audit queue persists and replays them.
+    #[test]
+    fn test_internal_event_names_are_rejected_in_notification_xml() {
+        for event in [
+            "s3:ObjectCreated:CreateMultipartUpload",
+            "s3:ObjectRemoved:AbortMultipartUpload",
+            "s3:ObjectRemoved:DeleteObjects",
+            "audit:S3:Operation",
+        ] {
+            let xml = format!(
+                r#"<?xml version="1.0" encoding="UTF-8"?>
+<NotificationConfiguration>
+    <QueueConfiguration>
+        <Id>internal</Id>
+        <Queue>arn:rustfs:sqs:ap-northeast-1:primary:webhook</Queue>
+        <Event>{event}</Event>
+    </QueueConfiguration>
+</NotificationConfiguration>"#
+            );
+            let arn_list = vec!["arn:rustfs:sqs:ap-northeast-1:primary:webhook".to_string()];
+            assert!(
+                BucketNotificationConfig::from_xml(Cursor::new(xml.as_bytes()), "ap-northeast-1", &arn_list).is_err(),
+                "{event} must be rejected in notification XML"
+            );
+        }
+    }
+
     /// Test specific event type instead of compound
     #[test]
     fn test_specific_event_type_xml() {

crates/s3-types/src/event_name.rs

@@ -116,6 +116,12 @@ pub enum EventName {
     // Splitting these per-operation would need `mask()` widened first.
     IamIdentityCredentialChanged,
     IamIdentityAuthChallenge,
+
+    // Audit-only S3 API operations that have no bucket notification event, such as
+    // `UploadPart` and the bucket configuration writes. One coarse variant covers
+    // all of them for the same mask-budget reason as the IAM block above: the
+    // specific operation lives in `AuditEntry::api.name`. It must stay last.
+    AuditS3Operation,
 }
 
 // Single event type sequential array for Everything.expand()
@@ -170,7 +176,7 @@ const LAST_SINGLE_TYPE_VALUE: u32 = EventName::IntelligentTiering as u32;
 /// meaningful: `mask()` turns a leaf variant's discriminant `v` into the bit
 /// `1 << (v - 1)`, so the highest discriminant is also the highest bit index in
 /// use. Keep this pointing at whatever variant is declared last.
-const LAST_EVENT_NAME_VALUE: u32 = EventName::IamIdentityAuthChallenge as u32;
+const LAST_EVENT_NAME_VALUE: u32 = EventName::AuditS3Operation as u32;
 
 /// `mask()` returns a `u64`, so discriminants may run from 1 to 64 inclusive.
 ///
@@ -264,6 +270,23 @@ impl EventName {
         }
     }
 
+    /// Parses the string form of the internal events that `parse` rejects.
+    ///
+    /// These events are never valid in a bucket notification configuration, so
+    /// `parse` and the default `Deserialize` stay strict. They do reach the audit
+    /// sink and are persisted by the store targets, which read them back through
+    /// [`deserialize_persisted_event_name`].
+    fn parse_internal(s: &str) -> Option<Self> {
+        match s {
+            "s3:ObjectRemoved:AbortMultipartUpload" => Some(EventName::ObjectRemovedAbortMultipartUpload),
+            "s3:ObjectCreated:CreateMultipartUpload" => Some(EventName::ObjectCreatedCreateMultipartUpload),
+            "s3:ObjectRemoved:DeleteObjects" => Some(EventName::ObjectRemovedDeleteObjects),
+            // Audit-only S3 operations: reach the audit sink, never a notification rule.
+            "audit:S3:Operation" => Some(EventName::AuditS3Operation),
+            _ => None,
+        }
+    }
+
     /// Parses an event string into an EventName with explicit error handling.
     #[inline]
     pub fn try_from_event_str(s: &str) -> Result<Self, ParseEventNameError> {
@@ -336,6 +359,7 @@ impl EventName {
             EventName::KmsServiceStopped => "kms:Service:Stopped",
             EventName::IamIdentityCredentialChanged => "iam:Identity:CredentialChanged",
             EventName::IamIdentityAuthChallenge => "iam:Identity:AuthChallenge",
+            EventName::AuditS3Operation => "audit:S3:Operation",
         }
     }
 
@@ -536,6 +560,20 @@ impl serde::ser::Serialize for EventName {
     }
 }
 
+/// `deserialize_with` function for fields that hold an `EventName` the server itself
+/// persisted, such as queued audit payload metadata. Unlike the default `Deserialize`,
+/// which also backs user-supplied notification XML, it accepts the internal events.
+pub fn deserialize_persisted_event_name<'de, D>(deserializer: D) -> Result<EventName, D::Error>
+where
+    D: serde::de::Deserializer<'de>,
+{
+    let s = <String as serde::Deserialize>::deserialize(deserializer)?;
+    EventName::parse(&s)
+        .ok()
+        .or_else(|| EventName::parse_internal(&s))
+        .ok_or_else(|| serde::de::Error::custom(ParseEventNameError(s)))
+}
+
 impl<'de> serde::de::Deserialize<'de> for EventName {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where
@@ -728,7 +766,8 @@ mod tests {
             | EventName::KmsServiceStarted
             | EventName::KmsServiceStopped
             | EventName::IamIdentityCredentialChanged
-            | EventName::IamIdentityAuthChallenge => {}
+            | EventName::IamIdentityAuthChallenge
+            | EventName::AuditS3Operation => {}
         }
     }
 
@@ -798,6 +837,7 @@ mod tests {
         EventName::KmsServiceStopped,
         EventName::IamIdentityCredentialChanged,
         EventName::IamIdentityAuthChallenge,
+        EventName::AuditS3Operation,
     ];
 
     /// Every KMS management-plane event.
@@ -886,7 +926,7 @@ mod tests {
     /// only as one element of an array someone may forget to extend.
     #[test]
     fn test_last_variant_still_gets_its_own_bit() {
-        let last = EventName::IamIdentityAuthChallenge;
+        let last = EventName::AuditS3Operation;
         assert_ne!(last.mask(), 0, "the last variant's mask overflowed to zero");
         assert_eq!(
             last.mask(),
@@ -950,6 +990,7 @@ mod tests {
             EventName::ObjectRemovedAbortMultipartUpload,
             EventName::ObjectCreatedCreateMultipartUpload,
             EventName::ObjectRemovedDeleteObjects,
+            EventName::AuditS3Operation,
         ];
 
         for ev in ALL_EVENT_NAMES {
@@ -967,6 +1008,40 @@ mod tests {
         }
     }
 
+    /// Audit entries are persisted by the store targets and read back through
+    /// `deserialize_persisted_event_name`, so every event that can be serialized, other than
+    /// `Everything`, must come back as itself. The default `Deserialize` also backs
+    /// user-supplied notification XML and must keep rejecting the internal events.
+    #[test]
+    fn test_persisted_round_trip_for_every_serializable_variant() {
+        #[derive(serde::Deserialize)]
+        struct Persisted(#[serde(deserialize_with = "deserialize_persisted_event_name")] EventName);
+
+        for ev in ALL_EVENT_NAMES {
+            if *ev == EventName::Everything {
+                continue;
+            }
+            let json = serde_json::to_string(ev).unwrap();
+            let back: Persisted = serde_json::from_str(&json)
+                .unwrap_or_else(|err| panic!("{ev:?} serialized as {json} does not deserialize: {err}"));
+            assert_eq!(back.0, *ev, "persisted round-trip changed {ev:?}");
+        }
+
+        for internal in [
+            EventName::ObjectRemovedAbortMultipartUpload,
+            EventName::ObjectCreatedCreateMultipartUpload,
+            EventName::ObjectRemovedDeleteObjects,
+            EventName::AuditS3Operation,
+        ] {
+            assert!(EventName::parse(internal.as_str()).is_err(), "parse must reject {internal:?}");
+            let json = serde_json::to_string(&internal).unwrap();
+            assert!(
+                serde_json::from_str::<EventName>(&json).is_err(),
+                "default Deserialize must reject {internal:?}"
+            );
+        }
+    }
+
     /// `ObjectScannerAll` specifically must round-trip via "s3:Scanner:*".
     #[test]
     fn test_object_scanner_all_round_trips() {
@@ -1055,6 +1130,23 @@ mod tests {
         }
     }
 
+    /// The audit-only S3 operation event must stay out of the `s3:` namespace and out of
+    /// every compound selector, so a bucket notification can never subscribe to it,
+    /// and it must survive the audit queue round trip.
+    #[test]
+    fn test_audit_s3_operation_is_audit_only_and_round_trips() {
+        let ev = EventName::AuditS3Operation;
+        assert!(ev.as_str().starts_with("audit:"), "unexpected name {:?}", ev.as_str());
+        // Notification configuration and the default `Deserialize` reject it; only the
+        // persisted path (queued audit entries) accepts it.
+        assert!(EventName::parse(ev.as_str()).is_err());
+        assert!(serde_json::from_str::<EventName>(&serde_json::to_string(&ev).unwrap()).is_err());
+        assert_eq!(ev.expand(), vec![ev]);
+        for other in ALL_EVENT_NAMES.iter().filter(|other| **other != ev) {
+            assert_eq!(other.mask() & ev.mask(), 0, "{other} shares a mask bit with {ev}");
+        }
+    }
+
     /// Each IAM event must own a distinct mask bit that no S3 selector shares.
     #[test]
     fn test_iam_event_masks_do_not_collide() {

crates/s3-types/src/lib.rs

@@ -14,4 +14,4 @@
 
 mod event_name;
 
-pub use event_name::{EventName, ParseEventNameError, event_schema_version};
+pub use event_name::{EventName, ParseEventNameError, deserialize_persisted_event_name, event_schema_version};

crates/targets/src/target/mod.rs

@@ -357,6 +357,7 @@ where
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct QueuedPayloadMeta {
+    #[serde(deserialize_with = "rustfs_s3_types::deserialize_persisted_event_name")]
     pub event_name: EventName,
     pub bucket_name: String,
     pub object_name: String,
@@ -1177,6 +1178,24 @@ mod tests {
         assert_eq!(decoded.dedup_id, "minted-id");
     }
 
+    /// Regression: the internal multipart events were dropped on replay because their
+    /// persisted name did not parse back.
+    #[test]
+    fn queued_payload_round_trips_internal_event_names() {
+        for event in [
+            EventName::ObjectCreatedCreateMultipartUpload,
+            EventName::ObjectRemovedAbortMultipartUpload,
+            EventName::ObjectRemovedDeleteObjects,
+            EventName::AuditS3Operation,
+        ] {
+            let body = br#"{"ok":true}"#.to_vec();
+            let meta = QueuedPayloadMeta::new(event, "bucket-a".to_string(), "obj".to_string(), "application/json", body.len());
+            let encoded = QueuedPayload::new(meta, body).encode().unwrap();
+            let decoded = QueuedPayload::decode(&encoded).unwrap_or_else(|err| panic!("{event:?} dropped on replay: {err}"));
+            assert_eq!(decoded.meta.event_name, event);
+        }
+    }
+
     #[test]
     fn queued_payload_round_trips_meta_and_body() {
         let body = br#"{"ok":true}"#.to_vec();

rustfs/src/storage/ecfs.rs

@@ -24,7 +24,7 @@ use super::{
 use crate::admin::handlers::site_replication::site_replication_bucket_meta_hook;
 use crate::error::{ApiError, ApiErrorDiagnostic};
 use crate::storage::access::{apply_bucket_generation_guard, bucket_config_mutation_incarnation, has_bypass_governance_header};
-use crate::storage::helper::OperationHelper;
+use crate::storage::helper::{OperationHelper, audit_only_operation};
 use crate::storage::options::get_opts;
 use crate::storage::s3_api::{self, acl};
 use crate::storage::storage_api::ecfs_consumer::contract::{
@@ -536,6 +536,190 @@ pub(crate) fn validate_object_lock_configuration_input(input_cfg: &ObjectLockCon
     Ok(())
 }
 
+/// Handler bodies that [`S3`] runs through `audit_only_operation`.
+impl FS {
+    async fn delete_bucket_replication_inner(
+        &self,
+        req: S3Request<DeleteBucketReplicationInput>,
+    ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
+        let (site_peers, contract) = site_replication_edit_context().await?;
+        let usecase = s3_api::bucket_usecase_for(self);
+        usecase.execute_delete_bucket_replication(req, site_peers, contract).await
+    }
+
+    async fn put_bucket_replication_inner(
+        &self,
+        req: S3Request<PutBucketReplicationInput>,
+    ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
+        deny_replication_config_edit_for_non_owner(&req).await?;
+        let (site_peers, contract) = site_replication_edit_context().await?;
+        let usecase = s3_api::bucket_usecase_for(self);
+        usecase.execute_put_bucket_replication(req, site_peers, contract).await
+    }
+
+    async fn put_bucket_acl_inner(&self, req: S3Request<PutBucketAclInput>) -> S3Result<S3Response<PutBucketAclOutput>> {
+        let PutBucketAclInput {
+            bucket,
+            access_control_policy,
+            ..
+        } = req.input;
+        record_s3_op(S3Operation::PutBucketAcl);
+
+        let Some(store) = self.server_ctx.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        if access_control_policy.is_some() {
+            return Err(s3_error!(
+                NotImplemented,
+                "ACL XML grants are not supported; use canned ACL headers or omit ACL"
+            ));
+        }
+
+        Ok(S3Response::new(PutBucketAclOutput::default()))
+    }
+
+    async fn put_bucket_logging_inner(
+        &self,
+        req: S3Request<PutBucketLoggingInput>,
+    ) -> S3Result<S3Response<PutBucketLoggingOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        record_s3_op(S3Operation::PutBucketLogging);
+        let Some(store) = self.server_ctx.object_store() else {
+            return Err(s3_error!(InternalError, "Not init"));
+        };
+        store
+            .get_bucket_info(&req.input.bucket, &BucketOptions::default())
+            .await
+            .map_err(crate::error::ApiError::from)?;
+
+        let logging_config = serialize(&req.input.bucket_logging_status)
+            .map_err(|err| S3Error::with_message(S3ErrorCode::MalformedXML, format!("{err}")))?;
+        update_bucket_metadata_config_if_incarnation(
+            &req.input.bucket,
+            BUCKET_LOGGING_CONFIG,
+            logging_config,
+            expected_incarnation_id,
+        )
+        .await
+        .map_err(crate::error::ApiError::from)?;
+
+        Ok(S3Response::new(PutBucketLoggingOutput::default()))
+    }
+
+    async fn put_object_lock_configuration_inner(
+        &self,
+        req: S3Request<PutObjectLockConfigurationInput>,
+    ) -> S3Result<S3Response<PutObjectLockConfigurationOutput>> {
+        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
+        let PutObjectLockConfigurationInput {
+            bucket,
+            object_lock_configuration,
+            ..
+        } = req.input;
+
+        let Some(input_cfg) = object_lock_configuration else { return Err(s3_error!(InvalidArgument)) };
+
+        let Some(store) = self.server_ctx.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        store
+            .get_bucket_info(&bucket, &BucketOptions::default())
+            .await
+            .map_err(ApiError::from)?;
+
+        validate_object_lock_configuration_input(&input_cfg)?;
+
+        match get_bucket_object_lock_config(&bucket).await {
+            Ok(_) => {}
+            Err(err) => {
+                if err == StorageError::ConfigNotFound {
+                    if !BucketVersioningSys::enabled(&bucket).await {
+                        return Err(S3Error::with_message(
+                            S3ErrorCode::InvalidBucketState,
+                            "Object Lock configuration cannot be enabled on existing buckets".to_string(),
+                        ));
+                    }
+                } else {
+                    warn!("get_object_lock_config err {:?}", err);
+                    return Err(S3Error::with_message(
+                        S3ErrorCode::InternalError,
+                        "Failed to get bucket ObjectLockConfiguration".to_string(),
+                    ));
+                }
+            }
+        };
+
+        let data = serialize(&input_cfg).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
+        let object_lock_config =
+            String::from_utf8(data.clone()).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
+
+        let updated_at = update_bucket_metadata_config_if_incarnation(&bucket, OBJECT_LOCK_CONFIG, data, expected_incarnation_id)
+            .await
+            .map_err(ApiError::from)?;
+
+        // When Object Lock is enabled, automatically enable versioning if not already enabled.
+        // This matches S3-compatible behavior.
+        let versioning_config = BucketVersioningSys::get(&bucket).await.map_err(ApiError::from)?;
+        if !versioning_config.enabled() {
+            let enable_versioning_config = VersioningConfiguration {
+                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
+                ..Default::default()
+            };
+            let versioning_data = serialize(&enable_versioning_config)
+                .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
+            update_bucket_metadata_config_if_incarnation(
+                &bucket,
+                BUCKET_VERSIONING_CONFIG,
+                versioning_data,
+                expected_incarnation_id,
+            )
+            .await
+            .map_err(ApiError::from)?;
+        }
+
+        if let Some(notification_sys) =
+            runtime_sources::current_notification_system_for_context(self.server_ctx.app_context().as_deref())
+        {
+            propagate_object_lock_peer_reload(notification_sys.load_bucket_metadata(&bucket).await)?;
+        }
+
+        if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta {
+            bucket: bucket.clone(),
+            r#type: "object-lock-config".to_string(),
+            object_lock_config: Some(object_lock_config),
+            updated_at: Some(updated_at),
+            api_version: Some(SITE_REPL_API_VERSION.to_string()),
+            ..Default::default()
+        })
+        .await
+        {
+            warn!(
+                component = LOG_COMPONENT_STORAGE,
+                subsystem = LOG_SUBSYSTEM_OBJECT_LOCK,
+                event = "put_object_lock_configuration",
+                bucket = %bucket,
+                result = "site_replication_hook_failed",
+                error = ?err,
+                "storage object lock state"
+            );
+        }
+
+        rustfs_scanner::record_dirty_usage_bucket_from_producer(
+            &bucket,
+            rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
+        );
+        Ok(S3Response::new(PutObjectLockConfigurationOutput::default()))
+    }
+}
+
 #[async_trait::async_trait]
 impl S3 for FS {
     #[instrument(level = "debug", skip(self))]
@@ -544,7 +728,7 @@ impl S3 for FS {
         req: S3Request<AbortMultipartUploadInput>,
     ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
         let usecase = s3_api::multipart_usecase_for(self);
-        usecase.execute_abort_multipart_upload(req).await
+        audit_only_operation(req, S3Operation::AbortMultipartUpload, |req| usecase.execute_abort_multipart_upload(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
@@ -594,7 +778,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self))]
     async fn delete_bucket_cors(&self, req: S3Request<DeleteBucketCorsInput>) -> S3Result<S3Response<DeleteBucketCorsOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_cors(req).await
+        audit_only_operation(req, S3Operation::DeleteBucketCors, |req| usecase.execute_delete_bucket_cors(req)).await
     }
 
     async fn delete_bucket_encryption(
@@ -602,7 +786,10 @@ impl S3 for FS {
         req: S3Request<DeleteBucketEncryptionInput>,
     ) -> S3Result<S3Response<DeleteBucketEncryptionOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_encryption(req).await
+        audit_only_operation(req, S3Operation::DeleteBucketEncryption, |req| {
+            usecase.execute_delete_bucket_encryption(req)
+        })
+        .await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -611,7 +798,10 @@ impl S3 for FS {
         req: S3Request<DeleteBucketLifecycleInput>,
     ) -> S3Result<S3Response<DeleteBucketLifecycleOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_lifecycle(req).await
+        audit_only_operation(req, S3Operation::DeleteBucketLifecycle, |req| {
+            usecase.execute_delete_bucket_lifecycle(req)
+        })
+        .await
     }
 
     async fn delete_bucket_policy(
@@ -619,17 +809,14 @@ impl S3 for FS {
         req: S3Request<DeleteBucketPolicyInput>,
     ) -> S3Result<S3Response<DeleteBucketPolicyOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_policy(req).await
+        audit_only_operation(req, S3Operation::DeleteBucketPolicy, |req| usecase.execute_delete_bucket_policy(req)).await
     }
 
     async fn delete_bucket_replication(
         &self,
         req: S3Request<DeleteBucketReplicationInput>,
     ) -> S3Result<S3Response<DeleteBucketReplicationOutput>> {
-        deny_replication_config_edit_for_non_owner(&req).await?;
-        let (site_peers, contract) = site_replication_edit_context().await?;
-        let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_replication(req, site_peers, contract).await
+        audit_only_operation(req, S3Operation::DeleteBucketReplication, |req| self.delete_bucket_replication_inner(req)).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -638,7 +825,7 @@ impl S3 for FS {
         req: S3Request<DeleteBucketTaggingInput>,
     ) -> S3Result<S3Response<DeleteBucketTaggingOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_bucket_tagging(req).await
+        audit_only_operation(req, S3Operation::DeleteBucketTagging, |req| usecase.execute_delete_bucket_tagging(req)).await
     }
 
     async fn delete_bucket_website(
@@ -668,7 +855,10 @@ impl S3 for FS {
         req: S3Request<DeletePublicAccessBlockInput>,
     ) -> S3Result<S3Response<DeletePublicAccessBlockOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_delete_public_access_block(req).await
+        audit_only_operation(req, S3Operation::DeletePublicAccessBlock, |req| {
+            usecase.execute_delete_public_access_block(req)
+        })
+        .await
     }
 
     /// Delete an object
@@ -1384,30 +1574,7 @@ impl S3 for FS {
     }
 
     async fn put_bucket_acl(&self, req: S3Request<PutBucketAclInput>) -> S3Result<S3Response<PutBucketAclOutput>> {
-        let PutBucketAclInput {
-            bucket,
-            access_control_policy,
-            ..
-        } = req.input;
-        record_s3_op(S3Operation::PutBucketAcl);
-
-        let Some(store) = self.server_ctx.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
-
-        store
-            .get_bucket_info(&bucket, &BucketOptions::default())
-            .await
-            .map_err(ApiError::from)?;
-
-        if access_control_policy.is_some() {
-            return Err(s3_error!(
-                NotImplemented,
-                "ACL XML grants are not supported; use canned ACL headers or omit ACL"
-            ));
-        }
-
-        Ok(S3Response::new(PutBucketAclOutput::default()))
+        audit_only_operation(req, S3Operation::PutBucketAcl, |req| self.put_bucket_acl_inner(req)).await
     }
 
     async fn put_bucket_accelerate_configuration(
@@ -1440,7 +1607,7 @@ impl S3 for FS {
     #[instrument(level = "debug", skip(self))]
     async fn put_bucket_cors(&self, req: S3Request<PutBucketCorsInput>) -> S3Result<S3Response<PutBucketCorsOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_cors(req).await
+        audit_only_operation(req, S3Operation::PutBucketCors, |req| usecase.execute_put_bucket_cors(req)).await
     }
 
     async fn get_bucket_logging(&self, req: S3Request<GetBucketLoggingInput>) -> S3Result<S3Response<GetBucketLoggingOutput>> {
@@ -1463,28 +1630,7 @@ impl S3 for FS {
     }
 
     async fn put_bucket_logging(&self, req: S3Request<PutBucketLoggingInput>) -> S3Result<S3Response<PutBucketLoggingOutput>> {
-        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
-        record_s3_op(S3Operation::PutBucketLogging);
-        let Some(store) = self.server_ctx.object_store() else {
-            return Err(s3_error!(InternalError, "Not init"));
-        };
-        store
-            .get_bucket_info(&req.input.bucket, &BucketOptions::default())
-            .await
-            .map_err(crate::error::ApiError::from)?;
-
-        let logging_config = serialize(&req.input.bucket_logging_status)
-            .map_err(|err| S3Error::with_message(S3ErrorCode::MalformedXML, format!("{err}")))?;
-        update_bucket_metadata_config_if_incarnation(
-            &req.input.bucket,
-            BUCKET_LOGGING_CONFIG,
-            logging_config,
-            expected_incarnation_id,
-        )
-        .await
-        .map_err(crate::error::ApiError::from)?;
-
-        Ok(S3Response::new(PutBucketLoggingOutput::default()))
+        audit_only_operation(req, S3Operation::PutBucketLogging, |req| self.put_bucket_logging_inner(req)).await
     }
 
     async fn put_bucket_encryption(
@@ -1492,7 +1638,7 @@ impl S3 for FS {
         req: S3Request<PutBucketEncryptionInput>,
     ) -> S3Result<S3Response<PutBucketEncryptionOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_encryption(req).await
+        audit_only_operation(req, S3Operation::PutBucketEncryption, |req| usecase.execute_put_bucket_encryption(req)).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -1501,7 +1647,10 @@ impl S3 for FS {
         req: S3Request<PutBucketLifecycleConfigurationInput>,
     ) -> S3Result<S3Response<PutBucketLifecycleConfigurationOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_lifecycle_configuration(req).await
+        audit_only_operation(req, S3Operation::PutBucketLifecycleConfiguration, |req| {
+            usecase.execute_put_bucket_lifecycle_configuration(req)
+        })
+        .await
     }
 
     async fn put_bucket_notification_configuration(
@@ -1509,22 +1658,22 @@ impl S3 for FS {
         req: S3Request<PutBucketNotificationConfigurationInput>,
     ) -> S3Result<S3Response<PutBucketNotificationConfigurationOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_notification_configuration(req).await
+        audit_only_operation(req, S3Operation::PutBucketNotificationConfiguration, |req| {
+            usecase.execute_put_bucket_notification_configuration(req)
+        })
+        .await
     }
 
     async fn put_bucket_policy(&self, req: S3Request<PutBucketPolicyInput>) -> S3Result<S3Response<PutBucketPolicyOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_policy(req).await
+        audit_only_operation(req, S3Operation::PutBucketPolicy, |req| usecase.execute_put_bucket_policy(req)).await
     }
 
     async fn put_bucket_replication(
         &self,
         req: S3Request<PutBucketReplicationInput>,
     ) -> S3Result<S3Response<PutBucketReplicationOutput>> {
-        deny_replication_config_edit_for_non_owner(&req).await?;
-        let (site_peers, contract) = site_replication_edit_context().await?;
-        let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_replication(req, site_peers, contract).await
+        audit_only_operation(req, S3Operation::PutBucketReplication, |req| self.put_bucket_replication_inner(req)).await
     }
 
     async fn put_bucket_request_payment(
@@ -1560,13 +1709,13 @@ impl S3 for FS {
         req: S3Request<PutPublicAccessBlockInput>,
     ) -> S3Result<S3Response<PutPublicAccessBlockOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_public_access_block(req).await
+        audit_only_operation(req, S3Operation::PutPublicAccessBlock, |req| usecase.execute_put_public_access_block(req)).await
     }
 
     #[instrument(level = "debug", skip(self))]
     async fn put_bucket_tagging(&self, req: S3Request<PutBucketTaggingInput>) -> S3Result<S3Response<PutBucketTaggingOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_tagging(req).await
+        audit_only_operation(req, S3Operation::PutBucketTagging, |req| usecase.execute_put_bucket_tagging(req)).await
     }
 
     #[instrument(level = "debug", skip(self))]
@@ -1575,7 +1724,7 @@ impl S3 for FS {
         req: S3Request<PutBucketVersioningInput>,
     ) -> S3Result<S3Response<PutBucketVersioningOutput>> {
         let usecase = s3_api::bucket_usecase_for(self);
-        usecase.execute_put_bucket_versioning(req).await
+        audit_only_operation(req, S3Operation::PutBucketVersioning, |req| usecase.execute_put_bucket_versioning(req)).await
     }
 
     async fn put_bucket_website(&self, req: S3Request<PutBucketWebsiteInput>) -> S3Result<S3Response<PutBucketWebsiteOutput>> {
@@ -1746,106 +1895,10 @@ impl S3 for FS {
         &self,
         req: S3Request<PutObjectLockConfigurationInput>,
     ) -> S3Result<S3Response<PutObjectLockConfigurationOutput>> {
-        let expected_incarnation_id = bucket_config_mutation_incarnation(&req, &req.input.bucket)?;
-        let PutObjectLockConfigurationInput {
-            bucket,
-            object_lock_configuration,
-            ..
-        } = req.input;
-
-        let Some(input_cfg) = object_lock_configuration else { return Err(s3_error!(InvalidArgument)) };
-
-        let Some(store) = self.server_ctx.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
-
-        store
-            .get_bucket_info(&bucket, &BucketOptions::default())
-            .await
-            .map_err(ApiError::from)?;
-
-        validate_object_lock_configuration_input(&input_cfg)?;
-
-        match get_bucket_object_lock_config(&bucket).await {
-            Ok(_) => {}
-            Err(err) => {
-                if err == StorageError::ConfigNotFound {
-                    if !BucketVersioningSys::enabled(&bucket).await {
-                        return Err(S3Error::with_message(
-                            S3ErrorCode::InvalidBucketState,
-                            "Object Lock configuration cannot be enabled on existing buckets".to_string(),
-                        ));
-                    }
-                } else {
-                    warn!("get_object_lock_config err {:?}", err);
-                    return Err(S3Error::with_message(
-                        S3ErrorCode::InternalError,
-                        "Failed to get bucket ObjectLockConfiguration".to_string(),
-                    ));
-                }
-            }
-        };
-
-        let data = serialize(&input_cfg).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
-        let object_lock_config =
-            String::from_utf8(data.clone()).map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
-
-        let updated_at = update_bucket_metadata_config_if_incarnation(&bucket, OBJECT_LOCK_CONFIG, data, expected_incarnation_id)
-            .await
-            .map_err(ApiError::from)?;
-
-        // When Object Lock is enabled, automatically enable versioning if not already enabled.
-        // This matches S3-compatible behavior.
-        let versioning_config = BucketVersioningSys::get(&bucket).await.map_err(ApiError::from)?;
-        if !versioning_config.enabled() {
-            let enable_versioning_config = VersioningConfiguration {
-                status: Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)),
-                ..Default::default()
-            };
-            let versioning_data = serialize(&enable_versioning_config)
-                .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("{err}")))?;
-            update_bucket_metadata_config_if_incarnation(
-                &bucket,
-                BUCKET_VERSIONING_CONFIG,
-                versioning_data,
-                expected_incarnation_id,
-            )
-            .await
-            .map_err(ApiError::from)?;
-        }
-
-        if let Some(notification_sys) =
-            runtime_sources::current_notification_system_for_context(self.server_ctx.app_context().as_deref())
-        {
-            propagate_object_lock_peer_reload(notification_sys.load_bucket_metadata(&bucket).await)?;
-        }
-
-        if let Err(err) = site_replication_bucket_meta_hook(SRBucketMeta {
-            bucket: bucket.clone(),
-            r#type: "object-lock-config".to_string(),
-            object_lock_config: Some(object_lock_config),
-            updated_at: Some(updated_at),
-            api_version: Some(SITE_REPL_API_VERSION.to_string()),
-            ..Default::default()
+        audit_only_operation(req, S3Operation::PutObjectLockConfiguration, |req| {
+            self.put_object_lock_configuration_inner(req)
         })
         .await
-        {
-            warn!(
-                component = LOG_COMPONENT_STORAGE,
-                subsystem = LOG_SUBSYSTEM_OBJECT_LOCK,
-                event = "put_object_lock_configuration",
-                bucket = %bucket,
-                result = "site_replication_hook_failed",
-                error = ?err,
-                "storage object lock state"
-            );
-        }
-
-        rustfs_scanner::record_dirty_usage_bucket_from_producer(
-            &bucket,
-            rustfs_scanner::SegmentInvalidationProducerIdentity::BucketMetadata,
-        );
-        Ok(S3Response::new(PutObjectLockConfigurationOutput::default()))
     }
 
     async fn put_object_retention(
@@ -2104,14 +2157,47 @@ impl S3 for FS {
         crate::hp_guard!("S3::upload_part");
         record_s3_op(S3Operation::UploadPart);
         let usecase = s3_api::multipart_usecase_for(self);
-        usecase.execute_upload_part(req).await
+        audit_only_operation(req, S3Operation::UploadPart, |req| usecase.execute_upload_part(req)).await
     }
 
     #[instrument(level = "debug", skip(self, req))]
     async fn upload_part_copy(&self, req: S3Request<UploadPartCopyInput>) -> S3Result<S3Response<UploadPartCopyOutput>> {
         record_s3_op(S3Operation::UploadPartCopy);
         let usecase = s3_api::multipart_usecase_for(self);
-        Box::pin(usecase.execute_upload_part_copy(req)).await
+        audit_only_operation(req, S3Operation::UploadPartCopy, |req| Box::pin(usecase.execute_upload_part_copy(req))).await
+    }
+}
+
+/// Every mutating S3 operation that has no bucket notification event reaches the audit
+/// stream only through `audit_only_operation`; a new such operation, or a dropped
+/// wrapper, must fail here rather than silently go unaudited.
+#[cfg(test)]
+mod audit_coverage_tests {
+    use rustfs_s3_ops::S3Operation;
+
+    const SOURCE: &str = include_str!("ecfs.rs");
+
+    fn is_read_only(op: S3Operation) -> bool {
+        let name = op.as_str().trim_start_matches("s3:");
+        ["Get", "Head", "List", "Select"]
+            .iter()
+            .any(|prefix| name.starts_with(prefix))
+    }
+
+    #[test]
+    fn mutating_operations_without_a_notification_event_are_wrapped_for_audit() {
+        let mut expected: Vec<S3Operation> = S3Operation::ALL
+            .into_iter()
+            .filter(|op| !is_read_only(*op) && op.to_event_name().is_none())
+            .collect();
+        // AbortMultipartUpload has an internal event but no handler-built helper.
+        expected.push(S3Operation::AbortMultipartUpload);
+        assert_eq!(expected.len(), 22, "an operation was added or removed: {expected:?}");
+
+        for op in expected {
+            let wrapper = format!("audit_only_operation(req, S3Operation::{op:?},");
+            assert!(SOURCE.contains(&wrapper), "{} is not run through audit_only_operation", op.as_str());
+        }
     }
 }
 

rustfs/src/storage/helper.rs

@@ -119,7 +119,23 @@ pub struct EnabledOperationHelper {
 impl OperationHelper {
     /// Create a new OperationHelper for S3 requests.
     pub fn new(req: &S3Request<impl Send + Sync>, event: EventName, op: S3Operation) -> Self {
-        let op_event_matches = operation_matches_event_name(op, event);
+        Self::build(req, event, op, false)
+    }
+
+    /// Create an OperationHelper that only emits an audit entry, never a bucket notification.
+    ///
+    /// For operations that have no notification event of their own (`UploadPart`, the
+    /// bucket configuration writes) or whose event is internal (`AbortMultipartUpload`).
+    /// Unlike `new` it does not call `record_s3_op`, so per-operation metrics stay as the
+    /// handlers already report them.
+    pub fn audit_only(req: &S3Request<impl Send + Sync>, op: S3Operation) -> Self {
+        let event = op.to_event_name().unwrap_or(EventName::AuditS3Operation);
+        Self::build(req, event, op, true)
+    }
+
+    fn build(req: &S3Request<impl Send + Sync>, event: EventName, op: S3Operation, audit_only: bool) -> Self {
+        // `audit_only` derives the event from the operation, so there is nothing to cross-check.
+        let op_event_matches = audit_only || operation_matches_event_name(op, event);
         debug_assert!(op_event_matches, "operation/event mismatch: op={} event={}", op.as_str(), event.as_str());
         if !op_event_matches {
             counter!(
@@ -136,7 +152,11 @@ impl OperationHelper {
         }
 
         let audit_enabled = is_audit_module_enabled();
-        let notify_enabled = should_build_notification_event(is_notify_module_enabled());
+        let notify_enabled = !audit_only && should_build_notification_event(is_notify_module_enabled());
+        if audit_only && !audit_enabled {
+            // Nothing to record and no metric to bump: skip the path parsing below.
+            return Self::Disabled;
+        }
 
         let path = req.uri.path().trim_start_matches('/');
         let mut segs = path.splitn(2, '/');
@@ -148,7 +168,9 @@ impl OperationHelper {
             .filter(|value| !value.is_empty())
             .unwrap_or(path_bucket);
 
-        record_s3_op(op);
+        if !audit_only {
+            record_s3_op(op);
+        }
 
         // Fast path: when both chains are disabled, avoid audit/notify builder work.
         if !audit_enabled && !notify_enabled {
@@ -418,6 +440,22 @@ impl OperationHelper {
     }
 }
 
+/// Runs `handler` on `req` and records the outcome as an audit-only entry for `op`.
+///
+/// The helper is built before the request is handed over so it sees the original headers
+/// and extensions, and it completes with the handler's result, so failed requests are audited too.
+pub(crate) async fn audit_only_operation<I, O, F, Fut>(req: S3Request<I>, op: S3Operation, handler: F) -> S3Result<S3Response<O>>
+where
+    I: Send + Sync,
+    F: FnOnce(S3Request<I>) -> Fut,
+    Fut: Future<Output = S3Result<S3Response<O>>>,
+{
+    let helper = OperationHelper::audit_only(&req, op);
+    let result = handler(req).await;
+    let _helper = helper.complete(&result);
+    result
+}
+
 fn should_build_notification_event(notify_module_enabled: bool) -> bool {
     notify_module_enabled || rustfs_notify::notification_system().is_some_and(|system| system.has_live_listeners())
 }
@@ -470,7 +508,7 @@ mod tests {
     use rustfs_s3_types::EventName;
     use rustfs_utils::http::headers::{AMZ_REQUEST_ID, REQUEST_ID_HEADER};
     use s3s::dto::{DeleteObjectTaggingInput, DeleteObjectTaggingOutput, GetObjectInput, GetObjectOutput};
-    use s3s::{S3Error, S3ErrorCode, S3Request, S3Response};
+    use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
     use std::sync::{Arc, Mutex};
     use temp_env::{async_with_vars, with_vars};
 
@@ -740,6 +778,151 @@ mod tests {
         );
     }
 
+    /// Operations without a notification event still produce an audit entry, and the
+    /// audit-only constructor never builds a notification event, even with notify on.
+    #[test]
+    fn audit_only_helper_audits_operations_without_a_notification_event() {
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
+            ],
+            || {
+                refresh_notify_module_enabled();
+                refresh_audit_module_enabled();
+
+                for (op, uri, expected_event) in [
+                    (S3Operation::UploadPart, "/audit-bucket/object.txt", EventName::AuditS3Operation),
+                    (S3Operation::UploadPartCopy, "/audit-bucket/object.txt", EventName::AuditS3Operation),
+                    (S3Operation::PutBucketPolicy, "/audit-bucket", EventName::AuditS3Operation),
+                    (S3Operation::PutBucketVersioning, "/audit-bucket", EventName::AuditS3Operation),
+                    (
+                        S3Operation::AbortMultipartUpload,
+                        "/audit-bucket/object.txt",
+                        EventName::ObjectRemovedAbortMultipartUpload,
+                    ),
+                ] {
+                    let input = DeleteObjectTaggingInput::builder()
+                        .bucket("audit-bucket".to_string())
+                        .key("object.txt".to_string())
+                        .build()
+                        .unwrap();
+                    let mut req = build_request(input, Method::PUT, Uri::from_static(uri));
+                    req.extensions.insert(ReqInfo {
+                        cred: Some(Credentials {
+                            access_key: "auditor".to_string(),
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    });
+
+                    let result = Ok(S3Response::new(DeleteObjectTaggingOutput::default()));
+                    let mut helper = OperationHelper::audit_only(&req, op).complete(&result);
+                    assert!(!helper.wants_object_info(), "{} must not build a notification event", op.as_str());
+                    assert!(helper.event_args().is_none(), "{} must not build a notification event", op.as_str());
+                    let OperationHelper::Enabled(state) = &mut helper else {
+                        panic!("helper should be enabled when the audit switch is on");
+                    };
+                    let entry = state.audit_builder.take().expect("audit builder should exist").build();
+                    assert_eq!(entry.event, expected_event, "{}", op.as_str());
+                    assert_eq!(entry.api.name.as_deref(), Some(op.as_str()));
+                    assert_eq!(entry.api.bucket.as_deref(), Some("audit-bucket"));
+                    assert_eq!(entry.api.status.as_deref(), Some("success"));
+                    assert_eq!(entry.access_key.as_deref(), Some("auditor"));
+                }
+            },
+        );
+    }
+
+    /// With audit off, the audit-only helper does no work: it is the disabled helper.
+    #[test]
+    fn audit_only_helper_is_disabled_when_audit_is_off() {
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("false")),
+            ],
+            || {
+                refresh_notify_module_enabled();
+                refresh_audit_module_enabled();
+                let input = DeleteObjectTaggingInput::builder()
+                    .bucket("audit-bucket".to_string())
+                    .key("object.txt".to_string())
+                    .build()
+                    .unwrap();
+                let req = build_request(input, Method::PUT, Uri::from_static("/audit-bucket/object.txt"));
+                assert!(matches!(
+                    OperationHelper::audit_only(&req, S3Operation::UploadPart),
+                    OperationHelper::Disabled
+                ));
+            },
+        );
+    }
+
+    /// A failed audit-only operation is audited with a failure status, the error message and
+    /// the error's status code.
+    #[test]
+    fn audit_only_helper_records_failure_status_and_error() {
+        with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("false")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
+            ],
+            || {
+                refresh_notify_module_enabled();
+                refresh_audit_module_enabled();
+
+                let input = DeleteObjectTaggingInput::builder()
+                    .bucket("audit-bucket".to_string())
+                    .key("object.txt".to_string())
+                    .build()
+                    .unwrap();
+                let req = build_request(input, Method::PUT, Uri::from_static("/audit-bucket"));
+                let result: S3Result<S3Response<DeleteObjectTaggingOutput>> =
+                    Err(S3Error::with_message(S3ErrorCode::AccessDenied, "policy refused".to_string()));
+                let mut helper = OperationHelper::audit_only(&req, S3Operation::PutBucketPolicy).complete(&result);
+                let OperationHelper::Enabled(state) = &mut helper else {
+                    panic!("helper should be enabled when the audit switch is on");
+                };
+                let entry = state.audit_builder.take().expect("audit builder should exist").build();
+                assert_eq!(entry.api.status.as_deref(), Some("failure"));
+                assert_eq!(entry.api.status_code, Some(403));
+                assert_eq!(entry.error.as_deref(), Some("policy refused"));
+            },
+        );
+    }
+
+    /// A failed request is audited with its failure status, and the wrapper hands the
+    /// handler's result back unchanged.
+    #[tokio::test]
+    async fn audit_only_operation_returns_the_handler_result_unchanged() {
+        async_with_vars(
+            [
+                (rustfs_config::ENV_NOTIFY_ENABLE, Some("true")),
+                (rustfs_config::ENV_AUDIT_ENABLE, Some("true")),
+            ],
+            async {
+                refresh_notify_module_enabled();
+                refresh_audit_module_enabled();
+
+                let input = DeleteObjectTaggingInput::builder()
+                    .bucket("audit-bucket".to_string())
+                    .key("object.txt".to_string())
+                    .build()
+                    .unwrap();
+                let req = build_request(input, Method::PUT, Uri::from_static("/audit-bucket"));
+                let err = super::audit_only_operation(req, S3Operation::PutBucketPolicy, |req| async move {
+                    assert_eq!(req.input.bucket, "audit-bucket");
+                    Err::<S3Response<DeleteObjectTaggingOutput>, _>(S3Error::new(S3ErrorCode::AccessDenied))
+                })
+                .await
+                .unwrap_err();
+                assert_eq!(*err.code(), S3ErrorCode::AccessDenied);
+            },
+        )
+        .await;
+    }
+
     #[test]
     fn operation_helper_audits_the_authenticated_request_identity() {
         with_vars(