diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 9deaec6d3..72aa5b037 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -2446,6 +2446,7 @@ pub struct ManualTransitionRunOptions { } #[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] pub struct ManualTransitionRunReport { pub bucket: String, pub prefix: String, @@ -2480,6 +2481,19 @@ pub struct ManualTransitionRunExecution { pub cancelled: bool, } +impl ManualTransitionRunExecution { + fn completed(report: ManualTransitionRunReport) -> Self { + Self { + report, + cancelled: false, + } + } + + fn cancelled(report: ManualTransitionRunReport) -> Self { + Self { report, cancelled: true } + } +} + impl ManualTransitionRunReport { fn new(bucket: &str, options: &ManualTransitionRunOptions) -> Self { Self { @@ -2543,10 +2557,7 @@ pub async fn enqueue_transition_for_existing_objects_scoped_with_cancel( let mut report = ManualTransitionRunReport::new(bucket, &options); let Some(lc) = runtime_sources::bucket_lifecycle_config(bucket).await else { - return Ok(ManualTransitionRunExecution { - report, - cancelled: false, - }); + return Ok(ManualTransitionRunExecution::completed(report)); }; report.lifecycle_config_found = true; let mut marker = options.marker.clone(); @@ -2558,7 +2569,7 @@ pub async fn enqueue_transition_for_existing_objects_scoped_with_cancel( loop { if manual_transition_cancelled(cancel_token.as_ref()) { - return Ok(ManualTransitionRunExecution { report, cancelled: true }); + return Ok(ManualTransitionRunExecution::cancelled(report)); } let page = api @@ -2570,26 +2581,20 @@ pub async fn enqueue_transition_for_existing_objects_scoped_with_cancel( if manual_transition_cancelled(cancel_token.as_ref()) { report.next_marker.clone_from(&previous_marker); report.next_version_idmarker.clone_from(&previous_version_marker); - return Ok(ManualTransitionRunExecution { report, cancelled: true }); + return Ok(ManualTransitionRunExecution::cancelled(report)); } if manual_transition_duration_elapsed(deadline) { report.truncated_by_duration = true; report.next_marker.clone_from(&previous_marker); report.next_version_idmarker.clone_from(&previous_version_marker); - return Ok(ManualTransitionRunExecution { - report, - cancelled: false, - }); + return Ok(ManualTransitionRunExecution::completed(report)); } report.scanned = report.scanned.saturating_add(1); enqueue_transition_with_lifecycle_report(object, &lc, &src, &options, &mut report).await; if report.has_partial_enqueue() { report.next_marker.clone_from(&previous_marker); report.next_version_idmarker.clone_from(&previous_version_marker); - return Ok(ManualTransitionRunExecution { - report, - cancelled: false, - }); + return Ok(ManualTransitionRunExecution::completed(report)); } if options.max_objects.is_some_and(|max_objects| report.scanned >= max_objects) { if manual_transition_has_more_after_limit(index, page.objects.len(), page.is_truncated) { @@ -2597,29 +2602,20 @@ pub async fn enqueue_transition_for_existing_objects_scoped_with_cancel( report.next_marker = Some(object.name.clone()); report.next_version_idmarker = Some(manual_transition_version_marker(object)); } - return Ok(ManualTransitionRunExecution { - report, - cancelled: false, - }); + return Ok(ManualTransitionRunExecution::completed(report)); } previous_marker = Some(object.name.clone()); previous_version_marker = Some(manual_transition_version_marker(object)); } if !page.is_truncated { - return Ok(ManualTransitionRunExecution { - report, - cancelled: false, - }); + return Ok(ManualTransitionRunExecution::completed(report)); } if manual_transition_duration_elapsed(deadline) { report.truncated_by_duration = true; report.next_marker.clone_from(&previous_marker); report.next_version_idmarker.clone_from(&previous_version_marker); - return Ok(ManualTransitionRunExecution { - report, - cancelled: false, - }); + return Ok(ManualTransitionRunExecution::completed(report)); } marker = page.next_marker; diff --git a/rustfs/src/admin/handlers/ilm_transition.rs b/rustfs/src/admin/handlers/ilm_transition.rs index 2ea82fc7d..a63de1437 100644 --- a/rustfs/src/admin/handlers/ilm_transition.rs +++ b/rustfs/src/admin/handlers/ilm_transition.rs @@ -454,7 +454,10 @@ fn manual_transition_job_config_path(job_id: &str) -> String { } fn parse_manual_transition_job_id(job_id: &str) -> S3Result { - Uuid::parse_str(job_id).map_err(|_| s3_error!(InvalidArgument, "invalid manual transition job id"))?; + let job_id = Uuid::parse_str(job_id).map_err(|_| s3_error!(InvalidArgument, "invalid manual transition job id"))?; + if job_id.is_nil() { + return Err(s3_error!(InvalidArgument, "invalid manual transition job id")); + } Ok(job_id.to_string()) } @@ -555,9 +558,7 @@ async fn save_manual_transition_job_record(store: Arc, record: &ManualT async fn read_manual_transition_job_record(store: Arc, job_id: &str) -> S3Result> { match read_admin_config(store, &manual_transition_job_config_path(job_id)).await { - Ok(data) => serde_json::from_slice(&data).map(Some).map_err(|err| { - S3Error::with_message(S3ErrorCode::InternalError, format!("failed to decode manual transition job: {err}")) - }), + Ok(data) => decode_manual_transition_job_record(job_id, &data).map(Some), Err(StorageError::ConfigNotFound) => Ok(None), Err(err) => Err(S3Error::with_message( S3ErrorCode::InternalError, @@ -566,6 +567,32 @@ async fn read_manual_transition_job_record(store: Arc, job_id: &str) -> } } +fn decode_manual_transition_job_record(job_id: &str, data: &[u8]) -> S3Result { + let mut record: ManualTransitionJobRecord = serde_json::from_slice(data).map_err(|err| { + S3Error::with_message(S3ErrorCode::InternalError, format!("failed to decode manual transition job: {err}")) + })?; + if record.schema_version != MANUAL_TRANSITION_JOB_SCHEMA_VERSION { + return Err(S3Error::with_message( + S3ErrorCode::InternalError, + "unsupported manual transition job schema version", + )); + } + if record.job_id != job_id { + return Err(S3Error::with_message( + S3ErrorCode::InternalError, + "manual transition job id does not match persisted path", + )); + } + record.status_endpoint = manual_transition_job_status_endpoint(job_id); + Ok(record) +} + +fn mark_manual_transition_job_unknown(record: &mut ManualTransitionJobRecord) { + record.status = ManualTransitionJobStatus::Unknown; + record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc())); + record.failure_reason = Some("manual transition job owner is not active on this node".to_string()); +} + async fn load_manual_transition_job_record(store: Arc, job_id: &str) -> S3Result { if let Some(record) = in_memory_manual_transition_job_record(job_id) { return Ok(record); @@ -574,9 +601,7 @@ async fn load_manual_transition_job_record(store: Arc, job_id: &str) -> return Err(s3_error!(NoSuchKey, "manual transition job not found")); }; if !record.status.is_terminal() { - record.status = ManualTransitionJobStatus::Unknown; - record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc())); - record.failure_reason = Some("manual transition job owner is not active on this node".to_string()); + mark_manual_transition_job_unknown(&mut record); } Ok(record) } @@ -762,10 +787,8 @@ impl Operation for ManualTransitionJobCancelHandler { return Err(s3_error!(NoSuchKey, "manual transition job not found")); }; if !record.status.is_terminal() { - record.status = ManualTransitionJobStatus::Unknown; + mark_manual_transition_job_unknown(&mut record); record.cancel_requested = true; - record.finished_at = Some(manual_transition_timestamp(OffsetDateTime::now_utc())); - record.failure_reason = Some("manual transition job owner is not active on this node".to_string()); save_manual_transition_job_record(store, &record).await?; } record @@ -1050,6 +1073,80 @@ mod tests { assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); } + #[test] + fn manual_transition_job_id_rejects_nil_uuid() { + let err = + parse_manual_transition_job_id("00000000-0000-0000-0000-000000000000").expect_err("nil job id must be rejected"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidArgument); + } + + #[test] + fn manual_transition_job_id_is_canonicalized_for_storage_paths() { + let job_id = parse_manual_transition_job_id("11111111111141118111111111111111").expect("compact UUID should parse"); + + assert_eq!(job_id, "11111111-1111-4111-8111-111111111111"); + } + + #[test] + fn manual_transition_job_decode_rejects_mismatched_persisted_id() { + let record = ManualTransitionJobRecord { + job_id: "22222222-2222-4222-8222-222222222222".to_string(), + status_endpoint: manual_transition_job_status_endpoint("22222222-2222-4222-8222-222222222222"), + ..new_manual_transition_job_record( + "11111111-1111-4111-8111-111111111111".to_string(), + "data", + &ManualTransitionRunOptions::default(), + OffsetDateTime::UNIX_EPOCH, + ) + }; + let data = serde_json::to_vec(&record).expect("job record should serialize"); + + let err = decode_manual_transition_job_record("11111111-1111-4111-8111-111111111111", &data) + .expect_err("persisted job id mismatch must fail closed"); + + assert_eq!(err.code(), &S3ErrorCode::InternalError); + } + + #[test] + fn manual_transition_job_decode_rejects_unsupported_schema_version() { + let mut record = new_manual_transition_job_record( + "11111111-1111-4111-8111-111111111111".to_string(), + "data", + &ManualTransitionRunOptions::default(), + OffsetDateTime::UNIX_EPOCH, + ); + record.schema_version = MANUAL_TRANSITION_JOB_SCHEMA_VERSION + 1; + let data = serde_json::to_vec(&record).expect("job record should serialize"); + + let err = decode_manual_transition_job_record("11111111-1111-4111-8111-111111111111", &data) + .expect_err("unsupported schema must fail closed"); + + assert_eq!(err.code(), &S3ErrorCode::InternalError); + } + + #[test] + fn manual_transition_job_decode_rejects_unknown_report_fields() { + let record = new_manual_transition_job_record( + "11111111-1111-4111-8111-111111111111".to_string(), + "data", + &ManualTransitionRunOptions::default(), + OffsetDateTime::UNIX_EPOCH, + ); + let mut value = serde_json::to_value(record).expect("job record should serialize"); + value + .pointer_mut("/report") + .and_then(serde_json::Value::as_object_mut) + .expect("report should be a JSON object") + .insert("unexpected_counter".to_string(), serde_json::json!(1)); + let data = serde_json::to_vec(&value).expect("job record JSON should encode"); + + let err = decode_manual_transition_job_record("11111111-1111-4111-8111-111111111111", &data) + .expect_err("unknown report fields must fail closed"); + + assert_eq!(err.code(), &S3ErrorCode::InternalError); + } + #[test] fn manual_transition_cancel_marks_active_job_and_token() { let (bucket, options, _run_mode) =