fix(kms): fail closed on missing or corrupt key material (#5475)

crates/kms/src/backends/local.rs

@@ -658,7 +658,20 @@ impl LocalKmsClient {
         }
 
         let content = fs::read(&key_path).await?;
-        let stored_key: StoredMasterKey = serde_json::from_slice(&content)?;
+
+        // Two-stage parse so an unrecognised protection marker is reported as an
+        // unsupported format (a newer build may still read the key) instead of being
+        // folded into generic corruption with every other malformed record.
+        let raw: serde_json::Value = serde_json::from_slice(&content)
+            .map_err(|e| KmsError::material_corrupt(key_id, format!("stored key record is not valid JSON: {e}")))?;
+        if let Some(marker) = raw.get("at_rest_protection")
+            && serde_json::from_value::<StoredKeyProtection>(marker.clone()).is_err()
+        {
+            let version = marker.as_str().map(str::to_owned).unwrap_or_else(|| marker.to_string());
+            return Err(KmsError::unsupported_format_version(key_id, version));
+        }
+        let stored_key: StoredMasterKey = serde_json::from_value(raw)
+            .map_err(|e| KmsError::material_corrupt(key_id, format!("stored key record does not deserialize: {e}")))?;
         if stored_key.key_id != key_id {
             return Err(KmsError::invalid_key(format!(
                 "Local KMS key file identity mismatch: expected {key_id:?}, found {:?}",
@@ -666,9 +679,15 @@ impl LocalKmsClient {
             )));
         }
 
+        // An empty material field is a damaged record, whatever the protection marker
+        // says. Fail closed: reads must never backfill or regenerate master key material.
+        if stored_key.encrypted_key_material.is_empty() {
+            return Err(KmsError::material_missing(key_id));
+        }
+
         let encrypted_bytes = BASE64
             .decode(&stored_key.encrypted_key_material)
-            .map_err(|e| KmsError::cryptographic_error("base64_decode", e.to_string()))?;
+            .map_err(|e| KmsError::material_corrupt(key_id, format!("stored key material is not valid base64: {e}")))?;
 
         let effective_protection = if stored_key.at_rest_protection == StoredKeyProtection::LegacyUnspecified {
             if stored_key.nonce.is_empty() {
@@ -692,7 +711,10 @@ impl LocalKmsClient {
                     ))
                 })?;
                 if stored_key.nonce.len() != 12 {
-                    return Err(KmsError::cryptographic_error("nonce", "Invalid nonce length"));
+                    return Err(KmsError::material_corrupt(
+                        key_id,
+                        format!("stored nonce has invalid length ({} bytes, expected 12)", stored_key.nonce.len()),
+                    ));
                 }
 
                 let mut nonce_array = [0u8; 12];
@@ -701,7 +723,7 @@ impl LocalKmsClient {
 
                 match cipher.decrypt(&nonce, encrypted_bytes.as_ref()) {
                     Ok(key_material) => key_material,
-                    Err(current_error) if stored_key.at_rest_protection == StoredKeyProtection::LegacyUnspecified => {
+                    Err(_) if stored_key.at_rest_protection == StoredKeyProtection::LegacyUnspecified => {
                         let legacy_cipher = self.legacy_master_cipher.as_ref().ok_or_else(|| {
                             KmsError::configuration_error(format!(
                                 "Local KMS key {key_id} is encrypted at rest and requires a configured master key"
@@ -709,9 +731,9 @@ impl LocalKmsClient {
                         })?;
                         legacy_cipher
                             .decrypt(&nonce, encrypted_bytes.as_ref())
-                            .map_err(|_| KmsError::cryptographic_error("decrypt", current_error.to_string()))?
+                            .map_err(|_| KmsError::material_authentication_failed(key_id))?
                     }
-                    Err(error) => return Err(KmsError::cryptographic_error("decrypt", error.to_string())),
+                    Err(_) => return Err(KmsError::material_authentication_failed(key_id)),
                 }
             }
             StoredKeyProtection::PlaintextDevOnly | StoredKeyProtection::LegacyUnspecified => {
@@ -1597,6 +1619,124 @@ mod tests {
         assert_eq!(decrypted, plaintext, "master key material must survive status transitions");
     }
 
+    /// Snapshot every file in the key directory as (name, content SHA-256, mtime).
+    /// Comparing snapshots proves the read paths performed no persistent write at all —
+    /// no rewrite, no "repair", no temp-file leftovers.
+    async fn snapshot_key_dir(dir: &std::path::Path) -> Vec<(String, Vec<u8>, std::time::SystemTime)> {
+        let mut entries = fs::read_dir(dir).await.expect("read key dir");
+        let mut snapshot = Vec::new();
+        while let Some(entry) = entries.next_entry().await.expect("next key dir entry") {
+            let content = fs::read(entry.path()).await.expect("read key dir file");
+            let modified = entry
+                .metadata()
+                .await
+                .expect("key dir file metadata")
+                .modified()
+                .expect("key dir file mtime");
+            snapshot.push((
+                entry.file_name().to_string_lossy().into_owned(),
+                Sha256::digest(&content).to_vec(),
+                modified,
+            ));
+        }
+        snapshot.sort();
+        snapshot
+    }
+
+    /// Poison matrix guard: every corruption class must surface its precise typed error
+    /// from every read path (get_key_material / describe_key / decrypt), and the key
+    /// directory must stay byte-for-byte identical. Restoring any historical "self-heal"
+    /// behaviour (regenerating or rewriting material on a failed read) flips either the
+    /// error assertion (read succeeds) or the snapshot assertion (directory changed).
+    #[tokio::test]
+    async fn read_paths_fail_closed_never_write_on_poisoned_key_files() {
+        let (client, temp_dir) = create_test_client().await;
+        let key_id = "poisoned-key";
+        client.create_key(key_id, "AES_256", None).await.expect("create key");
+
+        // Wrap a DEK while the key is healthy so the decrypt path can be exercised
+        // against each poisoned state of its master key.
+        let request = GenerateKeyRequest::new(key_id.to_string(), "AES_256".to_string());
+        let data_key = client.generate_data_key(&request, None).await.expect("generate data key");
+        let envelope_ciphertext = data_key.ciphertext.clone();
+
+        let key_path = client.master_key_path(key_id).expect("valid key id");
+        let pristine: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read pristine key file")).expect("decode pristine record");
+        let pristine_bytes = serde_json::to_vec_pretty(&pristine).expect("encode pristine record");
+
+        let with_field = |field: &str, value: serde_json::Value| {
+            let mut record = pristine.clone();
+            record[field] = value;
+            serde_json::to_vec_pretty(&record).expect("encode poisoned record")
+        };
+
+        let tampered_material = {
+            let mut material = BASE64
+                .decode(pristine["encrypted_key_material"].as_str().expect("material is a string"))
+                .expect("decode pristine material");
+            *material.last_mut().expect("material is not empty") ^= 0x01;
+            BASE64.encode(&material)
+        };
+
+        type PoisonCase = (&'static str, Vec<u8>, fn(&KmsError) -> bool);
+        let poisons: Vec<PoisonCase> = vec![
+            ("empty material", with_field("encrypted_key_material", serde_json::json!("")), |e| {
+                matches!(e, KmsError::MaterialMissing { .. })
+            }),
+            ("truncated JSON", pristine_bytes[..pristine_bytes.len() / 2].to_vec(), |e| {
+                matches!(e, KmsError::MaterialCorrupt { .. })
+            }),
+            (
+                "invalid base64",
+                with_field("encrypted_key_material", serde_json::json!("!!!not-base64!!!")),
+                |e| matches!(e, KmsError::MaterialCorrupt { .. }),
+            ),
+            ("wrong nonce length", with_field("nonce", serde_json::json!([0, 1, 2])), |e| {
+                matches!(e, KmsError::MaterialCorrupt { .. })
+            }),
+            (
+                "tampered AEAD",
+                with_field("encrypted_key_material", serde_json::json!(tampered_material)),
+                |e| matches!(e, KmsError::MaterialAuthenticationFailed { .. }),
+            ),
+            (
+                "unknown protection marker",
+                with_field("at_rest_protection", serde_json::json!("post-quantum-v2")),
+                |e| matches!(e, KmsError::UnsupportedFormatVersion { version, .. } if version == "post-quantum-v2"),
+            ),
+        ];
+
+        for (name, poisoned_content, expected) in poisons {
+            fs::write(&key_path, &poisoned_content).await.expect("write poisoned record");
+            let before = snapshot_key_dir(temp_dir.path()).await;
+
+            let error = client
+                .get_key_material(key_id)
+                .await
+                .expect_err("get_key_material must fail on poisoned material");
+            assert!(expected(&error), "{name}: get_key_material returned wrong variant: {error:?}");
+
+            let error = client
+                .describe_key(key_id, None)
+                .await
+                .expect_err("describe_key must fail on poisoned material");
+            assert!(expected(&error), "{name}: describe_key returned wrong variant: {error:?}");
+
+            let error = client
+                .decrypt(&DecryptRequest::new(envelope_ciphertext.clone()), None)
+                .await
+                .expect_err("decrypt must fail on poisoned material");
+            assert!(expected(&error), "{name}: decrypt returned wrong variant: {error:?}");
+
+            assert_eq!(
+                snapshot_key_dir(temp_dir.path()).await,
+                before,
+                "{name}: read paths must not write to the key directory"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn test_encryption_operations() {
         let (client, _temp_dir) = create_test_client().await;
@@ -1651,7 +1791,7 @@ mod tests {
             Ok(_) => panic!("wrong master key must fail initialization"),
             Err(error) => error,
         };
-        assert!(matches!(wrong_master_error, KmsError::CryptographicError { .. }));
+        assert!(matches!(wrong_master_error, KmsError::MaterialAuthenticationFailed { .. }));
     }
 
     #[tokio::test]
@@ -1906,7 +2046,7 @@ mod tests {
             .get_key_material("beta5-explicit-key")
             .await
             .expect_err("explicit current protection must not fall back to the beta.5 KDF");
-        assert!(matches!(explicit_error, KmsError::CryptographicError { .. }));
+        assert!(matches!(explicit_error, KmsError::MaterialAuthenticationFailed { .. }));
 
         let wrong_key_error = match LocalKmsClient::new(LocalConfig {
             key_dir: temp_dir.path().to_path_buf(),
@@ -1918,7 +2058,7 @@ mod tests {
             Ok(_) => panic!("wrong beta.5 master key must fail initialization"),
             Err(error) => error,
         };
-        assert!(matches!(wrong_key_error, KmsError::CryptographicError { .. }));
+        assert!(matches!(wrong_key_error, KmsError::MaterialAuthenticationFailed { .. }));
     }
 
     /// R03-CAN-072 / R03-CAN-073: key identifiers arrive from request input, so every path

crates/kms/src/backends/vault.rs

@@ -66,6 +66,35 @@ struct VaultKeyData {
     encrypted_key_material: String,
 }
 
+/// Decode and validate the stored master key material of a [`VaultKeyData`] record.
+///
+/// This is the single read-side gate for KV2 key material: missing or undecodable
+/// material must fail closed with a typed error and must never be regenerated or
+/// written back (regenerating would orphan every DEK wrapped by the original key).
+/// Kept synchronous and free of Vault I/O so the poison matrix is unit-testable
+/// without a live Vault.
+fn decode_stored_key_material(key_id: &str, encrypted_material: &str) -> Result<Vec<u8>> {
+    if encrypted_material.is_empty() {
+        return Err(KmsError::material_missing(key_id));
+    }
+
+    // Mirrors `decrypt_key_material`: stored material is currently base64 without an
+    // additional encryption layer.
+    let key_material = general_purpose::STANDARD
+        .decode(encrypted_material)
+        .map_err(|e| KmsError::material_corrupt(key_id, format!("stored key material is not valid base64: {e}")))?;
+
+    // Key material must be exactly 32 bytes for AES-256.
+    if key_material.len() != 32 {
+        return Err(KmsError::material_corrupt(
+            key_id,
+            format!("stored key material has invalid length ({} bytes, expected 32)", key_material.len()),
+        ));
+    }
+
+    Ok(key_material)
+}
+
 impl VaultKmsClient {
     /// Create a new Vault KMS client
     ///
@@ -139,47 +168,10 @@ impl VaultKmsClient {
 
     /// Get the actual key material for a master key
     async fn get_key_material(&self, key_id: &str) -> Result<Vec<u8>> {
-        let mut key_data = self.get_key_data(key_id).await?;
-
-        // If encrypted_key_material is empty, generate and store it (fix for old keys)
-        if key_data.encrypted_key_material.is_empty() {
-            warn!(key_id, "Vault KMS key material missing; regenerating");
-            let key_material = generate_key_material(&key_data.algorithm)?;
-            key_data.encrypted_key_material = self.encrypt_key_material(&key_material).await?;
-            // Store the updated key data back to Vault
-            self.store_key_data(key_id, &key_data).await?;
-            return Ok(key_material);
-        }
-
-        let key_material = match self.decrypt_key_material(&key_data.encrypted_key_material).await {
-            Ok(km) => km,
-            Err(e) => {
-                // Never regenerate/overwrite the master key on a decrypt failure: that would
-                // destroy the original material and make every DEK wrapped by this key
-                // permanently undecryptable. Surface the error so the read fails recoverably
-                // instead of causing silent data loss.
-                warn!(key_id, error = %e, "Vault KMS key material could not be decoded");
-                return Err(KmsError::cryptographic_error(
-                    "decrypt",
-                    format!("Stored key material for {key_id} is corrupted: {e}"),
-                ));
-            }
-        };
-
-        // Validate key material length (should be 32 bytes for AES-256).
-        if key_material.len() != 32 {
-            // As above: do not overwrite the stored key. Report the fault instead.
-            warn!(key_id, len = key_material.len(), "Vault KMS key material has invalid length");
-            return Err(KmsError::cryptographic_error(
-                "decrypt",
-                format!(
-                    "Stored key material for {key_id} has invalid length ({} bytes, expected 32)",
-                    key_material.len()
-                ),
-            ));
-        }
-
-        Ok(key_material)
+        let key_data = self.get_key_data(key_id).await?;
+        decode_stored_key_material(key_id, &key_data.encrypted_key_material).inspect_err(|error| {
+            warn!(key_id, %error, "Vault KMS key material failed validation");
+        })
     }
 
     /// Encrypt data using a master key
@@ -213,14 +205,15 @@ impl VaultKmsClient {
 
         // Get existing key data to preserve encrypted_key_material and other fields
         // This is called after create_key, so the key should already exist
-        let mut existing_key_data = self.get_key_data(key_id).await?;
+        let existing_key_data = self.get_key_data(key_id).await?;
 
-        // If encrypted_key_material is empty, generate it (this handles the case where
-        // an old key was created without proper key material)
+        // A key that was just created must already carry material; an empty value means
+        // the create flow failed to persist it. Fail closed instead of minting replacement
+        // material: silently generating a new key here would mask the broken create and
+        // orphan any DEK already wrapped by a different copy of this key.
         if existing_key_data.encrypted_key_material.is_empty() {
             warn!(key_id, "Vault KMS key metadata missing encrypted key material");
-            let key_material = generate_key_material(&existing_key_data.algorithm)?;
-            existing_key_data.encrypted_key_material = self.encrypt_key_material(&key_material).await?;
+            return Err(KmsError::material_missing(key_id));
         }
 
         // Update only the metadata fields, preserving the encrypted_key_material
@@ -623,6 +616,14 @@ impl VaultKmsBackend {
         // Get the current key data from Vault
         let mut key_data = self.client.get_key_data(key_id).await?;
 
+        // This is a read-modify-write of the whole VaultKeyData document. Refuse to write
+        // back a record whose key material is missing: persisting it would cement the
+        // empty-material state under a fresh document version. A damaged key must go
+        // through an explicit repair operation, not a metadata update.
+        if key_data.encrypted_key_material.is_empty() {
+            return Err(KmsError::material_missing(key_id));
+        }
+
         // Update the status based on the new metadata
         key_data.status = match metadata.key_state {
             KeyState::Enabled => KeyStatus::Active,
@@ -843,6 +844,46 @@ mod tests {
     use super::*;
     use crate::config::{VaultAuthMethod, VaultConfig};
 
+    /// Poison matrix for the read-side material gate. Every corruption class must fail
+    /// closed with its typed error; reintroducing any "self-heal" (regenerate on empty or
+    /// undecodable material) turns one of these expected errors into an Ok and fails the
+    /// test. Offline on purpose: `decode_stored_key_material` has no Vault I/O.
+    #[test]
+    fn decode_stored_key_material_fails_closed_on_poisoned_values() {
+        // Empty material means the record lost its key, not that a new one may be minted.
+        assert!(matches!(
+            decode_stored_key_material("poisoned", ""),
+            Err(KmsError::MaterialMissing { key_id }) if key_id == "poisoned"
+        ));
+
+        // Invalid base64.
+        assert!(matches!(
+            decode_stored_key_material("poisoned", "!!!not-base64!!!"),
+            Err(KmsError::MaterialCorrupt { key_id, .. }) if key_id == "poisoned"
+        ));
+
+        // Truncated material: valid base64 of fewer than 32 bytes.
+        let truncated = general_purpose::STANDARD.encode([0x42u8; 16]);
+        assert!(matches!(
+            decode_stored_key_material("poisoned", &truncated),
+            Err(KmsError::MaterialCorrupt { key_id, .. }) if key_id == "poisoned"
+        ));
+
+        // Oversized material: valid base64 of more than 32 bytes.
+        let oversized = general_purpose::STANDARD.encode([0x42u8; 33]);
+        assert!(matches!(
+            decode_stored_key_material("poisoned", &oversized),
+            Err(KmsError::MaterialCorrupt { key_id, .. }) if key_id == "poisoned"
+        ));
+
+        // Well-formed material still decodes.
+        let valid = general_purpose::STANDARD.encode([0x42u8; 32]);
+        assert_eq!(
+            decode_stored_key_material("healthy", &valid).expect("valid material must decode"),
+            vec![0x42u8; 32]
+        );
+    }
+
     #[tokio::test]
     #[ignore] // Requires a running Vault instance
     async fn test_vault_client_integration() {
@@ -928,9 +969,13 @@ mod tests {
         client.store_key_data(&key_id, &key_data).await.expect("store corrupt");
 
         // Reading the material must now ERROR, not silently regenerate + overwrite.
+        let error = client
+            .get_key_material(&key_id)
+            .await
+            .expect_err("corrupted key material must yield an error, not a fresh key");
         assert!(
-            client.get_key_material(&key_id).await.is_err(),
-            "corrupted key material must yield an error, not a fresh key"
+            matches!(error, KmsError::MaterialCorrupt { .. }),
+            "expected MaterialCorrupt, got {error:?}"
         );
 
         // And the stored (corrupted) material must be UNCHANGED.
@@ -941,6 +986,41 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[ignore] // Requires a running Vault instance (dev mode)
+    async fn test_empty_key_material_does_not_regenerate() {
+        // Regression: get_key_material previously treated empty stored material as a
+        // bootstrap case and silently generated + persisted a fresh master key on the
+        // read path. Empty material must instead fail closed as MaterialMissing and
+        // leave the stored record untouched.
+        let client = VaultKmsClient::new(integration_vault_config(), Duration::from_secs(30))
+            .await
+            .expect("client");
+
+        let key_id = format!("empty-{}", uuid::Uuid::new_v4());
+        client.create_key(&key_id, "AES_256", None).await.expect("create");
+
+        let mut key_data = client.get_key_data(&key_id).await.expect("read");
+        key_data.encrypted_key_material = String::new();
+        client.store_key_data(&key_id, &key_data).await.expect("store empty");
+
+        let error = client
+            .get_key_material(&key_id)
+            .await
+            .expect_err("empty key material must yield an error, not a fresh key");
+        assert!(
+            matches!(error, KmsError::MaterialMissing { .. }),
+            "expected MaterialMissing, got {error:?}"
+        );
+
+        // The stored record must still hold the empty value: no regeneration, no write.
+        let after = client.get_key_data(&key_id).await.expect("reread");
+        assert!(
+            after.encrypted_key_material.is_empty(),
+            "get_key_material must not backfill missing master key material"
+        );
+    }
+
     #[tokio::test]
     #[ignore] // Requires a running Vault instance (dev mode)
     async fn test_vault_cancel_key_deletion_persists_state() {

crates/kms/src/backends/vault_transit.rs

@@ -307,10 +307,17 @@ impl VaultTransitKmsClient {
             return Ok(persisted);
         }
 
+        // Deliberate exemption from the "read paths never write" rule (rustfs#4256 /
+        // rustfs#4262): transit keys created before metadata persistence existed have no
+        // KV record at all, so failing closed here would brick every pre-existing transit
+        // key. The synthesised record only describes metadata — key material lives solely
+        // inside Vault's transit engine and is never generated or written by this path.
+        //
         // Verify the transit key actually exists in Vault before synthesising.
         self.read_transit_key(key_id).await?;
         let metadata = TransitKeyMetadata::synthesized();
-        // Persist the synthesised metadata so future cache misses pick it up.
+        // Persist the synthesised metadata so future cache misses pick it up (best
+        // effort: the KV write failing must not fail the read).
         let _ = self.write_metadata_to_kv(key_id, &metadata).await;
         self.metadata_cache.write().await.insert(key_id.to_string(), metadata.clone());
         Ok(metadata)

crates/kms/src/error.rs

@@ -98,6 +98,28 @@ pub enum KmsError {
     /// Backend operation aborted by cancellation or shutdown
     #[error("Operation cancelled: {message}")]
     OperationCancelled { message: String },
+
+    // New variants must be appended below (never inserted above) so that
+    // concurrent additions rebase without conflicts.
+    /// Persisted key material is absent from an otherwise readable key record
+    #[error(
+        "Key material missing for key {key_id}: the stored record has no key material; restore it from backup or repair the key explicitly"
+    )]
+    MaterialMissing { key_id: String },
+
+    /// Persisted key material exists but cannot be decoded
+    #[error("Key material corrupt for key {key_id}: {message}")]
+    MaterialCorrupt { key_id: String, message: String },
+
+    /// Persisted key material failed authenticated decryption
+    #[error(
+        "Key material authentication failed for key {key_id}: the stored material cannot be decrypted with the configured master key"
+    )]
+    MaterialAuthenticationFailed { key_id: String },
+
+    /// Persisted key record uses a format version unknown to this build
+    #[error("Unsupported key format version {version:?} for key {key_id}")]
+    UnsupportedFormatVersion { key_id: String, version: String },
 }
 
 impl KmsError {
@@ -205,6 +227,32 @@ impl KmsError {
     pub fn operation_cancelled<S: Into<String>>(message: S) -> Self {
         Self::OperationCancelled { message: message.into() }
     }
+
+    /// Create a material missing error
+    pub fn material_missing<S: Into<String>>(key_id: S) -> Self {
+        Self::MaterialMissing { key_id: key_id.into() }
+    }
+
+    /// Create a material corrupt error
+    pub fn material_corrupt<S1: Into<String>, S2: Into<String>>(key_id: S1, message: S2) -> Self {
+        Self::MaterialCorrupt {
+            key_id: key_id.into(),
+            message: message.into(),
+        }
+    }
+
+    /// Create a material authentication failed error
+    pub fn material_authentication_failed<S: Into<String>>(key_id: S) -> Self {
+        Self::MaterialAuthenticationFailed { key_id: key_id.into() }
+    }
+
+    /// Create an unsupported format version error
+    pub fn unsupported_format_version<S1: Into<String>, S2: Into<String>>(key_id: S1, version: S2) -> Self {
+        Self::UnsupportedFormatVersion {
+            key_id: key_id.into(),
+            version: version.into(),
+        }
+    }
 }
 
 /// Convert from standard library errors

rustfs/src/admin/handlers/kms_keys.rs

@@ -851,6 +851,13 @@ impl Operation for DeleteKmsKeyHandler {
                 let status = match &e {
                     KmsError::KeyNotFound { .. } => StatusCode::NOT_FOUND,
                     KmsError::InvalidOperation { .. } | KmsError::ValidationError { .. } => StatusCode::BAD_REQUEST,
+                    // Damaged or missing key material is an integrity fault of an existing
+                    // key: it must surface as a server error, never as NOT_FOUND (the key
+                    // exists) and never as a retryable backend outage.
+                    KmsError::MaterialMissing { .. }
+                    | KmsError::MaterialCorrupt { .. }
+                    | KmsError::MaterialAuthenticationFailed { .. }
+                    | KmsError::UnsupportedFormatVersion { .. } => StatusCode::INTERNAL_SERVER_ERROR,
                     _ => StatusCode::INTERNAL_SERVER_ERROR,
                 };
                 let response = DeleteKmsKeyResponse {
@@ -1265,6 +1272,13 @@ impl Operation for DescribeKmsKeyHandler {
                 let status = match &e {
                     KmsError::KeyNotFound { .. } => StatusCode::NOT_FOUND,
                     KmsError::InvalidOperation { .. } => StatusCode::BAD_REQUEST,
+                    // Damaged or missing key material is an integrity fault of an existing
+                    // key: it must surface as a server error, never as NOT_FOUND (the key
+                    // exists) and never as a retryable backend outage.
+                    KmsError::MaterialMissing { .. }
+                    | KmsError::MaterialCorrupt { .. }
+                    | KmsError::MaterialAuthenticationFailed { .. }
+                    | KmsError::UnsupportedFormatVersion { .. } => StatusCode::INTERNAL_SERVER_ERROR,
                     _ => StatusCode::INTERNAL_SERVER_ERROR,
                 };
 

rustfs/src/error.rs

@@ -538,6 +538,29 @@ mod tests {
         assert_eq!(api_error.code, S3ErrorCode::InternalError);
     }
 
+    #[test]
+    fn test_kms_material_faults_map_to_internal_error_not_retryable_or_not_found() {
+        // Missing/corrupt key material is a persistent integrity fault of an existing key.
+        // It must not be disguised as a retryable backend outage (503 invites pointless
+        // retries) nor as NoSuchKey/404 (which suggests the key can be recreated —
+        // recreating it would orphan every DEK wrapped by the original material).
+        let material_faults = [
+            rustfs_kms::KmsError::material_missing("key-a"),
+            rustfs_kms::KmsError::material_corrupt("key-a", "truncated record"),
+            rustfs_kms::KmsError::material_authentication_failed("key-a"),
+            rustfs_kms::KmsError::unsupported_format_version("key-a", "v99"),
+        ];
+
+        for fault in material_faults {
+            let description = fault.to_string();
+            let api_error = ApiError::from(StorageError::other(fault));
+
+            assert_eq!(api_error.code, S3ErrorCode::InternalError, "wrong code for: {description}");
+            assert_ne!(api_error.code, S3ErrorCode::ServiceUnavailable, "must not be retryable: {description}");
+            assert_ne!(api_error.code, S3ErrorCode::NoSuchKey, "must not report key-not-found: {description}");
+        }
+    }
+
     #[test]
     fn test_api_error_from_storage_error_mappings() {
         let test_cases = vec![