diff --git a/crates/ecstore/src/bucket/lifecycle/lifecycle.rs b/crates/ecstore/src/bucket/lifecycle/lifecycle.rs index ee2afe002..236e3a745 100644 --- a/crates/ecstore/src/bucket/lifecycle/lifecycle.rs +++ b/crates/ecstore/src/bucket/lifecycle/lifecycle.rs @@ -46,6 +46,9 @@ const ERR_LIFECYCLE_BUCKET_LOCKED: &str = "ExpiredObjectAllVersions element and DelMarkerExpiration action cannot be used on an retention bucket"; const ERR_LIFECYCLE_TOO_MANY_RULES: &str = "Lifecycle configuration should have at most 1000 rules"; const ERR_LIFECYCLE_INVALID_EXPIRATION_DAYS: &str = "Lifecycle expiration days must be greater than 0"; +const ERR_LIFECYCLE_INVALID_EXPIRATION_DATE_NOT_MIDNIGHT: &str = "Expiration.Date must be at midnight UTC"; +const ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG: &str = "Rule ID must be at most 255 characters"; +const ERR_LIFECYCLE_INVALID_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled"; pub use rustfs_common::metrics::IlmAction; @@ -139,6 +142,7 @@ pub trait Lifecycle { async fn validate(&self, lr: &ObjectLockConfiguration) -> Result<(), std::io::Error>; async fn filter_rules(&self, obj: &ObjectOpts) -> Option>; async fn eval(&self, obj: &ObjectOpts) -> Event; + async fn predict_expiration(&self, obj: &ObjectOpts) -> Event; async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, newer_noncurrent_versions: usize) -> Event; //fn set_prediction_headers(&self, w: http.ResponseWriter, obj: ObjectOpts); async fn noncurrent_versions_expiration_limit(self: Arc, obj: &ObjectOpts) -> Event; @@ -233,13 +237,30 @@ impl Lifecycle for BucketLifecycleConfiguration { } for r in &self.rules { + if r.status != ExpirationStatus::from_static(ExpirationStatus::ENABLED) + && r.status != ExpirationStatus::from_static(ExpirationStatus::DISABLED) + { + return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_STATUS)); + } + if let Some(expiration) = &r.expiration { + if let Some(expiration_date) = &expiration.date { + let date = OffsetDateTime::from(expiration_date.clone()); + if date.hour() != 0 || date.minute() != 0 || date.second() != 0 || date.nanosecond() != 0 { + return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_EXPIRATION_DATE_NOT_MIDNIGHT)); + } + } if let Some(days) = expiration.days { if days <= 0 { return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_EXPIRATION_DAYS)); } } } + if let Some(id) = &r.id { + if id.len() > 255 { + return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG)); + } + } r.validate()?; /*if let Some(object_lock_enabled) = lr.object_lock_enabled.as_ref() { if let Some(expiration) = r.expiration.as_ref() { @@ -257,8 +278,10 @@ impl Lifecycle for BucketLifecycleConfiguration { } let other_rules = &self.rules[i + 1..]; for other_rule in other_rules { - if self.rules[i].id == other_rule.id { - return Err(std::io::Error::other(ERR_LIFECYCLE_DUPLICATE_ID)); + if let (Some(id1), Some(id2)) = (&self.rules[i].id, &other_rule.id) { + if id1 == id2 { + return Err(std::io::Error::other(ERR_LIFECYCLE_DUPLICATE_ID)); + } } } } @@ -295,6 +318,61 @@ impl Lifecycle for BucketLifecycleConfiguration { self.eval_inner(obj, OffsetDateTime::now_utc(), 0).await } + async fn predict_expiration(&self, obj: &ObjectOpts) -> Event { + let mod_time = match obj.mod_time { + Some(time) => { + if time.unix_timestamp() == 0 { + return Event::default(); + } + time + } + None => return Event::default(), + }; + + if obj.delete_marker || !(obj.is_latest || obj.version_id.is_none_or(|v| v.is_nil())) { + return Event::default(); + } + + let Some(lc_rules) = self.filter_rules(obj).await else { + return Event::default(); + }; + + let mut event: Option = None; + for rule in lc_rules { + let Some(expiration) = rule.expiration else { + continue; + }; + if expiration.expired_object_delete_marker.is_some_and(|v| v) { + continue; + } + let Some(due) = (match (&expiration.date, expiration.days) { + (Some(date), _) => Some(OffsetDateTime::from(date.clone())), + (None, Some(days)) => Some(expected_expiry_time(mod_time, days)), + _ => None, + }) else { + continue; + }; + + let predicted = Event { + action: IlmAction::DeleteAction, + rule_id: rule.id.clone().unwrap_or_default(), + due: Some(due), + noncurrent_days: 0, + newer_noncurrent_versions: 0, + storage_class: "".into(), + }; + let predicted_due = predicted.due.unwrap_or_else(|| OffsetDateTime::UNIX_EPOCH).unix_timestamp(); + let should_replace = event + .as_ref() + .is_none_or(|e| predicted_due < e.due.unwrap_or_else(|| OffsetDateTime::UNIX_EPOCH).unix_timestamp()); + if should_replace { + event = Some(predicted); + } + } + + event.unwrap_or_default() + } + async fn eval_inner(&self, obj: &ObjectOpts, now: OffsetDateTime, newer_noncurrent_versions: usize) -> Event { let mut events = Vec::::new(); info!( @@ -833,4 +911,167 @@ mod tests { .await .expect("expected validation to pass"); } + + #[tokio::test] + async fn validate_rejects_non_midnight_expiration_date() { + let lc = BucketLifecycleConfiguration { + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + date: Some(time::OffsetDateTime::from_unix_timestamp(20_000_101).unwrap().into()), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: None, + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }; + + let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err(); + + assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_EXPIRATION_DATE_NOT_MIDNIGHT); + } + + #[tokio::test] + async fn predict_expiration_selects_closest_expiry_for_put_object() { + let base_time = OffsetDateTime::from_unix_timestamp(1_000_000).unwrap(); + let lc = BucketLifecycleConfiguration { + rules: vec![ + LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: Some("rule-days".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }, + LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + date: Some((base_time + Duration::days(1)).into()), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: Some("rule-date".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }, + ], + }; + let opts = ObjectOpts { + name: "obj".to_string(), + mod_time: Some(base_time), + is_latest: true, + version_id: None, + ..Default::default() + }; + + let event = lc.predict_expiration(&opts).await; + let expected = base_time + Duration::days(1); + + assert_eq!(event.action, IlmAction::DeleteAction); + assert_eq!(event.rule_id, "rule-date"); + assert_eq!(event.due, Some(expected)); + } + + #[tokio::test] + async fn validate_accepts_multiple_rules_without_ids() { + let lc = BucketLifecycleConfiguration { + rules: vec![ + LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: None, + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }, + LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(31), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: None, + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }, + ], + }; + + lc.validate(&ObjectLockConfiguration::default()) + .await + .expect("expected validation to pass"); + } + + #[tokio::test] + async fn validate_rejects_rule_id_too_long() { + let lc = BucketLifecycleConfiguration { + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: Some("a".repeat(256)), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }; + + let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err(); + + assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_RULE_ID_TOO_LONG); + } + + #[tokio::test] + async fn validate_rejects_invalid_status_case_sensitive() { + let lc = BucketLifecycleConfiguration { + rules: vec![LifecycleRule { + status: ExpirationStatus::from_static("enabled"), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: None, + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }], + }; + + let err = lc.validate(&ObjectLockConfiguration::default()).await.unwrap_err(); + + assert_eq!(err.to_string(), ERR_LIFECYCLE_INVALID_RULE_STATUS); + } } diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 84968f0e2..15f450a05 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -60,7 +60,7 @@ use s3s::dto::*; use s3s::region::Region; use s3s::xml; use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; -use std::{fmt::Display, sync::Arc}; +use std::{collections::HashSet, fmt::Display, sync::Arc}; use tracing::{debug, error, info, instrument, warn}; use urlencoding::encode; @@ -81,6 +81,45 @@ fn resolve_notification_region(global_region: Option, request_region: Op global_region.or(request_region).unwrap_or_else(default_region) } +const ERR_LIFECYCLE_RULE_STATUS: &str = "Rule status must be either Enabled or Disabled"; + +fn assign_lifecycle_rule_ids(rules: &mut [LifecycleRule]) { + let mut rule_ids: HashSet = HashSet::new(); + + for rule in rules.iter() { + if let Some(id) = rule.id.as_ref() { + rule_ids.insert(id.to_string()); + } + } + + for (idx, rule) in rules.iter_mut().enumerate() { + if rule.id.is_none() { + let mut suffix = 0usize; + let mut generated_id = format!("rule-{}", idx); + + while rule_ids.contains(&generated_id) { + suffix += 1; + generated_id = format!("rule-{idx}-{suffix}"); + } + + rule_ids.insert(generated_id.clone()); + rule.id = Some(generated_id); + } + } +} + +fn validate_lifecycle_rule_status(rules: &[LifecycleRule]) -> Result<(), &'static str> { + for rule in rules { + if rule.status != ExpirationStatus::from_static(ExpirationStatus::ENABLED) + && rule.status != ExpirationStatus::from_static(ExpirationStatus::DISABLED) + { + return Err(ERR_LIFECYCLE_RULE_STATUS); + } + } + + Ok(()) +} + #[derive(Clone, Default)] pub struct DefaultBucketUsecase { context: Option>, @@ -1080,7 +1119,11 @@ impl DefaultBucketUsecase { .. } = req.input; - let Some(input_cfg) = lifecycle_configuration else { return Err(s3_error!(InvalidArgument)) }; + let Some(mut input_cfg) = lifecycle_configuration else { return Err(s3_error!(InvalidArgument)) }; + assign_lifecycle_rule_ids(&mut input_cfg.rules); + if let Err(err) = validate_lifecycle_rule_status(&input_cfg.rules) { + return Err(S3Error::with_message(S3ErrorCode::MalformedXML, format!("Malformed XML: {err}"))); + } let rcfg = match metadata_sys::get_object_lock_config(&bucket).await { Ok((cfg, _)) => cfg, @@ -1912,6 +1955,80 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InternalError); } + #[test] + fn normalize_lifecycle_rules_generate_rule_ids_for_missing_values() { + let mut rules = vec![ + LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: None, + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }, + LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(60), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: Some("rule-1".to_string()), + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }, + LifecycleRule { + status: ExpirationStatus::from_static(ExpirationStatus::ENABLED), + expiration: Some(LifecycleExpiration { + days: Some(90), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: None, + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }, + ]; + + assign_lifecycle_rule_ids(&mut rules); + + assert_eq!(rules[0].id.as_deref(), Some("rule-0")); + assert_eq!(rules[1].id.as_deref(), Some("rule-1")); + assert_eq!(rules[2].id.as_deref(), Some("rule-2")); + } + + #[test] + fn validate_lifecycle_rule_status_rejects_invalid_status() { + let rules = vec![LifecycleRule { + status: ExpirationStatus::from_static("enabled"), + expiration: Some(LifecycleExpiration { + days: Some(30), + ..Default::default() + }), + abort_incomplete_multipart_upload: None, + filter: None, + id: None, + noncurrent_version_expiration: None, + noncurrent_version_transitions: None, + prefix: None, + transitions: None, + }]; + + assert_eq!(validate_lifecycle_rule_status(&rules).unwrap_err(), ERR_LIFECYCLE_RULE_STATUS); + } + #[tokio::test] async fn execute_list_buckets_returns_internal_error_when_store_uninitialized() { let input = ListBucketsInput::builder().build().unwrap(); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 554f9ead3..38c2ed8f9 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -42,7 +42,7 @@ use rustfs_ecstore::bucket::quota::checker::QuotaChecker; use rustfs_ecstore::bucket::{ lifecycle::{ bucket_lifecycle_ops::{RestoreRequestOps, post_restore_opts}, - lifecycle::{self, TransitionOptions}, + lifecycle::{self, Lifecycle, TransitionOptions}, }, metadata::{BUCKET_VERSIONING_CONFIG, OBJECT_LOCK_CONFIG}, metadata_sys, @@ -172,6 +172,36 @@ fn normalize_delete_objects_version_id(version_id: Option) -> Result<(Op } } +fn build_put_object_expiration_header(event: &lifecycle::Event) -> Option { + if !event.action.delete() { + return None; + } + + let expire_time = event.due?; + + if event.rule_id.is_empty() || expire_time == OffsetDateTime::UNIX_EPOCH { + return None; + } + + let expiry_date = expire_time.format(&Rfc3339).ok()?; + Some(format!("expiry-date=\"{}\", rule-id=\"{}\"", expiry_date, event.rule_id)) +} + +async fn resolve_put_object_expiration(bucket: &str, obj_info: &ObjectInfo) -> Option { + let Ok((lifecycle_config, _)) = metadata_sys::get_lifecycle_config(bucket).await else { + debug!("resolve_put_object_expiration: lifecycle config not found for bucket {bucket}"); + return None; + }; + + let obj_opts = lifecycle::ObjectOpts::from_object_info(obj_info); + let event = lifecycle_config.predict_expiration(&obj_opts).await; + debug!( + "resolve_put_object_expiration: bucket={bucket}, action={:?}, rule_id={}, due={:?}", + event.action, event.rule_id, event.due + ); + build_put_object_expiration_header(&event) +} + #[derive(Clone, Default)] pub struct DefaultObjectUsecase { context: Option>, @@ -516,9 +546,10 @@ impl DefaultObjectUsecase { get_must_replicate_options(&mt2, "".to_string(), ReplicationStatusType::Empty, ReplicationType::Object, opts); let dsc = must_replicate(&bucket, &key, repoptions).await; + let expiration = resolve_put_object_expiration(&bucket, &obj_info).await; if dsc.replicate_any() { - schedule_replication(obj_info, store, dsc, ReplicationType::Object).await; + schedule_replication(obj_info.clone(), store, dsc, ReplicationType::Object).await; } let mut checksums = PutObjectChecksums { @@ -540,6 +571,7 @@ impl DefaultObjectUsecase { sse_customer_algorithm: sse_customer_algorithm.clone(), sse_customer_key_md5: sse_customer_key_md5.clone(), ssekms_key_id: effective_kms_key_id, + expiration, checksum_crc32: checksums.crc32, checksum_crc32c: checksums.crc32c, checksum_sha1: checksums.sha1, @@ -3832,6 +3864,62 @@ mod tests { assert!(!object_attributes_requested(&object_attributes, ObjectAttributes::OBJECT_SIZE)); } + #[test] + fn build_put_object_expiration_header_returns_none_for_non_delete_events() { + let event = lifecycle::Event { + action: lifecycle::IlmAction::TransitionAction, + rule_id: "rule-1".to_string(), + due: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap()), + noncurrent_days: 0, + newer_noncurrent_versions: 0, + storage_class: String::new(), + }; + + assert!(build_put_object_expiration_header(&event).is_none()); + } + + #[test] + fn build_put_object_expiration_header_formats_expected_value() { + let expire_time = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(); + let event = lifecycle::Event { + action: lifecycle::IlmAction::DeleteAction, + rule_id: "rule-1".to_string(), + due: Some(expire_time), + noncurrent_days: 0, + newer_noncurrent_versions: 0, + storage_class: String::new(), + }; + + let expiry_date = expire_time.format(&Rfc3339).unwrap(); + let expected = format!("expiry-date=\"{}\", rule-id=\"rule-1\"", expiry_date); + assert_eq!(build_put_object_expiration_header(&event), Some(expected)); + } + + #[test] + fn build_put_object_expiration_header_requires_rule_id_and_due_time() { + let event = lifecycle::Event { + action: lifecycle::IlmAction::DeleteAction, + rule_id: String::new(), + due: Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap()), + noncurrent_days: 0, + newer_noncurrent_versions: 0, + storage_class: String::new(), + }; + + assert!(build_put_object_expiration_header(&event).is_none()); + + let event = lifecycle::Event { + action: lifecycle::IlmAction::DeleteAction, + rule_id: "rule-1".to_string(), + due: Some(OffsetDateTime::UNIX_EPOCH), + noncurrent_days: 0, + newer_noncurrent_versions: 0, + storage_class: String::new(), + }; + + assert!(build_put_object_expiration_header(&event).is_none()); + } + #[tokio::test] async fn execute_get_object_legal_hold_returns_internal_error_when_store_uninitialized() { let input = GetObjectLegalHoldInput::builder()