fix(kms): version local key records safely

Cargo.lock

@@ -9642,6 +9642,7 @@ dependencies = [
  "tokio",
  "tokio-util",
  "tracing",
+ "tracing-subscriber",
  "url",
  "uuid",
  "vaultrs",

crates/kms/Cargo.toml

@@ -99,6 +99,9 @@ tokio = { workspace = true, features = ["net", "test-util"] }
 # Replays canned AWS KMS HTTP exchanges so the AWS backend tests stay offline.
 aws-smithy-http-client = { workspace = true, default-features = false, features = ["test-util"] }
 http = { workspace = true }
+# Captures warning events in format-compatibility tests without installing a
+# process-wide subscriber.
+tracing-subscriber = { workspace = true, features = ["fmt"] }
 
 [features]
 default = []

crates/kms/src/backends/local.rs

@@ -32,6 +32,7 @@ use async_trait::async_trait;
 use base64::{Engine as _, engine::general_purpose::STANDARD as BASE64};
 use jiff::Zoned;
 use rand::RngExt;
+use serde::de::IgnoredAny;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
 use std::collections::HashMap;
@@ -511,8 +512,12 @@ pub(crate) fn unknown_protection_marker(record: &[u8]) -> serde_json::Result<Opt
 }
 
 /// Serializable representation of a master key stored on disk
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize)]
 struct StoredMasterKey {
+    /// Persisted record schema version. Records written before this field was
+    /// introduced default to version 1 during deserialization.
+    #[serde(default = "default_stored_master_key_format_version")]
+    format_version: u32,
     key_id: String,
     version: u32,
     algorithm: String,
@@ -537,6 +542,81 @@ struct StoredMasterKey {
     at_rest_protection: StoredKeyProtection,
 }
 
+pub(crate) const STORED_MASTER_KEY_FORMAT_VERSION: u32 = 1;
+
+fn default_stored_master_key_format_version() -> u32 {
+    STORED_MASTER_KEY_FORMAT_VERSION
+}
+
+/// Read only the schema marker before attempting the complete key-record
+/// decode. A future record may add or remove required fields, but its version
+/// still needs to be reported as unsupported rather than as generic corruption.
+pub(crate) fn stored_master_key_format_version(record: &[u8]) -> serde_json::Result<u32> {
+    #[derive(Deserialize)]
+    struct FormatProbe {
+        #[serde(default = "default_stored_master_key_format_version")]
+        format_version: u32,
+    }
+
+    Ok(serde_json::from_slice::<FormatProbe>(record)?.format_version)
+}
+
+impl<'de> Deserialize<'de> for StoredMasterKey {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct Wire {
+            #[serde(default = "default_stored_master_key_format_version")]
+            format_version: u32,
+            key_id: String,
+            version: u32,
+            algorithm: String,
+            usage: KeyUsage,
+            status: KeyStatus,
+            description: Option<String>,
+            metadata: HashMap<String, String>,
+            #[serde(with = "crate::time_serde::zoned")]
+            created_at: Zoned,
+            #[serde(with = "crate::time_serde::option_zoned")]
+            rotated_at: Option<Zoned>,
+            created_by: Option<String>,
+            #[serde(default, with = "crate::time_serde::option_zoned")]
+            deletion_date: Option<Zoned>,
+            encrypted_key_material: String,
+            nonce: Vec<u8>,
+            #[serde(default)]
+            at_rest_protection: StoredKeyProtection,
+            #[serde(flatten)]
+            unknown_fields: HashMap<String, IgnoredAny>,
+        }
+
+        let wire = Wire::deserialize(deserializer)?;
+        for field in wire.unknown_fields.keys() {
+            tracing::warn!(key_id = %wire.key_id, field = %field, "Local KMS key record contains an unknown field");
+        }
+
+        Ok(Self {
+            format_version: wire.format_version,
+            key_id: wire.key_id,
+            version: wire.version,
+            algorithm: wire.algorithm,
+            usage: wire.usage,
+            status: wire.status,
+            description: wire.description,
+            metadata: wire.metadata,
+            created_at: wire.created_at,
+            rotated_at: wire.rotated_at,
+            created_by: wire.created_by,
+            deletion_date: wire.deletion_date,
+            encrypted_key_material: wire.encrypted_key_material,
+            nonce: wire.nonce,
+            at_rest_protection: wire.at_rest_protection,
+        })
+    }
+}
+
 impl LocalKmsClient {
     /// Create a new local KMS client
     pub async fn new(config: LocalConfig) -> Result<Self> {
@@ -851,6 +931,12 @@ impl LocalKmsClient {
 
         let content = fs::read(&key_path).await?;
 
+        let format_version = stored_master_key_format_version(&content)
+            .map_err(|e| KmsError::material_corrupt(key_id, format!("stored key record is not a readable JSON object: {e}")))?;
+        if format_version > STORED_MASTER_KEY_FORMAT_VERSION {
+            return Err(KmsError::unsupported_format_version(key_id, format_version.to_string()));
+        }
+
         // 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.
@@ -1023,6 +1109,7 @@ impl LocalKmsClient {
         };
 
         let stored_key = StoredMasterKey {
+            format_version: STORED_MASTER_KEY_FORMAT_VERSION,
             key_id: master_key.key_id.clone(),
             version: master_key.version,
             algorithm: master_key.algorithm.clone(),
@@ -2428,6 +2515,121 @@ mod tests {
         assert_eq!(key_info.created_at.time_zone().iana_name(), Some("UTC"));
     }
 
+    #[tokio::test]
+    async fn stored_master_key_format_version_is_explicit_and_legacy_defaults_to_v1() {
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client.create_key("format-key", "AES_256", None).await.expect("create key");
+
+        let key_path = client.master_key_path("format-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        assert_eq!(record.get("format_version"), Some(&serde_json::json!(STORED_MASTER_KEY_FORMAT_VERSION)));
+
+        // A record from before the explicit field was added remains readable.
+        record
+            .as_object_mut()
+            .expect("key record is an object")
+            .remove("format_version")
+            .expect("current records carry format_version");
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode legacy key record"))
+            .await
+            .expect("write legacy key record");
+        let info = client
+            .describe_key("format-key", None)
+            .await
+            .expect("legacy key record should load");
+        assert_eq!(info.key_id, "format-key");
+    }
+
+    #[tokio::test]
+    async fn stored_master_key_accepts_an_older_numeric_format_version() {
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client
+            .create_key("older-format-key", "AES_256", None)
+            .await
+            .expect("create key");
+
+        let key_path = client.master_key_path("older-format-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        record["format_version"] = serde_json::json!(0);
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode older key record"))
+            .await
+            .expect("write older key record");
+
+        let info = client
+            .describe_key("older-format-key", None)
+            .await
+            .expect("older format version should remain readable");
+        assert_eq!(info.key_id, "older-format-key");
+    }
+
+    #[tokio::test]
+    async fn stored_master_key_rejects_a_newer_format_version_before_decrypting() {
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client
+            .create_key("future-format-key", "AES_256", None)
+            .await
+            .expect("create key");
+
+        let key_path = client.master_key_path("future-format-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        record["format_version"] = serde_json::json!(99);
+        record.as_object_mut().expect("key record is an object").remove("usage");
+        fs::write(&key_path, serde_json::to_vec_pretty(&record).expect("encode future key record"))
+            .await
+            .expect("write future key record");
+
+        let error = client
+            .describe_key("future-format-key", None)
+            .await
+            .expect_err("a newer key format must fail closed");
+        assert!(matches!(
+            error,
+            KmsError::UnsupportedFormatVersion { key_id, version }
+                if key_id == "future-format-key" && version == "99"
+        ));
+    }
+
+    #[tokio::test]
+    async fn stored_master_key_unknown_fields_remain_readable() {
+        const UNKNOWN_FIELD_VALUE: &str = "field value must not be logged";
+        let (client, _temp_dir) = create_dev_mode_client().await;
+        client
+            .create_key("unknown-field-key", "AES_256", None)
+            .await
+            .expect("create key");
+
+        let key_path = client.master_key_path("unknown-field-key").expect("valid key id");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&key_path).await.expect("read key record")).expect("decode key record");
+        record["future_field"] = serde_json::json!("field value must not be logged");
+        fs::write(
+            &key_path,
+            serde_json::to_vec_pretty(&record).expect("encode key record with unknown field"),
+        )
+        .await
+        .expect("write key record with unknown field");
+
+        let logs = crate::test_support::CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(logs.clone())
+            .finish();
+        let record = fs::read(&key_path).await.expect("read key record");
+        let stored: StoredMasterKey = tracing::subscriber::with_default(subscriber, || {
+            serde_json::from_slice(&record).expect("unknown fields must remain forward-compatible")
+        });
+        assert_eq!(stored.key_id, "unknown-field-key");
+
+        let output = logs.output();
+        assert!(output.contains("Local KMS key record contains an unknown field"));
+        assert!(output.contains("future_field"));
+        assert!(!output.contains(UNKNOWN_FIELD_VALUE));
+    }
+
     #[tokio::test]
     async fn test_load_master_key_accepts_legacy_encrypted_record_without_protection_field() {
         let (client, temp_dir) = create_test_client().await;

crates/kms/src/backup/dry_run.rs

@@ -72,6 +72,7 @@ impl From<&BackupError> for RestoreBlocker {
             BackupError::Corrupted { .. } => RestoreBlockerCode::BundleCorrupted,
             BackupError::Truncated { .. } => RestoreBlockerCode::BundleTruncated,
             BackupError::UnknownVersion { .. } => RestoreBlockerCode::UnknownFormatVersion,
+            BackupError::UnsupportedFormatVersion { .. } => RestoreBlockerCode::UnknownFormatVersion,
             BackupError::WrongKek { .. } => RestoreBlockerCode::WrongBackupKek,
             BackupError::MissingArtifact { .. } => RestoreBlockerCode::MissingArtifact,
             BackupError::IncompleteBundle { .. } => RestoreBlockerCode::IncompleteBundle,

crates/kms/src/backup/error.rs

@@ -51,6 +51,11 @@ pub enum BackupError {
         supplied_kek_version: u32,
     },
 
+    /// A bundled key record declares a format version this build does not
+    /// understand.
+    #[error("bundled key record '{key_id}' declares unsupported format version {version}; this build cannot restore it")]
+    UnsupportedFormatVersion { key_id: String, version: String },
+
     /// Manifest requires an artifact that is not present in the bundle.
     #[error("backup bundle is missing a required artifact: {artifact}")]
     MissingArtifact { artifact: String },
@@ -128,6 +133,15 @@ mod tests {
             "unknown backup manifest format version 9 (this build supports version 1)"
         );
 
+        assert_eq!(
+            BackupError::UnsupportedFormatVersion {
+                key_id: "alpha".to_string(),
+                version: "9".to_string(),
+            }
+            .to_string(),
+            "bundled key record 'alpha' declares unsupported format version 9; this build cannot restore it"
+        );
+
         assert_eq!(
             BackupError::missing_artifact("key-material").to_string(),
             "backup bundle is missing a required artifact: key-material"

crates/kms/src/backup/local_export.rs

@@ -48,7 +48,10 @@
 //! 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, unknown_protection_marker};
+use crate::backends::local::{
+    LocalKmsClient, STORED_MASTER_KEY_FORMAT_VERSION, StoredKeyProtection, stored_master_key_format_version,
+    unknown_protection_marker,
+};
 use crate::backup::capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
 use crate::backup::error::BackupError;
 use crate::backup::manifest::{
@@ -380,6 +383,12 @@ async fn collect_snapshot(client: &LocalKmsClient) -> Result<CollectedSnapshot>
         // 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 format_version = stored_master_key_format_version(&raw).map_err(|error| {
+            KmsError::material_corrupt(&stem, format!("stored key record is not a readable JSON object: {error}"))
+        })?;
+        if format_version > STORED_MASTER_KEY_FORMAT_VERSION {
+            return Err(KmsError::unsupported_format_version(&stem, format_version.to_string()));
+        }
         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}"))
         })?;
@@ -1150,4 +1159,26 @@ mod tests {
             "got {error:?}"
         );
     }
+
+    #[tokio::test]
+    async fn numeric_record_format_version_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["format_version"] = serde_json::json!(99);
+        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("a newer record format must abort the export");
+        assert!(
+            matches!(&error, KmsError::UnsupportedFormatVersion { key_id, version }
+                if key_id == "alpha" && version == "99"),
+            "got {error:?}"
+        );
+    }
 }

crates/kms/src/backup/local_restore.rs

@@ -54,7 +54,8 @@
 
 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, unknown_protection_marker, validate_key_id,
+    STORED_MASTER_KEY_FORMAT_VERSION, StoredKeyProtection, durable_file, is_orphan_commit_temp_name,
+    stored_master_key_format_version, unknown_protection_marker, validate_key_id,
 };
 use crate::backup::capability::AtRestProtection;
 use crate::backup::dry_run::{
@@ -608,6 +609,15 @@ fn decode_key_record(
     // 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 format_version = stored_master_key_format_version(&plaintext)
+        .map_err(|error| BackupError::corrupted(format!("bundled key record '{stem}' is not a readable JSON object: {error}")))?;
+    if format_version > STORED_MASTER_KEY_FORMAT_VERSION {
+        return Err(BackupError::UnsupportedFormatVersion {
+            key_id: stem,
+            version: format_version.to_string(),
+        }
+        .into());
+    }
     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 {
@@ -2102,6 +2112,33 @@ mod tests {
         );
     }
 
+    #[test]
+    fn bundled_record_format_version_from_a_newer_build_is_not_reported_as_corruption() {
+        let record = serde_json::json!({"format_version": 99});
+        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::UnsupportedFormatVersion { key_id, version }
+                if key_id == "alpha" && version == "99"),
+            "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.

crates/kms/src/encryption/dek.rs

@@ -36,7 +36,7 @@ use std::collections::HashMap;
 /// material. Envelopes written before versioning carry `None`; backends must resolve
 /// `None` to a deterministic baseline version recorded in key metadata, never
 /// implicitly to whatever version is current.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize)]
 pub struct DataKeyEnvelope {
     pub key_id: String,
     pub master_key_id: String,
@@ -54,6 +54,45 @@ pub struct DataKeyEnvelope {
     pub master_key_version: Option<u32>,
 }
 
+impl<'de> Deserialize<'de> for DataKeyEnvelope {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        struct Wire {
+            key_id: String,
+            master_key_id: String,
+            key_spec: String,
+            encrypted_key: Vec<u8>,
+            nonce: Vec<u8>,
+            encryption_context: HashMap<String, String>,
+            #[serde(with = "crate::time_serde::zoned")]
+            created_at: Zoned,
+            #[serde(default)]
+            master_key_version: Option<u32>,
+            #[serde(flatten)]
+            unknown_fields: HashMap<String, IgnoredAny>,
+        }
+
+        let wire = Wire::deserialize(deserializer)?;
+        for field in wire.unknown_fields.keys() {
+            tracing::warn!(field = %field, "KMS data-key envelope contains an unknown field");
+        }
+
+        Ok(Self {
+            key_id: wire.key_id,
+            master_key_id: wire.master_key_id,
+            key_spec: wire.key_spec,
+            encrypted_key: wire.encrypted_key,
+            nonce: wire.nonce,
+            encryption_context: wire.encryption_context,
+            created_at: wire.created_at,
+            master_key_version: wire.master_key_version,
+        })
+    }
+}
+
 #[derive(Deserialize)]
 struct DataKeyEnvelopeMarker {
     #[serde(rename = "key_id")]
@@ -368,6 +407,38 @@ mod tests {
         assert_eq!(deserialized.master_key_version, None);
     }
 
+    #[test]
+    fn test_data_key_envelope_unknown_fields_remain_readable() {
+        const UNKNOWN_FIELD_VALUE: &str = "field value must not be logged";
+        let envelope_json = r#"{
+            "key_id": "test-key-id",
+            "master_key_id": "master-key-id",
+            "key_spec": "AES_256",
+            "encrypted_key": [1, 2, 3, 4],
+            "nonce": [5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16],
+            "encryption_context": {"bucket": "test-bucket"},
+            "created_at": "2024-01-01T00:00:00+00:00[UTC]",
+            "future_field": "field value must not be logged"
+        }"#;
+
+        let logs = crate::test_support::CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .with_max_level(tracing::Level::WARN)
+            .with_writer(logs.clone())
+            .finish();
+        let deserialized: DataKeyEnvelope = tracing::subscriber::with_default(subscriber, || {
+            serde_json::from_str(envelope_json).expect("unknown fields must remain readable")
+        });
+        assert_eq!(deserialized.key_id, "test-key-id");
+        assert_eq!(deserialized.master_key_version, None);
+
+        let output = logs.output();
+        assert!(output.contains("KMS data-key envelope contains an unknown field"));
+        assert!(output.contains("future_field"));
+        assert!(!output.contains(UNKNOWN_FIELD_VALUE));
+    }
+
     #[test]
     fn test_data_key_envelope_none_version_serializes_without_field() {
         // A `None` version must keep the serialized envelope on the historical

crates/kms/src/lib.rs

@@ -82,6 +82,45 @@ pub mod service_manager;
 mod time_serde;
 pub mod types;
 
+#[cfg(test)]
+pub(crate) mod test_support {
+    use std::io::{self, Write};
+    use std::sync::{Arc, Mutex};
+
+    #[derive(Clone, Default)]
+    pub(crate) struct CapturedLogs {
+        output: Arc<Mutex<Vec<u8>>>,
+    }
+
+    pub(crate) struct CapturedWriter(CapturedLogs);
+
+    impl Write for CapturedWriter {
+        fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+            self.0.output.lock().expect("log buffer lock poisoned").extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs {
+        type Writer = CapturedWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            CapturedWriter(self.clone())
+        }
+    }
+
+    impl CapturedLogs {
+        pub(crate) fn output(&self) -> String {
+            String::from_utf8(self.output.lock().expect("log buffer lock poisoned").clone())
+                .expect("captured logs should be UTF-8")
+        }
+    }
+}
+
 // Re-export public API
 pub use api_types::{
     CacheSummary, ConfigureAwsKmsRequest, ConfigureKmsRequest, ConfigureKmsResponse, ConfigureLocalKmsRequest,

docs/operations/kms-backend-security.md

@@ -117,6 +117,7 @@ This section states only what is true of the current implementation. It is writt
 Nothing in this list requires a coordinated format cutover. The compatibility is deliberate and is covered by decode tests.
 
 - **DEK envelopes.** `DataKeyEnvelope::master_key_version` is optional and omitted when absent, so envelopes written by non-rotating backends stay byte-identical to the historical seven-field JSON shape. An upgraded node reading a pre-versioning envelope resolves `None` to the key's recorded baseline version, or — for a key that was never rotated, and so has no baseline — to the current version, which is exactly the pre-versioning behavior.
+- **Local key records.** Each `<key_id>.key` record carries `format_version: 1`; records written before that field existed default to version 1 when read. A reader accepts a record whose version is at most the version it understands, and rejects a newer version with `UnsupportedFormatVersion` before it attempts to decrypt key material. Unknown fields remain accepted for rollback compatibility, but their names are emitted at `warn` level without values.
 - **KV2 key records.** `baseline_version` is read with a serde default, so records written by older builds deserialize unchanged, and `None` correctly means "never rotated".
 - **Transit metadata records.** Metadata persisted in KV v2 by either build decodes on the other.