fix(kms): classify KMS/SSE error contracts and SSE-S3 headers (#7697)

* fix(sse): classify bare SSE-KMS writes when no KMS is available

A `aws:kms` request without a key id, on a bucket without a default key,
returned `500 InternalError` whenever no KMS service was running: the
"no KMS key available" branch exited with an untyped storage error before
the availability classification that the keyed form already received.

Route that branch through the same split: `503 ServiceUnavailable` while
a configured KMS is stopped, `400 InvalidRequest` when KMS was never
configured, and `400 InvalidRequest` naming the missing key id when a
running KMS has no default key. `CreateMultipartUpload` shares the path.

Adds a unit test for the bare form and an e2e module that stops KMS
through the admin API, runs a master-key-only node, and runs a Local KMS
without a default key; refreshes the e2e-full selection digests.

(cherry picked from commit c3259dadc3d603a9185a5b0ad9f83dfb884e61c8)

* fix(sse): keep KMS error classes on the encrypted read path

GetObject, CopyObject and UploadPartCopy on an SSE-KMS object whose key
no longer exists answered `500 InternalError` ("KMS key not found") while
PutObject under the same key already answered `400 KMS.NotFoundException`.
The read path carries its classification through ecstore's
`EncryptionResolutionErrorKind`, which had no kind for a missing key, a
denied KMS grant or a missing backend capability, so all three folded
onto `DecryptionFailed` and the S3 layer reported an internal fault.

Add `KeyNotFound`, `AccessDenied` and `NotImplemented` kinds, map them on
both sides of the boundary, and give an envelope the configured backend
cannot unwrap a diagnosable message while keeping its `500`.

Unit tests cover the kind round trip and the reader wrapping; a new e2e
test deletes a key immediately and checks GET/Copy return 400 with
`KMS.NotFoundException` while HEAD stays 200. The e2e-full selection
digests are refreshed from the current listing (the previous digests
predated the delete-authorization tests) and the e2e `create_default_key`
helper is updated to the accepted `EncryptDecrypt` spelling.

(cherry picked from commit 2523a9814e97caea318d4ff1a51bef3a4d4445b2)

* fix(kms): classify key-management errors on the admin routes

`POST /kms/keys`, the legacy `create-key` alias and `generate-data-key`
reported every backend refusal as `500`: a blank key name (which each
backend failed on differently, the Local backend by writing a key file
with an empty stem), a name already taken, an unknown key, a disabled key
and a capability the backend lacks. `delete` and the lifecycle routes
already classified the same errors.

Refuse a blank or whitespace name in `KmsManager::create_key` before any
backend sees it, and share one `KmsError` to status mapping across
create, delete and generate-data-key (400 for validation and key state,
404 for an unknown key, 409 for a taken name, 501 for a missing
capability, 500 only for damaged material). The XML-error routes carry
the same status explicitly since s3s derives none for a custom code.

The read-only Static backend now reports create, delete and
cancel-deletion as `UnsupportedCapability`, matching its rotate and
enable/disable answers, so the admin API returns 501 for all of them.

(cherry picked from commit e33cac5493c4d9d6662e0d2980b58ba2b24a6d1b)

* fix(sse): stop SSE-S3 responses from naming the wrapping KMS key

`x-amz-server-side-encryption-aws-kms-key-id` is defined for `aws:kms`
objects only, but PutObject, CopyObject, CreateMultipartUpload and
GetObject returned it for `AES256` objects too, carrying the KMS key that
wraps the SSE-S3 data key (the service default, or the literal `default`
on a node without KMS). The write paths copied `kms_key_id` from the
encryption material unconditionally, and the single-decrypt GET
classification did the same after resolving the key for authorization.

Add `EncryptionMaterial::response_kms_key_id`, which yields the id only
for SSE-KMS, use it at the four write-response sites, and gate the GET
classification the same way. CompleteMultipartUpload and HeadObject
already omitted the header.

Unit tests pin both directions; a new e2e test covers Put/Get/Head/Copy
and CreateMultipartUpload for AES256 with an aws:kms control. The
e2e-full selection digests are refreshed from the current listing.

(cherry picked from commit 29d793a63352b0b60fd53c565e80fdbede8964bb)

* fix(s3): validate PutBucketEncryption rules before storing them

A default-encryption rule naming an unknown `SSEAlgorithm` (for example
`AES128`), a rule without `ApplyServerSideEncryptionByDefault`, an empty
rule list, or a `KMSMasterKeyID` on an `AES256` rule was stored as
written: the only algorithm check on the route decided whether to fill
in the default KMS key. `GetBucketEncryption` then advertised that
configuration while the write path encrypted header-less writes under
its `AES256` fallback, so the bucket's declared and actual schemes
disagreed. Two comments claimed the route already refused unknown
algorithms.

Validate the configuration before any of it is applied: `MalformedXML`
for a malformed rule set or unknown algorithm, `InvalidArgument` for a
key id on a non-KMS rule, and nothing stored on refusal. Correct the two
comments to describe when the AES256 fallback is still reachable.

Unit tests cover every refusal and the accepted shapes; an e2e test
checks the refusals leave the previous configuration in place. The
e2e-full selection digests are refreshed from the current listing.

(cherry picked from commit 29e4486dce41197ed93f5253cdbabc57d27a4ddb)

* test(e2e): refresh e2e-full selection for the combined KMS/SSE fixes

* test: align two unit tests with the new KMS and bucket-encryption contracts

`scheduled_deletion_carries_a_deadline_and_can_be_cancelled` still
expects the state error (`InvalidOperation`) for cancelling a key that
is not pending deletion; only the Static backend's mutations moved to
`UnsupportedCapability`. The uninitialized-store PutBucketEncryption
test now sends a well-formed AES256 rule so it reaches the store lookup
instead of the new configuration validation.

(cherry picked from commit e2e6a2535a)

CHANGELOG.md

@@ -24,6 +24,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - **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.
+- **Bare SSE-KMS writes on a node without a running KMS**: `x-amz-server-side-encryption: aws:kms` without a key id (and no bucket default key) returned `500 InternalError` while KMS was stopped or never configured, because the "no key available" branch exited before the availability classification that the keyed form already received. `PutObject` and `CreateMultipartUpload` now return `503 ServiceUnavailable` while a configured KMS is stopped, `400 InvalidRequest` when KMS was never configured, and `400 InvalidRequest` naming the missing key id when a running KMS has no default key.
+- **Reading an object whose KMS key is gone returned `500`**: `GetObject`, `CopyObject` and `UploadPartCopy` on an SSE-KMS object whose key had been deleted reported `500 InternalError` ("KMS key not found"), while the same condition on `PutObject` already returned `400 KMS.NotFoundException`. The read path carried only four classifications across the storage boundary and folded a missing key, a denied KMS grant and a missing backend capability onto "decryption failed". Those reads now return `400 KMS.NotFoundException`, `403 AccessDenied` and `501 NotImplemented` respectively; `HeadObject` is unaffected because it never unwraps the data key. An envelope the configured backend cannot unwrap (the key was re-created under the same name, or the backend was switched) stays `500` but now says so instead of the generic internal-error text.
+- **KMS key-management routes answered `500` for client-side failures**: `POST /rustfs/admin/v3/kms/keys` and the legacy `create-key` alias reported every backend refusal as `500`, including a blank key name (each backend failed differently, the Local backend by writing a key file with an empty stem) and a name that already exists; `POST /rustfs/admin/v3/kms/generate-data-key` did the same for an unknown or disabled key, although `describe` and `delete` already classified those. A blank or whitespace key name is now refused before it reaches any backend (`400`), a taken name is `409`, an unknown key is `404` (`KMS.NotFoundException`), a disabled key `400`, and a capability the backend lacks `501`; damaged key material stays `500`. The read-only Static backend now reports create, delete and cancel-deletion as missing capabilities (`501`), matching its rotate and enable/disable answers, instead of `400`/`500`.
+- **SSE-S3 responses named the internal wrapping key**: `PutObject`, `CopyObject`, `CreateMultipartUpload` and `GetObject` for an `AES256` object returned `x-amz-server-side-encryption-aws-kms-key-id` carrying the KMS key that wraps the SSE-S3 data key (the service default key, or the literal `default` on a node without KMS), although the header is defined for `aws:kms` objects only and `CompleteMultipartUpload` and `HeadObject` already omitted it. Those responses now advertise a key id only for `aws:kms` objects.
+- **PutBucketEncryption accepted algorithms the server cannot honour**: a default-encryption rule naming an unknown `SSEAlgorithm` (for example `AES128`), a rule without `ApplyServerSideEncryptionByDefault`, an empty rule list, or a `KMSMasterKeyID` on an `AES256` rule was stored as written. `GetBucketEncryption` then reported that configuration while every header-less write was encrypted under the `AES256` fallback, so the bucket's advertised and actual schemes disagreed. Those configurations are now refused with `400` (`MalformedXML` for a malformed rule, `InvalidArgument` for a key id on a non-KMS rule) and nothing is stored.
 - **SSE-C on buckets with default encryption**: a `PutObject` carrying a valid SSE-C header triple on a bucket that has default encryption configured no longer fails with `400 InvalidArgument` ("The SSE-C and managed server-side encryption headers cannot be used together"). PUT and the POST-object/extract path resolved the bucket default with a hard-coded "no explicit SSE-C" flag, so the default was layered onto the request and then tripped the request's own mutual-exclusion check; an SSE-C request now suppresses the bucket default on all three write paths, matching COPY and AWS S3. Every bucket with default encryption previously refused SSE-C single PUTs outright, while `CreateMultipartUpload` on the same bucket succeeded.
 - **Explicit SSE-S3 on SSE-KMS-default buckets**: `x-amz-server-side-encryption: AES256` against a bucket whose default is `aws:kms` no longer fails with `400 InvalidArgument`. The bucket default's KMS key id was inherited independently of the effective algorithm, producing a self-contradictory `AES256` + key-id pair; the key id is now inherited only when the effective algorithm is `aws:kms`. `PutBucketEncryption` fills in a default key id automatically, so this affected nearly every SSE-KMS-default bucket.
 - **Restore of encrypted or compressed multipart objects (silent data corruption)**: restoring a multipart object from a remote tier addressed the tier in *plaintext* coordinates while the copy-back reads the *stored* representation. Every part received a misaligned slice of the remote object whose length still satisfied the range, the hash reader and the completion size check, so the restore reported success and replaced the object's bytes. Restore now accumulates stored part sizes, passes the stored length to the hash reader alongside the plaintext length, and validates against the stored size. Objects restored by an affected release must be re-restored from the tier or recovered from a backup — this release does not detect or repair them retroactively.

crates/e2e_test/src/kms/bucket_default_encryption_test.rs

@@ -21,6 +21,7 @@
 
 use super::common::LocalKMSTestEnvironment;
 use crate::common::{TEST_BUCKET, init_logging};
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
     ChecksumAlgorithm, ChecksumMode, CompletedMultipartUpload, CompletedPart, ServerSideEncryption,
@@ -631,3 +632,92 @@ async fn test_sse_kms_without_key_id_populates_default() -> Result<(), Box<dyn s
     info!("Test passed: SSE-KMS without key ID correctly populates default key '{}'", default_key_id);
     Ok(())
 }
+
+/// A default-encryption configuration the write path cannot honour as written
+/// must be refused rather than stored: the write path falls back to AES256 for
+/// any algorithm it does not know, so storing `AES128` would make
+/// GetBucketEncryption report a scheme no object is encrypted under.
+#[tokio::test]
+async fn test_put_bucket_encryption_rejects_unknown_algorithm_and_misplaced_key_id()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut kms_env = LocalKMSTestEnvironment::new().await?;
+    let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
+    kms_env.wait_for_kms_ready().await?;
+
+    let s3_client = kms_env.base_env.create_s3_client();
+    kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
+
+    let config_with = |algorithm: ServerSideEncryption, key_id: Option<&str>| {
+        let mut by_default = ServerSideEncryptionByDefault::builder().sse_algorithm(algorithm);
+        if let Some(key_id) = key_id {
+            by_default = by_default.kms_master_key_id(key_id);
+        }
+        ServerSideEncryptionConfiguration::builder()
+            .rules(
+                ServerSideEncryptionRule::builder()
+                    .apply_server_side_encryption_by_default(by_default.build().unwrap())
+                    .build(),
+            )
+            .build()
+            .unwrap()
+    };
+
+    // Baseline the bucket on AES256 so a refused update has something to leave untouched.
+    s3_client
+        .put_bucket_encryption()
+        .bucket(TEST_BUCKET)
+        .server_side_encryption_configuration(config_with(ServerSideEncryption::Aes256, None))
+        .send()
+        .await?;
+
+    let unknown = s3_client
+        .put_bucket_encryption()
+        .bucket(TEST_BUCKET)
+        .server_side_encryption_configuration(config_with(ServerSideEncryption::from("AES128"), None))
+        .send()
+        .await
+        .expect_err("an unknown SSEAlgorithm must be refused");
+    assert_eq!(unknown.raw_response().map(|response| response.status().as_u16()), Some(400));
+    assert_eq!(
+        unknown.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("MalformedXML"),
+        "unknown algorithm error was {unknown:?}"
+    );
+
+    let misplaced_key = s3_client
+        .put_bucket_encryption()
+        .bucket(TEST_BUCKET)
+        .server_side_encryption_configuration(config_with(ServerSideEncryption::Aes256, Some(&default_key_id)))
+        .send()
+        .await
+        .expect_err("KMSMasterKeyID with AES256 must be refused");
+    assert_eq!(misplaced_key.raw_response().map(|response| response.status().as_u16()), Some(400));
+    assert_eq!(
+        misplaced_key.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("InvalidArgument"),
+        "misplaced key id error was {misplaced_key:?}"
+    );
+
+    // The refused updates left the baseline in place.
+    let stored = s3_client.get_bucket_encryption().bucket(TEST_BUCKET).send().await?;
+    let by_default = stored
+        .server_side_encryption_configuration()
+        .and_then(|config| config.rules().first())
+        .and_then(|rule| rule.apply_server_side_encryption_by_default())
+        .expect("baseline configuration must still be present");
+    assert_eq!(by_default.sse_algorithm(), &ServerSideEncryption::Aes256);
+    assert_eq!(by_default.kms_master_key_id(), None);
+
+    // aws:kms with a key id stays accepted.
+    s3_client
+        .put_bucket_encryption()
+        .bucket(TEST_BUCKET)
+        .server_side_encryption_configuration(config_with(ServerSideEncryption::AwsKms, Some(&default_key_id)))
+        .send()
+        .await?;
+
+    kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
+    Ok(())
+}

crates/e2e_test/src/kms/common.rs

@@ -176,6 +176,30 @@ pub async fn start_kms(
     Ok(())
 }
 
+/// Stop the running KMS service via admin API, keeping its configuration so
+/// `start_kms` can bring it back.
+pub async fn stop_kms(
+    base_url: &str,
+    access_key: &str,
+    secret_key: &str,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let response = kms_admin_request(
+        base_url,
+        http::Method::POST,
+        "/rustfs/admin/v3/kms/stop",
+        Some("{}"),
+        access_key,
+        secret_key,
+    )
+    .await?;
+    let response: serde_json::Value = serde_json::from_str(&response)?;
+    if response["success"] != true {
+        return Err(format!("KMS stop failed: {}", response["message"].as_str().unwrap_or("unknown error")).into());
+    }
+    info!("KMS stopped successfully");
+    Ok(())
+}
+
 /// Get KMS status via admin API
 pub async fn get_kms_status(
     base_url: &str,
@@ -325,7 +349,7 @@ pub async fn create_default_key(
     secret_key: &str,
 ) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
     let create_key_body = serde_json::json!({
-        "key_usage": "ENCRYPT_DECRYPT",
+        "key_usage": "EncryptDecrypt",
         "description": "Default key for e2e testing"
     })
     .to_string();

crates/e2e_test/src/kms/encryption_metadata_test.rs

@@ -357,3 +357,97 @@ async fn test_multipart_upload_writes_encrypted_data() -> Result<(), Box<dyn std
 
     Ok(())
 }
+
+/// `x-amz-server-side-encryption-aws-kms-key-id` is defined for `aws:kms`
+/// objects only. SSE-S3 wraps its data key under the service default key too,
+/// but that key is internal: PutObject, CopyObject and CreateMultipartUpload
+/// responses for an `AES256` object must not name it.
+#[tokio::test]
+async fn test_sse_s3_write_responses_carry_no_kms_key_id() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut kms_env = LocalKMSTestEnvironment::new().await?;
+    let default_key = kms_env.start_rustfs_for_local_kms().await?;
+    kms_env.wait_for_kms_ready().await?;
+
+    let s3_client = kms_env.base_env.create_s3_client();
+    kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
+
+    let put = s3_client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key("sse-s3-put")
+        .body(ByteStream::from_static(b"sse-s3 payload"))
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .send()
+        .await?;
+    assert_eq!(put.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    assert_eq!(put.ssekms_key_id(), None, "PutObject AES256 must not advertise the wrapping key");
+
+    let get = s3_client.get_object().bucket(TEST_BUCKET).key("sse-s3-put").send().await?;
+    assert_eq!(get.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    assert_eq!(get.ssekms_key_id(), None, "GetObject AES256 must not advertise the wrapping key");
+    let head = s3_client.head_object().bucket(TEST_BUCKET).key("sse-s3-put").send().await?;
+    assert_eq!(head.ssekms_key_id(), None, "HeadObject AES256 must not advertise the wrapping key");
+
+    let copy = s3_client
+        .copy_object()
+        .bucket(TEST_BUCKET)
+        .key("sse-s3-copy")
+        .copy_source(format!("{TEST_BUCKET}/sse-s3-put"))
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .send()
+        .await?;
+    assert_eq!(copy.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    assert_eq!(copy.ssekms_key_id(), None, "CopyObject AES256 must not advertise the wrapping key");
+
+    let multipart = s3_client
+        .create_multipart_upload()
+        .bucket(TEST_BUCKET)
+        .key("sse-s3-multipart")
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .send()
+        .await?;
+    assert_eq!(multipart.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+    assert_eq!(
+        multipart.ssekms_key_id(),
+        None,
+        "CreateMultipartUpload AES256 must not advertise the wrapping key"
+    );
+    s3_client
+        .abort_multipart_upload()
+        .bucket(TEST_BUCKET)
+        .key("sse-s3-multipart")
+        .upload_id(multipart.upload_id().expect("upload id"))
+        .send()
+        .await?;
+
+    // Control: the same responses keep naming the key for an aws:kms object.
+    let kms_put = s3_client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key("sse-kms-put")
+        .body(ByteStream::from_static(b"sse-kms payload"))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await?;
+    assert_eq!(kms_put.ssekms_key_id(), Some(default_key.as_str()));
+    let kms_multipart = s3_client
+        .create_multipart_upload()
+        .bucket(TEST_BUCKET)
+        .key("sse-kms-multipart")
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await?;
+    assert_eq!(kms_multipart.ssekms_key_id(), Some(default_key.as_str()));
+    s3_client
+        .abort_multipart_upload()
+        .bucket(TEST_BUCKET)
+        .key("sse-kms-multipart")
+        .upload_id(kms_multipart.upload_id().expect("upload id"))
+        .send()
+        .await?;
+
+    kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
+    Ok(())
+}

crates/e2e_test/src/kms/kms_fault_recovery_test.rs

@@ -21,7 +21,7 @@
 //! - Corrupted key files
 //! - Recovery from transient failures
 
-use super::common::LocalKMSTestEnvironment;
+use super::common::{LocalKMSTestEnvironment, create_default_key, create_key_with_specific_id, kms_admin_request};
 use crate::common::{TEST_BUCKET, init_logging};
 use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::types::ServerSideEncryption;
@@ -534,3 +534,129 @@ async fn test_kms_concurrent_encryption_requests() -> Result<(), Box<dyn std::er
     kms_env.base_env.delete_test_bucket(TEST_BUCKET).await?;
     Ok(())
 }
+
+/// Once the key an object was wrapped under is deleted, the object cannot be
+/// read until the key is restored. That is the same `400 KMS.NotFoundException`
+/// a write under a missing key returns, not a `500`; `HeadObject` never unwraps
+/// the data key and keeps answering `200`.
+#[tokio::test]
+async fn test_reads_under_a_deleted_kms_key_report_key_not_found() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+
+    let mut kms_env = LocalKMSTestEnvironment::new().await?;
+    let default_key_id = "rustfs-e2e-test-default-key";
+    create_key_with_specific_id(&kms_env.kms_keys_dir, default_key_id).await?;
+    let key_dir = kms_env.kms_keys_dir.clone();
+    kms_env
+        .base_env
+        .start_rustfs_server_with_env(
+            vec![
+                "--kms-enable",
+                "--kms-backend",
+                "local",
+                "--kms-key-dir",
+                &key_dir,
+                "--kms-default-key-id",
+                default_key_id,
+            ],
+            &[
+                ("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true"),
+                // Immediate deletion is refused on a default server; the test
+                // needs the key gone now rather than after the waiting window.
+                ("RUSTFS_KMS_ALLOW_IMMEDIATE_DELETION", "true"),
+            ],
+        )
+        .await?;
+    kms_env.wait_for_kms_ready().await?;
+    let base = &kms_env.base_env;
+    let s3_client = base.create_s3_client();
+    base.create_test_bucket(TEST_BUCKET).await?;
+
+    let doomed_key_id = create_default_key(&base.url, &base.access_key, &base.secret_key).await?;
+    let object_key = "wrapped-under-doomed-key";
+    let payload = b"readable only while the key exists".to_vec();
+    let put = s3_client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key(object_key)
+        .body(aws_sdk_s3::primitives::ByteStream::from(payload.clone()))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .ssekms_key_id(&doomed_key_id)
+        .send()
+        .await?;
+    assert_eq!(put.ssekms_key_id(), Some(doomed_key_id.as_str()));
+
+    info!("🗑️ deleting {doomed_key_id} immediately");
+    kms_admin_request(
+        &base.url,
+        http::Method::DELETE,
+        "/rustfs/admin/v3/kms/keys/delete",
+        Some(
+            &serde_json::json!({
+                "key_id": doomed_key_id,
+                "force_immediate": true,
+                "confirm_key_id": doomed_key_id,
+            })
+            .to_string(),
+        ),
+        &base.access_key,
+        &base.secret_key,
+    )
+    .await?;
+    kms_admin_request(
+        &base.url,
+        http::Method::POST,
+        "/rustfs/admin/v3/kms/clear-cache",
+        Some("{}"),
+        &base.access_key,
+        &base.secret_key,
+    )
+    .await?;
+
+    let head = s3_client.head_object().bucket(TEST_BUCKET).key(object_key).send().await?;
+    assert_eq!(head.ssekms_key_id(), Some(doomed_key_id.as_str()));
+
+    let get_error = s3_client
+        .get_object()
+        .bucket(TEST_BUCKET)
+        .key(object_key)
+        .send()
+        .await
+        .expect_err("an object wrapped under a deleted key must not be readable");
+    assert_eq!(get_error.raw_response().map(|response| response.status().as_u16()), Some(400));
+    assert_eq!(
+        get_error.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("KMS.NotFoundException"),
+        "GetObject error was {get_error:?}"
+    );
+
+    let copy_error = s3_client
+        .copy_object()
+        .bucket(TEST_BUCKET)
+        .key("copied-from-doomed-source")
+        .copy_source(format!("{TEST_BUCKET}/{object_key}"))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await
+        .expect_err("copying from an object wrapped under a deleted key must fail the same way");
+    assert_eq!(copy_error.raw_response().map(|response| response.status().as_u16()), Some(400));
+    assert_eq!(
+        copy_error.as_service_error().and_then(ProvideErrorMetadata::code),
+        Some("KMS.NotFoundException"),
+        "CopyObject error was {copy_error:?}"
+    );
+
+    // The default key is untouched, so the node keeps serving other objects.
+    let unaffected = s3_client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key("wrapped-under-default-key")
+        .body(aws_sdk_s3::primitives::ByteStream::from_static(b"still fine"))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await?;
+    assert_eq!(unaffected.ssekms_key_id(), Some(default_key_id));
+
+    base.delete_test_bucket(TEST_BUCKET).await?;
+    Ok(())
+}

crates/e2e_test/src/kms/kms_service_stop_test.rs

@@ -0,0 +1,231 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! SSE-KMS writes against a node whose KMS is stopped, was never configured,
+//! or runs without a default key.
+//!
+//! `docs/operations/kms-backend-security.md` promises `503` for a configured
+//! KMS that is not running and `400 InvalidRequest` when KMS was never
+//! configured. The bare `aws:kms` form (no key id, no bucket default) used to
+//! miss both branches and surface as `500`, so every scenario here is
+//! exercised with and without a key id.
+
+use super::common::{
+    LocalKMSTestEnvironment, assert_s3_error, create_key_with_specific_id, start_kms, stop_kms, wait_for_kms_ready,
+};
+use crate::common::{RustFSTestEnvironment, TEST_BUCKET, init_logging};
+use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::ServerSideEncryption;
+use tracing::info;
+
+const SERVICE_UNAVAILABLE_MESSAGE: &str = "The service is unavailable. Please retry.";
+const KMS_NOT_CONFIGURED_MESSAGE: &str = "SSE-KMS requires a configured and running KMS service";
+const KMS_NO_DEFAULT_KEY_MESSAGE: &str =
+    "SSE-KMS requires a KMS key id: the request named none and the KMS service has no default key";
+
+/// Issue an SSE-KMS PutObject and a CreateMultipartUpload, each with and
+/// without a key id, and require every one of them to fail with `status`/`code`/`message`.
+async fn assert_sse_kms_writes_refused(client: &Client, key_prefix: &str, status: u16, code: &str, message: &str) {
+    for (label, key_id) in [("bare", None), ("keyed", Some("rustfs-e2e-refused-key"))] {
+        let object_key = format!("{key_prefix}-{label}");
+        let mut put = client
+            .put_object()
+            .bucket(TEST_BUCKET)
+            .key(&object_key)
+            .body(ByteStream::from_static(b"must not be published"))
+            .server_side_encryption(ServerSideEncryption::AwsKms);
+        if let Some(key_id) = key_id {
+            put = put.ssekms_key_id(key_id);
+        }
+        assert_s3_error(put.send().await, status, code, message, &format!("{label} SSE-KMS PutObject"));
+
+        let mut create = client
+            .create_multipart_upload()
+            .bucket(TEST_BUCKET)
+            .key(&object_key)
+            .server_side_encryption(ServerSideEncryption::AwsKms);
+        if let Some(key_id) = key_id {
+            create = create.ssekms_key_id(key_id);
+        }
+        assert_s3_error(
+            create.send().await,
+            status,
+            code,
+            message,
+            &format!("{label} SSE-KMS CreateMultipartUpload"),
+        );
+
+        let absence = client
+            .get_object()
+            .bucket(TEST_BUCKET)
+            .key(&object_key)
+            .send()
+            .await
+            .expect_err("a refused SSE-KMS write must not publish an object");
+        assert_eq!(absence.raw_response().map(|response| response.status().as_u16()), Some(404));
+        assert_eq!(absence.as_service_error().and_then(ProvideErrorMetadata::code), Some("NoSuchKey"));
+    }
+}
+
+/// A configured KMS that an operator stopped is a transient outage: every
+/// SSE-KMS write is refused with `503` until `kms/start`, after which the bare
+/// form resolves the service default key again and earlier objects read back.
+#[tokio::test]
+async fn test_sse_kms_writes_are_refused_with_503_while_kms_is_stopped() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+    let mut kms_env = LocalKMSTestEnvironment::new().await?;
+    let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
+    kms_env.wait_for_kms_ready().await?;
+    let base = &kms_env.base_env;
+    let client = base.create_s3_client();
+    base.create_test_bucket(TEST_BUCKET).await?;
+
+    let object_key = "written-before-stop";
+    let payload = b"encrypted under the service default key".to_vec();
+    let put = client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key(object_key)
+        .body(ByteStream::from(payload.clone()))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await?;
+    assert_eq!(put.server_side_encryption(), Some(&ServerSideEncryption::AwsKms));
+    assert_eq!(put.ssekms_key_id(), Some(default_key_id.as_str()));
+
+    info!("stopping KMS through the admin API");
+    stop_kms(&base.url, &base.access_key, &base.secret_key).await?;
+    assert_sse_kms_writes_refused(&client, "while-stopped", 503, "ServiceUnavailable", SERVICE_UNAVAILABLE_MESSAGE).await;
+
+    info!("starting KMS again");
+    start_kms(&base.url, &base.access_key, &base.secret_key).await?;
+    wait_for_kms_ready(&base.url, &base.access_key, &base.secret_key).await?;
+
+    let restored = client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key("written-after-start")
+        .body(ByteStream::from_static(b"service is back"))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await?;
+    assert_eq!(restored.ssekms_key_id(), Some(default_key_id.as_str()));
+
+    let read_back = client.get_object().bucket(TEST_BUCKET).key(object_key).send().await?;
+    assert_eq!(read_back.body.collect().await?.into_bytes().as_ref(), payload.as_slice());
+
+    base.delete_test_bucket(TEST_BUCKET).await?;
+    Ok(())
+}
+
+/// A node that only carries `RUSTFS_SSE_S3_MASTER_KEY` serves SSE-S3 but has
+/// no KMS to name: SSE-KMS is a client configuration error (`400`), and it
+/// must never be downgraded onto the local master key.
+#[tokio::test]
+async fn test_sse_kms_writes_are_refused_with_400_when_kms_was_never_configured()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+    let mut env = RustFSTestEnvironment::new().await?;
+    // base64 of 32 zero bytes: a valid master key shape for the SSE-S3 fallback.
+    env.start_rustfs_server_with_env(vec![], &[("RUSTFS_SSE_S3_MASTER_KEY", "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=")])
+        .await?;
+    let client = env.create_s3_client();
+    env.create_test_bucket(TEST_BUCKET).await?;
+
+    let sse_s3 = client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key("sse-s3-fallback")
+        .body(ByteStream::from_static(b"local master key still serves AES256"))
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .send()
+        .await?;
+    assert_eq!(sse_s3.server_side_encryption(), Some(&ServerSideEncryption::Aes256));
+
+    assert_sse_kms_writes_refused(&client, "no-kms", 400, "InvalidRequest", KMS_NOT_CONFIGURED_MESSAGE).await;
+
+    env.delete_test_bucket(TEST_BUCKET).await?;
+    Ok(())
+}
+
+/// A running KMS without a default key can serve a keyed request but has
+/// nothing to resolve a bare `aws:kms` request to; that is the caller's
+/// omission, not a server fault.
+#[tokio::test]
+async fn test_bare_sse_kms_write_is_refused_with_400_when_kms_has_no_default_key()
+-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    init_logging();
+    let mut kms_env = LocalKMSTestEnvironment::new().await?;
+    let named_key_id = "rustfs-e2e-named-key";
+    create_key_with_specific_id(&kms_env.kms_keys_dir, named_key_id).await?;
+    let key_dir = kms_env.kms_keys_dir.clone();
+    kms_env
+        .base_env
+        .start_rustfs_server_with_env(
+            vec!["--kms-enable", "--kms-backend", "local", "--kms-key-dir", &key_dir],
+            &[("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true")],
+        )
+        .await?;
+    kms_env.wait_for_kms_ready().await?;
+    let base = &kms_env.base_env;
+    let client = base.create_s3_client();
+    base.create_test_bucket(TEST_BUCKET).await?;
+
+    let keyed = client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key("keyed-without-default")
+        .body(ByteStream::from_static(b"a named key needs no default"))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .ssekms_key_id(named_key_id)
+        .send()
+        .await?;
+    assert_eq!(keyed.ssekms_key_id(), Some(named_key_id));
+
+    let bare = client
+        .put_object()
+        .bucket(TEST_BUCKET)
+        .key("bare-without-default")
+        .body(ByteStream::from_static(b"must not be published"))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await;
+    assert_s3_error(
+        bare,
+        400,
+        "InvalidRequest",
+        KMS_NO_DEFAULT_KEY_MESSAGE,
+        "bare SSE-KMS PutObject without default key",
+    );
+
+    let bare_multipart = client
+        .create_multipart_upload()
+        .bucket(TEST_BUCKET)
+        .key("bare-without-default")
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .send()
+        .await;
+    assert_s3_error(
+        bare_multipart,
+        400,
+        "InvalidRequest",
+        KMS_NO_DEFAULT_KEY_MESSAGE,
+        "bare SSE-KMS CreateMultipartUpload without default key",
+    );
+
+    base.delete_test_bucket(TEST_BUCKET).await?;
+    Ok(())
+}

crates/e2e_test/src/kms/mod.rs

@@ -39,6 +39,9 @@ mod kms_edge_cases_test;
 #[cfg(test)]
 mod kms_fault_recovery_test;
 
+#[cfg(test)]
+mod kms_service_stop_test;
+
 #[cfg(test)]
 mod bucket_default_encryption_test;
 

crates/ecstore/src/object_api/encryption.rs

@@ -34,6 +34,14 @@ pub enum EncryptionResolutionErrorKind {
     InvalidRequest,
     InvalidMetadata,
     ServiceUnavailable,
+    /// The key named by the object's metadata no longer exists in the KMS.
+    /// A client error on the read (the object is unreadable until the key is
+    /// restored), distinct from a damaged envelope.
+    KeyNotFound,
+    /// The KMS refused the unwrap for the caller's principal.
+    AccessDenied,
+    /// The configured KMS backend lacks the capability the unwrap needs.
+    NotImplemented,
     DecryptionFailed,
 }
 

crates/kms/src/backends/contract_tests.rs

@@ -299,14 +299,16 @@ async fn static_backend_stateless_contract() {
     assert_eq!(decrypted.plaintext, data_key.plaintext_key);
     assert_key_state(backend, key_id, KeyState::Enabled).await;
 
-    expect_invalid_key_state(backend.create_key(create_request("another-key".to_string())).await, "read-only");
-    expect_invalid_key_state(backend.delete_key(schedule_request(key_id)).await, "read-only");
-    expect_invalid_key_state(backend.cancel_key_deletion(cancel_request(key_id)).await, "read-only");
-    // Enable/disable, rotation and rewrap are capability gaps at the product
-    // surface, not state-machine rejections. A single fixed key has no second
-    // version to rewrap onto, so reporting the gap is the only honest answer —
-    // re-wrapping with the same material would look like progress while
-    // changing nothing.
+    // Every mutation of the key set is a capability gap at the product surface,
+    // not a state-machine rejection: the backend has exactly one externally
+    // supplied key and no way to add, remove or alter it, so the admin API
+    // reports 501 for all of them rather than 400 for some.
+    expect_unsupported(backend.create_key(create_request("another-key".to_string())).await);
+    expect_unsupported(backend.delete_key(schedule_request(key_id)).await);
+    expect_unsupported(backend.cancel_key_deletion(cancel_request(key_id)).await);
+    // A single fixed key has no second version to rewrap onto, so reporting the
+    // gap is the only honest answer: re-wrapping with the same material would
+    // look like progress while changing nothing.
     expect_unsupported(backend.enable_key(key_id).await);
     expect_unsupported(backend.disable_key(key_id).await);
     expect_unsupported(backend.rotate_key(key_id).await);

crates/kms/src/backends/static_kms.rs

@@ -335,7 +335,7 @@ impl KmsBackend for StaticKmsBackend {
         if key_name == self.key_id {
             return Err(KmsError::key_already_exists(&self.key_id));
         }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot create new keys"))
+        Err(KmsError::unsupported_capability("static", "create_key"))
     }
 
     async fn encrypt(&self, request: EncryptRequest) -> Result<EncryptResponse> {
@@ -405,14 +405,14 @@ impl KmsBackend for StaticKmsBackend {
         if request.key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.key_id));
         }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot delete keys"))
+        Err(KmsError::unsupported_capability("static", "delete_key"))
     }
 
     async fn cancel_key_deletion(&self, request: CancelKeyDeletionRequest) -> Result<CancelKeyDeletionResponse> {
         if request.key_id != self.key_id {
             return Err(KmsError::key_not_found(&request.key_id));
         }
-        Err(KmsError::invalid_operation("Static KMS is read-only: cannot cancel key deletion"))
+        Err(KmsError::unsupported_capability("static", "cancel_key_deletion"))
     }
 
     async fn health_check(&self) -> Result<bool> {
@@ -654,7 +654,7 @@ mod tests {
     async fn test_create_key_returns_error_for_other_keys() {
         let (backend, _key_id, _key) = create_test_backend().await;
 
-        // Creating any other key should return invalid operation (read-only)
+        // Creating any other key is a capability the read-only backend lacks.
         let result = KmsBackendTrait::create_key(
             &backend,
             CreateKeyRequest {
@@ -663,9 +663,8 @@ mod tests {
             },
         )
         .await;
-        assert!(result.is_err());
-        let err_msg = result.expect_err("should be Err").to_string();
-        assert!(err_msg.contains("read-only") || err_msg.contains("cannot create"));
+        let error = result.expect_err("should be Err");
+        assert!(matches!(error, KmsError::UnsupportedCapability { .. }), "got {error:?}");
     }
 
     #[tokio::test]
@@ -778,7 +777,8 @@ mod tests {
         )
         .await;
         assert!(result.is_err());
-        assert!(result.expect_err("should be Err").to_string().contains("read-only"));
+        let error = result.expect_err("should be Err");
+        assert!(matches!(error, KmsError::UnsupportedCapability { .. }), "got {error:?}");
     }
 
     #[tokio::test]

crates/kms/src/manager.rs

@@ -214,6 +214,13 @@ impl KmsManager {
     }
 
     async fn create_key_inner(&self, request: CreateKeyRequest) -> Result<CreateKeyResponse> {
+        // A blank name is neither "generate one" (that is `None`) nor a usable
+        // id: the Local backend would write a key file with an empty stem and
+        // the Vault backends would address their mount root, each failing with
+        // a different backend-specific error.
+        if request.key_name.as_deref().is_some_and(|name| name.trim().is_empty()) {
+            return Err(KmsError::validation_error("key name must not be empty or whitespace"));
+        }
         let response = self.backend.create_key(request).await?;
 
         // Cache the key metadata if enabled

crates/kms/tests/behavior_keys.rs

@@ -43,7 +43,7 @@ use common::{
 };
 use rustfs_kms::{
     CancelKeyDeletionRequest, CreateKeyRequest, DecryptRequest, DeleteKeyRequest, DescribeKeyRequest, EncryptRequest,
-    GenerateDataKeyRequest, KeySpec, KeyState, KeyStatus, KeyUsage, KmsManager, ListKeysRequest,
+    GenerateDataKeyRequest, KeySpec, KeyState, KeyStatus, KeyUsage, KmsError, KmsManager, ListKeysRequest,
 };
 
 async fn describe_state(kms: &KmsManager, key_id: &str) -> KeyState {
@@ -113,6 +113,29 @@ async fn created_key_is_enabled_and_fully_described() {
     );
 }
 
+/// A blank name is refused by the manager before any backend sees it, so every
+/// backend answers the same `ValidationError` instead of its own failure mode
+/// (an empty Local key file stem, a Vault mount root, a Transit 405).
+#[tokio::test]
+async fn create_key_refuses_a_blank_name_before_reaching_the_backend() {
+    for kms in [TestKms::local().await, TestKms::static_backend().await] {
+        let manager = kms.kms().await;
+        for name in ["", "   ", "\t\n"] {
+            let result = manager
+                .create_key(CreateKeyRequest {
+                    key_name: Some(name.to_string()),
+                    ..Default::default()
+                })
+                .await;
+            assert!(
+                matches!(result, Err(KmsError::ValidationError { .. })),
+                "{:?} with key_name {name:?} must be a ValidationError, got {result:?}",
+                kms.config().backend
+            );
+        }
+    }
+}
+
 #[tokio::test]
 async fn auto_generated_key_ids_are_unique() {
     let kms = TestKms::local().await;
@@ -607,14 +630,14 @@ async fn static_backend_refuses_every_lifecycle_mutation() {
         "static must advertise no lifecycle capability: {caps:?}"
     );
 
-    assert_invalid_operation(
+    assert_unsupported_capability(
         manager
             .create_key(CreateKeyRequest {
                 key_name: Some("another-key".to_string()),
                 ..Default::default()
             })
             .await,
-        "read-only",
+        "create_key",
     );
     // Re-creating the configured key is a conflict, not a generic refusal.
     assert_key_already_exists(
@@ -628,7 +651,7 @@ async fn static_backend_refuses_every_lifecycle_mutation() {
     );
 
     let key_id = kms.config().static_config().expect("static config").key_id.clone();
-    assert_invalid_operation(
+    assert_unsupported_capability(
         manager
             .delete_key(DeleteKeyRequest {
                 key_id: key_id.clone(),
@@ -637,13 +660,13 @@ async fn static_backend_refuses_every_lifecycle_mutation() {
                 confirm_key_id: None,
             })
             .await,
-        "read-only",
+        "delete_key",
     );
-    assert_invalid_operation(
+    assert_unsupported_capability(
         manager
             .cancel_key_deletion(CancelKeyDeletionRequest { key_id: key_id.clone() })
             .await,
-        "read-only",
+        "cancel_key_deletion",
     );
     assert_unsupported_capability(manager.enable_key(&key_id).await, "enable_key");
     assert_unsupported_capability(manager.disable_key(&key_id).await, "disable_key");

rustfs/src/admin/handlers/kms_keys.rs

@@ -389,7 +389,7 @@ impl Operation for CreateKeyHandler {
                     error = %e,
                     "admin kms keys state"
                 );
-                Err(s3_error!(InternalError, "failed to create key: {}", e))
+                Err(key_admin_s3_error("create key", &e))
             }
         }
     }
@@ -461,7 +461,7 @@ impl Operation for DescribeKeyHandler {
                     error = %e,
                     "admin kms keys state"
                 );
-                Err(s3_error!(InternalError, "failed to describe key: {}", e))
+                Err(key_admin_s3_error("describe key", &e))
             }
         }
     }
@@ -496,8 +496,8 @@ mod tests {
         CancelKmsKeyDeletionRequest, CancelKmsKeyDeletionResponse, CreateKeyApiRequest, CreateKeyApiResponse,
         CreateKmsKeyRequest, CreateKmsKeyResponse, DeleteKmsKeyRequest, DeleteKmsKeyResponse, DescribeKeyApiResponse,
         DescribeKmsKeyResponse, GenerateDataKeyApiRequest, GenerateDataKeyApiResponse, ListKeysApiResponse, ListKmsKeysResponse,
-        delete_key_error_status, delete_request_from_query, extract_key_id, extract_query_params, key_impact_if_requested,
-        key_list_filters, kms_create_key_actions, kms_delete_key_actions, kms_describe_key_actions,
+        delete_request_from_query, extract_key_id, extract_query_params, key_admin_error_status, key_admin_s3_error,
+        key_impact_if_requested, key_list_filters, kms_create_key_actions, kms_delete_key_actions, kms_describe_key_actions,
         kms_generate_data_key_actions, kms_list_keys_actions, legacy_create_key_name, parse_list_limit, scoped_key_id,
         stable_json_value, wants_key_impact,
     };
@@ -717,23 +717,59 @@ mod tests {
             KmsError::invalid_operation("immediate deletion of key key-a is not allowed"),
             KmsError::validation_error("bad input"),
         ] {
-            assert_eq!(delete_key_error_status(&error), StatusCode::BAD_REQUEST, "{error} must be a 400");
+            assert_eq!(key_admin_error_status(&error), StatusCode::BAD_REQUEST, "{error} must be a 400");
         }
 
-        assert_eq!(delete_key_error_status(&KmsError::key_not_found("key-a")), StatusCode::NOT_FOUND);
+        assert_eq!(key_admin_error_status(&KmsError::key_not_found("key-a")), StatusCode::NOT_FOUND);
         assert_eq!(
-            delete_key_error_status(&KmsError::backend_error("vault is down")),
+            key_admin_error_status(&KmsError::backend_error("vault is down")),
             StatusCode::INTERNAL_SERVER_ERROR
         );
     }
 
+    /// One mapping serves create, delete and generate-data-key: a backend
+    /// without the capability answers 501, a taken name 409, a blank or
+    /// malformed name 400, and only damaged material stays a server fault.
+    #[test]
+    fn key_admin_error_status_contract() {
+        assert_eq!(
+            key_admin_error_status(&KmsError::unsupported_capability("static", "create_key")),
+            StatusCode::NOT_IMPLEMENTED
+        );
+        assert_eq!(key_admin_error_status(&KmsError::key_already_exists("key-a")), StatusCode::CONFLICT);
+        assert_eq!(
+            key_admin_error_status(&KmsError::validation_error("key name must not be empty or whitespace")),
+            StatusCode::BAD_REQUEST
+        );
+        assert_eq!(key_admin_error_status(&KmsError::invalid_key("bad name")), StatusCode::BAD_REQUEST);
+        assert_eq!(
+            key_admin_error_status(&KmsError::material_corrupt("key-a", "truncated")),
+            StatusCode::INTERNAL_SERVER_ERROR
+        );
+
+        // The XML-error routes carry the same status explicitly, since s3s
+        // derives none for a custom code.
+        let missing = key_admin_s3_error("generate data key", &KmsError::key_not_found("key-a"));
+        assert_eq!(missing.status_code(), Some(StatusCode::NOT_FOUND));
+        assert_eq!(
+            *missing.code(),
+            super::s3::S3ErrorCode::Custom(crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE.into())
+        );
+        let unsupported = key_admin_s3_error("create key", &KmsError::unsupported_capability("static", "create_key"));
+        assert_eq!(unsupported.status_code(), Some(StatusCode::NOT_IMPLEMENTED));
+        assert_eq!(*unsupported.code(), super::s3::S3ErrorCode::NotImplemented);
+        let blank = key_admin_s3_error("create key", &KmsError::validation_error("key name must not be empty"));
+        assert_eq!(blank.status_code(), Some(StatusCode::BAD_REQUEST));
+        assert_eq!(*blank.code(), super::s3::S3ErrorCode::InvalidRequest);
+    }
+
     /// A key the deployment still points at is refused with 409, not 400: the
     /// request is well formed and the key exists, and what has to change to
     /// make it succeed is the configuration, not the request.
     #[test]
     fn a_still_referenced_key_reports_a_conflict() {
         let error = KmsError::key_still_referenced("key-a", vec!["bucket:sse-bucket".to_string()]);
-        assert_eq!(delete_key_error_status(&error), StatusCode::CONFLICT);
+        assert_eq!(key_admin_error_status(&error), StatusCode::CONFLICT);
     }
 
     #[test]
@@ -1600,7 +1636,7 @@ impl Operation for GenerateDataKeyHandler {
                     error = %e,
                     "admin kms keys state"
                 );
-                Err(s3_error!(InternalError, "failed to generate data key: {}", e))
+                Err(key_admin_s3_error("generate data key", &e))
             }
         }
     }
@@ -1728,6 +1764,7 @@ impl Operation for CreateKmsKeyHandler {
                     error = %e,
                     "admin kms keys state"
                 );
+                let status = key_admin_error_status(&e);
                 let response = CreateKmsKeyResponse {
                     success: false,
                     message: format!("failed to create key: {e}"),
@@ -1741,7 +1778,7 @@ impl Operation for CreateKmsKeyHandler {
                 let mut headers = HeaderMap::new();
                 headers.insert(CONTENT_TYPE, "application/json".parse().expect("operation should succeed"));
 
-                Ok(S3Response::with_headers((StatusCode::INTERNAL_SERVER_ERROR, Body::from(data)), headers))
+                Ok(S3Response::with_headers((status, Body::from(data)), headers))
             }
         }
     }
@@ -1831,10 +1868,22 @@ fn delete_request_from_query(uri: &hyper::Uri) -> Result<DeleteKmsKeyRequest, Bo
 /// A rejected waiting window and a refused immediate deletion both arrive as
 /// [`KmsError::InvalidOperation`], and both are the caller's input to fix, so
 /// they must surface as 400 rather than as a server fault.
-fn delete_key_error_status(error: &KmsError) -> StatusCode {
+/// HTTP status for a KMS error on a key-management route, where the key id is
+/// the resource being addressed (so a missing key is `404`, unlike the S3 data
+/// path where it is a request error). Shared by create, delete and
+/// generate-data-key so the same backend error does not read as a client
+/// error on one route and a server fault on another.
+fn key_admin_error_status(error: &KmsError) -> StatusCode {
     match error {
         KmsError::KeyNotFound { .. } => StatusCode::NOT_FOUND,
-        KmsError::InvalidOperation { .. } | KmsError::ValidationError { .. } => StatusCode::BAD_REQUEST,
+        KmsError::InvalidOperation { .. } | KmsError::ValidationError { .. } | KmsError::InvalidKey { .. } => {
+            StatusCode::BAD_REQUEST
+        }
+        // The request is well formed; the name is simply taken.
+        KmsError::KeyAlreadyExists { .. } => StatusCode::CONFLICT,
+        // A permanent gap in the configured backend (for example the read-only
+        // Static backend), never a missing resource and never retryable.
+        KmsError::UnsupportedCapability { .. } => StatusCode::NOT_IMPLEMENTED,
         // 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.
@@ -1850,6 +1899,23 @@ fn delete_key_error_status(error: &KmsError) -> StatusCode {
     }
 }
 
+/// The same classification for the routes that answer with an S3 error
+/// document instead of a JSON body. s3s derives no status for a custom code,
+/// so the status is set explicitly from `key_admin_error_status`.
+fn key_admin_s3_error(action: &str, error: &KmsError) -> s3::S3Error {
+    let status = key_admin_error_status(error);
+    let code = match status {
+        StatusCode::NOT_FOUND => s3::S3ErrorCode::Custom(crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+        StatusCode::BAD_REQUEST => s3::S3ErrorCode::InvalidRequest,
+        StatusCode::CONFLICT => s3::S3ErrorCode::Custom("KMS.AlreadyExistsException".into()),
+        StatusCode::NOT_IMPLEMENTED => s3::S3ErrorCode::NotImplemented,
+        _ => s3::S3ErrorCode::InternalError,
+    };
+    let mut s3_error = s3::error(code, format!("failed to {action}: {error}"));
+    s3_error.set_status_code(status);
+    s3_error
+}
+
 /// Delete a KMS key
 pub struct DeleteKmsKeyHandler;
 
@@ -1984,7 +2050,7 @@ impl Operation for DeleteKmsKeyHandler {
                     error = %e,
                     "admin kms keys state"
                 );
-                let status = delete_key_error_status(&e);
+                let status = key_admin_error_status(&e);
                 let response = DeleteKmsKeyResponse {
                     success: false,
                     message: format!("Failed to delete key: {e}"),

rustfs/src/app/bucket_usecase.rs

@@ -117,7 +117,8 @@ use s3s::dto::{
     PutBucketNotificationConfigurationInput, PutBucketNotificationConfigurationOutput, PutBucketPolicyInput,
     PutBucketPolicyOutput, PutBucketReplicationInput, PutBucketReplicationOutput, PutBucketTaggingInput, PutBucketTaggingOutput,
     PutBucketVersioningInput, PutBucketVersioningOutput, PutPublicAccessBlockInput, PutPublicAccessBlockOutput,
-    ReplicationConfiguration, ServerSideEncryption, Tagging, Timestamp, UserMetadata, VersioningConfiguration,
+    ReplicationConfiguration, ServerSideEncryption, ServerSideEncryptionConfiguration, Tagging, Timestamp, UserMetadata,
+    VersioningConfiguration,
 };
 use s3s::region::Region;
 use s3s::xml;
@@ -2233,6 +2234,8 @@ impl DefaultBucketUsecase {
             ..
         } = req.input;
 
+        validate_bucket_encryption_configuration(&server_side_encryption_configuration)?;
+
         // When SSE-KMS is set without a specific key ID, populate the default
         // KMS key so that GetBucketEncryption responses include it. Clients like
         // mc rely on the presence of KMSMasterKeyID to distinguish SSE-KMS from
@@ -2997,6 +3000,46 @@ impl DefaultBucketUsecase {
     }
 }
 
+/// Refuse a default-encryption configuration the write path could not honour
+/// as written. `bucket_default_write_sse` falls back to AES256 for any
+/// algorithm it does not know, so storing one would make GetBucketEncryption
+/// advertise a scheme no object is encrypted under. s3s parses `SSEAlgorithm`
+/// as an open string, so the schema check has to happen here.
+fn validate_bucket_encryption_configuration(config: &ServerSideEncryptionConfiguration) -> S3Result<()> {
+    if config.rules.is_empty() {
+        return Err(S3Error::with_message(
+            S3ErrorCode::MalformedXML,
+            "ServerSideEncryptionConfiguration must contain at least one Rule".to_string(),
+        ));
+    }
+    for rule in &config.rules {
+        let Some(by_default) = rule.apply_server_side_encryption_by_default.as_ref() else {
+            return Err(S3Error::with_message(
+                S3ErrorCode::MalformedXML,
+                "Rule must contain ApplyServerSideEncryptionByDefault".to_string(),
+            ));
+        };
+        let names_kms_key = by_default.kms_master_key_id.as_deref().is_some_and(|id| !id.is_empty());
+        match by_default.sse_algorithm.as_str() {
+            ServerSideEncryption::AWS_KMS => {}
+            ServerSideEncryption::AES256 if names_kms_key => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidArgument,
+                    "KMSMasterKeyID can only be specified when SSEAlgorithm is aws:kms".to_string(),
+                ));
+            }
+            ServerSideEncryption::AES256 => {}
+            other => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::MalformedXML,
+                    format!("SSEAlgorithm {other} is not supported; expected AES256 or aws:kms"),
+                ));
+            }
+        }
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -3005,7 +3048,8 @@ mod tests {
     use s3s::dto::{
         BucketVersioningStatus, CORSConfiguration, Destination, ExcludedPrefix, FilterRule, FilterRuleName, LifecycleExpiration,
         NoncurrentVersionTransition, PublicAccessBlockConfiguration, QueueConfiguration, ReplicationRule, S3KeyFilter,
-        ServerSideEncryptionConfiguration, Tag, Transition, TransitionStorageClass,
+        ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule, Tag, Transition,
+        TransitionStorageClass,
     };
     use std::sync::Arc;
     use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
@@ -3020,6 +3064,59 @@ mod tests {
             .unwrap_or_default()
     }
 
+    fn sse_config(rules: Vec<ServerSideEncryptionRule>) -> ServerSideEncryptionConfiguration {
+        ServerSideEncryptionConfiguration { rules }
+    }
+
+    fn sse_rule(algorithm: &str, kms_key_id: Option<&str>) -> ServerSideEncryptionRule {
+        ServerSideEncryptionRule {
+            apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
+                sse_algorithm: ServerSideEncryption::from(algorithm.to_string()),
+                kms_master_key_id: kms_key_id.map(|id| id.to_string()),
+            }),
+            blocked_encryption_types: None,
+            bucket_key_enabled: None,
+        }
+    }
+
+    /// The stored configuration must be one the write path honours as written:
+    /// only AES256 and aws:kms exist, and a key id belongs to aws:kms alone.
+    #[test]
+    fn put_bucket_encryption_refuses_configurations_the_write_path_cannot_honour() {
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", None)])).expect("AES256 is valid");
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("aws:kms", Some("bucket-key"))]))
+            .expect("aws:kms with a key is valid");
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("aws:kms", None)]))
+            .expect("aws:kms without a key is valid (the default key is filled in)");
+        validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", Some(""))]))
+            .expect("an empty key id on AES256 is how some clients spell 'none'");
+
+        let unknown = validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES128", None)]))
+            .expect_err("AES128 is not an algorithm this server encrypts with");
+        assert_eq!(*unknown.code(), S3ErrorCode::MalformedXML);
+
+        let misplaced = validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", Some("bucket-key"))]))
+            .expect_err("a key id only makes sense for aws:kms");
+        assert_eq!(*misplaced.code(), S3ErrorCode::InvalidArgument);
+
+        let empty = validate_bucket_encryption_configuration(&sse_config(Vec::new())).expect_err("no rule, no default");
+        assert_eq!(*empty.code(), S3ErrorCode::MalformedXML);
+
+        let bare_rule = validate_bucket_encryption_configuration(&sse_config(vec![ServerSideEncryptionRule {
+            apply_server_side_encryption_by_default: None,
+            blocked_encryption_types: None,
+            bucket_key_enabled: None,
+        }]))
+        .expect_err("a rule without ApplyServerSideEncryptionByDefault configures nothing");
+        assert_eq!(*bare_rule.code(), S3ErrorCode::MalformedXML);
+
+        // A second rule is checked too, so a malformed one cannot hide behind a valid first rule.
+        let second_bad =
+            validate_bucket_encryption_configuration(&sse_config(vec![sse_rule("AES256", None), sse_rule("garbage", None)]))
+                .expect_err("every rule is validated");
+        assert_eq!(*second_bad.code(), S3ErrorCode::MalformedXML);
+    }
+
     #[tokio::test]
     async fn bucket_usecase_task_finishes_post_commit_hooks_after_parent_cancellation() {
         let admission = Arc::new(Semaphore::new(1));
@@ -5033,9 +5130,11 @@ mod tests {
 
     #[tokio::test]
     async fn execute_put_bucket_encryption_returns_internal_error_when_store_uninitialized() {
+        // A well-formed rule, so the request reaches the store lookup instead
+        // of being refused by configuration validation first.
         let input = PutBucketEncryptionInput::builder()
             .bucket("test-bucket".to_string())
-            .server_side_encryption_configuration(ServerSideEncryptionConfiguration::default())
+            .server_side_encryption_configuration(sse_config(vec![sse_rule("AES256", None)]))
             .build()
             .unwrap();
 

rustfs/src/app/multipart_usecase.rs

@@ -1026,7 +1026,7 @@ impl DefaultMultipartUsecase {
         let (effective_sse, effective_kms_key_id) = match prepared_material {
             Some(material) => {
                 let server_side_encryption = Some(material.server_side_encryption.clone());
-                let ssekms_key_id = material.kms_key_id.clone();
+                let ssekms_key_id = material.response_kms_key_id();
 
                 let mut encryption_metadata = encryption_material_to_metadata(&material)?;
                 if material.key_kind == EncryptionKeyKind::Object {

rustfs/src/app/object/copy.rs

@@ -693,7 +693,7 @@ impl DefaultObjectUsecase {
 
         if let Some(material) = sse_encryption(encryption_request).await? {
             effective_sse = Some(material.server_side_encryption.clone());
-            effective_kms_key_id = material.kms_key_id.clone();
+            effective_kms_key_id = material.response_kms_key_id();
 
             write_plan = write_plan.with_encryption(material.write_encryption(None));
 

rustfs/src/app/object/extract.rs

@@ -2503,7 +2503,7 @@ impl DefaultObjectUsecase {
                 .await
             ) {
                 effective_sse = Some(material.server_side_encryption.clone());
-                effective_kms_key_id = material.kms_key_id.clone();
+                effective_kms_key_id = material.response_kms_key_id();
                 write_plan = write_plan.with_encryption(material.write_encryption(None));
                 let encryption_metadata = extract_try!(encryption_material_to_metadata(&material));
                 metadata.extend(encryption_metadata.clone());

rustfs/src/app/object/put.rs

@@ -1867,7 +1867,7 @@ impl DefaultObjectUsecase {
 
         if let Some(material) = encryption_material {
             effective_sse = Some(material.server_side_encryption.clone());
-            effective_kms_key_id = material.kms_key_id.clone();
+            effective_kms_key_id = material.response_kms_key_id();
 
             write_plan = write_plan.with_encryption(material.write_encryption(None));
 

rustfs/src/app/object/shared.rs

@@ -1424,9 +1424,10 @@ mod tests {
 
     #[test]
     fn resolve_bucket_default_sse_falls_back_to_aes256_for_an_unknown_algorithm() {
-        // Reachable only through corrupt or hand-edited bucket metadata;
-        // PutBucketEncryption rejects unknown algorithms. All three call sites
-        // now share this single decision (backlog#1826).
+        // PutBucketEncryption refuses unknown algorithms, so this is reachable
+        // only through a configuration stored before that check or through
+        // hand-edited bucket metadata. All three call sites share this single
+        // decision (backlog#1826).
         let config = bucket_sse_config_with("garbage", None);
 
         let (sse, kms_key_id) = resolve_bucket_default_sse(Some(&config), None, None, false);

rustfs/src/storage/sse.rs

@@ -220,8 +220,9 @@ pub struct SseConfiguration {
 /// malformed bucket default pass the `copy_changes_encryption` guard and take
 /// the metadata-only shortcut while this layer still encrypts: fresh DEK
 /// metadata is committed beside the untouched plaintext blocks and the object
-/// becomes unreadable. Reachable only via corrupt or hand-edited bucket
-/// metadata — PutBucketEncryption rejects unknown algorithms (backlog#1826).
+/// becomes unreadable. PutBucketEncryption refuses unknown algorithms, so this
+/// is reachable only through a configuration stored before that check or
+/// through hand-edited bucket metadata (backlog#1826).
 pub(crate) fn bucket_default_write_sse(sse: &ServerSideEncryptionByDefault) -> ServerSideEncryption {
     match sse.sse_algorithm.as_str() {
         "AES256" => ServerSideEncryption::from_static(ServerSideEncryption::AES256),
@@ -723,6 +724,11 @@ pub(crate) fn map_get_object_reader_error(err: StorageError) -> ApiError {
         let code = match resolution_error.kind() {
             EncryptionResolutionErrorKind::InvalidRequest => S3ErrorCode::InvalidRequest,
             EncryptionResolutionErrorKind::ServiceUnavailable => S3ErrorCode::ServiceUnavailable,
+            // Same code the write path returns for this key; `From<ApiError>`
+            // attaches the 400 that s3s cannot derive for a custom code.
+            EncryptionResolutionErrorKind::KeyNotFound => S3ErrorCode::Custom(crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+            EncryptionResolutionErrorKind::AccessDenied => S3ErrorCode::AccessDenied,
+            EncryptionResolutionErrorKind::NotImplemented => S3ErrorCode::NotImplemented,
             // A permanent property of the stored object, not a transient server
             // fault: 5xx would invite client retry storms against an object
             // this server can never decrypt.
@@ -1512,10 +1518,19 @@ fn normalize_encryption_metadata_case(
     Ok(Cow::Owned(normalized))
 }
 
+/// Carry the S3-level classification of a decryption failure through the
+/// ecstore boundary. Every code produced by `data_plane_kms_error` needs a
+/// kind here, otherwise the read path reports it as an internal fault even
+/// though the write path already reports the same KMS error to the client.
 fn map_encryption_resolution_error(error: ApiError) -> EncryptionResolutionError {
-    let kind = match error.code {
+    let kind = match &error.code {
         S3ErrorCode::InvalidArgument | S3ErrorCode::InvalidRequest => EncryptionResolutionErrorKind::InvalidRequest,
         S3ErrorCode::ServiceUnavailable => EncryptionResolutionErrorKind::ServiceUnavailable,
+        S3ErrorCode::Custom(code) if &**code == crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE => {
+            EncryptionResolutionErrorKind::KeyNotFound
+        }
+        S3ErrorCode::AccessDenied => EncryptionResolutionErrorKind::AccessDenied,
+        S3ErrorCode::NotImplemented => EncryptionResolutionErrorKind::NotImplemented,
         _ => EncryptionResolutionErrorKind::DecryptionFailed,
     };
     EncryptionResolutionError::new(kind, error.message)
@@ -1530,6 +1545,19 @@ pub struct ManagedSealedKey {
 }
 
 impl EncryptionMaterial {
+    /// The KMS key id a write response may advertise.
+    ///
+    /// `kms_key_id` is always set for managed SSE because SSE-S3 also wraps
+    /// its data key under the service default key, but that key is an
+    /// internal detail of an `AES256` object: only an `aws:kms` object names
+    /// a key the caller can act on, and `x-amz-server-side-encryption-aws-kms-key-id`
+    /// is defined only for that scheme.
+    pub fn response_kms_key_id(&self) -> Option<SSEKMSKeyId> {
+        matches!(self.sse_type, SSEType::SseKms)
+            .then(|| self.kms_key_id.clone())
+            .flatten()
+    }
+
     pub fn write_encryption(&self, multipart_part_number: Option<usize>) -> super::WriteEncryption {
         match (self.key_kind, multipart_part_number) {
             (EncryptionKeyKind::Object, Some(part_number)) => {
@@ -2473,7 +2501,9 @@ pub async fn classify_sse_read_response(request: DecryptionRequest<'_>) -> Resul
         server_side_encryption: ServerSideEncryption::from(managed_sse_public_header(sse_type).to_string()),
         sse_customer_algorithm: None,
         sse_customer_key_md5: None,
-        ssekms_key_id: Some(SSEKMSKeyId::from(kms_key_id)),
+        // The key id was needed above to authorize the read, but an AES256
+        // object's wrapping key is internal: only aws:kms objects advertise it.
+        ssekms_key_id: matches!(sse_type, SSEType::SseKms).then(|| SSEKMSKeyId::from(kms_key_id)),
     }))
 }
 
@@ -2751,9 +2781,21 @@ async fn apply_managed_encryption_material_inner(
         }
         (SSEType::SseKms, Some(kms_key_id)) => kms_key_id,
         (SSEType::SseKms, None) => {
-            return Err(ApiError::from(StorageError::other(
-                "No KMS key available for managed server-side encryption (required for SSE-KMS)",
-            )));
+            // Neither the request nor the bucket default named a key and no
+            // service default filled in. Without a service this is the same
+            // outage/misconfiguration the provider check below reports, so it
+            // must carry the same 503/400 split rather than an untyped
+            // internal error; with a running service that has no default key
+            // the caller simply has to name one.
+            if runtime_sources::current_encryption_service().await.is_none() {
+                return Err(sse_kms_unavailable_error(kms_configured_but_unavailable().await));
+            }
+            return Err(ApiError {
+                code: S3ErrorCode::InvalidRequest,
+                message: "SSE-KMS requires a KMS key id: the request named none and the KMS service has no default key"
+                    .to_string(),
+                source: None,
+            });
         }
         _ => unreachable!("managed SSE branch only supports SSE-S3 or SSE-KMS"),
     };
@@ -3354,6 +3396,25 @@ fn kms_operation_error(error: rustfs_kms::KmsError) -> ApiError {
     api_error
 }
 
+/// Classification for a failed data-key unwrap.
+///
+/// An AEAD failure stays `500` (the envelope is an integrity fault, not a
+/// request a retry or a different header can fix), but the generic internal
+/// error text hides the one diagnosis an operator needs: the configured
+/// backend holds different key material under this key id than the one that
+/// wrapped the object, typically after re-creating a key of the same name or
+/// switching backends.
+fn kms_unwrap_error(error: rustfs_kms::KmsError) -> ApiError {
+    let unwrap_rejected = matches!(error, rustfs_kms::KmsError::CryptographicError { .. });
+    let mut api_error = kms_operation_error(error);
+    if unwrap_rejected && api_error.code == S3ErrorCode::InternalError {
+        api_error.message = "The object's data key envelope could not be unwrapped by the configured KMS backend: the \
+                             key material under this key id differs from the one that wrapped it, or the envelope is damaged"
+            .to_string();
+    }
+    api_error
+}
+
 impl KmsSseDekProvider {
     /// Create a new KMS-backed provider
     pub async fn new() -> Result<Self, ApiError> {
@@ -3438,7 +3499,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let data_key = service
             .decrypt_data_key(encrypted_dek, context)
             .await
-            .map_err(kms_operation_error)?;
+            .map_err(kms_unwrap_error)?;
 
         Ok(data_key.plaintext_key)
     }
@@ -3457,7 +3518,7 @@ impl SseDekProvider for KmsSseDekProvider {
         let data_key = service
             .decrypt_legacy_data_key(encrypted_dek)
             .await
-            .map_err(kms_operation_error)?;
+            .map_err(kms_unwrap_error)?;
 
         Ok(data_key.plaintext_key)
     }
@@ -4409,6 +4470,56 @@ mod tests {
         assert_eq!(super::kms_data_plane_error_class(&missing), "key_not_found");
     }
 
+    /// The read path squeezes the S3 classification through ecstore's
+    /// resolution-error kinds; every KMS class the write path reports to the
+    /// client must survive that hop instead of collapsing onto `DecryptionFailed`
+    /// (which the S3 layer reports as `500`).
+    #[test]
+    fn encryption_resolution_kinds_preserve_kms_read_classification() {
+        let cases = [
+            (
+                rustfs_kms::KmsError::key_not_found("no-such-key"),
+                EncryptionResolutionErrorKind::KeyNotFound,
+            ),
+            (rustfs_kms::KmsError::access_denied("policy"), EncryptionResolutionErrorKind::AccessDenied),
+            (
+                rustfs_kms::KmsError::unsupported_capability("local", "decrypt_legacy"),
+                EncryptionResolutionErrorKind::NotImplemented,
+            ),
+            (
+                rustfs_kms::KmsError::backend_error("connection refused"),
+                EncryptionResolutionErrorKind::ServiceUnavailable,
+            ),
+            (
+                rustfs_kms::KmsError::invalid_operation("key is disabled"),
+                EncryptionResolutionErrorKind::InvalidRequest,
+            ),
+            (
+                rustfs_kms::KmsError::cryptographic_error("decrypt", "authentication failed"),
+                EncryptionResolutionErrorKind::DecryptionFailed,
+            ),
+        ];
+        for (error, expected) in cases {
+            let description = error.to_string();
+            let resolution = super::map_encryption_resolution_error(kms_operation_error(error));
+            assert_eq!(resolution.kind(), expected, "{description}");
+        }
+    }
+
+    /// An unwrap the backend rejects stays an internal error, but says why in
+    /// words an operator can act on rather than the generic 500 text.
+    #[test]
+    fn kms_unwrap_error_keeps_500_but_names_the_envelope_mismatch() {
+        let rejected = super::kms_unwrap_error(rustfs_kms::KmsError::cryptographic_error("decrypt", "authentication failed"));
+        assert_eq!(rejected.code, S3ErrorCode::InternalError);
+        assert!(rejected.message.contains("could not be unwrapped"), "message was {}", rejected.message);
+        assert_eq!(super::kms_data_plane_error_class(&rejected), "cryptographic");
+
+        let missing = super::kms_unwrap_error(rustfs_kms::KmsError::key_not_found("no-such-key"));
+        assert_eq!(missing.code, S3ErrorCode::Custom(crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE.into()));
+        assert_eq!(missing.message, "KMS key not found: no-such-key");
+    }
+
     #[test]
     fn sse_kms_never_falls_back_to_the_local_sse_s3_provider() {
         let unconfigured = super::sse_kms_unavailable_error(false);
@@ -4496,6 +4607,56 @@ mod tests {
         reset_sse_dek_provider();
     }
 
+    /// A bare `aws:kms` request (no key id, no bucket default) on a node with
+    /// no KMS has no key to resolve. It must get the same configuration
+    /// refusal as the keyed form, whether or not the SSE-S3 master key is
+    /// set, rather than an untyped internal error (backlog#2368 B4).
+    #[tokio::test]
+    async fn sse_kms_write_without_a_key_id_is_refused_like_the_keyed_form() {
+        let _guard = lock_sse_test_state().await;
+
+        for master_key in [None, Some(BASE64_STANDARD.encode_to_string([9u8; 32]))] {
+            reset_sse_dek_provider();
+            async_with_vars(
+                [
+                    ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
+                    ("RUSTFS_SSE_S3_MASTER_KEY", master_key.clone()),
+                ],
+                async {
+                    // Entered directly: `sse_encryption` consults the bucket
+                    // default first, which needs a bucket metadata store.
+                    let error = apply_managed_encryption_material(
+                        "finance",
+                        "ledger.csv",
+                        ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
+                        None,
+                        None,
+                        128,
+                        None,
+                    )
+                    .await
+                    .expect_err("SSE-KMS without a key id must be refused when no KMS is running");
+
+                    assert_eq!(
+                        error.code,
+                        S3ErrorCode::InvalidRequest,
+                        "master_key={master_key:?}: message was {}",
+                        error.message
+                    );
+                    assert!(error.message.contains("SSE-KMS requires"), "message was {}", error.message);
+                    assert!(
+                        !error.message.contains("RUSTFS_SSE_S3_MASTER_KEY"),
+                        "an SSE-KMS refusal must not name the SSE-S3 master key: {}",
+                        error.message
+                    );
+                },
+            )
+            .await;
+        }
+
+        reset_sse_dek_provider();
+    }
+
     /// The SSE-S3 local fallback itself is unchanged: refusing SSE-KMS must not
     /// take the documented no-KMS deployment down with it.
     #[tokio::test]
@@ -5834,7 +5995,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_sse_encryption_persists_aws_kms_header_for_kms_objects() {
-        let metadata = encryption_material_to_metadata(&EncryptionMaterial {
+        let material = EncryptionMaterial {
             sse_type: SSEType::SseKms,
             server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
             kms_key_id: Some("test-key".to_string()),
@@ -5847,8 +6008,10 @@ mod tests {
             key_kind: EncryptionKeyKind::Direct,
             managed_kms_context: None,
             managed_sealed_key: None,
-        })
-        .expect("managed SSE metadata should serialize");
+        };
+        // Only an aws:kms object names its key in write responses.
+        assert_eq!(material.response_kms_key_id().as_deref(), Some("test-key"));
+        let metadata = encryption_material_to_metadata(&material).expect("managed SSE metadata should serialize");
 
         assert_eq!(metadata.get("x-amz-server-side-encryption").map(String::as_str), Some("aws:kms"));
         assert_eq!(
@@ -6025,6 +6188,9 @@ mod tests {
                 let metadata = encryption_material_to_metadata(&material).expect("managed SSE-S3 metadata should serialize");
 
                 assert_eq!(material.kms_key_id.as_deref(), Some("default"));
+                // The wrapping key stays internal: no write response may
+                // advertise it for an AES256 object.
+                assert_eq!(material.response_kms_key_id(), None);
                 assert_eq!(metadata.get("x-amz-server-side-encryption").map(String::as_str), Some("AES256"));
                 assert!(!metadata.contains_key("x-amz-server-side-encryption-aws-kms-key-id"));
                 assert_eq!(metadata.get(INTERNAL_ENCRYPTION_KEY_ID_HEADER).map(String::as_str), Some("default"));
@@ -7528,6 +7694,32 @@ mod tests {
         assert_eq!(err.message, "KMS unavailable");
     }
 
+    /// The reader wraps the resolution error in an io error exactly like
+    /// `readers.rs` does; a missing key must come out as the same `400
+    /// KMS.NotFoundException` the write path returns, not `500`.
+    #[test]
+    fn test_map_get_object_reader_error_reports_missing_kms_key_as_client_error() {
+        let resolution_error =
+            super::EncryptionResolutionError::new(EncryptionResolutionErrorKind::KeyNotFound, "KMS key not found: finance-key");
+        let err = map_get_object_reader_error(StorageError::other(resolution_error));
+        assert_eq!(err.code, S3ErrorCode::Custom(crate::error::KMS_KEY_NOT_FOUND_ERROR_CODE.into()));
+        assert_eq!(err.message, "KMS key not found: finance-key");
+        let s3_error = s3s::S3Error::from(err);
+        assert_eq!(s3_error.status_code(), Some(http::StatusCode::BAD_REQUEST));
+
+        let denied = map_get_object_reader_error(StorageError::other(super::EncryptionResolutionError::new(
+            EncryptionResolutionErrorKind::AccessDenied,
+            "Access Denied",
+        )));
+        assert_eq!(denied.code, S3ErrorCode::AccessDenied);
+
+        let unsupported = map_get_object_reader_error(StorageError::other(super::EncryptionResolutionError::new(
+            EncryptionResolutionErrorKind::NotImplemented,
+            "backend cannot unwrap legacy envelopes",
+        )));
+        assert_eq!(unsupported.code, S3ErrorCode::NotImplemented);
+    }
+
     #[test]
     fn test_map_get_object_reader_error_maps_part_missing_to_slow_down_read() {
         let err = map_get_object_reader_error(StorageError::PartMissingOrCorrupt);
@@ -8083,6 +8275,32 @@ mod tests {
     // Read-side response classification (single-decrypt GET path)
     // ========================================================================
 
+    /// An AES256 object is read under the same key-id resolution as aws:kms
+    /// (authorization needs it), but the response must not advertise that
+    /// internal wrapping key.
+    #[tokio::test]
+    async fn classification_withholds_the_wrapping_key_for_sse_s3_reads() {
+        let metadata = HashMap::from([
+            ("x-amz-server-side-encryption".to_string(), ServerSideEncryption::AES256.to_string()),
+            (INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), "service-default".to_string()),
+            (INTERNAL_ENCRYPTION_KEY_HEADER.to_string(), BASE64_STANDARD.encode_to_string([1u8; 16])),
+            (INTERNAL_ENCRYPTION_IV_HEADER.to_string(), BASE64_STANDARD.encode_to_string([2u8; 12])),
+        ]);
+        let headers = super::classify_sse_read_response(DecryptionRequest {
+            bucket: "finance",
+            key: "ledger.csv",
+            metadata: &metadata,
+            sse_customer_key: None,
+            sse_customer_key_md5: None,
+            principal: None,
+        })
+        .await
+        .expect("sse-s3 classification should succeed")
+        .expect("managed metadata should classify");
+        assert_eq!(headers.server_side_encryption.as_str(), ServerSideEncryption::AES256);
+        assert_eq!(headers.ssekms_key_id, None);
+    }
+
     #[tokio::test]
     async fn classification_reproduces_managed_read_headers_and_audit_without_a_kms_unwrap() {
         use rustfs_kms::types::{CreateKeyRequest, KeyUsage};