feat: object retention (#1589)

rustfs/src/storage/access.rs

@@ -20,6 +20,7 @@ use rustfs_ecstore::bucket::policy_sys::PolicySys;
 use rustfs_iam::error::Error as IamError;
 use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_policy::policy::{Args, BucketPolicyArgs};
+use rustfs_utils::http::AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE;
 use s3s::access::{S3Access, S3AccessContext};
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Result, dto::*, s3_error};
 use std::collections::HashMap;
@@ -189,6 +190,15 @@ pub async fn authorize_request<T>(req: &mut S3Request<T>, action: Action) -> S3R
     Err(s3_error!(AccessDenied, "Access Denied"))
 }
 
+/// Check if the request has the x-amz-bypass-governance-retention header set to true
+pub fn has_bypass_governance_header(headers: &http::HeaderMap) -> bool {
+    headers
+        .get(AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE)
+        .and_then(|v| v.to_str().ok())
+        .map(|s| s.eq_ignore_ascii_case("true"))
+        .unwrap_or(false)
+}
+
 #[async_trait::async_trait]
 impl S3Access for FS {
     // /// Checks whether the current request has accesses to the resources.
@@ -448,7 +458,14 @@ impl S3Access for FS {
         req_info.object = Some(req.input.key.clone());
         req_info.version_id = req.input.version_id.clone();
 
-        authorize_request(req, Action::S3Action(S3Action::DeleteObjectAction)).await
+        authorize_request(req, Action::S3Action(S3Action::DeleteObjectAction)).await?;
+
+        // S3 Standard: When bypass_governance header is set, must have s3:BypassGovernanceRetention permission
+        if has_bypass_governance_header(&req.headers) {
+            authorize_request(req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await?;
+        }
+
+        Ok(())
     }
 
     /// Checks whether the DeleteObjectTagging request has accesses to the resources.
@@ -471,7 +488,15 @@ impl S3Access for FS {
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = None;
         req_info.version_id = None;
-        authorize_request(req, Action::S3Action(S3Action::DeleteObjectAction)).await
+
+        authorize_request(req, Action::S3Action(S3Action::DeleteObjectAction)).await?;
+
+        // S3 Standard: When bypass_governance header is set, must have s3:BypassGovernanceRetention permission
+        if has_bypass_governance_header(&req.headers) {
+            authorize_request(req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await?;
+        }
+
+        Ok(())
     }
 
     /// Checks whether the DeletePublicAccessBlock request has accesses to the resources.
@@ -1113,15 +1138,20 @@ impl S3Access for FS {
     }
 
     /// Checks whether the PutObjectRetention request has accesses to the resources.
-    ///
-    /// This method returns `Ok(())` by default.
     async fn put_object_retention(&self, req: &mut S3Request<PutObjectRetentionInput>) -> S3Result<()> {
         let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
         req_info.bucket = Some(req.input.bucket.clone());
         req_info.object = Some(req.input.key.clone());
         req_info.version_id = req.input.version_id.clone();
 
-        authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await
+        authorize_request(req, Action::S3Action(S3Action::PutObjectRetentionAction)).await?;
+
+        // S3 Standard: When bypass_governance header is set, must have s3:BypassGovernanceRetention permission
+        if has_bypass_governance_header(&req.headers) {
+            authorize_request(req, Action::S3Action(S3Action::BypassGovernanceRetentionAction)).await?;
+        }
+
+        Ok(())
     }
 
     /// Checks whether the PutObjectTagging request has accesses to the resources.

rustfs/src/storage/ecfs.rs

@@ -26,7 +26,7 @@ use crate::storage::entity;
 use crate::storage::helper::OperationHelper;
 use crate::storage::options::{filter_object_metadata, get_content_sha256};
 use crate::storage::{
-    access::{ReqInfo, authorize_request},
+    access::{ReqInfo, authorize_request, has_bypass_governance_header},
     options::{
         copy_dst_opts, copy_src_opts, del_opts, extract_metadata, extract_metadata_from_mime_with_object_name,
         get_complete_multipart_upload_opts, get_opts, parse_copy_source_range, put_opts,
@@ -53,7 +53,7 @@ use rustfs_ecstore::{
         },
         metadata_sys,
         metadata_sys::get_replication_config,
-        object_lock::objectlock_sys::BucketObjectLockSys,
+        object_lock::objectlock_sys::{BucketObjectLockSys, check_object_lock_for_deletion, check_retention_for_modification},
         policy_sys::PolicySys,
         quota::QuotaOperation,
         replication::{
@@ -551,12 +551,24 @@ fn parse_object_lock_retention(retention: Option<ObjectLockRetention>) -> S3Resu
             None => String::default(),
         };
 
-        let retain_until_date = v
-            .retain_until_date
-            .map(|v| OffsetDateTime::from(v).format(&Rfc3339).unwrap())
-            .unwrap_or_default();
-
         let now = OffsetDateTime::now_utc();
+
+        // Validate retain_until_date is in the future (S3 requirement)
+        // Only validate when both mode and date are provided (not clearing retention)
+        let retain_until_date = if let Some(date) = v.retain_until_date {
+            let retain_until = OffsetDateTime::from(date);
+            // Only validate future date when mode is set (not clearing retention)
+            if !mode.is_empty() && retain_until <= now {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidArgument,
+                    "The retain until date must be in the future".to_string(),
+                ));
+            }
+            retain_until.format(&Rfc3339).unwrap()
+        } else {
+            String::default()
+        };
+
         // This is intentional behavior. Empty string represents "retention cleared" which is different from "retention never set". Consistent with minio
         eval_metadata.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), mode);
         eval_metadata.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), retain_until_date);
@@ -1955,12 +1967,13 @@ impl S3 for FS {
 
         let metadata = extract_metadata(&req.headers);
 
+        // Clone version_id before it's moved
+        let version_id_clone = version_id.clone();
+
         let mut opts: ObjectOptions = del_opts(&bucket, &key, version_id, &req.headers, metadata)
             .await
             .map_err(ApiError::from)?;
 
-        // TODO: check object lock
-
         let lock_cfg = BucketObjectLockSys::get(&bucket).await;
         if lock_cfg.is_some() && opts.delete_prefix {
             return Err(S3Error::with_message(
@@ -1983,6 +1996,33 @@ impl S3 for FS {
             return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
         };
 
+        // Check Object Lock retention before deletion
+        // TODO: Future optimization (separate PR) - If performance becomes critical under high delete load:
+        // 1. Integrate OptimizedFileCache (file_cache.rs) into the read_version() path
+        // 2. Or add a lightweight get_object_lock_info() that only fetches retention metadata
+        // 3. Or use combined get-and-delete in storage layer with retention check callback
+        // Note: The project has OptimizedFileCache with moka, but get_object_info doesn't use it yet
+        let get_opts: ObjectOptions = get_opts(&bucket, &key, version_id_clone, None, &req.headers)
+            .await
+            .map_err(ApiError::from)?;
+
+        match store.get_object_info(&bucket, &key, &get_opts).await {
+            Ok(obj_info) => {
+                // Check for bypass governance retention header (permission already verified in access.rs)
+                let bypass_governance = has_bypass_governance_header(&req.headers);
+
+                if let Some(block_reason) = check_object_lock_for_deletion(&bucket, &obj_info, bypass_governance).await {
+                    return Err(S3Error::with_message(S3ErrorCode::AccessDenied, block_reason.error_message()));
+                }
+            }
+            Err(err) => {
+                // If object not found, allow deletion to proceed (will return 204 No Content)
+                if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                    return Err(ApiError::from(err).into());
+                }
+            }
+        }
+
         let obj_info = {
             match store.delete_object(&bucket, &key, opts).await {
                 Ok(obj) => obj,
@@ -2101,6 +2141,9 @@ impl S3 for FS {
 
         let version_cfg = BucketVersioningSys::get(&bucket).await.unwrap_or_default();
 
+        // Check for bypass governance retention header (permission already verified in access.rs)
+        let bypass_governance = has_bypass_governance_header(&req.headers);
+
         #[derive(Default, Clone)]
         struct DeleteResult {
             delete_object: Option<rustfs_ecstore::store_api::DeletedObject>,
@@ -2176,6 +2219,21 @@ impl S3 for FS {
                 Err(e) => (ObjectInfo::default(), Some(e.to_string())),
             };
 
+            // Check Object Lock retention before deletion
+            // NOTE: Unlike single DeleteObject, this reuses the get_object_info result from quota
+            // tracking above, so no additional storage operation is required for the retention check.
+            if gerr.is_none()
+                && let Some(block_reason) = check_object_lock_for_deletion(&bucket, &goi, bypass_governance).await
+            {
+                delete_results[idx].error = Some(Error {
+                    code: Some("AccessDenied".to_string()),
+                    key: Some(obj_id.key.clone()),
+                    message: Some(block_reason.error_message()),
+                    version_id: version_id.clone(),
+                });
+                continue;
+            }
+
             // Store object size for quota tracking
             object_sizes.insert(object.object_name.clone(), goi.size);
 
@@ -5992,6 +6050,20 @@ impl S3 for FS {
             .await
             .map_err(ApiError::from)?;
 
+        // When Object Lock is enabled, automatically enable versioning if not already enabled
+        // This matches AWS S3 and MinIO 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 = try_!(serialize(&enable_versioning_config));
+            metadata_sys::update(&bucket, BUCKET_VERSIONING_CONFIG, versioning_data)
+                .await
+                .map_err(ApiError::from)?;
+        }
+
         Ok(S3Response::new(PutObjectLockConfigurationOutput::default()))
     }
 
@@ -6639,7 +6711,41 @@ impl S3 for FS {
         // check object lock
         validate_bucket_object_lock_enabled(&bucket).await?;
 
-        // TODO: check allow
+        // Extract new retain_until_date for validation
+        let new_retain_until = retention
+            .as_ref()
+            .and_then(|r| r.retain_until_date.as_ref())
+            .map(|d| OffsetDateTime::from(d.clone()));
+
+        // Check if object already has retention and if modification is allowed
+        // This follows AWS S3 and MinIO behavior:
+        // - COMPLIANCE mode: can only extend retention period, never shorten
+        // - GOVERNANCE mode: requires bypass header to modify/shorten retention
+        //
+        // TODO: Known race condition (fix in future PR)
+        // There's a TOCTOU (time-of-check-time-of-use) window between retention check here
+        // and the actual update at put_object_metadata. In theory:
+        //   Thread A: reads GOVERNANCE mode, checks bypass header
+        //   Thread B: updates retention to COMPLIANCE mode
+        //   Thread A: modifies retention, bypassing what is now COMPLIANCE mode
+        // This violates S3 spec that COMPLIANCE cannot be modified even with bypass.
+        // Fix options:
+        // 1. Pass expected retention mode to storage layer, verify before update
+        // 2. Use optimistic concurrency with version/etag checks
+        // 3. Perform check within the same lock scope as update in storage layer
+        // Current mitigation: Storage layer has fast_lock_manager which provides some protection
+        let check_opts: ObjectOptions = get_opts(&bucket, &key, version_id.clone(), None, &req.headers)
+            .await
+            .map_err(ApiError::from)?;
+
+        if let Ok(existing_obj_info) = store.get_object_info(&bucket, &key, &check_opts).await {
+            let bypass_governance = has_bypass_governance_header(&req.headers);
+            if let Some(block_reason) =
+                check_retention_for_modification(&existing_obj_info.user_defined, new_retain_until, bypass_governance)
+            {
+                return Err(S3Error::with_message(S3ErrorCode::AccessDenied, block_reason.error_message()));
+            }
+        }
 
         let eval_metadata = parse_object_lock_retention(retention)?;
 
@@ -7122,33 +7228,33 @@ mod tests {
         assert!(parse_object_lock_retention(None).is_ok());
         assert!(parse_object_lock_retention(None).unwrap().is_empty());
 
-        // [2] Normal case: Retention with valid COMPLIANCE mode
+        // [2] Normal case: Retention with valid COMPLIANCE mode (future date)
         let valid_compliance_retention = ObjectLockRetention {
             mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
-            retain_until_date: Some(datetime!(2025-01-01 00:00:00 UTC).into()),
+            retain_until_date: Some(datetime!(2030-01-01 00:00:00 UTC).into()),
         };
         let compliance_metadata = parse_object_lock_retention(Some(valid_compliance_retention)).unwrap();
         assert_eq!(compliance_metadata.get("x-amz-object-lock-mode").unwrap(), "COMPLIANCE");
         assert_eq!(
             compliance_metadata.get("x-amz-object-lock-retain-until-date").unwrap(),
-            "2025-01-01T00:00:00Z"
+            "2030-01-01T00:00:00Z"
         );
         assert!(
             compliance_metadata.contains_key(&format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, "objectlock-retention-timestamp"))
         );
 
-        // [3] Normal case: Retention with valid GOVERNANCE mode
+        // [3] Normal case: Retention with valid GOVERNANCE mode (future date)
         let valid_governance_retention = ObjectLockRetention {
             mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::GOVERNANCE)),
-            retain_until_date: Some(datetime!(2025-01-01 00:00:00 UTC).into()),
+            retain_until_date: Some(datetime!(2030-01-01 00:00:00 UTC).into()),
         };
         let governance_metadata = parse_object_lock_retention(Some(valid_governance_retention)).unwrap();
         assert_eq!(governance_metadata.get("x-amz-object-lock-mode").unwrap(), "GOVERNANCE");
 
-        // [4] Normal case: Retention with None mode (empty string for mode)
+        // [4] Normal case: Retention with None mode (empty string for mode, date not validated)
         let none_mode_retention = ObjectLockRetention {
             mode: None,
-            retain_until_date: Some(datetime!(2025-01-01 00:00:00 UTC).into()),
+            retain_until_date: Some(datetime!(2030-01-01 00:00:00 UTC).into()),
         };
         let none_mode_metadata = parse_object_lock_retention(Some(none_mode_retention)).unwrap();
         assert_eq!(none_mode_metadata.get("x-amz-object-lock-mode").unwrap(), "");
@@ -7164,7 +7270,7 @@ mod tests {
         // [6] Error case: Retention with invalid mode (non COMPLIANCE/GOVERNANCE)
         let invalid_mode_retention = ObjectLockRetention {
             mode: Some(ObjectLockRetentionMode::from_static("INVALID_MODE")),
-            retain_until_date: Some(datetime!(2025-01-01 00:00:00 UTC).into()),
+            retain_until_date: Some(datetime!(2030-01-01 00:00:00 UTC).into()),
         };
         let err = parse_object_lock_retention(Some(invalid_mode_retention)).unwrap_err();
         assert_eq!(err.code().as_str(), S3ErrorCode::MalformedXML.as_str());
@@ -7172,6 +7278,15 @@ mod tests {
             err.message(),
             Some("The XML you provided was not well-formed or did not validate against our published schema")
         );
+
+        // [7] Error case: Retention with past date should fail
+        let past_date_retention = ObjectLockRetention {
+            mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
+            retain_until_date: Some(datetime!(2020-01-01 00:00:00 UTC).into()),
+        };
+        let err = parse_object_lock_retention(Some(past_date_retention)).unwrap_err();
+        assert_eq!(err.code().as_str(), S3ErrorCode::InvalidArgument.as_str());
+        assert_eq!(err.message(), Some("The retain until date must be in the future"));
     }
 
     #[test]

rustfs/src/storage/options.rs

@@ -444,6 +444,10 @@ static SUPPORTED_HEADERS: LazyLock<Vec<&'static str>> = LazyLock::new(|| {
         "x-amz-tagging",
         "expires",
         "x-amz-replication-status",
+        // Object Lock headers - required for S3 Object Lock functionality
+        "x-amz-object-lock-mode",
+        "x-amz-object-lock-retain-until-date",
+        "x-amz-object-lock-legal-hold",
     ]
 });
 
@@ -1021,10 +1025,13 @@ mod tests {
             "x-amz-tagging",
             "expires",
             "x-amz-replication-status",
+            "x-amz-object-lock-mode",
+            "x-amz-object-lock-retain-until-date",
+            "x-amz-object-lock-legal-hold",
         ];
 
         assert_eq!(*SUPPORTED_HEADERS, expected_headers);
-        assert_eq!(SUPPORTED_HEADERS.len(), 9);
+        assert_eq!(SUPPORTED_HEADERS.len(), 12);
     }
 
     #[test]