feat(kms): bulk DEK rekey sweep with admin API and kms:Rekey action (#6654)

2026-08-30 00:47:13 +00:00 · 2026-08-26 18:19:28 +08:00
parent ff47714363
commit 286626c1bd
16 changed files with 823 additions and 8 deletions
@@ -14,7 +14,7 @@

 //! KMS admin handlers for HTTP API

-use super::{kms_backup, kms_dynamic, kms_key_lifecycle, kms_key_metadata, kms_keys, kms_management};
+use super::{kms_backup, kms_dynamic, kms_key_lifecycle, kms_key_metadata, kms_keys, kms_management, kms_rekey};
 use crate::admin::router::{AdminOperation, S3Router};

 pub fn register_kms_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
@@ -24,5 +24,6 @@ pub fn register_kms_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<(
    kms_key_lifecycle::register_kms_key_lifecycle_route(r)?;
    kms_backup::register_kms_backup_route(r)?;
    kms_key_metadata::register_kms_key_metadata_route(r)?;
+    kms_rekey::register_kms_rekey_route(r)?;
    Ok(())
 }
@@ -0,0 +1,209 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Admin API for the bulk DEK rekey sweep: start, status and cancel.
+//!
+//! All three endpoints require `kms:Rekey`, a cluster-scoped action: the sweep
+//! walks and rewrites object metadata across buckets, so no per-key role
+//! template confers it.
+
+use crate::admin::auth::authorize_admin_request;
+use crate::admin::router::{AdminOperation, Operation, S3Router};
+use crate::admin::runtime_sources::{current_kms_runtime_service_manager, current_object_store_handle};
+use crate::admin::storage_api::s3::{self, Body, S3ErrorCode, S3Request, S3Response, S3Result};
+use crate::kms_rekey::{self, RekeyStartError};
+use crate::server::ADMIN_PREFIX;
+use hyper::{HeaderMap, Method, StatusCode};
+use matchit::Params;
+use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
+use rustfs_policy::policy::action::{Action, KmsAction};
+use serde::Deserialize;
+
+fn kms_rekey_actions() -> Vec<Action> {
+    vec![Action::KmsAction(KmsAction::RekeyAction)]
+}
+
+async fn authorize_kms_rekey_request(req: &S3Request<Body>) -> S3Result<()> {
+    if req.credentials.is_none() {
+        return Err(s3::error(S3ErrorCode::InvalidRequest, "authentication required"));
+    }
+    authorize_admin_request(req, kms_rekey_actions()).await?;
+    Ok(())
+}
+
+fn json_response(status: StatusCode, body: Vec<u8>) -> S3Response<(StatusCode, Body)> {
+    let mut headers = HeaderMap::new();
+    headers.insert(s3::header::CONTENT_TYPE, "application/json".parse().expect("static content type"));
+    S3Response::with_headers((status, Body::from(body)), headers)
+}
+
+fn snapshot_response(status: StatusCode, snapshot: &kms_rekey::RekeyJobSnapshot) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let body = serde_json::to_vec(snapshot)
+        .map_err(|e| s3::error(S3ErrorCode::InternalError, format!("failed to serialize rekey status: {e}")))?;
+    Ok(json_response(status, body))
+}
+
+/// Body of `POST /v3/kms/keys/rekey`. An empty body sweeps every bucket.
+#[derive(Debug, Default, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct StartKmsRekeyRequest {
+    /// Buckets to sweep; every bucket when absent or empty.
+    #[serde(default)]
+    buckets: Option<Vec<String>>,
+    /// Object key prefix to restrict the sweep to.
+    #[serde(default)]
+    prefix: Option<String>,
+}
+
+/// `POST /v3/kms/keys/rekey` — start a sweep.
+pub struct StartKmsRekeyHandler;
+
+#[async_trait::async_trait]
+impl Operation for StartKmsRekeyHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_kms_rekey_request(&req).await?;
+
+        let mut req = req;
+        let body = req
+            .input
+            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
+            .await
+            .map_err(|e| s3::error(S3ErrorCode::InvalidRequest, format!("failed to read request body: {e}")))?;
+        let request: StartKmsRekeyRequest = if body.is_empty() {
+            StartKmsRekeyRequest::default()
+        } else {
+            serde_json::from_slice(&body)
+                .map_err(|e| s3::error(S3ErrorCode::InvalidRequest, format!("invalid rekey request body: {e}")))?
+        };
+
+        // Refuse up front when the backend cannot rewrap at all: a sweep would
+        // count every encrypted object as failed while changing nothing.
+        // `UnsupportedCapability` is a permanent gap of the configured backend,
+        // so it surfaces as 501, mirroring the key lifecycle handlers.
+        let Some(manager) = current_kms_runtime_service_manager() else {
+            return Err(s3::error(S3ErrorCode::InternalError, "KMS service not initialized"));
+        };
+        let Some(service) = manager.get_encryption_service().await else {
+            return Err(s3::error(S3ErrorCode::InternalError, "KMS service not running"));
+        };
+        if !service.backend_capabilities().rewrap {
+            return Ok(json_response(
+                StatusCode::NOT_IMPLEMENTED,
+                serde_json::json!({
+                    "error": "the configured KMS backend does not support rewrapping data-key envelopes"
+                })
+                .to_string()
+                .into_bytes(),
+            ));
+        }
+
+        let Some(store) = current_object_store_handle() else {
+            return Err(s3::error(S3ErrorCode::InternalError, "object store is not ready"));
+        };
+
+        match kms_rekey::start(store, request.buckets, request.prefix.unwrap_or_default()).await {
+            Ok(snapshot) => snapshot_response(StatusCode::OK, &snapshot),
+            Err(RekeyStartError::AlreadyRunning(job_id)) => Ok(json_response(
+                StatusCode::CONFLICT,
+                serde_json::json!({ "error": "a rekey sweep is already running", "job_id": job_id })
+                    .to_string()
+                    .into_bytes(),
+            )),
+            Err(RekeyStartError::Storage(message)) => Err(s3::error(S3ErrorCode::InternalError, message)),
+        }
+    }
+}
+
+/// `GET /v3/kms/keys/rekey/status` — progress of the current or last sweep.
+pub struct KmsRekeyStatusHandler;
+
+#[async_trait::async_trait]
+impl Operation for KmsRekeyStatusHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_kms_rekey_request(&req).await?;
+
+        match kms_rekey::status() {
+            Some(snapshot) => snapshot_response(StatusCode::OK, &snapshot),
+            None => Ok(json_response(
+                StatusCode::NOT_FOUND,
+                serde_json::json!({ "error": "no rekey sweep has run" })
+                    .to_string()
+                    .into_bytes(),
+            )),
+        }
+    }
+}
+
+/// `POST /v3/kms/keys/rekey/cancel` — request cancellation of a running sweep.
+pub struct CancelKmsRekeyHandler;
+
+#[async_trait::async_trait]
+impl Operation for CancelKmsRekeyHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_kms_rekey_request(&req).await?;
+
+        match kms_rekey::cancel() {
+            Some(snapshot) => snapshot_response(StatusCode::OK, &snapshot),
+            None => Ok(json_response(
+                StatusCode::NOT_FOUND,
+                serde_json::json!({ "error": "no rekey sweep has run" })
+                    .to_string()
+                    .into_bytes(),
+            )),
+        }
+    }
+}
+
+pub fn register_kms_rekey_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    r.insert(
+        Method::POST,
+        format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/rekey").as_str(),
+        AdminOperation(&StartKmsRekeyHandler {}),
+    )?;
+
+    r.insert(
+        Method::GET,
+        format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/rekey/status").as_str(),
+        AdminOperation(&KmsRekeyStatusHandler {}),
+    )?;
+
+    r.insert(
+        Method::POST,
+        format!("{}{}", ADMIN_PREFIX, "/v3/kms/keys/rekey/cancel").as_str(),
+        AdminOperation(&CancelKmsRekeyHandler {}),
+    )?;
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn rekey_endpoints_require_the_cluster_scoped_action() {
+        assert_eq!(kms_rekey_actions(), vec![Action::KmsAction(KmsAction::RekeyAction)]);
+    }
+
+    #[test]
+    fn start_request_parses_and_rejects_unknown_fields() {
+        let parsed: StartKmsRekeyRequest =
+            serde_json::from_slice(br#"{"buckets": ["a", "b"], "prefix": "photos/"}"#).expect("valid body must parse");
+        assert_eq!(parsed.buckets.as_deref(), Some(["a".to_string(), "b".to_string()].as_slice()));
+        assert_eq!(parsed.prefix.as_deref(), Some("photos/"));
+
+        serde_json::from_slice::<StartKmsRekeyRequest>(br#"{"bucket": "typo"}"#)
+            .expect_err("unknown fields must be rejected, not silently ignored");
+    }
+}
@@ -41,6 +41,7 @@ pub mod kms_key_lifecycle;
 pub mod kms_key_metadata;
 pub mod kms_keys;
 pub mod kms_management;
+pub mod kms_rekey;
 pub mod metrics;
 pub mod mfa;
 pub mod module_switch;
@@ -65,6 +65,7 @@ const KMS_DISABLE_KEY: AdminActionRef = AdminActionRef::new("kms:DisableKey");
 const KMS_ENABLE_KEY: AdminActionRef = AdminActionRef::new("kms:EnableKey");
 const KMS_GENERATE_DATA_KEY: AdminActionRef = AdminActionRef::new("kms:GenerateDataKey");
 const KMS_LIST_KEYS: AdminActionRef = AdminActionRef::new("kms:ListKeys");
+const KMS_REKEY: AdminActionRef = AdminActionRef::new("kms:Rekey");
 const KMS_RESTORE: AdminActionRef = AdminActionRef::new("kms:Restore");
 const KMS_ROTATE_KEY: AdminActionRef = AdminActionRef::new("kms:RotateKey");
 const KMS_SERVICE_CONTROL: AdminActionRef = AdminActionRef::new("kms:ServiceControl");
@@ -821,6 +822,21 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
        RouteRiskLevel::High,
    ),
    admin(HttpMethod::Post, "/rustfs/admin/v3/kms/keys/rotate", KMS_ROTATE_KEY, RouteRiskLevel::High),
+    // The rekey sweep walks and rewrites object metadata across buckets, so it
+    // carries its own cluster-scoped action rather than reusing a per-key one.
+    admin(HttpMethod::Post, "/rustfs/admin/v3/kms/keys/rekey", KMS_REKEY, RouteRiskLevel::High),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/kms/keys/rekey/status",
+        KMS_REKEY,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/kms/keys/rekey/cancel",
+        KMS_REKEY,
+        RouteRiskLevel::High,
+    ),
    // Backup and restore act on the material of every key at once, so they
    // carry their own actions rather than reusing any per-key one.
    admin(HttpMethod::Get, "/rustfs/admin/v3/kms/backup", KMS_BACKUP, RouteRiskLevel::Sensitive),
@@ -357,6 +357,9 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
        admin_route(Method::POST, "/v3/kms/keys/enable"),
        admin_route(Method::POST, "/v3/kms/keys/disable"),
        admin_route(Method::POST, "/v3/kms/keys/rotate"),
+        admin_route(Method::POST, "/v3/kms/keys/rekey"),
+        admin_route(Method::GET, "/v3/kms/keys/rekey/status"),
+        admin_route(Method::POST, "/v3/kms/keys/rekey/cancel"),
        admin_route(Method::GET, "/v3/kms/backup"),
        admin_route(Method::POST, "/v3/kms/backup"),
        admin_route(Method::POST, "/v3/kms/restore/dry-run"),
@@ -39,6 +39,12 @@ expression: kms_route_contract()
    "action": "kms:ListKeys",
    "risk": "Sensitive"
  },
+  {
+    "method": "GET",
+    "path": "/rustfs/admin/v3/kms/keys/rekey/status",
+    "action": "kms:Rekey",
+    "risk": "Sensitive"
+  },
  {
    "method": "GET",
    "path": "/rustfs/admin/v3/kms/keys/{key_id}",
@@ -123,6 +129,18 @@ expression: kms_route_contract()
    "action": "kms:EnableKey",
    "risk": "High"
  },
+  {
+    "method": "POST",
+    "path": "/rustfs/admin/v3/kms/keys/rekey",
+    "action": "kms:Rekey",
+    "risk": "High"
+  },
+  {
+    "method": "POST",
+    "path": "/rustfs/admin/v3/kms/keys/rekey/cancel",
+    "action": "kms:Rekey",
+    "risk": "High"
+  },
  {
    "method": "POST",
    "path": "/rustfs/admin/v3/kms/keys/rotate",
@@ -0,0 +1,354 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Bulk DEK rekey sweep: walk stored objects and rewrap their KMS-wrapped
+//! data keys onto current master key versions, touching only `xl.meta`.
+//!
+//! The sweep is convergent rather than checkpointed: every step is idempotent
+//! (an already-current envelope costs one describe-shaped KMS call and no
+//! write), so recovery from a crash, a cancel, or partial failure is simply
+//! running the sweep again. Failures are counted and logged per object, never
+//! silently dropped, and never abort the sweep — a sweep that stopped at the
+//! first unreadable object would hide every object behind it.
+//!
+//! One sweep runs at a time. Concurrent sweeps would double every KMS
+//! round-trip for zero extra coverage and interleave their metadata writes.
+//!
+//! Cross-site note: replication strips encryption metadata in transit, so a
+//! rewrap here never propagates to a replica site — each site runs its own
+//! sweep.
+
+use crate::storage_api::kms::contract::bucket::{BucketOperations, BucketOptions};
+use crate::storage_api::kms::contract::list::ListOperations;
+use crate::storage_api::kms::contract::object::ObjectOperations;
+use crate::storage_api::kms::{ECStore, ObjectDekRewrapOutcome, StorageObjectOptions, rewrap_object_encryption_metadata};
+use serde::Serialize;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, LazyLock, Mutex};
+use tokio_util::sync::CancellationToken;
+use tracing::{info, warn};
+
+const LOG_COMPONENT: &str = "kms";
+const LOG_SUBSYSTEM: &str = "rekey";
+
+/// Terminal and non-terminal states of a sweep.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub(crate) enum RekeyJobState {
+    Running,
+    Completed,
+    Cancelled,
+}
+
+/// Wire shape of the status response; also the internal progress record.
+#[derive(Debug, Clone, Serialize)]
+pub(crate) struct RekeyJobSnapshot {
+    pub job_id: String,
+    pub state: RekeyJobState,
+    /// Buckets the sweep covers, in sweep order.
+    pub buckets: Vec<String>,
+    /// Bucket currently being walked; `None` before the first and after the last.
+    pub current_bucket: Option<String>,
+    /// Object versions the walk yielded.
+    pub scanned: u64,
+    /// Envelopes rewrapped and persisted.
+    pub rewrapped: u64,
+    /// Envelopes already on the current version and format.
+    pub already_current: u64,
+    /// Versions with nothing to rewrap (plaintext, SSE-C, MinIO-sealed).
+    pub not_applicable: u64,
+    /// Versions whose rewrap or metadata write failed; details are in the log.
+    /// A re-run retries exactly these, because everything else converges to
+    /// `already_current`.
+    pub failed: u64,
+}
+
+struct RekeyJob {
+    id: String,
+    cancel: CancellationToken,
+    buckets: Vec<String>,
+    scanned: AtomicU64,
+    rewrapped: AtomicU64,
+    already_current: AtomicU64,
+    not_applicable: AtomicU64,
+    failed: AtomicU64,
+    /// `(state, current_bucket)` under one lock so a snapshot never pairs a
+    /// terminal state with a bucket still marked in progress.
+    progress: Mutex<(RekeyJobState, Option<String>)>,
+}
+
+impl RekeyJob {
+    fn snapshot(&self) -> RekeyJobSnapshot {
+        let (state, current_bucket) = self.progress.lock().expect("rekey progress lock").clone();
+        RekeyJobSnapshot {
+            job_id: self.id.clone(),
+            state,
+            buckets: self.buckets.clone(),
+            current_bucket,
+            scanned: self.scanned.load(Ordering::Relaxed),
+            rewrapped: self.rewrapped.load(Ordering::Relaxed),
+            already_current: self.already_current.load(Ordering::Relaxed),
+            not_applicable: self.not_applicable.load(Ordering::Relaxed),
+            failed: self.failed.load(Ordering::Relaxed),
+        }
+    }
+
+    fn set_progress(&self, state: RekeyJobState, current_bucket: Option<String>) {
+        *self.progress.lock().expect("rekey progress lock") = (state, current_bucket);
+    }
+}
+
+/// The single sweep slot. Holding the finished job (rather than clearing it)
+/// keeps the final counters queryable until the next sweep starts.
+static ACTIVE_JOB: LazyLock<Mutex<Option<Arc<RekeyJob>>>> = LazyLock::new(|| Mutex::new(None));
+
+/// Why a sweep could not be started.
+#[derive(Debug)]
+pub(crate) enum RekeyStartError {
+    /// A sweep is already running; its id is carried for the error message.
+    AlreadyRunning(String),
+    /// The requested bucket list could not be resolved.
+    Storage(String),
+}
+
+/// Start a sweep over `buckets` (all buckets when `None`) under `prefix`.
+///
+/// Returns the job id. The sweep runs on a background task; progress and the
+/// terminal state are read through [`status`].
+pub(crate) async fn start(
+    store: Arc<ECStore>,
+    buckets: Option<Vec<String>>,
+    prefix: String,
+) -> Result<RekeyJobSnapshot, RekeyStartError> {
+    let buckets = match buckets {
+        Some(buckets) if !buckets.is_empty() => buckets,
+        _ => store
+            .list_bucket(&BucketOptions::default())
+            .await
+            .map_err(|e| RekeyStartError::Storage(format!("failed to list buckets: {e}")))?
+            .into_iter()
+            .map(|bucket| bucket.name)
+            .collect(),
+    };
+
+    let job = {
+        let mut slot = ACTIVE_JOB.lock().expect("rekey job slot lock");
+        if let Some(existing) = slot.as_ref()
+            && existing.snapshot().state == RekeyJobState::Running
+        {
+            return Err(RekeyStartError::AlreadyRunning(existing.id.clone()));
+        }
+        let job = Arc::new(RekeyJob {
+            id: uuid::Uuid::new_v4().to_string(),
+            cancel: CancellationToken::new(),
+            buckets,
+            scanned: AtomicU64::new(0),
+            rewrapped: AtomicU64::new(0),
+            already_current: AtomicU64::new(0),
+            not_applicable: AtomicU64::new(0),
+            failed: AtomicU64::new(0),
+            progress: Mutex::new((RekeyJobState::Running, None)),
+        });
+        *slot = Some(job.clone());
+        job
+    };
+
+    info!(
+        component = LOG_COMPONENT,
+        subsystem = LOG_SUBSYSTEM,
+        event = "kms_rekey_started",
+        job_id = %job.id,
+        buckets = job.buckets.len(),
+        "KMS rekey sweep started"
+    );
+    let task_job = job.clone();
+    tokio::spawn(async move { run_sweep(store, task_job, prefix).await });
+
+    Ok(job.snapshot())
+}
+
+/// Progress of the current or most recent sweep, if any.
+pub(crate) fn status() -> Option<RekeyJobSnapshot> {
+    ACTIVE_JOB
+        .lock()
+        .expect("rekey job slot lock")
+        .as_ref()
+        .map(|job| job.snapshot())
+}
+
+/// Request cancellation of the running sweep. Returns the snapshot the caller
+/// can report, or `None` when no sweep exists.
+pub(crate) fn cancel() -> Option<RekeyJobSnapshot> {
+    let slot = ACTIVE_JOB.lock().expect("rekey job slot lock");
+    let job = slot.as_ref()?;
+    job.cancel.cancel();
+    Some(job.snapshot())
+}
+
+async fn run_sweep(store: Arc<ECStore>, job: Arc<RekeyJob>, prefix: String) {
+    for bucket in job.buckets.clone() {
+        if job.cancel.is_cancelled() {
+            break;
+        }
+        job.set_progress(RekeyJobState::Running, Some(bucket.clone()));
+        sweep_bucket(&store, &job, &bucket, &prefix).await;
+    }
+
+    let state = if job.cancel.is_cancelled() {
+        RekeyJobState::Cancelled
+    } else {
+        RekeyJobState::Completed
+    };
+    job.set_progress(state, None);
+    let snapshot = job.snapshot();
+    info!(
+        component = LOG_COMPONENT,
+        subsystem = LOG_SUBSYSTEM,
+        event = "kms_rekey_finished",
+        job_id = %job.id,
+        state = ?snapshot.state,
+        scanned = snapshot.scanned,
+        rewrapped = snapshot.rewrapped,
+        already_current = snapshot.already_current,
+        not_applicable = snapshot.not_applicable,
+        failed = snapshot.failed,
+        "KMS rekey sweep finished"
+    );
+}
+
+async fn sweep_bucket(store: &Arc<ECStore>, job: &Arc<RekeyJob>, bucket: &str, prefix: &str) {
+    let (tx, mut rx) = tokio::sync::mpsc::channel(256);
+    let walk_cancel = job.cancel.child_token();
+    // Every version's envelope needs rewrapping, not just the latest.
+    type WalkOptionsOf = <ECStore as ListOperations>::WalkOptions;
+    let walk_options = WalkOptionsOf {
+        latest_only: false,
+        ..Default::default()
+    };
+    let walk_store = store.clone();
+    let walk_bucket = bucket.to_string();
+    let walk_prefix = prefix.to_string();
+    let walk_task = tokio::spawn(async move {
+        walk_store
+            .walk(walk_cancel, &walk_bucket, &walk_prefix, tx, walk_options)
+            .await
+    });
+
+    while let Some(entry) = rx.recv().await {
+        if job.cancel.is_cancelled() {
+            break;
+        }
+        if let Some(error) = entry.err {
+            job.failed.fetch_add(1, Ordering::Relaxed);
+            warn!(
+                component = LOG_COMPONENT,
+                subsystem = LOG_SUBSYSTEM,
+                event = "kms_rekey_walk_entry_failed",
+                job_id = %job.id,
+                bucket,
+                error = %error,
+                "KMS rekey sweep could not list an object"
+            );
+            continue;
+        }
+        let Some(object_info) = entry.item else {
+            continue;
+        };
+        if object_info.delete_marker || object_info.is_dir {
+            continue;
+        }
+        job.scanned.fetch_add(1, Ordering::Relaxed);
+
+        let object = object_info.name.clone();
+        match rewrap_object_encryption_metadata(bucket, &object, object_info.user_defined.as_ref()).await {
+            Ok(ObjectDekRewrapOutcome::NotApplicable) => {
+                job.not_applicable.fetch_add(1, Ordering::Relaxed);
+            }
+            Ok(ObjectDekRewrapOutcome::AlreadyCurrent) => {
+                job.already_current.fetch_add(1, Ordering::Relaxed);
+            }
+            Ok(ObjectDekRewrapOutcome::Rewrapped { metadata }) => {
+                let options = StorageObjectOptions {
+                    version_id: object_info.version_id.map(|version| version.to_string()),
+                    eval_metadata: Some(metadata),
+                    ..Default::default()
+                };
+                match store.put_object_metadata(bucket, &object, &options).await {
+                    Ok(_) => {
+                        job.rewrapped.fetch_add(1, Ordering::Relaxed);
+                    }
+                    Err(error) => {
+                        job.failed.fetch_add(1, Ordering::Relaxed);
+                        warn!(
+                            component = LOG_COMPONENT,
+                            subsystem = LOG_SUBSYSTEM,
+                            event = "kms_rekey_persist_failed",
+                            job_id = %job.id,
+                            bucket,
+                            object = %object,
+                            error = %error,
+                            "KMS rekey sweep rewrapped an envelope but could not persist it; the object keeps its \
+                             previous (still valid) wrapping and a re-run retries it"
+                        );
+                    }
+                }
+            }
+            Err(error) => {
+                job.failed.fetch_add(1, Ordering::Relaxed);
+                warn!(
+                    component = LOG_COMPONENT,
+                    subsystem = LOG_SUBSYSTEM,
+                    event = "kms_rekey_object_failed",
+                    job_id = %job.id,
+                    bucket,
+                    object = %object,
+                    error = %error,
+                    "KMS rekey sweep could not rewrap an object's data key; a re-run retries it"
+                );
+            }
+        }
+    }
+    // Dropping the receiver ends the walk on early break; a cancelled walk is
+    // not a sweep failure.
+    drop(rx);
+    match walk_task.await {
+        Ok(Ok(())) => {}
+        Ok(Err(error)) => {
+            if !job.cancel.is_cancelled() {
+                job.failed.fetch_add(1, Ordering::Relaxed);
+                warn!(
+                    component = LOG_COMPONENT,
+                    subsystem = LOG_SUBSYSTEM,
+                    event = "kms_rekey_walk_failed",
+                    job_id = %job.id,
+                    bucket,
+                    error = %error,
+                    "KMS rekey sweep walk ended with an error; coverage of this bucket is incomplete"
+                );
+            }
+        }
+        Err(error) => {
+            job.failed.fetch_add(1, Ordering::Relaxed);
+            warn!(
+                component = LOG_COMPONENT,
+                subsystem = LOG_SUBSYSTEM,
+                event = "kms_rekey_walk_failed",
+                job_id = %job.id,
+                bucket,
+                error = %error,
+                "KMS rekey sweep walk task failed; coverage of this bucket is incomplete"
+            );
+        }
+    }
+}
@@ -89,6 +89,7 @@ pub mod error;
 pub mod init;
 pub mod inspect;
 pub(crate) mod kms_deletion_gate;
+pub(crate) mod kms_rekey;
 pub mod license;
 pub mod memory_observability;
 pub mod module_switches;
@@ -109,10 +109,10 @@ pub(crate) use super::ecfs_extend::{
    validate_list_object_unordered_with_delimiter, validate_object_key, wrap_response_with_cors,
 };
 pub(crate) use super::sse::{
-    DecryptionRequest, EncryptionRequest, PrepareEncryptionRequest, SseKmsPrincipal, authorize_sse_kms_object_read,
-    classify_sse_read_response, extract_server_side_encryption_from_headers, sse_decryption, sse_encryption,
-    sse_prepare_encryption, strip_managed_encryption_metadata, validate_sse_headers_for_read, validate_sse_headers_for_write,
-    validate_ssec_for_read,
+    DecryptionRequest, EncryptionRequest, ObjectDekRewrapOutcome, PrepareEncryptionRequest, SseKmsPrincipal,
+    authorize_sse_kms_object_read, classify_sse_read_response, extract_server_side_encryption_from_headers,
+    rewrap_object_encryption_metadata, sse_decryption, sse_encryption, sse_prepare_encryption, strip_managed_encryption_metadata,
+    validate_sse_headers_for_read, validate_sse_headers_for_write, validate_ssec_for_read,
 };

 pub(crate) mod access_consumer {
@@ -87,9 +87,19 @@ pub(crate) mod kms {
        pub(crate) mod bucket {
            pub(crate) use super::super::super::storage_contracts::{BucketOperations, BucketOptions};
        }
+
+        pub(crate) mod list {
+            pub(crate) use super::super::super::storage_contracts::ListOperations;
+        }
+
+        pub(crate) mod object {
+            pub(crate) use super::super::super::storage_contracts::ObjectOperations;
+        }
    }

    pub(crate) use crate::storage::storage_api::{ECStore, StorageError};
+    pub(crate) type StorageObjectOptions = crate::storage::storage_api::StorageObjectOptions;
+    pub(crate) use crate::storage::storage_api::{ObjectDekRewrapOutcome, rewrap_object_encryption_metadata};

    /// Bucket SSE configuration for the KMS deletion reference gate;
    /// `Err(StorageError::ConfigNotFound)` when the bucket has none.