feat(kms): add the data key rewrap primitive (#5607)

* feat(kms): add data key rewrap and wrapping inspection primitives

Rewrap re-protects an existing data key envelope with the master key's
current version without touching the data key itself, which is the
precondition for ever retiring an older version: until every envelope a
version wrapped has been moved off it, destroying that version orphans
every object whose data key it wrapped.

Adds KmsBackend::rewrap_data_key and its read-only counterpart
describe_data_key_wrapping, both gated by a new BackendCapabilities::rewrap
flag and defaulting to UnsupportedCapability. Vault KV2 unwraps with the
frozen version record that wrapped the envelope and re-wraps with the
current material; Vault Transit uses the native transit/rewrap endpoint so
the data key never enters this process.

No read or write path changes: nothing calls these yet.

* test(kms): cover the rewrap primitive against a scripted Vault

* fix(kms): resolve both key materials before the data key is unwrapped

Keeps every fallible step out of the window in which the plaintext data
key exists, so no error path can drop it without zeroizing it first.

crates/kms/src/audit.rs

@@ -157,6 +157,7 @@ pub fn error_class(error: &KmsError) -> &'static str {
         KmsError::CredentialsUnavailable { .. } => "credentials_unavailable",
         KmsError::BaselineVersionLost { .. } => "baseline_version_lost",
         KmsError::KeyStillReferenced { .. } => "key_still_referenced",
+        KmsError::RewrapWouldExposePlaintext { .. } => "rewrap_would_expose_plaintext",
     }
 }
 

crates/kms/src/backends/contract_tests.rs

@@ -37,8 +37,9 @@ use crate::error::{KmsError, Result};
 use crate::manager::KmsManager;
 use crate::service::ObjectEncryptionService;
 use crate::types::{
-    CancelKeyDeletionRequest, CreateKeyRequest, DecryptRequest, DeleteKeyRequest, DescribeKeyRequest, EncryptRequest,
-    GenerateDataKeyRequest, KeySpec, KeyState, KeyUsage, ObjectEncryptionContext,
+    CancelKeyDeletionRequest, CreateKeyRequest, DecryptRequest, DeleteKeyRequest, DescribeDataKeyWrappingRequest,
+    DescribeKeyRequest, EncryptRequest, GenerateDataKeyRequest, KeySpec, KeyState, KeyUsage, ObjectEncryptionContext,
+    RewrapDataKeyRequest,
 };
 use base64::Engine as _;
 use base64::engine::general_purpose::STANDARD as BASE64;
@@ -64,6 +65,23 @@ async fn expect_rotate_rejected(backend: &dyn KmsBackend, key_id: &str) {
     }
 }
 
+/// Rewrap while not Enabled: it produces a new wrapping, so backends that
+/// support it must gate it through the state machine exactly as encryption is
+/// gated; backends without version history report the capability gap instead.
+async fn expect_rewrap_rejected(backend: &dyn KmsBackend, ciphertext: Vec<u8>) {
+    let result = backend
+        .rewrap_data_key(RewrapDataKeyRequest {
+            ciphertext,
+            encryption_context: context(),
+        })
+        .await;
+    if backend.capabilities().rewrap {
+        expect_invalid_key_state(result, "");
+    } else {
+        expect_unsupported(result);
+    }
+}
+
 fn expect_invalid_key_state<T: std::fmt::Debug>(result: Result<T>, expected_fragment: &str) {
     match result {
         Err(KmsError::InvalidOperation { message }) => assert!(
@@ -158,6 +176,7 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
     expect_invalid_key_state(backend.encrypt(encrypt_request(key_id)).await, "disabled");
     expect_invalid_key_state(backend.generate_data_key(generate_request(key_id)).await, "disabled");
     expect_rotate_rejected(backend, key_id).await;
+    expect_rewrap_rejected(backend, data_key.ciphertext_blob.clone()).await;
     // ...but decryption of existing data keeps working (explicit AWS deviation)...
     let decrypted = backend
         .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
@@ -187,6 +206,7 @@ async fn assert_state_machine_contract(backend: &dyn KmsBackend, key_id: &str) {
     expect_invalid_key_state(backend.enable_key(key_id).await, "pending deletion");
     expect_invalid_key_state(backend.disable_key(key_id).await, "pending deletion");
     expect_rotate_rejected(backend, key_id).await;
+    expect_rewrap_rejected(backend, data_key.ciphertext_blob.clone()).await;
     expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "pending deletion");
     let decrypted = backend
         .decrypt(decrypt_request(data_key.ciphertext_blob.clone()))
@@ -282,11 +302,30 @@ async fn static_backend_stateless_contract() {
     expect_invalid_key_state(backend.create_key(create_request("another-key".to_string())).await, "read-only");
     expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "read-only");
     expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "read-only");
-    // Enable/disable and rotation are capability gaps at the product
-    // surface, not state-machine rejections.
+    // Enable/disable, rotation and rewrap are capability gaps at the product
+    // surface, not state-machine rejections. A single fixed key has no second
+    // version to rewrap onto, so reporting the gap is the only honest answer —
+    // re-wrapping with the same material would look like progress while
+    // changing nothing.
     expect_unsupported(backend.enable_key(key_id).await);
     expect_unsupported(backend.disable_key(key_id).await);
     expect_unsupported(backend.rotate_key(key_id).await);
+    expect_unsupported(
+        backend
+            .rewrap_data_key(RewrapDataKeyRequest {
+                ciphertext: data_key.ciphertext_blob.clone(),
+                encryption_context: context(),
+            })
+            .await,
+    );
+    expect_unsupported(
+        backend
+            .describe_data_key_wrapping(DescribeDataKeyWrappingRequest {
+                ciphertext: data_key.ciphertext_blob,
+                encryption_context: context(),
+            })
+            .await,
+    );
 }
 
 fn vault_dev_config(constructor: fn(url::Url, String) -> KmsConfig) -> KmsConfig {

crates/kms/src/backends/mod.rs

@@ -126,6 +126,39 @@ pub(crate) fn ensure_tag_keys_are_mutable<'a>(tag_keys: impl IntoIterator<Item =
     Ok(())
 }
 
+/// Reject a rewrap whose caller cannot reproduce the envelope's encryption
+/// context.
+///
+/// Deliberately the same rule the decrypt paths apply — every context entry the
+/// envelope carries must be matched, and an entirely empty request context is
+/// accepted for envelopes written before contexts were recorded. Rewrap must
+/// never be *stricter* than decrypt: a retirement sweep has to be able to
+/// rewrap every envelope that still reads, or the old master key version it is
+/// trying to retire stays referenced forever by objects that are perfectly
+/// readable.
+///
+/// It must not be *laxer* either. The context binds an envelope to one
+/// bucket/object pair, and a rewrap that skipped the check would let a caller
+/// launder an envelope it has no claim on into a freshly wrapped one.
+pub(crate) fn ensure_rewrap_context_matches(
+    envelope_context: &HashMap<String, String>,
+    request_context: &HashMap<String, String>,
+) -> Result<()> {
+    for (key, expected_value) in envelope_context {
+        match request_context.get(key) {
+            Some(actual_value) if actual_value == expected_value => {}
+            Some(actual_value) => {
+                return Err(KmsError::context_mismatch(format!(
+                    "Context mismatch for key '{key}': expected '{expected_value}', got '{actual_value}'"
+                )));
+            }
+            None if request_context.is_empty() => {}
+            None => return Err(KmsError::context_mismatch(format!("Missing context key '{key}'"))),
+        }
+    }
+    Ok(())
+}
+
 /// Page size used when a [`ListKeysRequest`] does not ask for one.
 pub(crate) const DEFAULT_LIST_KEYS_PAGE_SIZE: u32 = 100;
 
@@ -271,6 +304,75 @@ pub trait KmsBackend: Send + Sync {
         Err(KmsError::unsupported_capability("backend without rotation support", "rotate_key"))
     }
 
+    /// Re-wrap an existing data key envelope under the key's current version,
+    /// returning an envelope that protects the same data key.
+    ///
+    /// This is the primitive that makes retiring an old master key version
+    /// possible at all: until every envelope a version wrapped has been moved
+    /// onto the current version, destroying that version's material orphans
+    /// every object whose data key it wrapped. Rotation alone does not shrink
+    /// the blast radius of a leaked master key version, because envelopes
+    /// written before it stay decryptable under the leaked material forever.
+    ///
+    /// Contract for implementations:
+    ///
+    /// - The plaintext data key must not be persisted, logged, or returned. It
+    ///   is either never materialized at all (backends with a native rewrap) or
+    ///   held only for the length of the re-wrap.
+    /// - The destination version is always the key's *current* version. Wrapping
+    ///   to any older version would create objects that nodes which resolve
+    ///   envelopes to the current version cannot read.
+    /// - Everything except the wrapping is carried over verbatim, encryption
+    ///   context included, so the result is the same data key rewrapped and not
+    ///   a new one. An envelope must never be moved between objects.
+    /// - Idempotence: an envelope already wrapped by the current version comes
+    ///   back byte for byte with [`RewrapDataKeyResponse::rewrapped`] false, so
+    ///   re-running a sweep converges and performs no metadata writes.
+    ///
+    /// Only backends that advertise [`BackendCapabilities::rewrap`] may override
+    /// this method; the default rejects the operation. A backend without
+    /// retained version history has nothing to rewrap *to*, so it reports the
+    /// capability gap rather than performing a re-wrap that changes no version
+    /// and buys no security.
+    async fn rewrap_data_key(&self, _request: RewrapDataKeyRequest) -> Result<RewrapDataKeyResponse> {
+        Err(KmsError::unsupported_capability(
+            "backend without versioned key material to rewrap onto",
+            "rewrap_data_key",
+        ))
+    }
+
+    /// Report which master key version wraps an existing data key envelope, and
+    /// whether that is the key's current version.
+    ///
+    /// The read-only counterpart of [`Self::rewrap_data_key`], and the only
+    /// supported way to ask the question: each rotating backend records the
+    /// version somewhere else, so every caller that parsed envelopes itself
+    /// would carry its own copy of that knowledge outside the KMS boundary.
+    ///
+    /// [`DescribeDataKeyWrappingResponse::is_current`] must be the exact
+    /// negation of what [`RewrapDataKeyResponse::rewrapped`] would report for
+    /// the same envelope. The two answers drive a single loop — scan to find
+    /// work, rewrap to do it, scan again to prove it is done — and a
+    /// disagreement would make that loop either never terminate or declare
+    /// success while envelopes still reference a version somebody is about to
+    /// destroy.
+    ///
+    /// Deliberately not gated on key state: an inventory has to be able to
+    /// count envelopes under keys that are disabled or already scheduled for
+    /// deletion, which are precisely the keys whose retirement is in question.
+    ///
+    /// Gated by the same [`BackendCapabilities::rewrap`] flag, because a backend
+    /// that cannot rewrap has no version to report progress against.
+    async fn describe_data_key_wrapping(
+        &self,
+        _request: DescribeDataKeyWrappingRequest,
+    ) -> Result<DescribeDataKeyWrappingResponse> {
+        Err(KmsError::unsupported_capability(
+            "backend without versioned key material to rewrap onto",
+            "describe_data_key_wrapping",
+        ))
+    }
+
     /// Replace a key's free-form description; `None` clears it.
     ///
     /// Backends that advertise [`BackendCapabilities::update_key_metadata`]
@@ -387,6 +489,8 @@ pub struct BackendCapabilities {
     pub physical_delete: bool,
     /// Updating a key's description and tags after creation
     pub update_key_metadata: bool,
+    /// Re-wrapping an existing data key envelope onto the key's current version
+    pub rewrap: bool,
 }
 
 impl BackendCapabilities {
@@ -404,6 +508,7 @@ impl BackendCapabilities {
             versioning: false,
             physical_delete: false,
             update_key_metadata: false,
+            rewrap: false,
         }
     }
 
@@ -460,6 +565,12 @@ impl BackendCapabilities {
         self.update_key_metadata = update_key_metadata;
         self
     }
+
+    /// Set whether envelope rewrap onto the current key version is supported
+    pub const fn with_rewrap(mut self, rewrap: bool) -> Self {
+        self.rewrap = rewrap;
+        self
+    }
 }
 
 impl Default for BackendCapabilities {
@@ -539,6 +650,7 @@ mod tests {
         assert!(!capabilities.versioning);
         assert!(!capabilities.physical_delete);
         assert!(!capabilities.update_key_metadata);
+        assert!(!capabilities.rewrap);
     }
 
     #[tokio::test]
@@ -553,6 +665,26 @@ mod tests {
             ),
             ("tag_key", MinimalBackend.tag_key("any-key", &HashMap::new()).await),
             ("untag_key", MinimalBackend.untag_key("any-key", &[]).await),
+            (
+                "rewrap_data_key",
+                MinimalBackend
+                    .rewrap_data_key(RewrapDataKeyRequest {
+                        ciphertext: b"{}".to_vec(),
+                        encryption_context: HashMap::new(),
+                    })
+                    .await
+                    .map(|_| ()),
+            ),
+            (
+                "describe_data_key_wrapping",
+                MinimalBackend
+                    .describe_data_key_wrapping(DescribeDataKeyWrappingRequest {
+                        ciphertext: b"{}".to_vec(),
+                        encryption_context: HashMap::new(),
+                    })
+                    .await
+                    .map(|_| ()),
+            ),
         ] {
             let error = result.expect_err("backends must opt in to lifecycle operations by overriding them");
             assert!(
@@ -574,6 +706,40 @@ mod tests {
         );
     }
 
+    /// The rewrap context guard has to sit exactly where the decrypt guard
+    /// sits: strict enough that an envelope cannot be laundered under a
+    /// different context, lax enough that everything decrypt accepts can still
+    /// be rewrapped — otherwise a retirement sweep stalls on readable objects.
+    #[test]
+    fn rewrap_context_guard_matches_the_decrypt_rule() {
+        let envelope = HashMap::from([
+            ("bucket".to_string(), "b/o".to_string()),
+            ("tenant".to_string(), "acme".to_string()),
+        ]);
+
+        ensure_rewrap_context_matches(&envelope, &envelope).expect("an exact context must be accepted");
+        // An empty request context is the pre-context-recording compatibility
+        // case decrypt allows; rewrap must not be stricter.
+        ensure_rewrap_context_matches(&envelope, &HashMap::new()).expect("an empty request context must stay accepted");
+        // Extra entries the envelope does not carry are ignored, as on decrypt.
+        let mut superset = envelope.clone();
+        superset.insert("extra".to_string(), "ignored".to_string());
+        ensure_rewrap_context_matches(&envelope, &superset).expect("extra request context entries must be ignored");
+
+        let mut tampered = envelope.clone();
+        tampered.insert("bucket".to_string(), "other/o".to_string());
+        assert!(
+            matches!(ensure_rewrap_context_matches(&envelope, &tampered), Err(KmsError::ContextMismatch { .. })),
+            "a tampered context value must be rejected"
+        );
+
+        let partial = HashMap::from([("tenant".to_string(), "acme".to_string())]);
+        assert!(
+            matches!(ensure_rewrap_context_matches(&envelope, &partial), Err(KmsError::ContextMismatch { .. })),
+            "a non-empty context missing an envelope entry must be rejected"
+        );
+    }
+
     #[test]
     fn identity_tag_is_rejected_by_metadata_updates() {
         let error = ensure_tag_keys_are_mutable([RESERVED_KEY_NAME_TAG])

crates/kms/src/backends/snapshots/rustfs_kms__backends__aws__tests__aws_backend_capabilities.snap

@@ -8,6 +8,7 @@ expression: capabilities_snapshot(backend.capabilities())
   "encrypt": true,
   "generate_data_key": true,
   "physical_delete": false,
+  "rewrap": false,
   "rotate": true,
   "schedule_deletion": true,
   "update_key_metadata": false,

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__local_backend_capabilities.snap

@@ -8,6 +8,7 @@ expression: capabilities_snapshot(backend.capabilities())
   "encrypt": true,
   "generate_data_key": true,
   "physical_delete": true,
+  "rewrap": false,
   "rotate": false,
   "schedule_deletion": true,
   "update_key_metadata": true,

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__static_backend_capabilities.snap

@@ -8,6 +8,7 @@ expression: capabilities_snapshot(backend.capabilities())
   "encrypt": true,
   "generate_data_key": true,
   "physical_delete": false,
+  "rewrap": false,
   "rotate": false,
   "schedule_deletion": false,
   "update_key_metadata": false,

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_kv2_backend_capabilities.snap

@@ -8,6 +8,7 @@ expression: capabilities_snapshot(backend.capabilities())
   "encrypt": true,
   "generate_data_key": true,
   "physical_delete": true,
+  "rewrap": true,
   "rotate": true,
   "schedule_deletion": true,
   "update_key_metadata": true,

crates/kms/src/backends/snapshots/rustfs_kms__backends__tests__vault_transit_backend_capabilities.snap

@@ -8,6 +8,7 @@ expression: capabilities_snapshot(backend.capabilities())
   "encrypt": true,
   "generate_data_key": true,
   "physical_delete": true,
+  "rewrap": true,
   "rotate": true,
   "schedule_deletion": true,
   "update_key_metadata": true,

crates/kms/src/backends/vault.rs

@@ -20,7 +20,7 @@ use crate::backends::vault_credentials::{
 };
 use crate::backends::{
     BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, empty_key_page, ensure_key_state_permits,
-    ensure_key_status_permits, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys,
+    ensure_key_status_permits, ensure_rewrap_context_matches, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys,
 };
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{AesDekCrypto, DataKeyEnvelope, DekCrypto, generate_key_material};
@@ -38,6 +38,7 @@ use std::time::Duration;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, info, warn};
 use vaultrs::{api::kv2::requests::SetSecretRequestOptions, error::ClientError, kv2};
+use zeroize::Zeroize as _;
 
 /// Vault KMS client implementation
 pub struct VaultKmsClient {
@@ -893,6 +894,137 @@ impl VaultKmsClient {
         Ok(plaintext)
     }
 
+    /// Report which master key version wraps an envelope, and whether that is
+    /// the key's current version.
+    ///
+    /// Reads the key record only; it never unwraps anything, so it works on keys
+    /// whose state forbids new cryptographic use — which is exactly the
+    /// population a retirement inventory has to cover.
+    pub(crate) async fn describe_data_key_wrapping(
+        &self,
+        request: &DescribeDataKeyWrappingRequest,
+    ) -> Result<DescribeDataKeyWrappingResponse> {
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
+            .map_err(|e| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {e}")))?;
+        ensure_rewrap_context_matches(&envelope.encryption_context, &request.encryption_context)?;
+
+        let key_data = self.get_key_data(&envelope.master_key_id).await?;
+        let current_version = key_data.version;
+
+        Ok(DescribeDataKeyWrappingResponse {
+            key_id: envelope.master_key_id,
+            key_version: Some(resolve_envelope_master_key_version(
+                envelope.master_key_version,
+                key_data.baseline_version,
+                current_version,
+            )),
+            current_key_version: Some(current_version),
+            // Deliberately not `key_version == current_version`. A
+            // pre-versioning envelope resolves to the current version while
+            // saying nothing, and `rewrap_data_key` rewrites exactly those to
+            // stamp the version — so reporting them as current here would leave
+            // the sweep and the scan permanently disagreeing.
+            is_current: envelope.master_key_version == Some(current_version),
+        })
+    }
+
+    /// Re-wrap an existing envelope with the key's current master key version.
+    ///
+    /// The plaintext data key is unwrapped with the version that actually
+    /// wrapped it and immediately re-wrapped with the current material. It is
+    /// zeroized before this returns, never persisted, never logged and never
+    /// handed to the caller: the whole point of rewrap is that the data key is
+    /// re-protected without anyone above this layer holding it.
+    ///
+    /// Everything except the wrapping is carried over verbatim — the data key's
+    /// own id, its spec, its encryption context and its creation time — so the
+    /// result is the same data key under new wrapping. That keeps the DEK's
+    /// recorded age honest and makes the operation invisible to every consumer
+    /// of the envelope other than the version stamp.
+    ///
+    /// A pre-versioning envelope (no `master_key_version`) is rewrapped even
+    /// when it resolves to the current version and its bytes would be unwrapped
+    /// with the very material they are about to be re-wrapped with. That is not
+    /// wasted work: the version stamp is the only evidence a retirement scan can
+    /// read, and an envelope that does not state its version can never be
+    /// counted as migrated.
+    pub(crate) async fn rewrap_data_key(&self, request: &RewrapDataKeyRequest) -> Result<RewrapDataKeyResponse> {
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
+            .map_err(|e| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {e}")))?;
+        ensure_rewrap_context_matches(&envelope.encryption_context, &request.encryption_context)?;
+
+        // Single read of the key record: the version that unwraps, the material
+        // that re-wraps and the version stamped into the result must all come
+        // from one snapshot. Reading them separately would let a concurrent
+        // rotation produce an envelope stamped with a version whose material
+        // never wrapped it — an envelope that then fails to decrypt forever.
+        let key_data = self.get_key_data(&envelope.master_key_id).await?;
+        ensure_key_status_permits(&envelope.master_key_id, &key_data.status, StateGatedOperation::Encrypt)?;
+        let current_version = key_data.version;
+
+        if envelope.master_key_version == Some(current_version) {
+            // Already on the current version and saying so. Hand the input back
+            // untouched rather than producing an equivalent envelope with a
+            // fresh nonce: a re-run of a sweep must converge to zero writes, and
+            // the storage layer keys its write decision off these bytes.
+            return Ok(RewrapDataKeyResponse {
+                ciphertext: request.ciphertext.clone(),
+                key_id: envelope.master_key_id,
+                source_key_version: Some(current_version),
+                destination_key_version: Some(current_version),
+                rewrapped: false,
+            });
+        }
+
+        let source_version =
+            resolve_envelope_master_key_version(envelope.master_key_version, key_data.baseline_version, current_version);
+        // Both materials are resolved before anything is unwrapped, so no
+        // fallible step sits between the plaintext data key coming into
+        // existence and the zeroize that removes it again.
+        let source_material = self
+            .get_key_material_for_version(&envelope.master_key_id, &key_data, source_version)
+            .await?;
+        let destination_material = decode_stored_key_material(&envelope.master_key_id, &key_data.encrypted_key_material)
+            .inspect_err(|error| warn!(key_id = %envelope.master_key_id, %error, "Vault KMS key material failed validation"))?;
+
+        let mut plaintext_key = match self
+            .dek_crypto
+            .decrypt(&source_material, &envelope.encrypted_key, &envelope.nonce)
+            .await
+        {
+            Ok(plaintext) => plaintext,
+            Err(error) => {
+                return Err(self
+                    .explain_unwrap_failure(&envelope.master_key_id, &key_data, envelope.master_key_version, error)
+                    .await);
+            }
+        };
+        let rewrapped = self.dek_crypto.encrypt(&destination_material, &plaintext_key).await;
+        plaintext_key.zeroize();
+        let (encrypted_key, nonce) = rewrapped?;
+
+        let rewrapped_envelope = DataKeyEnvelope {
+            key_id: envelope.key_id,
+            master_key_id: envelope.master_key_id,
+            key_spec: envelope.key_spec,
+            encrypted_key,
+            nonce,
+            encryption_context: envelope.encryption_context,
+            created_at: envelope.created_at,
+            master_key_version: Some(current_version),
+        };
+        let ciphertext = serde_json::to_vec(&rewrapped_envelope)?;
+
+        debug!(key_id = %rewrapped_envelope.master_key_id, source_version, current_version, "Vault KMS data key rewrapped");
+        Ok(RewrapDataKeyResponse {
+            ciphertext,
+            key_id: rewrapped_envelope.master_key_id,
+            source_key_version: Some(source_version),
+            destination_key_version: Some(current_version),
+            rewrapped: true,
+        })
+    }
+
     /// Re-report a failure to unwrap a data key as the lost-baseline diagnosis
     /// when that is what the key record shows.
     ///
@@ -1449,6 +1581,17 @@ impl KmsBackend for VaultKmsBackend {
         })
     }
 
+    async fn rewrap_data_key(&self, request: RewrapDataKeyRequest) -> Result<RewrapDataKeyResponse> {
+        self.client.rewrap_data_key(&request).await
+    }
+
+    async fn describe_data_key_wrapping(
+        &self,
+        request: DescribeDataKeyWrappingRequest,
+    ) -> Result<DescribeDataKeyWrappingResponse> {
+        self.client.describe_data_key_wrapping(&request).await
+    }
+
     async fn generate_data_key(&self, request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
         let generate_request = GenerateKeyRequest {
             master_key_id: request.key_id.clone(),
@@ -1658,7 +1801,9 @@ impl KmsBackend for VaultKmsBackend {
         // Rotation freezes the outgoing material as an immutable version
         // record before switching the current pointer, and envelopes resolve
         // their wrapping version on decrypt, so every historical version
-        // stays decryptable after a rotation.
+        // stays decryptable after a rotation. Those same immutable records are
+        // what lets an envelope be unwrapped with the version that wrapped it
+        // and re-wrapped onto the current one.
         BackendCapabilities::minimal()
             .with_rotate(true)
             .with_enable_disable(true)
@@ -1666,6 +1811,7 @@ impl KmsBackend for VaultKmsBackend {
             .with_versioning(true)
             .with_physical_delete(true)
             .with_update_key_metadata(true)
+            .with_rewrap(true)
     }
 
     async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {
@@ -3988,4 +4134,246 @@ mod tests {
             );
         }
     }
+
+    /// Rewrap against a scripted Vault serving `state`, scripting the key record
+    /// plus every version record the state holds, so the implementation — not
+    /// the harness — decides which of them it needs. Returns the response
+    /// together with the requests the rewrap made.
+    async fn rewrap_scripted(state: &KeyState, ciphertext: &[u8]) -> (RewrapDataKeyResponse, Vec<String>) {
+        let mut responses = vec![ScriptedResponse::ok(kv2_read_data(&state.key_data))];
+        responses.extend(
+            state
+                .version_records
+                .iter()
+                .map(|record| ScriptedResponse::ok(kv2_read_version_record_data(record))),
+        );
+        let (vault, client) = scripted_client(responses).await;
+
+        let response = client
+            .rewrap_data_key(&RewrapDataKeyRequest {
+                ciphertext: ciphertext.to_vec(),
+                encryption_context: HashMap::new(),
+            })
+            .await
+            .expect("rewrap must produce an envelope on the current version");
+        (response, vault.requests())
+    }
+
+    /// Describe the wrapping of `ciphertext` against a scripted Vault serving
+    /// `state`.
+    async fn describe_wrapping_scripted(state: &KeyState, ciphertext: &[u8]) -> DescribeDataKeyWrappingResponse {
+        let (_vault, client) = scripted_client(vec![ScriptedResponse::ok(kv2_read_data(&state.key_data))]).await;
+        client
+            .describe_data_key_wrapping(&DescribeDataKeyWrappingRequest {
+                ciphertext: ciphertext.to_vec(),
+                encryption_context: HashMap::new(),
+            })
+            .await
+            .expect("describing an envelope's wrapping must succeed")
+    }
+
+    /// The whole point of the primitive: an envelope wrapped by a superseded
+    /// master key version comes back wrapped by the current one, carrying the
+    /// same data key.
+    ///
+    /// Two independent things pin that the *old* material did the unwrapping.
+    /// The frozen version-1 record is read — an implementation that reached for
+    /// the current material would never ask for it — and the wrapping is
+    /// AES-256-GCM, so unwrapping with the wrong material cannot yield the
+    /// original data key at all, only an authentication failure. The closing
+    /// decrypt then shows the result is genuinely bound to version 2: it
+    /// resolves on the key record alone, with no version record in sight.
+    #[tokio::test]
+    async fn wired_kv2_rewrap_moves_an_old_envelope_onto_the_current_version() {
+        const PLAINTEXT: &[u8] = b"data-key-material-that-must-survive";
+
+        let state_v1 = KeyState::new(healthy_key_data());
+        let encrypted_v1 = encrypt_scripted(&state_v1, PLAINTEXT).await;
+        let state_v2 = rotate_scripted(&state_v1).await;
+        assert_ne!(
+            state_v2.key_data.encrypted_key_material, state_v1.key_data.encrypted_key_material,
+            "the rotation must have replaced the current material, or this test proves nothing"
+        );
+
+        let before = describe_wrapping_scripted(&state_v2, &encrypted_v1.ciphertext).await;
+        assert_eq!(before.key_version, Some(1));
+        assert_eq!(before.current_key_version, Some(2));
+        assert!(!before.is_current, "a version-1 envelope on a version-2 key is not current");
+
+        let (response, requests) = rewrap_scripted(&state_v2, &encrypted_v1.ciphertext).await;
+        assert!(response.rewrapped);
+        assert_eq!(response.source_key_version, Some(1));
+        assert_eq!(response.destination_key_version, Some(2));
+        assert_eq!(
+            requests,
+            vec![
+                "GET /v1/secret/data/rustfs/kms/keys/wired-key".to_string(),
+                "GET /v1/secret/data/rustfs/kms/keys/wired-key/versions/1".to_string(),
+            ],
+            "the unwrap must resolve the frozen version-1 material: {requests:?}"
+        );
+
+        let original: DataKeyEnvelope = serde_json::from_slice(&encrypted_v1.ciphertext).expect("envelope must parse");
+        let rewrapped: DataKeyEnvelope = serde_json::from_slice(&response.ciphertext).expect("rewrapped envelope must parse");
+        assert_eq!(rewrapped.master_key_version, Some(2), "the result must name the current version");
+        assert_ne!(rewrapped.encrypted_key, original.encrypted_key, "the wrapping must actually change");
+        // Everything but the wrapping is carried over, so the data key keeps its
+        // identity and its recorded age.
+        assert_eq!(rewrapped.key_id, original.key_id);
+        assert_eq!(rewrapped.key_spec, original.key_spec);
+        assert_eq!(rewrapped.encryption_context, original.encryption_context);
+        assert_eq!(rewrapped.created_at, original.created_at);
+
+        let (plaintext, decrypt_requests) = decrypt_scripted(&state_v2, &response.ciphertext).await;
+        assert_eq!(
+            plaintext, PLAINTEXT,
+            "the rewrapped envelope must yield the original data key byte for byte"
+        );
+        assert_eq!(
+            decrypt_requests.len(),
+            1,
+            "the rewrapped envelope must resolve on the current record alone: {decrypt_requests:?}"
+        );
+
+        let after = describe_wrapping_scripted(&state_v2, &response.ciphertext).await;
+        assert!(after.is_current, "the scan must agree the envelope no longer needs rewrapping");
+    }
+
+    /// Re-running a sweep must converge. An envelope already on the current
+    /// version comes back byte for byte with nothing to persist, rather than as
+    /// an equivalent envelope with a fresh nonce that would make every pass
+    /// rewrite every object's metadata forever.
+    #[tokio::test]
+    async fn wired_kv2_rewrap_of_a_current_envelope_is_a_no_op() {
+        let state_v1 = KeyState::new(healthy_key_data());
+        let state_v2 = rotate_scripted(&state_v1).await;
+        let encrypted_v2 = encrypt_scripted(&state_v2, b"written-after-the-rotation").await;
+
+        let described = describe_wrapping_scripted(&state_v2, &encrypted_v2.ciphertext).await;
+        assert!(described.is_current);
+
+        let (response, requests) = rewrap_scripted(&state_v2, &encrypted_v2.ciphertext).await;
+        assert!(!response.rewrapped, "an already-current envelope has nothing to rewrap");
+        assert_eq!(
+            response.ciphertext, encrypted_v2.ciphertext,
+            "a no-op rewrap must hand the input back unchanged"
+        );
+        assert_eq!(response.source_key_version, Some(2));
+        assert_eq!(response.destination_key_version, Some(2));
+        assert_eq!(requests.len(), 1, "a no-op must not reach for any version record: {requests:?}");
+
+        // Idempotence in the literal sense: feeding the result back in changes
+        // nothing again.
+        let (again, _) = rewrap_scripted(&state_v2, &response.ciphertext).await;
+        assert!(!again.rewrapped);
+        assert_eq!(again.ciphertext, encrypted_v2.ciphertext);
+    }
+
+    /// A pre-versioning envelope carries no version at all, so it can never
+    /// satisfy a retirement scan however current its material happens to be.
+    /// Rewrap therefore rewrites it to stamp the version — including on a
+    /// never-rotated key, where the material it is unwrapped with and the
+    /// material it is re-wrapped with are the same bytes.
+    #[tokio::test]
+    async fn wired_kv2_rewrap_stamps_a_pre_versioning_envelope() {
+        const PLAINTEXT: &[u8] = b"written-before-versioning-existed";
+
+        let state_v1 = KeyState::new(healthy_key_data());
+        let encrypted_v1 = encrypt_scripted(&state_v1, PLAINTEXT).await;
+        let legacy = strip_master_key_version(&encrypted_v1.ciphertext);
+        let legacy_envelope: DataKeyEnvelope = serde_json::from_slice(&legacy).expect("legacy envelope must parse");
+        assert_eq!(legacy_envelope.master_key_version, None);
+
+        // Never rotated: the resolved version is already the current one, and
+        // the envelope is still rewritten purely to record it.
+        let described = describe_wrapping_scripted(&state_v1, &legacy).await;
+        assert_eq!(described.key_version, Some(1));
+        assert_eq!(described.current_key_version, Some(1));
+        assert!(
+            !described.is_current,
+            "an envelope that does not state its version can never count as migrated"
+        );
+
+        let (response, requests) = rewrap_scripted(&state_v1, &legacy).await;
+        assert!(response.rewrapped);
+        assert_eq!(response.source_key_version, Some(1));
+        assert_eq!(response.destination_key_version, Some(1));
+        assert_eq!(requests.len(), 1, "a never-rotated key has no version record to read: {requests:?}");
+        let stamped: DataKeyEnvelope = serde_json::from_slice(&response.ciphertext).expect("stamped envelope must parse");
+        assert_eq!(stamped.master_key_version, Some(1));
+        let (plaintext, _) = decrypt_scripted(&state_v1, &response.ciphertext).await;
+        assert_eq!(plaintext, PLAINTEXT);
+
+        // After a rotation the same legacy envelope resolves through the frozen
+        // baseline instead, and lands on the rotated version.
+        let state_v2 = rotate_scripted(&state_v1).await;
+        assert_eq!(state_v2.key_data.baseline_version, Some(1));
+        let (rotated_response, rotated_requests) = rewrap_scripted(&state_v2, &legacy).await;
+        assert_eq!(rotated_response.source_key_version, Some(1), "the baseline is what wrapped it");
+        assert_eq!(rotated_response.destination_key_version, Some(2));
+        assert!(
+            rotated_requests[1].ends_with("/versions/1"),
+            "the unwrap must resolve the baseline material: {rotated_requests:?}"
+        );
+        let (rotated_plaintext, _) = decrypt_scripted(&state_v2, &rotated_response.ciphertext).await;
+        assert_eq!(rotated_plaintext, PLAINTEXT);
+    }
+
+    /// Drop the `master_key_version` field to produce the envelope shape a
+    /// pre-versioning build wrote.
+    fn strip_master_key_version(ciphertext: &[u8]) -> Vec<u8> {
+        let mut value: serde_json::Value = serde_json::from_slice(ciphertext).expect("envelope must parse");
+        value
+            .as_object_mut()
+            .expect("envelope is a JSON object")
+            .remove("master_key_version");
+        serde_json::to_vec(&value).expect("serialize legacy envelope")
+    }
+
+    /// The encryption context binds an envelope to one object. A caller that
+    /// cannot reproduce it is refused before any Vault read, so rewrap cannot be
+    /// used to launder an envelope onto a fresh wrapping.
+    #[tokio::test]
+    async fn wired_kv2_rewrap_rejects_a_tampered_encryption_context() {
+        let context = HashMap::from([("bucket".to_string(), "photos/cat.jpg".to_string())]);
+        let (vault, client) = scripted_client(vec![ScriptedResponse::ok(kv2_read_data(&healthy_key_data()))]).await;
+
+        let encrypted = client
+            .encrypt(
+                &EncryptRequest {
+                    key_id: "wired-key".to_string(),
+                    plaintext: b"bound-to-one-object".to_vec(),
+                    encryption_context: context.clone(),
+                    grant_tokens: Vec::new(),
+                },
+                None,
+            )
+            .await
+            .expect("encrypt must produce an envelope");
+
+        let error = client
+            .rewrap_data_key(&RewrapDataKeyRequest {
+                ciphertext: encrypted.ciphertext.clone(),
+                encryption_context: HashMap::from([("bucket".to_string(), "photos/other.jpg".to_string())]),
+            })
+            .await
+            .expect_err("a context that does not match the envelope must be refused");
+        assert!(matches!(error, KmsError::ContextMismatch { .. }), "got {error:?}");
+
+        let error = client
+            .describe_data_key_wrapping(&DescribeDataKeyWrappingRequest {
+                ciphertext: encrypted.ciphertext,
+                encryption_context: HashMap::from([("bucket".to_string(), "photos/other.jpg".to_string())]),
+            })
+            .await
+            .expect_err("the read-only accessor must apply the same guard");
+        assert!(matches!(error, KmsError::ContextMismatch { .. }), "got {error:?}");
+
+        assert_eq!(
+            vault.requests().len(),
+            1,
+            "the context guard must run before any Vault read: {:?}",
+            vault.requests()
+        );
+    }
 }

crates/kms/src/backends/vault_transit.rs

@@ -20,7 +20,7 @@ use crate::backends::vault_credentials::{
 };
 use crate::backends::{
     BackendCapabilities, ExpiredKeyRemoval, KmsBackend, StateGatedOperation, empty_key_page, ensure_key_state_permits,
-    ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys,
+    ensure_rewrap_context_matches, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys,
 };
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
@@ -45,6 +45,7 @@ use vaultrs::{
         requests::{
             CreateKeyRequestBuilder, DecryptDataRequestBuilder, EncryptDataRequestBuilder, UpdateKeyConfigurationRequestBuilder,
         },
+        responses::ReadKeyData,
     },
     error::ClientError,
     kv2,
@@ -73,6 +74,19 @@ const METADATA_CACHE_TTL: Duration = Duration::from_secs(300);
 /// grow process memory without limit.
 const METADATA_CACHE_CAPACITY: u64 = 1024;
 
+/// Read the key version out of a Transit ciphertext's `vault:vN:` prefix.
+///
+/// Transit ciphertext self-describes the version that wrapped it, which is why
+/// [`DataKeyEnvelope::master_key_version`] stays `None` on this backend. The
+/// prefix is therefore the only place a rewrap can learn whether it changed
+/// anything. `None` means the ciphertext is not in a shape this backend
+/// produced, and callers must treat the version as unknown rather than assume
+/// one.
+fn transit_ciphertext_version(ciphertext: &str) -> Option<u32> {
+    let (version, _) = ciphertext.strip_prefix("vault:v")?.split_once(':')?;
+    version.parse().ok()
+}
+
 /// Whether a KV2 write failed its check-and-set precondition.
 ///
 /// Mirrors the helper of the same name in `vault.rs`; the two backends keep
@@ -358,6 +372,47 @@ impl VaultTransitKmsClient {
             .map_err(|e| KmsError::cryptographic_error("base64_decode", e.to_string()))
     }
 
+    /// Re-encrypt a Transit ciphertext under the key's latest version without
+    /// the plaintext ever leaving Vault.
+    ///
+    /// Classified as an idempotent read because it is one: Vault mutates
+    /// nothing, and a replayed attempt only produces another ciphertext of the
+    /// same data key under the same version.
+    async fn transit_rewrap(&self, key_id: &str, ciphertext: &str) -> Result<String> {
+        let response = self
+            .run("vault_transit_rewrap", OpClass::ReadIdempotent, move || async move {
+                let vault = self.vault().map_err(AttemptError::fatal)?;
+                data::rewrap(&vault.client, &self.config.mount_path, key_id, ciphertext, None)
+                    .await
+                    .map_err(|e| AttemptError::from_vaultrs(e, |e| Self::map_vault_error(key_id, e, "rewrap")))
+            })
+            .await?;
+
+        Ok(response.ciphertext)
+    }
+
+    /// Vault's own newest version of a transit key.
+    ///
+    /// `transit/keys/:name` reports retained versions as a version-number to
+    /// creation-time map rather than as a single "latest" field, so the newest
+    /// version is the largest entry in it.
+    ///
+    /// Read from Vault rather than taken from the RustFS metadata record's
+    /// `current_version` counter: that counter only advances when a rotation
+    /// goes through this process, while `transit/rewrap` always targets Vault's
+    /// notion of latest. The scan that decides whether a rewrap is still needed
+    /// and the rewrap that acts on it must answer to the same authority, or an
+    /// operator-side `vault write -f transit/keys/x/rotate` would leave the two
+    /// permanently disagreeing.
+    async fn latest_transit_key_version(&self, key_id: &str) -> Result<Option<u32>> {
+        let response = self.read_transit_key(key_id).await?;
+        let latest = match &response.keys {
+            ReadKeyData::Symmetric(versions) => versions.keys().filter_map(|version| version.parse::<u32>().ok()).max(),
+            ReadKeyData::Asymmetric(versions) => versions.keys().filter_map(|version| version.parse::<u32>().ok()).max(),
+        };
+        Ok(latest)
+    }
+
     fn metadata_key_path(&self, key_id: &str) -> String {
         format!("{}/{}", self.metadata_key_prefix, key_id)
     }
@@ -758,6 +813,125 @@ impl VaultTransitKmsClient {
         }
     }
 
+    /// Report which transit key version wraps an envelope, and whether that is
+    /// the key's latest version.
+    ///
+    /// Reads only key metadata, never the ciphertext's contents, so it answers
+    /// for AAD-bound envelopes that [`Self::rewrap_data_key`] has to refuse — an
+    /// inventory must be able to count exactly the envelopes that are stuck.
+    pub(crate) async fn describe_data_key_wrapping(
+        &self,
+        request: &DescribeDataKeyWrappingRequest,
+    ) -> Result<DescribeDataKeyWrappingResponse> {
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
+            .map_err(|e| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {e}")))?;
+        ensure_rewrap_context_matches(&envelope.encryption_context, &request.encryption_context)?;
+
+        let source_ciphertext = std::str::from_utf8(&envelope.encrypted_key)
+            .map_err(|e| KmsError::cryptographic_error("utf8", format!("Invalid Transit ciphertext: {e}")))?;
+        let key_version = transit_ciphertext_version(source_ciphertext);
+        let current_key_version = self.latest_transit_key_version(&envelope.master_key_id).await?;
+
+        Ok(DescribeDataKeyWrappingResponse {
+            key_id: envelope.master_key_id,
+            key_version,
+            current_key_version,
+            // An unreadable prefix on either side means the version is unknown,
+            // and unknown must never read as "already current" — that is the
+            // answer that lets an operator destroy a version still in use.
+            is_current: key_version.is_some() && key_version == current_key_version,
+        })
+    }
+
+    /// Re-wrap an existing envelope onto the transit key's latest version using
+    /// Vault's native rewrap endpoint.
+    ///
+    /// The data key is never decrypted into this process: Vault re-encrypts the
+    /// ciphertext internally and hands back only the new ciphertext, so no
+    /// `transit/decrypt` is issued and no plaintext data key exists here to
+    /// leak, log or persist.
+    ///
+    /// # Envelopes bound to an encryption context cannot be rewrapped
+    ///
+    /// This backend binds the encryption context into the wrapping as AEAD
+    /// associated data ([`Self::transit_encrypt`]), and Vault's `transit/rewrap`
+    /// endpoint accepts no `associated_data` parameter — the only way to move
+    /// such a ciphertext onto a newer version is `transit/decrypt` followed by
+    /// `transit/encrypt`, which materializes the plaintext data key inside
+    /// RustFS. That trade is refused here rather than made silently: it would
+    /// hand back a valid envelope while dropping the very property that makes a
+    /// backend-side rewrap worth having. Every object-level envelope carries a
+    /// bucket/object context, so in practice this rejects them all until the
+    /// context binding or the endpoint changes.
+    ///
+    /// The context guard still runs first, so a caller that cannot reproduce the
+    /// envelope's context is told that rather than being told about the AAD
+    /// limitation of an envelope it has no claim on.
+    pub(crate) async fn rewrap_data_key(&self, request: &RewrapDataKeyRequest) -> Result<RewrapDataKeyResponse> {
+        let envelope: DataKeyEnvelope = serde_json::from_slice(&request.ciphertext)
+            .map_err(|e| KmsError::cryptographic_error("parse", format!("Failed to parse data key envelope: {e}")))?;
+        ensure_rewrap_context_matches(&envelope.encryption_context, &request.encryption_context)?;
+        self.ensure_key_state_allows(&envelope.master_key_id, StateGatedOperation::Encrypt)
+            .await?;
+
+        if !envelope.encryption_context.is_empty() {
+            return Err(KmsError::rewrap_would_expose_plaintext(
+                &envelope.master_key_id,
+                "the envelope binds its encryption context as AEAD associated data, which Vault Transit's rewrap endpoint \
+                 cannot carry; rewrapping it would require decrypting the data key inside RustFS",
+            ));
+        }
+
+        let source_ciphertext = std::str::from_utf8(&envelope.encrypted_key)
+            .map_err(|e| KmsError::cryptographic_error("utf8", format!("Invalid Transit ciphertext: {e}")))?;
+        let source_key_version = transit_ciphertext_version(source_ciphertext);
+
+        let rewrapped_ciphertext = match self.transit_rewrap(&envelope.master_key_id, source_ciphertext).await {
+            Ok(ciphertext) => ciphertext,
+            Err(error) => {
+                self.invalidate_metadata_on_state_error(&envelope.master_key_id, &error).await;
+                return Err(error);
+            }
+        };
+        let destination_key_version = transit_ciphertext_version(&rewrapped_ciphertext);
+
+        // Vault re-encrypts unconditionally, so an already-current ciphertext
+        // comes back changed but no newer. Report that as "nothing to persist"
+        // and hand the input back untouched, or a repeated sweep would rewrite
+        // every object's metadata on every pass forever.
+        if source_key_version.is_some() && source_key_version == destination_key_version {
+            return Ok(RewrapDataKeyResponse {
+                ciphertext: request.ciphertext.clone(),
+                key_id: envelope.master_key_id,
+                source_key_version,
+                destination_key_version,
+                rewrapped: false,
+            });
+        }
+
+        let rewrapped_envelope = DataKeyEnvelope {
+            key_id: envelope.key_id,
+            master_key_id: envelope.master_key_id,
+            key_spec: envelope.key_spec,
+            encrypted_key: rewrapped_ciphertext.into_bytes(),
+            nonce: envelope.nonce,
+            encryption_context: envelope.encryption_context,
+            created_at: envelope.created_at,
+            // Transit ciphertext still self-describes its version, so the
+            // envelope field stays absent exactly as generate_data_key leaves it.
+            master_key_version: None,
+        };
+        let ciphertext = serde_json::to_vec(&rewrapped_envelope)?;
+
+        Ok(RewrapDataKeyResponse {
+            ciphertext,
+            key_id: rewrapped_envelope.master_key_id,
+            source_key_version,
+            destination_key_version,
+            rewrapped: true,
+        })
+    }
+
     /// Test-only lifecycle driver: the product path goes through [`KmsBackend`].
     #[cfg(test)]
     pub(crate) async fn create_key(
@@ -1166,6 +1340,17 @@ impl KmsBackend for VaultTransitKmsBackend {
         })
     }
 
+    async fn rewrap_data_key(&self, request: RewrapDataKeyRequest) -> Result<RewrapDataKeyResponse> {
+        self.client.rewrap_data_key(&request).await
+    }
+
+    async fn describe_data_key_wrapping(
+        &self,
+        request: DescribeDataKeyWrappingRequest,
+    ) -> Result<DescribeDataKeyWrappingResponse> {
+        self.client.describe_data_key_wrapping(&request).await
+    }
+
     async fn generate_data_key(&self, request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
         let generate_request = GenerateKeyRequest {
             master_key_id: request.key_id.clone(),
@@ -1308,7 +1493,11 @@ impl KmsBackend for VaultTransitKmsBackend {
     fn capabilities(&self) -> BackendCapabilities {
         // Vault Transit natively supports version-retaining rotation, keeps
         // prior versions addressable for decryption, and allows physical
-        // deletion once a key is pending deletion.
+        // deletion once a key is pending deletion. Rewrap is advertised because
+        // the endpoint exists and works; envelopes whose encryption context is
+        // bound as associated data are still refused per envelope (see
+        // `VaultTransitKmsClient::rewrap_data_key`), which is a property of the
+        // envelope rather than of the backend.
         BackendCapabilities::minimal()
             .with_rotate(true)
             .with_enable_disable(true)
@@ -1316,6 +1505,7 @@ impl KmsBackend for VaultTransitKmsBackend {
             .with_versioning(true)
             .with_physical_delete(true)
             .with_update_key_metadata(true)
+            .with_rewrap(true)
     }
 
     async fn remove_expired_key(&self, key_id: &str, now: &Zoned) -> Result<ExpiredKeyRemoval> {
@@ -2368,4 +2558,242 @@ mod tests {
             "the pre-rotation ciphertext must reach Vault unchanged: {decrypt_body}"
         );
     }
+
+    /// Transit read-key payload for a key that has been rotated up to `latest`.
+    fn transit_key_read_data_up_to(key_id: &str, latest: u32) -> serde_json::Value {
+        let mut response: serde_json::Value = transit_key_read_data(key_id);
+        let keys: serde_json::Map<String, serde_json::Value> = (1..=latest)
+            .map(|version| (version.to_string(), serde_json::json!(1_700_000_000_u64 + u64::from(version))))
+            .collect();
+        response["keys"] = serde_json::Value::Object(keys);
+        response
+    }
+
+    fn wired_key_request(context: HashMap<String, String>) -> GenerateKeyRequest {
+        GenerateKeyRequest {
+            master_key_id: "wired-key".to_string(),
+            key_spec: "AES_256".to_string(),
+            key_length: Some(32),
+            encryption_context: context,
+            grant_tokens: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn transit_ciphertext_version_reads_only_a_well_formed_prefix() {
+        assert_eq!(transit_ciphertext_version("vault:v1:abc"), Some(1));
+        assert_eq!(transit_ciphertext_version("vault:v27:abc"), Some(27));
+        // Anything else leaves the version unknown rather than guessing one; an
+        // invented version is what would let a still-referenced key version be
+        // reported as retired.
+        assert_eq!(transit_ciphertext_version("vault:v:abc"), None);
+        assert_eq!(transit_ciphertext_version("vault:vx:abc"), None);
+        assert_eq!(transit_ciphertext_version("vault:v1"), None);
+        assert_eq!(transit_ciphertext_version("v1:abc"), None);
+        assert_eq!(transit_ciphertext_version(""), None);
+    }
+
+    /// The property that justifies having a Transit-specific rewrap at all: the
+    /// data key is re-encrypted by Vault, so no `transit/decrypt` is issued and
+    /// no plaintext data key ever exists inside this process.
+    #[tokio::test]
+    async fn wired_transit_rewrap_uses_the_native_endpoint_and_never_decrypts() {
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let (vault, client) = scripted_client(vec![
+            // generate_data_key: metadata state gate, then the transit encrypt.
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v1:scripted" })),
+            // rewrap: the state gate hits the metadata cache, so the only call
+            // is the native rewrap.
+            ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v3:rewrapped" })),
+        ])
+        .await;
+
+        let data_key = client
+            .generate_data_key(&wired_key_request(HashMap::new()), None)
+            .await
+            .expect("generate_data_key must produce an envelope");
+        let original: DataKeyEnvelope = serde_json::from_slice(&data_key.ciphertext).expect("envelope must parse");
+
+        let response = client
+            .rewrap_data_key(&RewrapDataKeyRequest {
+                ciphertext: data_key.ciphertext.clone(),
+                encryption_context: HashMap::new(),
+            })
+            .await
+            .expect("rewrap must move the ciphertext onto the latest version");
+
+        assert!(response.rewrapped);
+        assert_eq!(response.source_key_version, Some(1));
+        assert_eq!(response.destination_key_version, Some(3));
+
+        let rewrapped: DataKeyEnvelope = serde_json::from_slice(&response.ciphertext).expect("rewrapped envelope must parse");
+        assert_eq!(rewrapped.encrypted_key, b"vault:v3:rewrapped".to_vec());
+        assert_eq!(
+            rewrapped.master_key_version, None,
+            "Transit ciphertext self-describes its version, so the envelope field must stay absent"
+        );
+        assert_eq!(rewrapped.key_id, original.key_id);
+        assert_eq!(rewrapped.created_at, original.created_at);
+        assert_eq!(rewrapped.encryption_context, original.encryption_context);
+
+        let requests = vault.requests();
+        assert!(
+            requests.contains(&"POST /v1/transit/rewrap/wired-key".to_string()),
+            "the native rewrap endpoint must be used: {requests:?}"
+        );
+        assert!(
+            !requests.iter().any(|request| request.contains("/transit/decrypt/")),
+            "no decrypt may be issued: the plaintext data key must never enter this process: {requests:?}"
+        );
+
+        // The ciphertext Vault was asked to rewrap is the one that was stored,
+        // byte for byte.
+        let bodies = vault.request_bodies();
+        let rewrap_index = requests
+            .iter()
+            .position(|request| request == "POST /v1/transit/rewrap/wired-key")
+            .expect("the rewrap request must be recorded");
+        let body: serde_json::Value = serde_json::from_str(&bodies[rewrap_index]).expect("rewrap body must be JSON");
+        assert_eq!(body["ciphertext"], serde_json::json!("vault:v1:scripted"), "{body}");
+    }
+
+    /// Vault re-encrypts unconditionally, so an already-latest ciphertext comes
+    /// back different but no newer. That must report as "nothing to persist", or
+    /// every sweep pass would rewrite every object's metadata forever.
+    #[tokio::test]
+    async fn wired_transit_rewrap_of_a_current_ciphertext_is_a_no_op() {
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let (_vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v1:scripted" })),
+            // Same version back, different bytes.
+            ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v1:re-encrypted" })),
+        ])
+        .await;
+
+        let data_key = client
+            .generate_data_key(&wired_key_request(HashMap::new()), None)
+            .await
+            .expect("generate_data_key must produce an envelope");
+
+        let response = client
+            .rewrap_data_key(&RewrapDataKeyRequest {
+                ciphertext: data_key.ciphertext.clone(),
+                encryption_context: HashMap::new(),
+            })
+            .await
+            .expect("rewrap must succeed");
+
+        assert!(!response.rewrapped, "a ciphertext already on the latest version is a no-op");
+        assert_eq!(
+            response.ciphertext, data_key.ciphertext,
+            "a no-op must hand the stored envelope back unchanged, not Vault's fresh re-encryption"
+        );
+        assert_eq!(response.source_key_version, Some(1));
+        assert_eq!(response.destination_key_version, Some(1));
+    }
+
+    /// Vault's `transit/rewrap` endpoint takes no `associated_data` parameter,
+    /// and this backend binds the encryption context as exactly that. The only
+    /// remaining route would decrypt the data key inside RustFS, so the request
+    /// is refused rather than silently downgraded — and refused without any call
+    /// to Vault at all.
+    #[tokio::test]
+    async fn wired_transit_rewrap_refuses_an_aad_bound_envelope() {
+        let context = HashMap::from([("bucket".to_string(), "photos/cat.jpg".to_string())]);
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v1:scripted" })),
+            // Only the read-only accessor below is allowed to consume this.
+            ScriptedResponse::ok(transit_key_read_data_up_to("wired-key", 2)),
+        ])
+        .await;
+
+        let data_key = client
+            .generate_data_key(&wired_key_request(context.clone()), None)
+            .await
+            .expect("generate_data_key must produce an envelope");
+
+        let error = client
+            .rewrap_data_key(&RewrapDataKeyRequest {
+                ciphertext: data_key.ciphertext.clone(),
+                encryption_context: context.clone(),
+            })
+            .await
+            .expect_err("an AAD-bound envelope must not be rewrapped by decrypting it here");
+        assert!(
+            matches!(&error, KmsError::RewrapWouldExposePlaintext { key_id, .. } if key_id == "wired-key"),
+            "got {error:?}"
+        );
+
+        // The stuck envelope must still be countable, or an inventory could not
+        // report how much of the key version is unmigratable.
+        let described = client
+            .describe_data_key_wrapping(&DescribeDataKeyWrappingRequest {
+                ciphertext: data_key.ciphertext.clone(),
+                encryption_context: context,
+            })
+            .await
+            .expect("describing the wrapping must work even when rewrapping it cannot");
+        assert_eq!(described.key_version, Some(1));
+        assert_eq!(described.current_key_version, Some(2));
+        assert!(!described.is_current);
+
+        let requests = vault.requests();
+        assert!(
+            !requests.iter().any(|request| request.contains("/transit/rewrap/")),
+            "the refusal must happen before any rewrap call: {requests:?}"
+        );
+        assert!(
+            !requests.iter().any(|request| request.contains("/transit/decrypt/")),
+            "and above all before any decrypt: {requests:?}"
+        );
+    }
+
+    /// The current version comes from Vault's own key record rather than from
+    /// the RustFS metadata counter, which only advances on rotations this
+    /// process performed.
+    #[tokio::test]
+    async fn wired_transit_describe_wrapping_reads_vaults_latest_version() {
+        let metadata = TransitKeyMetadata::from_create_request(&CreateKeyRequest::default());
+        assert_eq!(metadata.current_version, 1, "the RustFS counter still says version 1");
+
+        let (vault, client) = scripted_client(vec![
+            ScriptedResponse::ok(metadata_read_data(&metadata)),
+            ScriptedResponse::ok(serde_json::json!({ "ciphertext": "vault:v1:scripted" })),
+            // Vault has been rotated behind RustFS's back.
+            ScriptedResponse::ok(transit_key_read_data_up_to("wired-key", 4)),
+        ])
+        .await;
+
+        let data_key = client
+            .generate_data_key(&wired_key_request(HashMap::new()), None)
+            .await
+            .expect("generate_data_key must produce an envelope");
+
+        let described = client
+            .describe_data_key_wrapping(&DescribeDataKeyWrappingRequest {
+                ciphertext: data_key.ciphertext.clone(),
+                encryption_context: HashMap::new(),
+            })
+            .await
+            .expect("describing the wrapping must succeed");
+
+        assert_eq!(described.key_id, "wired-key");
+        assert_eq!(described.key_version, Some(1));
+        assert_eq!(
+            described.current_key_version,
+            Some(4),
+            "the latest version must come from Vault, not from the RustFS metadata counter"
+        );
+        assert!(!described.is_current);
+
+        let requests = vault.requests();
+        assert!(
+            requests.contains(&"GET /v1/transit/keys/wired-key".to_string()),
+            "the latest version must be read from the transit key record: {requests:?}"
+        );
+    }
 }

crates/kms/src/error.rs

@@ -154,6 +154,15 @@ pub enum KmsError {
         .references.join(", ")
     )]
     KeyStillReferenced { key_id: String, references: Vec<String> },
+
+    /// The only available way to rewrap this envelope would pull the plaintext
+    /// data key into the RustFS process. Refused rather than performed: the
+    /// point of a backend-side rewrap is that the data key stays inside the
+    /// backend, so silently falling back to unwrap-then-rewrap would hand back
+    /// a correct envelope while quietly dropping the property that justified
+    /// the operation.
+    #[error("Cannot rewrap a data key of key {key_id} without exposing its plaintext: {reason}")]
+    RewrapWouldExposePlaintext { key_id: String, reason: String },
 }
 
 impl KmsError {
@@ -328,6 +337,14 @@ impl KmsError {
             oldest_version,
         }
     }
+
+    /// Create a rewrap-would-expose-plaintext error
+    pub fn rewrap_would_expose_plaintext<S1: Into<String>, S2: Into<String>>(key_id: S1, reason: S2) -> Self {
+        Self::RewrapWouldExposePlaintext {
+            key_id: key_id.into(),
+            reason: reason.into(),
+        }
+    }
 }
 
 /// Convert from standard library errors

crates/kms/src/manager.rs

@@ -23,8 +23,10 @@ use crate::error::{KmsError, Result};
 use crate::types::{
     CancelKeyDeletionRequest, CancelKeyDeletionResponse, CreateKeyRequest, CreateKeyResponse,
     DEFAULT_PENDING_DELETION_WINDOW_DAYS, DecryptRequest, DecryptResponse, DeleteKeyRequest, DeleteKeyResponse,
-    DescribeKeyRequest, DescribeKeyResponse, EncryptRequest, EncryptResponse, GenerateDataKeyRequest, GenerateDataKeyResponse,
-    ListKeysRequest, ListKeysResponse, MAX_PENDING_DELETION_WINDOW_DAYS, MIN_PENDING_DELETION_WINDOW_DAYS, OperationContext,
+    DescribeDataKeyWrappingRequest, DescribeDataKeyWrappingResponse, DescribeKeyRequest, DescribeKeyResponse, EncryptRequest,
+    EncryptResponse, GenerateDataKeyRequest, GenerateDataKeyResponse, ListKeysRequest, ListKeysResponse,
+    MAX_PENDING_DELETION_WINDOW_DAYS, MIN_PENDING_DELETION_WINDOW_DAYS, OperationContext, RewrapDataKeyRequest,
+    RewrapDataKeyResponse,
 };
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -167,6 +169,29 @@ impl KmsManager {
         self.backend.generate_data_key(request).await
     }
 
+    /// Re-wrap an existing data key envelope onto the master key's current
+    /// version, leaving the data key — and therefore every object body it
+    /// protects — untouched.
+    ///
+    /// Backends without retained version history reject this with
+    /// [`KmsError::UnsupportedCapability`]; check
+    /// [`Self::backend_capabilities`] before offering it.
+    pub async fn rewrap_data_key(&self, request: RewrapDataKeyRequest) -> Result<RewrapDataKeyResponse> {
+        self.backend.rewrap_data_key(request).await
+    }
+
+    /// Report which master key version wraps an existing data key envelope.
+    ///
+    /// The read-only side of [`Self::rewrap_data_key`], and the supported way to
+    /// ask that question: where the version is recorded differs per backend, so
+    /// callers must not inspect envelopes themselves.
+    pub async fn describe_data_key_wrapping(
+        &self,
+        request: DescribeDataKeyWrappingRequest,
+    ) -> Result<DescribeDataKeyWrappingResponse> {
+        self.backend.describe_data_key_wrapping(request).await
+    }
+
     /// Describe a key
     ///
     /// Audited as an internal operation; callers serving an authenticated

crates/kms/src/types.rs

@@ -959,3 +959,80 @@ impl Drop for DataKeyInfo {
         self.clear_plaintext();
     }
 }
+
+/// Request to rewrap an existing data key envelope under the current version of
+/// the master key that already wraps it.
+///
+/// Rewrap changes only the wrapping. The data key inside is never replaced, so
+/// object ciphertext, IV derivation and ETags are untouched — which is what
+/// makes it possible to retire an old master key version without rewriting a
+/// single object body.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RewrapDataKeyRequest {
+    /// The existing wrapped data key, exactly as it was persisted
+    pub ciphertext: Vec<u8>,
+    /// Encryption context the envelope was created with; the backend rejects
+    /// the request when the envelope's own context is not reproduced here, so a
+    /// caller cannot rewrap an envelope it could not have decrypted
+    pub encryption_context: HashMap<String, String>,
+}
+
+/// Request to report where an existing data key envelope sits relative to its
+/// master key's current version.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DescribeDataKeyWrappingRequest {
+    /// The existing wrapped data key, exactly as it was persisted
+    pub ciphertext: Vec<u8>,
+    /// Encryption context the envelope was created with, checked exactly as
+    /// [`RewrapDataKeyRequest`] checks it
+    pub encryption_context: HashMap<String, String>,
+}
+
+/// Where an existing wrapped data key sits relative to its master key's current
+/// version.
+///
+/// This is the only supported way to ask that question. The answer lives in a
+/// different place for every backend that rotates — a JSON field on the envelope
+/// for Vault KV2, the `vault:vN:` prefix of the ciphertext for Vault Transit,
+/// nowhere at all for backends that do not rotate — and a caller that reached
+/// into the envelope itself would be re-implementing that table, on the wrong
+/// side of the KMS boundary, once per call site.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct DescribeDataKeyWrappingResponse {
+    /// Master key wrapping the data key
+    pub key_id: String,
+    /// Version that wrapped the data key, when the backend can name it
+    pub key_version: Option<u32>,
+    /// The master key's current version, when the backend can name it
+    pub current_key_version: Option<u32>,
+    /// Whether the envelope is already wrapped by the current version, which is
+    /// exactly the condition under which a rewrap would be a no-op.
+    ///
+    /// Callers must read this rather than compare the two version fields. A
+    /// backend may leave either version unset while still knowing the answer,
+    /// and the two must never disagree: this flag is what a retirement scan
+    /// counts, and a rewrap sweep that rewrote envelopes the scan already
+    /// considered done would never converge.
+    pub is_current: bool,
+}
+
+/// Result of [`RewrapDataKeyRequest`].
+///
+/// The plaintext data key is deliberately absent: rewrap exists so that the
+/// data key can be re-wrapped without the caller ever holding it.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RewrapDataKeyResponse {
+    /// The wrapped data key to persist in place of the request's ciphertext.
+    /// Byte-identical to the request when `rewrapped` is false.
+    pub ciphertext: Vec<u8>,
+    /// Master key that wraps the data key; unchanged by a rewrap
+    pub key_id: String,
+    /// Master key version that wrapped the input, when the backend can name it
+    pub source_key_version: Option<u32>,
+    /// Master key version that wraps `ciphertext`, when the backend can name it
+    pub destination_key_version: Option<u32>,
+    /// Whether the wrapping actually changed. False means the input was already
+    /// wrapped by the current version and callers must not persist anything —
+    /// re-running a rewrap sweep has to converge without further writes.
+    pub rewrapped: bool,
+}