fix(kms): fail closed on Local key records that cannot be interpreted (#5606)

* fix(kms): fail closed on Local key records this build cannot interpret

The Local backend's protection marker is the only version discriminator its
key records have, and three readers walked past it.

`ensure_missing_salt_can_be_generated` skipped every record it could not read
or parse, so a directory whose protection state is unknown still got a fresh
salt published before startup validation failed. That write is the
irreversible step: the next startup finds a salt file, never re-enters the
guard, and the evidence that the real salt was lost is gone. Every record the
guard now rejects already failed startup key validation a few lines later, so
no directory that initializes today stops initializing.

Backup export and restore folded an unknown marker into "material corrupt" /
"bundle corrupted". The record is intact and a newer build reads it fine, so
the operator response is a version change, not a disaster recovery. Both now
classify the marker before their schema parse, sharing one probe with the
backend reader.

`list_keys` dropped any record it could not decode from the page. Concurrent
removal stays a skip; anything else fails the listing rather than answering
"these are your keys" with a set that silently omits one.

* test(kms): cover every fail-closed path around the Local protection marker

Each test fails on the pre-fix code in the way the fix is about: the salt
cases because a replacement salt is published before startup validation
fails, the export and restore cases because the verdict comes back as
corruption, and the listing case because the record is edited out of the page.

The restore commit marker's unknown-version branch had no test at all,
unlike its Vault counterpart; it is now driven from the decoder, from the
restore entry point, and from backend startup.

Also states the widened salt guard in the Local backend operations doc,
including the operator recovery path for an unrecognized record.

* fix(kms): say 'not a readable JSON object' when the marker probe cannot parse

The probe now fails on any input that is not a JSON object, not only on
malformed JSON, so the message must cover both.

* test(kms): assert the salt file before the error variant

The replacement salt is written before the error the guard reports, so the
file assertion is the one that fails on a regression.

* test(kms): guard the new backup error variant's display string

crates/kms/src/backends/local.rs

@@ -480,6 +480,36 @@ pub(crate) enum StoredKeyProtection {
     PlaintextDevOnly,
 }
 
+/// The record's `at_rest_protection` value when this build cannot interpret
+/// it, rendered for diagnostics. `Ok(None)` means the marker is absent
+/// (pre-beta.9 records) or names a protection mode this build implements.
+///
+/// Every reader of a stored key record must consult this before its own
+/// schema parse. Letting a strict [`StoredKeyProtection`] field fail inside a
+/// larger struct collapses "written by a newer build" into "corrupt", and an
+/// operator who reads corruption starts a disaster recovery instead of a
+/// version rollback. The probe deliberately ignores every other field, so the
+/// verdict is available even for records whose schema this build cannot
+/// satisfy, and no key material is copied out of the caller's buffer.
+///
+/// `Err` carries the JSON error so callers can keep their own classification
+/// for bytes that are not a record at all.
+pub(crate) fn unknown_protection_marker(record: &[u8]) -> serde_json::Result<Option<String>> {
+    #[derive(Deserialize)]
+    struct MarkerProbe {
+        #[serde(default)]
+        at_rest_protection: Option<serde_json::Value>,
+    }
+
+    let Some(marker) = serde_json::from_slice::<MarkerProbe>(record)?.at_rest_protection else {
+        return Ok(None);
+    };
+    if serde_json::from_value::<StoredKeyProtection>(marker.clone()).is_ok() {
+        return Ok(None);
+    }
+    Ok(Some(marker.as_str().map(str::to_owned).unwrap_or_else(|| marker.to_string())))
+}
+
 /// Serializable representation of a master key stored on disk
 #[derive(Debug, Clone, Serialize, Deserialize)]
 struct StoredMasterKey {
@@ -741,10 +771,18 @@ impl LocalKmsClient {
     /// real problem (restore the salt file or the whole directory) instead of
     /// a generic decrypt failure.
     ///
-    /// Files that do not parse are ignored here — startup key validation
-    /// reports them with their own errors right after. Legacy pre-marker files
-    /// are also ignored: pre-beta.9 directories legitimately have no salt file
-    /// yet, and an empty directory must keep initializing as before.
+    /// A record this build cannot read or cannot interpret blocks generation
+    /// just as hard. Skipping it would publish a fresh salt over a directory
+    /// whose protection state is unknown, and that publication is
+    /// irreversible in the only way that matters: the next startup finds a
+    /// salt file, never re-enters this guard, and the evidence that the real
+    /// salt was missing is gone. Every record this rejects also fails startup
+    /// key validation a few lines later, so no directory that initializes
+    /// today stops initializing — the salt is simply no longer written first.
+    ///
+    /// Legacy pre-marker files stay allowed: they parse here, pre-beta.9
+    /// directories legitimately have no salt file yet, and an empty directory
+    /// must keep initializing as before.
     async fn ensure_missing_salt_can_be_generated(config: &LocalConfig) -> Result<()> {
         #[derive(Deserialize)]
         struct ProtectionProbe {
@@ -758,12 +796,23 @@ impl LocalKmsClient {
             if path.extension().is_none_or(|extension| extension != "key") {
                 continue;
             }
-            let Ok(content) = fs::read(&path).await else {
-                continue;
-            };
-            let Ok(probe) = serde_json::from_slice::<ProtectionProbe>(&content) else {
-                continue;
-            };
+            let content = fs::read(&path).await.map_err(|error| {
+                KmsError::configuration_error(format!(
+                    "Local KMS master key salt at {} is missing and key record {} cannot be read ({error}); \
+                     refusing to generate a replacement salt while a record's protection state is unknown",
+                    Self::master_key_salt_path(config).display(),
+                    path.display()
+                ))
+            })?;
+            let probe = serde_json::from_slice::<ProtectionProbe>(&content).map_err(|error| {
+                KmsError::configuration_error(format!(
+                    "Local KMS master key salt at {} is missing and key record {} is not interpretable by this build ({error}); \
+                     refusing to generate a replacement salt — restore the salt file from backup, or run a build that \
+                     understands the record",
+                    Self::master_key_salt_path(config).display(),
+                    path.display()
+                ))
+            })?;
             if probe.at_rest_protection == StoredKeyProtection::EncryptedMasterKey {
                 return Err(KmsError::configuration_error(format!(
                     "Local KMS master key salt at {} is missing but {} is marked encrypted-master-key; \
@@ -805,15 +854,12 @@ impl LocalKmsClient {
         // 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());
+        let unknown_marker = unknown_protection_marker(&content)
+            .map_err(|e| KmsError::material_corrupt(key_id, format!("stored key record is not a readable JSON object: {e}")))?;
+        if let Some(version) = unknown_marker {
             return Err(KmsError::unsupported_format_version(key_id, version));
         }
-        let stored_key: StoredMasterKey = serde_json::from_value(raw)
+        let stored_key: StoredMasterKey = serde_json::from_slice(&content)
             .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!(
@@ -1254,16 +1300,21 @@ impl LocalKmsClient {
         // by the requested limit rather than by the size of the key set.
         let mut keys = Vec::with_capacity(page.items.len());
         for key_id in page.items {
-            // A key that vanished or cannot be decoded is dropped from the page
-            // instead of failing it: concurrent removal is normal, and the
-            // cursor is derived from the identifier list, so the listing still
-            // advances past it.
             let key_info = match self.describe_key(key_id, None).await {
                 Ok(key_info) => key_info,
-                Err(error) => {
-                    debug!(key_id, %error, "skipping unreadable key while listing");
+                // A key that vanished between the scan and the read is dropped
+                // from the page: concurrent removal is normal, and the cursor
+                // is derived from the identifier list, so the listing still
+                // advances past it.
+                Err(KmsError::KeyNotFound { .. }) => {
+                    debug!(key_id, "skipping key removed while listing");
                     continue;
                 }
+                // Anything else means the record is still there and this build
+                // cannot interpret it. Dropping it would answer "these are
+                // your keys" with a set that silently omits one, and the
+                // deletion sweep would count a census it never fully saw.
+                Err(error) => return Err(error),
             };
 
             if let Some(ref status_filter) = request.status_filter
@@ -2962,6 +3013,129 @@ mod tests {
         );
     }
 
+    /// The salt guard must fail closed on every record it cannot interpret,
+    /// not just on the ones that spell out `encrypted-master-key`.
+    ///
+    /// Skipping such a record publishes a fresh salt, and that write is the
+    /// irreversible step: the next startup sees a salt file, never re-enters
+    /// the guard, and the only signal that the real salt was lost is gone.
+    /// Every input below also fails startup key validation, so this rejects
+    /// nothing that used to initialize — it only stops the salt from being
+    /// written before that failure.
+    #[tokio::test]
+    async fn missing_salt_with_uninterpretable_key_records_fails_closed_without_generating_a_salt() {
+        let (client, temp_dir) = create_test_client().await;
+        client.create_key("sealed-key", "AES_256", None).await.expect("create key");
+        let config = client.config.clone();
+        let key_path = client.master_key_path("sealed-key").expect("valid key id");
+        let salt_path = LocalKmsClient::master_key_salt_path(&config);
+        let pristine: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key file")).expect("decode record");
+        drop(client);
+
+        let mut newer_build_record = pristine.clone();
+        newer_build_record["at_rest_protection"] = serde_json::json!("post-quantum-v2");
+        let newer_build_record = serde_json::to_vec_pretty(&newer_build_record).expect("encode record");
+        let truncated = {
+            let bytes = serde_json::to_vec_pretty(&pristine).expect("encode record");
+            bytes[..bytes.len() / 2].to_vec()
+        };
+
+        for (name, content) in [
+            ("record from a newer build", newer_build_record),
+            ("record that does not decode", truncated),
+        ] {
+            fs::write(&key_path, &content).await.expect("write record");
+            fs::remove_file(&salt_path).await.ok();
+
+            let error = match LocalKmsClient::new(config.clone()).await {
+                Ok(_) => panic!("{name}: a missing salt must not be papered over"),
+                Err(error) => error,
+            };
+            // Asserted first: publishing a replacement salt is the
+            // irreversible step, and it happens before any error is produced.
+            assert!(!salt_path.exists(), "{name}: a replacement salt must never be generated");
+            assert!(
+                matches!(error, KmsError::ConfigurationError { .. }),
+                "{name}: expected a salt-specific configuration error, got {error:?}"
+            );
+            assert!(error.to_string().contains("salt"), "{name}: error must point at the salt: {error}");
+            assert_eq!(
+                sorted_dir_file_names(temp_dir.path()).await,
+                vec!["sealed-key.key".to_string()],
+                "{name}: the failed startup must not modify the key directory"
+            );
+        }
+    }
+
+    /// Compatibility floor for the guard above: a record written before the
+    /// protection marker existed carries no marker at all, and such a
+    /// directory legitimately has no salt yet. It must keep initializing.
+    /// A record this build cannot interpret must not be edited out of a
+    /// listing. The page would claim to describe the key set while omitting a
+    /// key that is still on disk, and the deletion sweep — which counts the
+    /// lifecycle gauges out of the pages it lists — would report a census it
+    /// never fully saw as complete.
+    #[tokio::test]
+    async fn list_keys_fails_closed_on_a_record_it_cannot_interpret() {
+        let (client, _temp_dir) = create_test_client().await;
+        client.create_key("alpha", "AES_256", None).await.expect("create alpha");
+        client.create_key("beta", "AES_256", None).await.expect("create beta");
+
+        let key_path = client.master_key_path("beta").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read record")).expect("decode record");
+        record["at_rest_protection"] = serde_json::json!("post-quantum-v2");
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode record"))
+            .await
+            .expect("write record");
+
+        let error = client
+            .list_keys(&ListKeysRequest::default(), None)
+            .await
+            .expect_err("a listing must not quietly omit a key it cannot read");
+        assert!(
+            matches!(&error, KmsError::UnsupportedFormatVersion { key_id, version }
+                if key_id == "beta" && version == "post-quantum-v2"),
+            "got {error:?}"
+        );
+    }
+
+    #[tokio::test]
+    async fn missing_salt_with_pre_marker_key_records_still_initializes() {
+        let (dev_client, temp_dir) = create_dev_mode_client().await;
+        dev_client
+            .create_key("pre-marker-key", "AES_256", None)
+            .await
+            .expect("create key");
+        let key_path = dev_client.master_key_path("pre-marker-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key file")).expect("decode record");
+        drop(dev_client);
+
+        // Pre-beta.9 records carry no protection marker at all, and their
+        // directories legitimately hold no salt file yet.
+        record
+            .as_object_mut()
+            .expect("record is an object")
+            .remove("at_rest_protection");
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode record"))
+            .await
+            .expect("write record");
+
+        let client = LocalKmsClient::new(test_config(temp_dir.path()))
+            .await
+            .expect("a pre-marker directory must keep initializing");
+        assert!(
+            LocalKmsClient::master_key_salt_path(&client.config).exists(),
+            "first-time salt creation must still happen"
+        );
+        client
+            .describe_key("pre-marker-key", None)
+            .await
+            .expect("the pre-marker record must stay readable");
+    }
+
     #[tokio::test]
     async fn missing_salt_with_only_plaintext_dev_keys_still_initializes() {
         let (dev_client, temp_dir) = create_dev_mode_client().await;

crates/kms/src/backup/dry_run.rs

@@ -27,8 +27,10 @@ use serde::{Deserialize, Serialize};
 
 /// Machine-readable category of a restore blocker.
 ///
-/// The first six codes mirror the [`BackupError`] variants; the remaining
-/// codes cover preflight conditions that are not bundle defects.
+/// The first six codes mirror the [`BackupError`] variants — an unknown
+/// format version reports the same code whether it came from the manifest or
+/// from a bundled key record — and the remaining codes cover preflight
+/// conditions that are not bundle defects.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "kebab-case")]
 pub enum RestoreBlockerCode {
@@ -73,6 +75,7 @@ impl From<&BackupError> for RestoreBlocker {
             BackupError::WrongKek { .. } => RestoreBlockerCode::WrongBackupKek,
             BackupError::MissingArtifact { .. } => RestoreBlockerCode::MissingArtifact,
             BackupError::IncompleteBundle { .. } => RestoreBlockerCode::IncompleteBundle,
+            BackupError::UnsupportedRecordVersion { .. } => RestoreBlockerCode::UnknownFormatVersion,
         };
         Self {
             code,

crates/kms/src/backup/error.rs

@@ -60,6 +60,13 @@ pub enum BackupError {
     /// restored, regardless of how much of it is readable.
     #[error("backup bundle is incomplete ({reason}); incomplete bundles must never be restored")]
     IncompleteBundle { reason: String },
+
+    /// A bundled key record declares an at-rest protection mode this build
+    /// does not implement. Deliberately distinct from [`Self::Corrupted`]:
+    /// the record is intact and a newer build reads it fine, so the operator
+    /// response is a version change, not a disaster recovery.
+    #[error("bundled key record '{key_id}' declares unsupported at-rest protection {version:?}; this build cannot restore it")]
+    UnsupportedRecordVersion { key_id: String, version: String },
 }
 
 impl BackupError {
@@ -129,5 +136,14 @@ mod tests {
             BackupError::incomplete_bundle("manifest has no completeness marker").to_string(),
             "backup bundle is incomplete (manifest has no completeness marker); incomplete bundles must never be restored"
         );
+
+        let unsupported_record = BackupError::UnsupportedRecordVersion {
+            key_id: "alpha".to_string(),
+            version: "post-quantum-v2".to_string(),
+        };
+        assert_eq!(
+            unsupported_record.to_string(),
+            "bundled key record 'alpha' declares unsupported at-rest protection \"post-quantum-v2\"; this build cannot restore it"
+        );
     }
 }

crates/kms/src/backup/local_export.rs

@@ -48,7 +48,7 @@
 //! published. A crash at any earlier point leaves a bundle without a
 //! manifest, which decodes as an incomplete bundle and can never be restored.
 
-use crate::backends::local::{LocalKmsClient, StoredKeyProtection};
+use crate::backends::local::{LocalKmsClient, StoredKeyProtection, unknown_protection_marker};
 use crate::backup::capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
 use crate::backup::error::BackupError;
 use crate::backup::manifest::{
@@ -376,7 +376,16 @@ async fn collect_snapshot(client: &LocalKmsClient) -> Result<CollectedSnapshot>
 
         let raw = Zeroizing::new(fs::read(&path).await?);
         // Any unreadable record aborts the export: a bundle silently missing
-        // one key is worse than no bundle at all.
+        // one key is worse than no bundle at all. The protection marker is
+        // classified first so a record from a newer build keeps its own
+        // verdict — an operator who reads "material corrupt" starts a
+        // disaster recovery for what is only a version mismatch.
+        let unknown_marker = unknown_protection_marker(&raw).map_err(|error| {
+            KmsError::material_corrupt(&stem, format!("stored key record is not a readable JSON object: {error}"))
+        })?;
+        if let Some(version) = unknown_marker {
+            return Err(KmsError::unsupported_format_version(&stem, version));
+        }
         let probe: StoredRecordProbe = serde_json::from_slice(&raw)
             .map_err(|error| KmsError::material_corrupt(&stem, format!("stored key record does not deserialize: {error}")))?;
         if probe.key_id != stem {
@@ -1115,4 +1124,30 @@ mod tests {
             .expect_err("identity mismatch must abort the export");
         assert!(matches!(error, KmsError::InvalidKey { .. }), "got {error:?}");
     }
+
+    /// A record written by a newer build must abort the export as an
+    /// unsupported format, not as corrupt material: the two verdicts send the
+    /// operator down completely different runbooks, and only one of them is
+    /// true here.
+    #[tokio::test]
+    async fn record_from_a_newer_build_aborts_export_as_unsupported_format() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("alpha", "AES_256", None).await.expect("create key");
+
+        let record_path = client.key_directory().join("alpha.key");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&record_path).expect("read record")).expect("decode record");
+        record["at_rest_protection"] = serde_json::json!("post-quantum-v2");
+        std::fs::write(&record_path, serde_json::to_vec_pretty(&record).expect("encode record")).expect("write record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("an uninterpretable record must abort the export");
+        assert!(
+            matches!(&error, KmsError::UnsupportedFormatVersion { key_id, version }
+                if key_id == "alpha" && version == "post-quantum-v2"),
+            "got {error:?}"
+        );
+    }
 }

crates/kms/src/backup/local_restore.rs

@@ -54,7 +54,7 @@
 
 use crate::backends::local::{
     LOCAL_KMS_MASTER_KEY_SALT_FILE, LOCAL_KMS_MASTER_KEY_SALT_LEN, LOCAL_RESTORE_COMMIT_MARKER_FILE, LocalKmsClient,
-    StoredKeyProtection, durable_file, is_orphan_commit_temp_name, validate_key_id,
+    StoredKeyProtection, durable_file, is_orphan_commit_temp_name, unknown_protection_marker, validate_key_id,
 };
 use crate::backup::capability::AtRestProtection;
 use crate::backup::dry_run::{
@@ -605,6 +605,14 @@ fn decode_key_record(
         .to_string();
     validate_key_id(&stem)?;
 
+    // Classify the protection marker before the schema parse: a record from a
+    // newer build is not a damaged bundle, and reporting it as corruption
+    // sends the operator into disaster recovery instead of a version change.
+    let unknown_marker = unknown_protection_marker(&plaintext)
+        .map_err(|error| BackupError::corrupted(format!("bundled key record '{stem}' is not a readable JSON object: {error}")))?;
+    if let Some(version) = unknown_marker {
+        return Err(BackupError::UnsupportedRecordVersion { key_id: stem, version }.into());
+    }
     let probe: RestoredRecordProbe = serde_json::from_slice(&plaintext)
         .map_err(|error| BackupError::corrupted(format!("bundled key record '{stem}' does not deserialize: {error}")))?;
     if probe.key_id != stem {
@@ -2056,4 +2064,82 @@ mod tests {
             .find(|blocker| blocker.code == RestoreBlockerCode::BundleCorrupted)
             .expect("a tampered config artifact must be reported as corruption");
     }
+
+    /// A bundled record from a newer build is not a damaged bundle: the bytes
+    /// are intact and the build that wrote them reads them back. Reporting it
+    /// as corruption sends the operator into a disaster recovery for what a
+    /// version change fixes, so the verdict must stay separate all the way
+    /// out to the dry-run blocker code.
+    #[test]
+    fn bundled_record_from_a_newer_build_is_not_reported_as_corruption() {
+        let record = serde_json::json!({
+            "key_id": "alpha",
+            "at_rest_protection": "post-quantum-v2",
+            "encrypted_key_material": "AAAAAAAAAAAAAAAAAAAAAA==",
+            "nonce": vec![0u8; 12],
+        });
+        let error = match decode_key_record(
+            "artifacts/keys/alpha.key.enc",
+            Zeroizing::new(serde_json::to_vec(&record).expect("encode record")),
+            &[AtRestProtection::EncryptedMasterKey],
+        ) {
+            Ok(_) => panic!("a record this build cannot interpret must be rejected"),
+            Err(error) => error,
+        };
+
+        let KmsError::Backup(inner) = &error else {
+            panic!("expected a backup error, got {error:?}");
+        };
+        assert!(
+            matches!(inner, BackupError::UnsupportedRecordVersion { key_id, version }
+                if key_id == "alpha" && version == "post-quantum-v2"),
+            "got {inner:?}"
+        );
+        assert_eq!(
+            RestoreBlocker::from(inner).code,
+            RestoreBlockerCode::UnknownFormatVersion,
+            "a dry run must report a version blocker, not a corruption blocker"
+        );
+    }
+
+    /// The commit marker's format-version branch, symmetric with the Vault
+    /// restore marker's: an unknown version is refused outright everywhere the
+    /// marker is read, and the backend refuses to start while it is present.
+    #[tokio::test]
+    async fn restore_commit_marker_from_a_newer_build_is_refused() {
+        let mut marker = serde_json::to_value(RestoreCommitMarker {
+            format_version: RESTORE_MARKER_FORMAT_VERSION,
+            backup_id: BACKUP_ID.to_string(),
+            manifest_digest: ContentDigest::sha256_of(b"manifest"),
+            files: vec![LOCAL_KMS_MASTER_KEY_SALT_FILE.to_string()],
+        })
+        .expect("marker json");
+        marker["format_version"] = serde_json::json!(99);
+        let marker = serde_json::to_vec_pretty(&marker).expect("marker bytes");
+
+        let error = RestoreCommitMarker::decode(&marker).expect_err("an unknown marker version must be refused");
+        assert!(error.to_string().contains("unknown format version 99"), "got {error}");
+
+        // Reachable from the restore entry point, not just from the decoder.
+        let (client, _source_dir) = source_with_keys(Some(TEST_MASTER_KEY), &["alpha"]).await;
+        let work = TempDir::new().expect("work dir");
+        let (bundle, _manifest) = export_bundle(&client, work.path()).await;
+        let target = TempDir::new().expect("target dir");
+        fs::write(target.path().join(LOCAL_RESTORE_COMMIT_MARKER_FILE), &marker)
+            .await
+            .expect("seed marker");
+        let before = snapshot_tree(target.path());
+        let error = restore_local_backup(&test_kek(), &restore_request(&bundle, target.path()))
+            .await
+            .expect_err("an unknown marker version must block the restore");
+        assert!(error.to_string().contains("unknown format version 99"), "got {error}");
+        assert_eq!(snapshot_tree(target.path()), before, "the refused restore must not write");
+
+        // And the backend itself refuses to start on any marker at all.
+        let error = match LocalKmsClient::new(local_config(target.path(), Some(TEST_MASTER_KEY))).await {
+            Ok(_) => panic!("a restore marker must block startup"),
+            Err(error) => error,
+        };
+        assert!(error.to_string().contains("unfinished restore"), "got {error}");
+    }
 }

docs/operations/kms-backend-security.md

@@ -249,7 +249,7 @@ On startup the backend then:
 
 - **Removes orphaned commit temp files.** The matcher is strict (`<prefix>.tmp-<uuid>`, never anything ending in `.key`), so published key files — including a key the user named to look like a temp file — are never touched. Publishing is atomic, so a matching leftover can only be an unpublished remnant of an interrupted commit.
 - **Validates every published `.key` file.** A record that fails to decode fails startup rather than being silently skipped.
-- **Guards the salt file.** If `.master-key.salt` is missing but the directory contains keys marked `encrypted-master-key`, initialization fails closed with a configuration error naming the salt path. A regenerated salt derives a different master key and can never decrypt those keys, so the correct recovery is to **restore the salt file (or the whole directory) from backup**, never to let a fresh salt be generated. An empty directory, or a legacy directory predating the salt file, still initializes normally.
+- **Guards the salt file.** If `.master-key.salt` is missing but the directory contains keys marked `encrypted-master-key`, initialization fails closed with a configuration error naming the salt path. A regenerated salt derives a different master key and can never decrypt those keys, so the correct recovery is to **restore the salt file (or the whole directory) from backup**, never to let a fresh salt be generated. The guard is equally strict about a record it cannot read or cannot interpret — for example one written by a newer RustFS that names an at-rest protection this build does not implement: such a directory's protection state is unknown, so no replacement salt is generated for it either. Recovery is to restore the salt file, run a build that understands the record, or move the unrecognized file out of `key_dir` after confirming it is not needed. An empty directory, or a legacy directory predating the salt file, still initializes normally.
 
 ### Backing up the key directory