fix(kms): refuse key ids that leave the key prefix on the Vault backends (#7727)

Only the Local backend refused a key identifier containing a path
separator. On Vault KV2 a create with the name bad/name succeeded and
produced a nested KV2 path that the listing then reported as a
directory rather than a key, and an identifier containing .. addressed
a record outside the configured key prefix once the HTTP client
normalised the URL; Vault Transit built its transit key name and its
metadata path from the same unchecked identifier.

Lift the Local backend's containment rule into a shared segment check
(empty, /, backslash, NUL, . and ..) and apply it at the single point
where each backend turns the identifier into a path or a Transit key
name, so create, describe, encrypt, delete and the metadata writes all
refuse with InvalidKey before any request reaches Vault. The admin API
already maps that to 400. The AWS backend is untouched: it addresses
keys by ARN and alias, both of which contain /.

Refs rustfs/backlog#2474 (KMS-213 CreateNegatives on vault-kv2).

Co-authored-by: Hauser <housemecn@gmail.com>

CHANGELOG.md

@@ -21,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **Presigned URLs honour only signed headers** (GHSA-g8w9-qw9q-fghr): a SigV4 presigned request that carries an `x-amz-*` request header not listed in `X-Amz-SignedHeaders` is now rejected with `403 AccessDenied` ("There were headers present in the request which were not signed"), matching AWS S3. Previously the holder of a presigned `PutObject` URL could add unsigned `x-amz-tagging`, `x-amz-storage-class`, `x-amz-website-redirect-location`, ACL, metadata, Object Lock or SSE headers and have them applied. Presigners that intend a property must set it before signing so the SDK lists the header in `SignedHeaders`; `x-amz-cf-id` (CloudFront) remains tolerated unsigned. Header-signed SigV4 and SigV2 requests are unchanged.
 
 ### Fixed
+- **Vault backends accepted key names that escape the key prefix**: only the Local backend refused a key identifier containing a path separator, so on Vault KV2 `POST /rustfs/admin/v3/kms/keys` with a name such as `bad/name` succeeded and created a nested KV2 path that the key listing then reported as a directory rather than a key, and an identifier containing `..` addressed a record outside the configured key prefix once the HTTP client normalised the URL. Both Vault backends now apply the Local backend's containment rule at the point where the identifier becomes a path or a Transit key name: an empty identifier, one containing `/`, `\\` or NUL, or the dot segments `.` and `..` is refused with `400` (`InvalidKey`) before any request reaches Vault. The AWS backend is unaffected, since it addresses keys by ARN and alias. A Vault deployment that created such a key on an earlier build must re-create it under a plain name; objects encrypted under the old name stay readable only until that key is refused, so migrate them first. Refs rustfs/backlog#2474.
 - **Fresh multi-pool bootstrap with distinct format creators**: a new deployment whose pools have their first endpoint on different nodes (for example two single-node pools) could never publish its initial `pool.bin`: each node held fresh-bootstrap proof only for the pool it formatted, the deployment-wide proof collapsed to none, and every node died with `pool metadata recovery required: no durable bootstrap identity or pool.bin replica is available` after the startup retry budget. The first pool's creator now mints the pending cluster identity on its own pool, every other creator copies that nonce-bound identity onto the pool it formatted first-hand, and the elected writer publishes `pool.bin` once every pool replica carries the same pending identity. Corrupt or disagreeing replicas, pools that merely have a format, expansion pools joining an initialized deployment, and restarts without first-hand proof still fail closed. Non-elected nodes that start before `pool.bin` exists, and the elected writer while it waits for the other creators, no longer latch their pool-metadata write gate for the life of the process. Refs rustfs/backlog#2338, rustfs/backlog#2375.
 - **Lock RPC timeout storms** (#7363): the remote lock client no longer evicts and re-dials the shared internode HTTP/2 channel on every request deadline. A timeout evicts only when the peer has not completed any lock RPC for two deadlines, evictions and transport-failure re-dials are rate limited per peer (`RUSTFS_OBJECT_LOCK_RPC_EVICTION_COOLDOWN_MS`, default 5 s), and a timed-out request is left running instead of being reset (bounded per peer by `RUSTFS_OBJECT_LOCK_RPC_DETACHED_LIMIT`, default 256), so a slow lock endpoint can no longer drive the `RST_STREAM`/`GOAWAY too_many_resets`/reconnect loop. A lock granted after its caller timed out is released immediately, and unlocks that fail the quick retries continue on a deferred 1/2/4/8/16 s schedule before the server lease reclaims them. New `rustfs_remote_lock_*` metrics cover timeouts, evictions, suppressed evictions, detached streams, late completions and late releases per peer. Operator guide at `docs/operations/lock-rpc-storm-protection.md`.
 - **KMS failures on the S3 data path carry an actionable status**: only "key not found" and a backend outage were classified; every other KMS failure — a disabled or pending-deletion key, a denied KMS grant, an encryption-context mismatch, an unsupported algorithm, a credential or timeout failure, a capability the backend does not have — collapsed onto `500 InternalError`. SDKs therefore applied exponential backoff to configuration errors that no retry can fix, and monitoring filed every one of them as a server fault. Unusable-key and request-side failures now return `400`, a denied grant `403`, transient backend failures `503` — including a key store the backend could not read, so an outage stays distinguishable from a missing key all the way to the client — and a missing backend capability `501`. Damaged or unreadable key material still returns `500`, which is what it is.

crates/kms/src/backends/local.rs

@@ -17,6 +17,7 @@
 use crate::backends::{
     BackendCapabilities, ExpiredKeyRemoval, KmsBackend, ListedKeyFailure, StateGatedOperation, UnreadableKeys,
     classify_listed_key_failure, ensure_key_status_permits, ensure_tag_keys_are_mutable, paginate_keys, started_at_the_first_key,
+    validate_key_id_segment,
 };
 use crate::config::KmsConfig;
 use crate::config::LocalConfig;
@@ -62,14 +63,9 @@ use zeroize::Zeroizing;
 /// pub(crate) because the backup restore path applies the same containment
 /// rule to key identifiers recovered from bundle artifacts.
 pub(crate) fn validate_key_id(key_id: &str) -> Result<()> {
-    if key_id.is_empty() {
-        return Err(KmsError::invalid_key("key identifier must not be empty"));
-    }
-    if key_id.contains('/') || key_id.contains('\\') || key_id.contains('\0') {
-        return Err(KmsError::invalid_key(format!(
-            "key identifier must not contain path separators or NUL: {key_id:?}"
-        )));
-    }
+    // The separator, NUL and dot-segment refusals are shared with the Vault
+    // backends; the component check below is the filesystem-specific half.
+    validate_key_id_segment(key_id)?;
 
     // Catches `.`, `..`, absolute paths, and platform-specific forms such as Windows
     // drive prefixes, all of which would move the join outside key_dir.
@@ -3405,8 +3401,8 @@ mod tests {
 
         // The invariant is containment, so assert that directly: whatever the input, the
         // result is either refused or a path whose parent is exactly the key directory.
-        // Note `.` and `..` are contained rather than refused — the `.key` suffix turns
-        // them into the ordinary filenames `..key` and `...key`.
+        // `.` and `..` would be contained by the `.key` suffix alone, but the shared
+        // segment rule refuses them so every backend answers alike.
         for candidate in [
             "../escape",
             "../../etc/rustfs",
@@ -3431,7 +3427,7 @@ mod tests {
         }
 
         // The traversal forms specifically must be refused, not merely contained.
-        for escaping in ["../escape", "sub/dir", "/absolute", "back\\slash", "nul\0byte", ""] {
+        for escaping in ["../escape", "sub/dir", "/absolute", "back\\slash", "nul\0byte", "", ".", ".."] {
             let err = client.master_key_path(escaping).expect_err("traversal must be refused");
             assert!(
                 matches!(err, KmsError::InvalidKey { .. }),

crates/kms/src/backends/mod.rs

@@ -32,6 +32,35 @@ pub mod vault;
 pub(crate) mod vault_credentials;
 pub mod vault_transit;
 
+/// Refuse a key identifier that cannot be used as a single path segment.
+///
+/// Every path-addressed backend derives its storage location by joining the key
+/// identifier onto a configured prefix: the Local backend joins it onto `key_dir`,
+/// the Vault KV2 backend onto `key_path_prefix`, the Vault Transit backend onto
+/// both `transit/keys/` and the metadata prefix. A separator, a dot segment or a
+/// NUL byte in the identifier moves that join somewhere else — `../evil` reads and
+/// deletes a record outside the prefix, `a/b` addresses a nested path the listing
+/// reports as a directory rather than a key. The rule is containment, not a
+/// character allowlist, so identifiers already in use keep resolving.
+///
+/// Applied by each backend at the point where the identifier becomes a path or a
+/// Vault key name, never at the manager: the AWS backend addresses keys by ARN and
+/// alias, both of which legitimately contain `/`.
+pub(crate) fn validate_key_id_segment(key_id: &str) -> Result<()> {
+    if key_id.is_empty() {
+        return Err(KmsError::invalid_key("key identifier must not be empty"));
+    }
+    if key_id.contains('/') || key_id.contains('\\') || key_id.contains('\0') {
+        return Err(KmsError::invalid_key(format!(
+            "key identifier must not contain path separators or NUL: {key_id:?}"
+        )));
+    }
+    if key_id == "." || key_id == ".." {
+        return Err(KmsError::invalid_key(format!("key identifier must not be a dot segment: {key_id:?}")));
+    }
+    Ok(())
+}
+
 /// Operations whose availability depends on the key's lifecycle state.
 ///
 /// Decryption is deliberately absent: RustFS allows decryption with
@@ -724,6 +753,27 @@ mod tests {
     use crate::config::KmsConfig;
     use base64_simd::STANDARD as BASE64;
 
+    #[test]
+    fn key_id_segment_rule_refuses_every_form_that_leaves_the_prefix() {
+        for refused in [
+            "",
+            "bad/name",
+            "../escape",
+            "..",
+            ".",
+            "/absolute",
+            "back\\slash",
+            "nul\0byte",
+            "a/../b",
+        ] {
+            let err = validate_key_id_segment(refused).expect_err("must be refused");
+            assert!(matches!(err, KmsError::InvalidKey { .. }), "{refused:?}: {err:?}");
+        }
+        for accepted in ["k213", "a.b_c-1", "..leading-dots", "3f2504e0-4f89-11d3-9a0c-0305e82c3301"] {
+            validate_key_id_segment(accepted).unwrap_or_else(|e| panic!("{accepted:?} must be accepted: {e:?}"));
+        }
+    }
+
     /// Backend that implements only the trait-mandated operations and relies
     /// on the default `capabilities` implementation.
     struct MinimalBackend;

crates/kms/src/backends/vault.rs

@@ -22,6 +22,7 @@ use crate::backends::{
     BackendCapabilities, ExpiredKeyRemoval, KmsBackend, ListedKeyFailure, StateGatedOperation, UnreadableKeys,
     classify_listed_key_failure, empty_key_page, ensure_key_state_permits, ensure_key_status_permits,
     ensure_rewrap_context_matches, ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys, started_at_the_first_key,
+    validate_key_id_segment,
 };
 use crate::config::{KmsConfig, VaultConfig};
 use crate::encryption::{
@@ -671,9 +672,17 @@ impl VaultKmsClient {
         policy::execute(operation, class, &self.retry, &self.cancel, attempt).await
     }
 
-    /// Get the full path for a key in Vault
-    fn key_path(&self, key_id: &str) -> String {
-        format!("{}/{}", self.key_path_prefix, key_id)
+    /// Get the full path for a key in Vault.
+    ///
+    /// Every KV2 path this backend reads, writes or deletes is derived here, so
+    /// refusing an identifier that is not a single path segment at this one
+    /// point keeps `create`, `describe`, `delete` and the metadata writes inside
+    /// `key_path_prefix`: `../evil` would otherwise address a record outside it
+    /// once the HTTP client normalises the URL, and `a/b` a nested path that the
+    /// listing reports as a directory rather than a key.
+    fn key_path(&self, key_id: &str) -> Result<String> {
+        validate_key_id_segment(key_id)?;
+        Ok(format!("{}/{}", self.key_path_prefix, key_id))
     }
 
     /// Get the path of the immutable record holding one version's material
@@ -765,7 +774,7 @@ impl VaultKmsClient {
     /// Read the key record together with the KV2 secret version holding it, so a
     /// later write can be check-and-set against exactly this snapshot.
     async fn get_key_data_versioned(&self, key_id: &str) -> Result<(u32, VaultKeyData)> {
-        let path = self.key_path(key_id);
+        let path = self.key_path(key_id)?;
         let path = path.as_str();
 
         let metadata = self
@@ -806,7 +815,7 @@ impl VaultKmsClient {
     /// On success returns the secret version created by this write so a caller
     /// can chain further check-and-set writes.
     async fn try_cas_store_key_data(&self, key_id: &str, key_data: &VaultKeyData, cas: u32) -> Result<Option<u32>> {
-        let path = self.key_path(key_id);
+        let path = self.key_path(key_id)?;
         let path = path.as_str();
 
         // Single attempt: replaying a lost-response write would double-apply
@@ -910,7 +919,7 @@ impl VaultKmsClient {
     /// when a record already exists — i.e. a concurrent create committed
     /// first. An existing record is never overwritten.
     async fn try_create_key_data(&self, key_id: &str, key_data: &VaultKeyData) -> Result<bool> {
-        let path = self.key_path(key_id);
+        let path = self.key_path(key_id)?;
         let path = path.as_str();
 
         // Single attempt: the create-only CAS makes a duplicate replay fail
@@ -937,7 +946,7 @@ impl VaultKmsClient {
     /// records.
     #[cfg(test)]
     async fn store_key_data(&self, key_id: &str, key_data: &VaultKeyData) -> Result<()> {
-        let path = self.key_path(key_id);
+        let path = self.key_path(key_id)?;
         let path = path.as_str();
 
         self.run("vault_kv2_write_key", OpClass::MutatingNonIdempotent, move || async move {
@@ -984,7 +993,7 @@ impl VaultKmsClient {
 
     /// Retrieve key data from Vault
     async fn get_key_data(&self, key_id: &str) -> Result<VaultKeyData> {
-        let path = self.key_path(key_id);
+        let path = self.key_path(key_id)?;
         let path = path.as_str();
 
         let secret: VaultKeyData = self
@@ -1122,7 +1131,7 @@ impl VaultKmsClient {
 
     /// Physically delete a key from Vault storage
     async fn delete_key(&self, key_id: &str) -> Result<()> {
-        let path = self.key_path(key_id);
+        let path = self.key_path(key_id)?;
         let path = path.as_str();
 
         // Purge immutable version records first: if any purge fails, the top-level
@@ -2436,6 +2445,42 @@ mod tests {
     /// A caller asking for no keys gets an empty page, and the page arithmetic
     /// never reaches for the element before an empty page. The scripted key
     /// listing stays unused: a request for zero keys has nothing to ask Vault.
+    /// A key identifier becomes a KV2 path by string join, so one that is not a
+    /// single segment is refused before any request leaves the process: `../x`
+    /// would otherwise read, overwrite or delete a record outside the key
+    /// prefix once the URL is normalised, and `a/b` would create a nested path
+    /// the listing reports as a directory rather than a key.
+    #[tokio::test]
+    async fn path_addressed_operations_refuse_key_ids_that_leave_the_key_prefix() {
+        let (vault, client) = scripted_client(vec![]).await;
+
+        for key_id in ["bad/name", "../escape", "..", ".", "", "back\\slash", "nul\0byte"] {
+            let err = client
+                .create_key(key_id, "AES_256", None)
+                .await
+                .expect_err("create must refuse a non-segment key id");
+            assert!(matches!(err, KmsError::InvalidKey { .. }), "create {key_id:?}: {err:?}");
+
+            let err = client
+                .get_key_data(key_id)
+                .await
+                .expect_err("read must refuse a non-segment key id");
+            assert!(matches!(err, KmsError::InvalidKey { .. }), "read {key_id:?}: {err:?}");
+
+            let err = client
+                .delete_key(key_id)
+                .await
+                .expect_err("delete must refuse a non-segment key id");
+            assert!(matches!(err, KmsError::InvalidKey { .. }), "delete {key_id:?}: {err:?}");
+        }
+
+        assert!(
+            vault.requests().is_empty(),
+            "a refused key id must never reach Vault: {:?}",
+            vault.requests()
+        );
+    }
+
     #[tokio::test]
     async fn zero_limit_list_returns_an_empty_page_without_calling_vault() {
         let (vault, client) =
@@ -3003,7 +3048,7 @@ mod tests {
             .await
             .expect("client");
 
-        assert_eq!(client.key_path("my-key"), "rustfs/kms/keys/my-key");
+        assert_eq!(client.key_path("my-key").expect("valid key id"), "rustfs/kms/keys/my-key");
         assert_eq!(client.key_versions_dir("my-key"), "rustfs/kms/keys/my-key/versions");
         assert_eq!(client.key_version_path("my-key", 3), "rustfs/kms/keys/my-key/versions/3");
     }

crates/kms/src/backends/vault_transit.rs

@@ -22,7 +22,7 @@ use crate::backends::vault_credentials::{
 use crate::backends::{
     BackendCapabilities, ExpiredKeyRemoval, KmsBackend, ListedKeyFailure, StateGatedOperation, UnreadableKeys,
     classify_listed_key_failure, empty_key_page, ensure_key_state_permits, ensure_rewrap_context_matches,
-    ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys, started_at_the_first_key,
+    ensure_tag_keys_are_mutable, list_keys_page_size, paginate_keys, started_at_the_first_key, validate_key_id_segment,
 };
 use crate::config::{KmsConfig, VaultTransitConfig};
 use crate::encryption::{DataKeyEnvelope, generate_key_material};
@@ -491,6 +491,7 @@ impl VaultTransitKmsClient {
     }
 
     async fn read_transit_key(&self, key_id: &str) -> Result<vaultrs::api::transit::responses::ReadKeyResponse> {
+        validate_key_id_segment(key_id)?;
         self.run("vault_transit_read_key", OpClass::ReadIdempotent, move || async move {
             let vault = self.vault().map_err(AttemptError::fatal)?;
             key::read(&vault.client, &self.config.mount_path, key_id)
@@ -501,6 +502,7 @@ impl VaultTransitKmsClient {
     }
 
     async fn create_transit_key(&self, key_id: &str) -> Result<()> {
+        validate_key_id_segment(key_id)?;
         // Single attempt: create carries external side effects and the caller
         // owns the read-confirm recovery for lost responses.
         self.run("vault_transit_create_key", OpClass::MutatingNonIdempotent, move || async move {
@@ -524,6 +526,7 @@ impl VaultTransitKmsClient {
         plaintext: &[u8],
         encryption_context: &HashMap<String, String>,
     ) -> Result<String> {
+        validate_key_id_segment(key_id)?;
         let plaintext_b64 = BASE64.encode_to_string(plaintext);
         let plaintext_b64 = plaintext_b64.as_str();
         let aad = Self::canonicalize_context(encryption_context)?;
@@ -551,6 +554,7 @@ impl VaultTransitKmsClient {
         ciphertext: &str,
         encryption_context: &HashMap<String, String>,
     ) -> Result<Vec<u8>> {
+        validate_key_id_segment(key_id)?;
         let aad = Self::canonicalize_context(encryption_context)?;
         let aad = aad.as_deref();
 
@@ -579,6 +583,7 @@ impl VaultTransitKmsClient {
     /// 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> {
+        validate_key_id_segment(key_id)?;
         let response = self
             .run("vault_transit_rewrap", OpClass::ReadIdempotent, move || async move {
                 let vault = self.vault().map_err(AttemptError::fatal)?;
@@ -613,12 +618,16 @@ impl VaultTransitKmsClient {
         Ok(latest)
     }
 
-    fn metadata_key_path(&self, key_id: &str) -> String {
-        format!("{}/{}", self.metadata_key_prefix, key_id)
+    /// KV2 path of a key's metadata record. Refuses identifiers that are not a
+    /// single path segment so the join cannot leave `metadata_key_prefix`; the
+    /// transit calls apply the same rule before naming the key to Vault.
+    fn metadata_key_path(&self, key_id: &str) -> Result<String> {
+        validate_key_id_segment(key_id)?;
+        Ok(format!("{}/{}", self.metadata_key_prefix, key_id))
     }
 
     async fn read_metadata_from_kv(&self, key_id: &str) -> Result<Option<TransitKeyMetadata>> {
-        let path = self.metadata_key_path(key_id);
+        let path = self.metadata_key_path(key_id)?;
         let path = path.as_str();
         self.run("vault_transit_read_metadata", OpClass::ReadIdempotent, move || async move {
             let vault = self.vault().map_err(AttemptError::fatal)?;
@@ -643,7 +652,7 @@ impl VaultTransitKmsClient {
     /// holding it, so a later write can be check-and-set against exactly this
     /// snapshot. `None` means no record exists (a pre-persistence key).
     async fn read_metadata_from_kv_versioned(&self, key_id: &str) -> Result<Option<(u32, TransitKeyMetadata)>> {
-        let path = self.metadata_key_path(key_id);
+        let path = self.metadata_key_path(key_id)?;
         let path = path.as_str();
 
         let kv_metadata = self
@@ -693,7 +702,7 @@ impl VaultTransitKmsClient {
     /// double-apply the mutation, and a CAS conflict is a normal concurrency
     /// signal, not a backend failure.
     async fn cas_write_metadata_to_kv(&self, key_id: &str, metadata: &TransitKeyMetadata, cas: u32) -> Result<bool> {
-        let path = self.metadata_key_path(key_id);
+        let path = self.metadata_key_path(key_id)?;
         let path = path.as_str();
         let persisted: TransitKeyMetadataPersisted = metadata.clone().into();
         let persisted = &persisted;
@@ -721,7 +730,7 @@ impl VaultTransitKmsClient {
     }
 
     async fn delete_metadata_from_kv(&self, key_id: &str) -> Result<()> {
-        let path = self.metadata_key_path(key_id);
+        let path = self.metadata_key_path(key_id)?;
         let path = path.as_str();
         self.run("vault_transit_delete_metadata", OpClass::MutatingNonIdempotent, move || async move {
             let vault = self.vault().map_err(AttemptError::fatal)?;
@@ -740,6 +749,7 @@ impl VaultTransitKmsClient {
 
     /// Flip `deletion_allowed` on the transit key so it can be deleted.
     async fn allow_transit_key_deletion(&self, key_id: &str) -> Result<()> {
+        validate_key_id_segment(key_id)?;
         self.run("vault_transit_allow_deletion", OpClass::MutatingNonIdempotent, move || async move {
             let vault = self.vault().map_err(AttemptError::fatal)?;
             let mut builder = UpdateKeyConfigurationRequestBuilder::default();
@@ -758,6 +768,7 @@ impl VaultTransitKmsClient {
 
     /// Physically delete the transit key material in Vault.
     async fn delete_transit_key(&self, key_id: &str) -> Result<()> {
+        validate_key_id_segment(key_id)?;
         self.run("vault_transit_delete_key", OpClass::MutatingNonIdempotent, move || async move {
             let vault = self.vault().map_err(AttemptError::fatal)?;
             key::delete(&vault.client, &self.config.mount_path, key_id)
@@ -1452,6 +1463,7 @@ impl VaultTransitKmsClient {
     }
 
     pub(crate) async fn rotate_key(&self, key_id: &str, _context: Option<&OperationContext>) -> Result<MasterKeyInfo> {
+        validate_key_id_segment(key_id)?;
         self.ensure_key_state_allows(key_id, StateGatedOperation::Rotate).await?;
 
         // Single attempt, never retried: replaying a rotate whose response was
@@ -2034,6 +2046,40 @@ mod tests {
     /// A caller asking for no keys gets an empty page, and the page arithmetic
     /// never reaches for the element before an empty page. The scripted key
     /// listing stays unused: a request for zero keys has nothing to ask Vault.
+    /// The identifier names both the transit key (`transit/keys/<id>`) and its
+    /// KV2 metadata record, so a non-segment id is refused before either path
+    /// is formed and no request reaches Vault.
+    #[tokio::test]
+    async fn transit_operations_refuse_key_ids_that_leave_the_key_prefix() {
+        let (vault, client) = scripted_client(vec![]).await;
+
+        for key_id in ["bad/name", "../escape", "..", ".", "", "back\\slash", "nul\0byte"] {
+            let err = client
+                .create_key(key_id, "AES_256", None)
+                .await
+                .expect_err("create must refuse a non-segment key id");
+            assert!(matches!(err, KmsError::InvalidKey { .. }), "create {key_id:?}: {err:?}");
+
+            let err = client
+                .describe_key(key_id, None)
+                .await
+                .expect_err("describe must refuse a non-segment key id");
+            assert!(matches!(err, KmsError::InvalidKey { .. }), "describe {key_id:?}: {err:?}");
+
+            let err = client
+                .transit_encrypt(key_id, b"plaintext", &HashMap::new())
+                .await
+                .expect_err("encrypt must refuse a non-segment key id");
+            assert!(matches!(err, KmsError::InvalidKey { .. }), "encrypt {key_id:?}: {err:?}");
+        }
+
+        assert!(
+            vault.requests().is_empty(),
+            "a refused key id must never reach Vault: {:?}",
+            vault.requests()
+        );
+    }
+
     #[tokio::test]
     async fn zero_limit_list_returns_an_empty_page_without_calling_vault() {
         let (vault, client) =

docs/operations/kms-admin-contract.md

@@ -19,7 +19,7 @@ The wire prefix is `/rustfs/admin/v3`. `GET /kms/status` and `GET /kms/service-s
 | `GET /kms/service-status` | `kms:ServiceControl` | sensitive | no | Carries `cluster_config` fingerprints and the `consistent` flag |
 | `GET /kms/config` | `kms:Configure` | sensitive | no | Contains operational paths; redact before display |
 | `POST /kms/clear-cache` | `kms:ClearCache` | high | no | `KmsClearCacheResponse` (`{status,message}`) |
-| `POST /kms/keys` | `kms:Configure` | high | no | Key creation shares the configure action |
+| `POST /kms/keys` | `kms:Configure` | high | no | Key creation shares the configure action. On the Local, Vault KV2 and Vault Transit backends the name must be a single path segment: empty names, names containing `/`, `\\` or NUL, and the dot segments `.` and `..` are refused with `400` before any backend request; the AWS backend takes ARNs and aliases and does not apply this rule |
 | `GET /kms/keys` | `kms:ListKeys` | sensitive | no | See the key listing contract below |
 | `GET /kms/keys/{key_id}` | `kms:DescribeKey` | sensitive | yes | `?impact=true` opts into the configuration-reference report |
 | `DELETE /kms/keys/delete` | `kms:DeleteKey` | critical | yes | JSON body; `force_immediate` also requires `confirm_key_id` and the server-side `RUSTFS_KMS_ALLOW_IMMEDIATE_DELETION` gate |