fix(ilm): validate persisted transition jobs

Refs #1479

Co-Authored-By: heihutu <heihutu@gmail.com>

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;

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<String> {
-    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<ECStore>, record: &ManualT
 
 async fn read_manual_transition_job_record(store: Arc<ECStore>, job_id: &str) -> S3Result<Option<ManualTransitionJobRecord>> {
     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<ECStore>, job_id: &str) ->
     }
 }
 
+fn decode_manual_transition_job_record(job_id: &str, data: &[u8]) -> S3Result<ManualTransitionJobRecord> {
+    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<ECStore>, job_id: &str) -> S3Result<ManualTransitionJobRecord> {
     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<ECStore>, 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) =