fix(sse): resolve 1.0.0 SSE/KMS blockers and P1 findings (#7511)

* fix(sse): resolve bucket default encryption per request PUT and the POST-object/extract path resolved a bucket's default encryption with a hard-coded "no explicit SSE-C" flag, so the default was layered onto a request that already carried an SSE-C header triple and then tripped that request's own mutual-exclusion check. Every bucket with default encryption refused SSE-C single PUTs with 400 InvalidArgument, while CreateMultipartUpload on the same bucket succeeded because it resolves SSE elsewhere. Both call sites now derive the flag from the request headers, as COPY already did. The bucket default's KMS key id was also inherited independently of the effective algorithm, so an explicit AES256 request against an aws:kms default bucket produced a self-contradictory algorithm/key-id pair and was rejected. The key id is now inherited only when the effective algorithm is aws:kms, matching the storage-layer resolver. Refs backlog#2368 B1, B2. * fix(sse): refuse SSE-KMS without a running KMS service A write requesting aws:kms on a node with no KMS service fell back to the node-local SSE-S3 provider: the data key was wrapped with RUSTFS_SSE_S3_MASTER_KEY while the object metadata still recorded aws:kms and the requested KMS key id. The stored object claimed a KMS protection it never had, under a key that was never consulted, and no signal distinguished it from a genuine SSE-KMS object. The managed-encryption path now asks the resolved DEK provider whether it wraps with a node-local master key and refuses SSE-KMS in that case: InvalidRequest when KMS was never configured, ServiceUnavailable when a configured service is not running. The check sits after the per-key authorization gate so an unauthorized caller still receives AccessDenied whatever the KMS runtime state is, and asks the provider rather than a parallel availability signal because the provider is what actually wraps the key. A missing master key no longer answers an SSE-KMS request with an SSE-S3-worded configuration error. The SSE-S3 local fallback is unchanged. Refs backlog#2368 B4. * fix(ecstore): restore and archive tiers in stored coordinates Multipart restore addressed the remote tier in plaintext coordinates while the copy-back reads the stored representation. Each 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 silently replaced the object's bytes. Encrypted and compressed multipart objects were both affected. Restore now accumulates stored part sizes, passes the stored length to the hash reader alongside the plaintext length, and validates against the stored size. The copy-back digests stored bytes, so its computed MD5 is not the object's public ETag. Restore now preserves the object ETag on both the single-part and multipart paths, and gives each restored part its own recorded part ETag rather than the object-level value. Transition also handed the tier the object's SSE headers and its RustFS-wrapped data key as request headers. Any S3 target rejected an SSE-C archive outright, an SSE-KMS archive asked the target to encrypt a second time under a key id it does not own, and the wrapped DEK left the cluster. The archive request now strips every SSE header and encryption marker with the predicate the replication path already uses; the local xl.meta keeps all of it, so read-through and restore are unaffected. Objects restored by an affected release are not detected or repaired retroactively and must be re-restored from the tier. Refs backlog#2368 B3, B5; backlog#2369 P7.1. * fix(rio): lock the v1 nonce layout within a segment Decrypting a v1 segment tried three historical nonce layouts per frame, independently for every frame. The last of them exists for streams written before 1.0.0-alpha.91, which reused a segment's part nonce for every block in it; because block zero's derived nonce equals that base nonce, a frame encrypted at index zero authenticated at any position. An attacker able to rewrite the underlying shards could replay it and have the forged plaintext returned with 200 and an unchanged length. Shard integrity uses a keyed-hash-free checksum, which such an attacker can recompute, so it is not a barrier. A segment now locks onto whichever layout decoded its first non-zero-index frame and rejects any later frame needing a different one. That leaves one residual shape: a stream built purely from repeats of frame zero has no later frame to disagree. New RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK (default true, so pre-alpha.91 objects keep decrypting) drops the third layout entirely when set to false, which closes it. Turning it off refuses pre-alpha.91 objects, so migrate them first by rewriting in place. Refs backlog#2369 P2. * fix(kms): reload a service that failed to start POST /rustfs/admin/v3/kms/reload short-circuited whenever the persisted configuration matched the in-memory one byte for byte. A node whose KMS failed to start keeps that configuration and sits in Error, so the documented recovery call returned "reloaded successfully" while leaving the node down. Peers reached the same path through the reload broadcast, so a cluster that lost Vault during a rolling restart had no working recovery route other than the node-local start endpoint. Reload now short-circuits only for a service that is actually running, and otherwise reconfigures, which starts a service that is not running. The AWS backend also advertised key-version enumeration through kms/status, which its own documentation says it cannot do; the capability and its golden snapshot now say false. Refs backlog#2369 P1, P7.3. * docs: record the SSE and KMS changes for 1.0.0 The Unreleased changelog section carried no entry for any encryption work merged since 1.0.0-rc.5, including three items with operational impact: the config-secret variable whose absence persists secrets in cleartext with only a warning, the v2 frame write switch and its rolling-upgrade constraint, and per-key authorization making a public bucket incompatible with SSE-KMS objects. Adds those plus this batch, including the SSE-KMS refusal as a breaking change with both routes out. Also corrects four places where documentation contradicted the code: the cleanup register still called encrypted range seek opt-in after its default flipped, the Helm README claimed vault_mount_path only applies to Transit while the template also feeds the KV2 mount, the disaster-recovery drill listed bundle contents for backends whose export is refused with 501, and the Chinese README capability table predated most of the feature set. Documents the SSE-S3 local master key as a first-class operational mode with its rotation dead end, and what the v1 frame layout does and does not authenticate. Refs backlog#2369 P5. * fix(kms): classify data-path KMS failures by what the caller can do Only "key not found" and a backend outage were classified; every other KMS failure that reached the S3 data path fell through to 500 InternalError with a generic message. A disabled or pending-deletion key, a denied KMS grant, an encryption-context mismatch, an unsupported algorithm, a credential or timeout failure, and a capability the configured backend does not have all looked identical to a server fault. SDKs therefore applied exponential backoff to configuration errors no retry can fix, and monitoring counted every one of them against the server's own error rate. Unusable-key and request-side failures now answer 400, a denied grant 403, transient backend failures 503, and a missing backend capability 501. Damaged, unreadable, or unknown-format key material keeps its 500: it is a server-side integrity fault, and existing tests pin it. The classifier is deliberately separate from the admin lifecycle mapping, which answers 404 for a missing key because there a key id is the resource being addressed; on the data path it arrives inside a request header or a bucket default. Messages either name what the caller asked for or stay generic, with deployment-side detail left on the error source the way the storage-IO mapping already does. Refs backlog#2368 B6. * fix(kms): track and renew static Vault tokens Token authentication hard-coded "this token carries no lease", so the renewal task never started, the remaining-TTL gauge was never published, and nothing looked wrong. `vault token create` grants a 768-hour TTL by default, so a cluster that had been healthy for a month turned every KMS call into a 403 and could not recover without a restart or a reconfigure. Production configuration validation only rejects the literal dev-token, so an ordinary expiring token reaches a whole cluster. The source now reads `auth/token/lookup-self` at login and adopts what Vault reports. A token with no expiry behaves exactly as before. An expiring renewable one is picked up by the existing renewal loop and renewed at half TTL like every other auth method. An expiring non-renewable one warns with its remaining lifetime and publishes the gauge, so the fail-closed window is visible before it arrives. The probe never fails the login: a policy that omits lookup-self, or a Vault that is briefly unreachable, warns and falls back to exactly the previous behaviour rather than taking down a deployment that works today. The scripted Vault test double answers the lookup out of band so existing scripts keep describing only the protocol under test. Refs backlog#2369 P3. * feat(sse): report SSE-C requests that arrive without TLS An SSE-C request carries the customer's AES key in a request header, so AWS S3 and MinIO both refuse one that did not arrive over TLS. RustFS accepted them on any transport: a plaintext hop hands the key to anyone on the path, and since the object cannot be read without that same key, the exposure lasts as long as the object does. Refusing outright is the correct end state but not a safe default to adopt inside a release window, because the project's own s3-tests and e2e lanes and most staging deployments speak plain HTTP. This release reports instead: each such request increments rustfs_ssec_plaintext_requests_total and logs one warning per process, so an operator can confirm nothing would break before the default flips. RUSTFS_SSE_C_REQUIRE_TLS=true opts into the AWS 400 now. The verdict is per connection rather than per deployment: the layer is built with whether this listener terminated TLS, and additionally accepts an https protocol forwarded by a proxy the trusted-proxy configuration already vetted. It sits beside the rate limiter, after the layer that makes a forwarded protocol trustworthy and after the request context, so a rejection can echo the request id. Refs backlog#2369 P7.2. * fix(kms): say what a node-local backend means for a cluster The Local backend keeps key material on each node's own disk and generates its Argon2id salt per node, so two nodes derive different keys from the same master_key and an object encrypted on one node cannot be decrypted on another. Behind a load balancer that surfaces as intermittent 500s on reads that succeeded moments earlier, with nothing tying the symptom to the cause: the only signal was a generic "development, testing and demos only" positioning warning that says nothing about what actually breaks. Configuring or reconfiguring Local while the deployment is distributed now logs a dedicated event and appends the consequence to the configure response, so the operator who made the change sees it. The product decision to warn rather than refuse is unchanged. Refs backlog#2369 P7.4. * docs: record the remaining SSE and KMS changes for 1.0.0 Adds changelog entries for the KMS data-path status classification, the Vault static-token lease probe, the SSE-C plaintext-transport report and its switch, and the node-local backend warning. Documents two things the backend security guide never stated: that SSE-C belongs on a secure transport, with the counter and switch to plan the change around, and that the Local backend cannot be shared by a multi-node deployment because each node derives different keys from the same master key. Refs backlog#2368 B6; backlog#2369 P3, P5, P7.2, P7.4. * fix(kms): report an unreadable key store as an outage on the S3 path A backend now distinguishes a key store it could not read from a key that is genuinely absent, but the S3 boundary collapsed the first one back onto 500 InternalError through the fallthrough for integrity faults. The distinction was therefore invisible to the client: a temporary key-directory outage looked exactly like a permanently damaged key record, and neither the status nor the metric said the request was worth retrying. An unreadable key store joins the retryable class and answers 503, next to a backend error and a credential failure. Damaged, unreadable or unknown-format key material keeps its 500. Refs backlog#2368 B6; builds on rustfs/rustfs#7470.
2026-09-08 21:25:59 +00:00 · 2026-09-08 22:37:53 +08:00
parent 929a9f27c4
commit ae9fe62fb1
27 changed files with 1760 additions and 90 deletions
@@ -13,6 +13,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 - **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.
+- **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.
+- **Restore no longer drifts the object ETag**: the copy-back digests stored (encrypted or compressed) bytes, so the recomputed MD5 is not the object's public ETag. Single-part and multipart restores now preserve the original object ETag, and each restored part keeps its own recorded part ETag.
+- **ILM archive no longer forwards encryption metadata to the tier**: transition requests carried the object's SSE headers and the RustFS-wrapped data key as request headers. Any S3 target rejected an SSE-C archive outright (`400`, no key supplied), an SSE-KMS archive asked the target to encrypt a second time under a key id it does not own, and the wrapped DEK left the cluster. The archive request now strips every SSE header and encryption marker using the same predicate the replication path uses; the local `xl.meta` keeps all of it, so read-through and restore are unaffected.
+- **KMS reload is no longer a no-op on a node whose KMS failed to start**: `POST /rustfs/admin/v3/kms/reload` short-circuited whenever the persisted configuration matched the in-memory one byte for byte. A node whose KMS failed to start (for example Vault briefly unreachable during a rolling restart) keeps that configuration and sits in `Error`, so the documented recovery call returned "reloaded successfully" while leaving the node down — and did the same on every peer through the reload broadcast. Reload now short-circuits only for a service that is actually running, and otherwise reconfigures, which starts the service.
+- **AWS KMS capability reporting**: the AWS backend no longer advertises `versioning` support through `GET /rustfs/admin/v3/kms/status`. AWS KMS key versions are not enumerable through this backend, as the backend documentation already stated.
 - **Multipart admission queue**: an `UploadPart` waiting for a foreground write permit now waits at most 10 s by default (`RUSTFS_PUT_MULTIPART_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS`, previously 30 s), so a queued part returns S3 `SlowDown` before the client's socket write timeout drops the connection. Separately, the API listener no longer forces a 4 MiB `SO_RCVBUF` on every accepted socket (kernel autotuning applies; `RUSTFS_HTTP_SOCKET_RECV_BUFFER_BYTES` restores a fixed size), so a queued part no longer lets up to 8 MiB of unread body accumulate in kernel memory per connection, which is what throttled whole nodes under SDK-default multipart concurrency. Fixes #7385.
 - **Helm Ingress**: `customAnnotations` are now merged with class-specific annotations (nginx/traefik) instead of being ignored when `ingress.className` is set.
 - **Per-pool erasure parity**: Erasure parity (STANDARD and reduced-redundancy) is now resolved independently for every pool instead of reusing the first pool's value. A heterogeneous topology — for example a 4-drive pool plus a 2-drive pool created during expansion — previously inherited the first pool's parity and could resolve to zero data shards in the smaller pool, panicking Reed-Solomon construction on write. Automatic parity now resolves per pool (for example `2+2` in the 4-drive pool and `1+1` in the 2-drive pool). Fixes #4801.
@@ -54,6 +62,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
  - Four-layer regression-prevention tests guard against silent feature deletion: compile-time module assertion, module-presence unit test, cross-module `Protocol` enum assertion, end-to-end SSH banner test against the running binary

 ### Changed
+- **Encryption and KMS work merged since `1.0.0-rc.5`** (entries were missing from this section):
+  - **Persisted KMS configuration secrets** are sealed field-by-field with `RUSTFS_KMS_CONFIG_SECRET`. **When the variable is unset the secrets are persisted in cleartext and the server only warns** (`persisted KMS configuration carries cleartext secrets`); it never refuses the write. Set it, identically, on every node, and re-save the configuration to seal an existing one.
+  - **New v2 ciphertext frame format** with per-frame index binding and final-frame authentication. Its write switch `RUSTFS_ENCRYPTION_FRAME_V2` is **off by default**: v2 frames are unreadable by nodes without v2 read support, and encrypted ciphertext travels verbatim through transition, decommission and SSE-C replication passthrough, so turn it on only after every node — and every RustFS warm/replication target that receives raw ciphertext — runs a release with v2 read support. Reading v2 objects needs no switch.
+  - **Per-key SSE-KMS authorization** (`RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY`, default `false`). With it on, anonymous callers hold no KMS grants, so **a public bucket serving SSE-KMS objects is an incompatible combination** and those reads return `AccessDenied`.
+  - Envelope context binding as KMS AAD (`ENV_KMS_ENVELOPE_AAD`, off by default; a node that predates the field cannot open bound envelopes).
+  - Vault custom CA and mutual TLS; object-level DEK rewrap plus a batch rekey admin API; a backend-locality runtime signal on `kms/status`.
+  - Single-pass decryption for encrypted GET, and encrypted single-part closed-range seek — the latter is now **on by default** (`RUSTFS_ENCRYPTED_RANGE_SEEK`, default `true`; the switch remains as a kill switch).
+- **Vault static tokens are now tracked and renewed**: with `Token` authentication RustFS hard-coded "this token has no lease", so the renewal task never started and no remaining-TTL gauge was published. `vault token create` grants a 768-hour TTL by default, which turned a healthy-looking cluster into one where every KMS call returned 403 about a month later, with no self-healing short of a restart or reconfigure. RustFS now calls `auth/token/lookup-self` at login and adopts what Vault reports: a non-expiring token behaves exactly as before, an expiring renewable one is renewed at half TTL like the other auth methods, and an expiring non-renewable one logs `vault_static_token_not_renewable` and publishes its remaining TTL. The probe never fails the login: a token whose policy omits `lookup-self` (Vault's `default` policy grants it), or a Vault that is unreachable at that moment, logs `vault_static_token_lookup_failed` and falls back to the previous no-lease behaviour, so no deployment that works today stops working.
+- **SSE-C over a plaintext transport is reported**: AWS S3 and MinIO refuse an SSE-C request that did not arrive over TLS, because the customer key travels in a request header. RustFS accepted them on any transport and still does by default — flipping to a rejection inside a release window would break plaintext staging and test deployments. Each such request now increments `rustfs_ssec_plaintext_requests_total` and logs one `ssec_request_without_tls` warning per process, and `RUSTFS_SSE_C_REQUIRE_TLS=true` opts into the AWS `400` now. The default is expected to flip in a later release; confirm the counter reads zero first. The verdict is per connection: a TLS listener satisfies it, and so does an `https` protocol forwarded by a proxy the trusted-proxy configuration accepts.
+- **Local KMS backend on a distributed deployment says what actually breaks**: the backend keeps key material and its Argon2id salt on each node's own disk, so two nodes derive different keys from the same `master_key` and an object encrypted on one node cannot be decrypted on another — intermittent 500s behind a load balancer. Configuring it while the deployment is distributed now logs `kms_node_local_backend_in_distributed_deployment` and appends that consequence to the `kms/configure` response, instead of only the generic "development only" positioning warning. It remains a warning, not a gate.
+- **SSE-KMS is refused when no KMS is running (breaking)**: a write requesting `x-amz-server-side-encryption: aws:kms` on a node with no KMS service no longer succeeds. Earlier releases wrapped the data key with the node-local `RUSTFS_SSE_S3_MASTER_KEY` while still writing `aws:kms` and the requested key id into the object metadata — metadata that claimed a KMS protection the object never had, under a key that was never consulted. Such a request now returns `400 InvalidRequest` when KMS was never configured and `503` when a configured service is not running; the refusal is evaluated after the per-key authorization gate, so an unauthorized caller still receives `403 AccessDenied`. **Upgrade note:** a deployment that relied on this write succeeding will start receiving 4xx/503. Either configure a KMS, or request `AES256` and keep the documented SSE-S3 local-master-key fallback, which is unchanged. Objects already written this way remain readable.
+- **Legacy ciphertext nonce layouts are now locked per segment**: while decrypting a v1 segment, the reader locks onto whichever of the three historical nonce layouts decoded the segment's first non-zero-index frame and rejects any later frame that needs a different one. Because a frame encrypted at block index zero authenticates under the pre-`1.0.0-alpha.91` reused-part-nonce layout at any position, an attacker able to rewrite the underlying shards could previously replay it and have the forged plaintext returned with `200`. New `RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK` (default `true`) drops that third layout entirely when set to `false`, which closes the residual case of a stream built purely from repeats of frame zero. Turn it off only after migrating pre-alpha.91 encrypted objects (rewrite in place with CopyObject); see [KMS backend security properties](docs/operations/kms-backend-security.md) for what the v1 frame layout does and does not authenticate.
 - **HTTP Server Stack**: Integrated `KeystoneAuthLayer` middleware from `rustfs-keystone` crate into service stack (positioned after ReadinessGateLayer)
 - **Storage-class validation on startup (upgrade note)**: A persisted explicit storage class (`RUSTFS_STORAGE_CLASS_STANDARD` / `RUSTFS_STORAGE_CLASS_RRS`, for example `EC:2`) is now validated against the actual per-pool drive counts at startup and rejected when a pool cannot satisfy it. This is fail-closed and correct, but a cluster that persisted a storage class larger than a small or heterogeneous pool can hold (for example `EC:2` alongside a 2-drive pool), which earlier releases accepted and silently resolved to an invalid layout, will now refuse to start after upgrade. To recover, unset `RUSTFS_STORAGE_CLASS_STANDARD` so the server derives a valid per-pool default automatically, or set it to a value every pool can satisfy.
 - **IAMAuth**: Enhanced `get_secret_key()` to return empty secret for Keystone credentials (bypasses signature validation)
@@ -10578,6 +10578,7 @@ dependencies = [
 "serde_json",
 "sha1 0.11.0",
 "sha2 0.11.0",
+ "temp-env",
 "thiserror 2.0.20",
 "tokio",
 "tokio-test",
@@ -46,14 +46,31 @@ RustFS 是一个基于 Rust 构建的高性能分布式对象存储系统。Rust
 - **完全开源**：采用 Apache 2.0 许可证，鼓励社区贡献和商业使用。
 - **简单易用**：设计简洁，易于部署和管理。

-| 功能               | 状态    | 功能                    | 状态      |
-| :----------------- | :------ | :---------------------- | :-------- |
-| **S3 核心功能**    | ✅ 可用 | **Bitrot (防数据腐烂)** | ✅ 可用   |
-| **上传 / 下载**    | ✅ 可用 | **单机模式**            | ✅ 可用   |
-| **版本控制**       | ✅ 可用 | **存储桶复制**          | ✅ 可用   |
-| **日志功能**       | ✅ 可用 | **生命周期管理**        | 🚧 测试中 |
-| **事件通知**       | ✅ 可用 | **分布式模式**          | 🚧 测试中 |
-| **K8s Helm Chart** | ✅ 可用 | **OPA (策略引擎)**      | 🚧 测试中 |
+状态说明：✅ 可用 —— 已发布并有 CI 门禁覆盖；🧪 预览 —— 已发布但需显式开关，或兼容性承诺有边界。
+
+| 功能                        | 状态    | 功能                     | 状态    |
+| :-------------------------- | :------ | :----------------------- | :------ |
+| **S3 核心功能**             | ✅ 可用 | **分布式模式**           | ✅ 可用 |
+| **上传 / 下载**             | ✅ 可用 | **单机模式**             | ✅ 可用 |
+| **版本控制**                | ✅ 可用 | **Bitrot (防数据腐烂)**  | ✅ 可用 |
+| **对象锁定 (WORM)**         | ✅ 可用 | **修复与扫描器**         | ✅ 可用 |
+| **服务端加密 (SSE)**        | ✅ 可用 | **存储池扩容 / 下线**    | ✅ 可用 |
+| **RustFS KMS**              | ✅ 可用 | **存储桶复制**           | ✅ 可用 |
+| **生命周期管理 (ILM)**      | ✅ 可用 | **站点复制**             | ✅ 可用 |
+| **ILM 分层 (远端 S3)**      | ✅ 可用 | **存储桶配额**           | ✅ 可用 |
+| **S3 Select**               | ✅ 可用 | **事件通知**             | ✅ 可用 |
+| **S3 Tables (Iceberg REST)**| 🧪 预览 | **审计日志**             | ✅ 可用 |
+| **IAM / 策略**              | ✅ 可用 | **日志与可观测性**       | ✅ 可用 |
+| **OIDC / SSO**              | ✅ 可用 | **Web 控制台**           | ✅ 可用 |
+| **Keystone 认证**           | ✅ 可用 | **K8s Helm Chart**       | ✅ 可用 |
+| **Swift API**               | ✅ 可用 | **FTPS / WebDAV**        | ✅ 可用 |
+| **多租户**                  | ✅ 可用 | **SFTP**                 | ✅ 可用 |
+| **MinIO 磁盘格式兼容**      | 🧪 预览 |                          |         |
+
+说明：
+
+- **服务端加密**：支持 SSE-C、SSE-S3 与 SSE-KMS。SSE-KMS 必须先配置 KMS 服务；未配置 KMS 时请求 `aws:kms` 会被拒绝，不会降级到本地主密钥。
+- **RustFS KMS**：生产环境支持 Vault（KV2 / Transit）与 AWS KMS 后端；`Local` 与 `Static` 后端仅供开发与测试使用，详见 [KMS 后端安全属性](docs/operations/kms-backend-security.md)。

 ## RustFS vs MinIO 性能对比

@@ -5165,6 +5165,9 @@ pub async fn put_restore_opts(
        user_defined: meta,
        version_id: oi.version_id.map(|e| e.to_string()),
        mod_time: oi.mod_time,
+        // Restore writes stored (possibly encrypted) bytes, so the writer's
+        // computed MD5 is not the object's public plaintext ETag.
+        preserve_etag: oi.etag.clone(),
        //expires:           oi.expires,
        ..Default::default()
    })
@@ -9209,6 +9209,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
        let dest_obj = transaction.remote_object.clone();
        let mut transition_meta = (*oi.user_defined).clone();
        rustfs_utils::http::remove_str(&mut transition_meta, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
+        // The tier holds opaque stored bytes. Its metadata must not be treated
+        // as a second object header set: forwarding SSE intent or wrapped DEKs
+        // would request a second encryption pass and disclose local envelope
+        // material to the remote provider.
+        transition_meta.retain(|key, _| !rustfs_utils::http::is_replication_stripped_encryption_key(key));
        transition_meta.insert("name".to_string(), object.to_string());
        rustfs_utils::http::metadata_compat::insert_str(
            &mut transition_meta,
@@ -9765,13 +9770,16 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                    part_opts.part_number = Some(part_info.number);
                    #[cfg(test)]
                    fail_restore_multipart_at(RestoreMultipartFailurePoint::InvalidPartSize)?;
-                    if part_info.actual_size <= 0 {
-                        return Err(Error::other(format!("invalid multipart restore part size {}", part_info.actual_size)));
+                    if part_info.size == 0 {
+                        return Err(Error::other(format!("invalid multipart restore stored part size {}", part_info.size)));
                    }
+                    let stored_part_size = i64::try_from(part_info.size).map_err(|_| {
+                        Error::other(format!("multipart restore stored part size exceeds i64: {}", part_info.size))
+                    })?;
                    #[cfg(test)]
                    fail_restore_multipart_at(RestoreMultipartFailurePoint::RangeOverflow)?;
                    let part_end = part_offset
-                        .checked_add(part_info.actual_size - 1)
+                        .checked_add(stored_part_size - 1)
                        .ok_or_else(|| Error::other("multipart restore part range overflow".to_string()))?;
                    let rs = Some(HTTPRangeSpec {
                        is_suffix_length: false,
@@ -9799,13 +9807,19 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                    #[cfg(test)]
                    fail_restore_multipart_at(RestoreMultipartFailurePoint::HashReader)?;
                    let hash_reader =
-                        HashReader::from_stream(reader, part_info.actual_size, part_info.actual_size, None, None, false)?;
+                        HashReader::from_stream(reader, stored_part_size, part_info.actual_size, None, None, false)?;
                    let mut p_reader = PutObjReader::new(hash_reader);
                    #[cfg(test)]
                    fail_restore_multipart_at(RestoreMultipartFailurePoint::PutPart)?;
+                    // `ropts` carries the object's ETag so the single-part copy-back
+                    // keeps it (the writer only ever sees stored bytes). A part write
+                    // must not inherit that object-level value, or every restored part
+                    // would be recorded under the same ETag; each part keeps its own.
+                    let mut part_write_opts = ropts.clone();
+                    part_write_opts.preserve_etag = Some(part_info.etag.clone()).filter(|etag| !etag.is_empty());
                    let p_info = self_
                        .clone()
-                        .put_object_part(bucket, object, &res.upload_id, part_info.number, &mut p_reader, &ropts)
+                        .put_object_part(bucket, object, &res.upload_id, part_info.number, &mut p_reader, &part_write_opts)
                        .await?;
                    #[cfg(test)]
                    let p_info = if restore_multipart_failure_is(RestoreMultipartFailurePoint::SizeMismatch) {
@@ -9815,7 +9829,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                    } else {
                        p_info
                    };
-                    if p_info.size as i64 != part_info.actual_size {
+                    if p_info.size as i64 != stored_part_size {
                        return Err(Error::other(ObjectApiError::InvalidObjectState(GenericError {
                            bucket: bucket.to_string(),
                            object: object.to_string(),
@@ -9847,6 +9861,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                    user_defined: restore_commit_metadata,
                    no_lock: false,
                    decommission_capacity_admission: opts.decommission_capacity_admission.clone(),
+                    // The composite ETag would otherwise be recomputed from the
+                    // parts as they were written back, which for an encrypted or
+                    // compressed object digests stored bytes rather than the
+                    // object's public ETag.
+                    preserve_etag: oi.etag.clone(),
                    ..Default::default()
                };
                self_
@@ -13488,6 +13507,303 @@ mod transition_commit_failure_tests {
        );
    }

+    /// backlog#2368 B5: the tier stores opaque bytes, so the archive request
+    /// must not carry the object's encryption headers. Forwarding them made
+    /// every S3 target reject an SSE-C archive outright, asked the target to
+    /// encrypt an SSE-KMS object a second time under a key id it does not own,
+    /// and handed the wrapped DEK to a third-party provider.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn transition_does_not_forward_encryption_metadata_to_the_tier() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transition-encryption-metadata-bucket";
+        let object = "object.bin";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let encryption_metadata = [
+            ("x-amz-server-side-encryption", "aws:kms"),
+            ("x-amz-server-side-encryption-aws-kms-key-id", "arn:aws:kms:us-east-1:123:key/abc"),
+            ("x-amz-server-side-encryption-customer-algorithm", "AES256"),
+            (rustfs_utils::http::INTERNAL_ENCRYPTION_KEY_HEADER, "d3JhcHBlZC1kZWs="),
+            (rustfs_utils::http::INTERNAL_ENCRYPTION_IV_HEADER, "AAAAAAAAAAAAAAAA"),
+            (rustfs_utils::http::INTERNAL_ENCRYPTION_ALGORITHM_HEADER, "AES256"),
+        ];
+        let mut user_defined: HashMap<String, String> = encryption_metadata
+            .iter()
+            .map(|(key, value)| ((*key).to_string(), (*value).to_string()))
+            .collect();
+        user_defined.insert("x-amz-meta-owner".to_string(), "finance".to_string());
+
+        let mut reader = PutObjReader::from_vec(b"stored bytes the tier keeps opaque ".repeat(64));
+        set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    user_defined,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("the encrypted source object should be written");
+        let original = set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("the source object should be readable");
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        set_disks
+            .transition_object(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    transition: TransitionOptions {
+                        status: TRANSITION_PENDING.to_string(),
+                        tier: tier_name,
+                        etag: original.etag.clone().unwrap_or_default(),
+                        ..Default::default()
+                    },
+                    version_id: original.version_id.map(|version| version.to_string()),
+                    mod_time: original.mod_time,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("the encrypted object should transition");
+
+        let transitioned = set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("the transitioned object should be readable");
+        let remote_metadata = backend
+            .metadata(&transitioned.transitioned_object.name)
+            .await
+            .expect("the tier must have received the object");
+
+        for (key, _) in encryption_metadata {
+            assert!(
+                !remote_metadata.keys().any(|stored| stored.eq_ignore_ascii_case(key)),
+                "transition must not forward {key} to the tier: {remote_metadata:?}"
+            );
+        }
+        assert!(
+            remote_metadata
+                .iter()
+                .any(|(key, value)| key.eq_ignore_ascii_case("x-amz-meta-owner") && value == "finance"),
+            "ordinary user metadata must still travel to the tier: {remote_metadata:?}"
+        );
+
+        // Read-through and restore both resolve encryption locally, so the
+        // stripped keys must survive untouched in the local metadata.
+        for (key, value) in encryption_metadata {
+            assert_eq!(
+                transitioned.user_defined.get(key).map(String::as_str),
+                Some(value),
+                "the local copy must keep {key}"
+            );
+        }
+    }
+
+    /// Deterministic bytes that do not repeat with a short period, so a slice
+    /// taken at the wrong offset cannot coincidentally compare equal.
+    fn stored_representation_bytes(seed: u32, len: usize) -> Vec<u8> {
+        (0..len)
+            .map(|index| {
+                let mixed = (index as u32).wrapping_add(seed).wrapping_mul(2_654_435_761);
+                (mixed >> 13) as u8
+            })
+            .collect()
+    }
+
+    /// Compares two stored representations without dumping megabytes of bytes
+    /// into the failure output.
+    fn assert_stored_representation_eq(actual: &[u8], expected: &[u8], what: &str) {
+        assert_eq!(actual.len(), expected.len(), "{what}: stored length differs");
+        if let Some(offset) = actual.iter().zip(expected).position(|(left, right)| left != right) {
+            panic!(
+                "{what}: stored bytes differ at offset {offset} (found {:#04x}, expected {:#04x})",
+                actual[offset], expected[offset]
+            );
+        }
+    }
+
+    async fn read_stored_representation(set_disks: &Arc<SetDisks>, bucket: &str, object: &str) -> Vec<u8> {
+        let mut reader = set_disks
+            .get_object_reader(
+                bucket,
+                object,
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    no_lock: true,
+                    raw_data_movement_read: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("stored-representation reader should open");
+        let mut body = Vec::new();
+        reader.stream.read_to_end(&mut body).await.expect("stored body should drain");
+        body
+    }
+
+    /// backlog#2368 B3: the multipart restore loop addresses the tier in STORED
+    /// coordinates. Accumulating each part's PLAINTEXT length instead handed
+    /// every part a misaligned slice of the remote object whose length still
+    /// satisfied the range, the `HashReader` and the completion size check, so
+    /// the copy-back reported success while silently replacing the bytes.
+    ///
+    /// The fixture reproduces the encrypted geometry — a stored form LONGER
+    /// than the plaintext it encodes — because that is what keeps a
+    /// plaintext-coordinate range inside the tier object and makes the
+    /// corruption silent rather than a short read.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn multipart_restore_copies_the_stored_representation_back_verbatim() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "restore-multipart-stored-coordinates-bucket";
+        let object = "object.bin";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        // The minimum-part-size gate reads the PLAINTEXT length, so part one
+        // clears 5 MiB there while its stored form carries encoding overhead.
+        let part_shapes = [(6 * 1024 * 1024_usize, 9_216_usize), (256 * 1024_usize, 512_usize)];
+        let mut user_defined = HashMap::new();
+        user_defined.insert(rustfs_utils::http::INTERNAL_ENCRYPTION_ALGORITHM_HEADER.to_string(), "AES256".to_string());
+        user_defined.insert(
+            rustfs_utils::http::INTERNAL_ENCRYPTION_IV_HEADER.to_string(),
+            "AAAAAAAAAAAAAAAA".to_string(),
+        );
+
+        let upload = set_disks
+            .new_multipart_upload(
+                bucket,
+                object,
+                &ObjectOptions {
+                    user_defined: user_defined.clone(),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("multipart upload should be created");
+
+        let mut uploaded_parts = Vec::new();
+        let mut expected_stored = Vec::new();
+        for (index, (plaintext_len, overhead)) in part_shapes.iter().enumerate() {
+            let stored = stored_representation_bytes(index as u32 * 7 + 1, plaintext_len + overhead);
+            expected_stored.extend_from_slice(&stored);
+            let stored_len = stored.len() as i64;
+            let hash_reader =
+                HashReader::from_stream(std::io::Cursor::new(stored), stored_len, *plaintext_len as i64, None, None, false)
+                    .expect("hash reader over the stored representation");
+            let mut reader = PutObjReader::new(hash_reader);
+            let info = set_disks
+                .put_object_part(bucket, object, &upload.upload_id, index + 1, &mut reader, &ObjectOptions::default())
+                .await
+                .expect("stored part should be staged");
+            assert_eq!(info.size as i64, stored_len, "a part is stored in its encoded length");
+            uploaded_parts.push(CompletePart {
+                part_num: info.part_num,
+                etag: info.etag,
+                ..Default::default()
+            });
+        }
+
+        let original = set_disks
+            .clone()
+            .complete_multipart_upload(bucket, object, &upload.upload_id, uploaded_parts, &ObjectOptions::default())
+            .await
+            .expect("source multipart upload should complete");
+        let original_parts: Vec<(usize, usize, i64, String)> = original
+            .parts
+            .iter()
+            .map(|part| (part.number, part.size, part.actual_size, part.etag.clone()))
+            .collect();
+        for (_, size, actual_size, _) in &original_parts {
+            assert!(
+                *size as i64 > *actual_size,
+                "the fixture must keep the two coordinate systems apart: stored {size} vs plaintext {actual_size}"
+            );
+        }
+        let stored_before = read_stored_representation(&set_disks, bucket, object).await;
+        assert_stored_representation_eq(&stored_before, &expected_stored, "the fixture must store its encoded bytes verbatim");
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        set_disks
+            .transition_object(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    transition: TransitionOptions {
+                        status: TRANSITION_PENDING.to_string(),
+                        tier: tier_name,
+                        etag: original.etag.clone().unwrap_or_default(),
+                        ..Default::default()
+                    },
+                    version_id: original.version_id.map(|version| version.to_string()),
+                    mod_time: original.mod_time,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("multipart source should transition before restore");
+
+        let operation_id = Uuid::new_v4();
+        set_disks
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    eval_metadata: Some(restore_metadata(operation_id, true)),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("the restore generation should be installed");
+        let mut restore_opts = ObjectOptions::default();
+        restore_opts.transition.restore_request.days = Some(1);
+        restore_opts.user_defined = restore_operation_id_metadata(operation_id);
+        set_disks
+            .clone()
+            .restore_transitioned_object(bucket, object, &restore_opts)
+            .await
+            .expect("multipart restore should complete");
+
+        let stored_after = read_stored_representation(&set_disks, bucket, object).await;
+        assert_stored_representation_eq(
+            &stored_after,
+            &expected_stored,
+            "a multipart restore must copy the stored representation back verbatim",
+        );
+
+        let restored = set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("the restored object should be readable");
+        let restored_parts: Vec<(usize, usize, i64, String)> = restored
+            .parts
+            .iter()
+            .map(|part| (part.number, part.size, part.actual_size, part.etag.clone()))
+            .collect();
+        assert_eq!(
+            restored_parts, original_parts,
+            "restore must rebuild the same part layout, sizes and part ETags"
+        );
+        assert_eq!(restored.size, original.size, "restore must keep the stored object size");
+        // backlog#2369 P7.1: the copy-back digests stored bytes, so the object's
+        // public ETag has to be carried over rather than recomputed.
+        assert_eq!(restored.etag, original.etag, "restore must preserve the object ETag");
+    }
+
    #[tokio::test]
    #[serial_test::serial]
    async fn restore_failure_after_snapshot_cleans_exact_generation_and_returns_primary_error() {
@@ -819,7 +819,9 @@ impl KmsBackend for AwsKmsBackend {
            .with_rotate(true)
            .with_enable_disable(true)
            .with_schedule_deletion(true)
-            .with_versioning(true)
+            // AWS KMS exposes rotation state but does not enumerate key
+            // versions through this backend's API contract.
+            .with_versioning(false)
            .with_physical_delete(false)
            .with_production_supported(true)
    }
@@ -74,6 +74,39 @@ impl ScriptedResponse {
    }
 }

+/// The `auth/token/lookup-self` answer every scripted Vault serves for free.
+///
+/// A Vault client now probes its token's remaining lifetime at login
+/// (backlog#2369 P3), which is credential plumbing rather than the protocol any
+/// of these tests is scripting. Answering it out of band keeps every existing
+/// script meaningful: `ttl` 0 is Vault's "this token does not expire", so the
+/// probe changes nothing about how a scripted test behaves.
+pub(crate) fn token_lookup_self_response() -> String {
+    serde_json::json!({
+        "data": {
+            "accessor": "scripted-accessor",
+            "creation_time": 1_700_000_000u64,
+            "creation_ttl": 0,
+            "display_name": "token",
+            "entity_id": "",
+            "explicit_max_ttl": 0,
+            "id": "scripted-token",
+            "num_uses": 0,
+            "orphan": true,
+            "path": "auth/token/create",
+            "policies": ["default"],
+            "renewable": false,
+            "ttl": 0
+        }
+    })
+    .to_string()
+}
+
+/// Whether a recorded request line addresses the token self-lookup.
+pub(crate) fn is_token_lookup_self(request_line: &str) -> bool {
+    request_line.contains("/v1/auth/token/lookup-self")
+}
+
 /// A scripted stand-in Vault listening on a loopback port.
 pub(crate) struct ScriptedVault {
    /// Base address (`http://127.0.0.1:port`) to point a Vault client at.
@@ -102,6 +135,19 @@ impl ScriptedVault {
                let Some((request_line, body, mut stream)) = read_request(stream).await else {
                    continue;
                };
+                if is_token_lookup_self(&request_line) {
+                    // Served out of band so the credential probe does not
+                    // consume a scripted response meant for the protocol under
+                    // test, and is not recorded as one of its requests.
+                    let body = token_lookup_self_response();
+                    let payload = format!(
+                        "HTTP/1.1 200 Scripted\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
+                        body.len(),
+                    );
+                    let _ = stream.write_all(payload.as_bytes()).await;
+                    let _ = stream.shutdown().await;
+                    continue;
+                }
                recorded
                    .lock()
                    .expect("scripted vault request log poisoned")
@@ -155,6 +201,19 @@ impl ScriptedVault {
                    let Some((request_line, body, stream)) = read_request(stream).await else {
                        return;
                    };
+                    if is_token_lookup_self(&request_line) {
+                        // Credential plumbing, not part of the KV2 protocol
+                        // this responder models; see token_lookup_self_response.
+                        write_response(
+                            stream,
+                            ScriptedResponse::Http {
+                                status: 200,
+                                body: token_lookup_self_response(),
+                            },
+                        )
+                        .await;
+                        return;
+                    }
                    recorded
                        .lock()
                        .expect("scripted vault request log poisoned")
@@ -13,5 +13,5 @@ expression: capabilities_snapshot(backend.capabilities())
  "rotate": true,
  "schedule_deletion": true,
  "update_key_metadata": false,
-  "versioning": true
+  "versioning": false
 }
@@ -114,7 +114,7 @@ impl fmt::Debug for SecretString {
 }

 /// Expiry attributes of a lease-bound token.
-#[derive(Debug, Clone, Copy)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) struct LeaseInfo {
    /// Time-to-live granted at issue or renewal.
    pub(crate) ttl: Duration,
@@ -207,31 +207,102 @@ pub(crate) trait TokenSource: fmt::Debug + Send + Sync {
 }

 /// Token source for [`VaultAuthMethod::Token`]: always yields the token fixed
-/// at configuration time. The token carries no lease, so it is never renewed
-/// and never expires from the provider's point of view.
+/// at configuration time.
+///
+/// The token itself is never re-issued, but it usually still expires:
+/// `vault token create` defaults to a 768-hour TTL. Hard-coding "no lease"
+/// here left the renewal task unstarted and published no remaining-TTL gauge,
+/// so a healthy-looking cluster turned every KMS call into a 403 a month later
+/// and could only be recovered by a restart or a reconfigure (backlog#2369 P3).
+/// The source therefore asks Vault what it is holding, once per client
+/// generation, and lets the existing renewal loop take over whenever the answer
+/// carries a TTL.
+/// Map a `lookup-self` answer onto a lease.
+///
+/// A zero TTL is Vault's answer for a token that never expires (root and
+/// periodic-root tokens), which keeps the pre-probe behaviour exactly: no
+/// lease, no renewal task, no expiry gate. A response that omits `renewable`
+/// is treated as not renewable, so the renewal loop falls back to re-reading
+/// the remaining TTL instead of assuming it can extend it.
+fn static_token_lease(ttl_secs: u64, renewable: Option<bool>) -> Option<LeaseInfo> {
+    (ttl_secs > 0).then_some(LeaseInfo {
+        ttl: Duration::from_secs(ttl_secs),
+        renewable: renewable.unwrap_or(false),
+    })
+}
+
 pub(crate) struct StaticToken {
    token: TokenLease,
+    /// Client authenticated with the configured token, used only for
+    /// `lookup-self`. Per-generation renewals use the generation's own client.
+    lookup_client: VaultClient,
 }

 impl StaticToken {
-    pub(crate) fn new(token: String) -> Self {
-        Self {
+    pub(crate) fn new(settings: &VaultConnectionSettings, token: String) -> Result<Self> {
+        let lookup_client = settings.build_client(&token)?;
+        Ok(Self {
            token: TokenLease::new(token, None),
-        }
+            lookup_client,
+        })
    }
 }

 #[async_trait]
 impl TokenSource for StaticToken {
    async fn acquire(&self) -> AttemptResult<TokenLease> {
-        Ok(self.token.clone())
+        // A lookup failure must not fail the login. The token itself may well
+        // be valid: a policy can omit `lookup-self`, and Vault may simply be
+        // unreachable for the moment. Failing here would take down deployments
+        // that work today, so the probe degrades to the pre-probe behaviour —
+        // no lease, no renewal — and says so loudly instead.
+        let lease = match vaultrs::token::lookup_self(&self.lookup_client).await {
+            Ok(lookup) => static_token_lease(lookup.ttl, lookup.renewable),
+            Err(error) => {
+                warn!(
+                    event = "vault_static_token_lookup_failed",
+                    error = %error,
+                    "Could not read the configured Vault token's remaining lifetime, so it will not be \
+                     renewed and its expiry will not be tracked. Grant the token `lookup-self` (Vault's \
+                     default policy does) or switch to AppRole, Kubernetes or an agent-managed token file"
+                );
+                None
+            }
+        };
+
+        if let Some(lease) = lease
+            && !lease.renewable
+        {
+            warn!(
+                event = "vault_static_token_not_renewable",
+                ttl_secs = lease.ttl.as_secs(),
+                "The configured Vault token expires and cannot be renewed; RustFS will fail closed as it \
+                 approaches expiry. Switch to AppRole, Kubernetes or an agent-managed token file, or \
+                 reconfigure with a fresh token before it lapses"
+            );
+        }
+
+        Ok(TokenLease::new(self.token.expose().to_string(), lease))
+    }
+
+    async fn renew(&self, client: &VaultClient) -> AttemptResult<TokenLease> {
+        // Vault refuses renew-self on a non-renewable token; the renewal loop
+        // then falls back to `acquire`, which re-reads the remaining TTL and
+        // keeps the gauge honest until the fail-closed window is reached.
+        let auth = vaultrs::token::renew_self(client, None)
+            .await
+            .map_err(|error| attempt_error("token renewal", error))?;
+        Ok(TokenLease::from_auth(auth))
    }
 }

 impl fmt::Debug for StaticToken {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        // TokenLease::fmt already redacts the token value.
-        f.debug_struct("StaticToken").field("token", &self.token).finish()
+        // TokenLease::fmt already redacts the token value; VaultClient embeds
+        // its settings, including the token, so it must stay out of Debug.
+        f.debug_struct("StaticToken")
+            .field("token", &self.token)
+            .finish_non_exhaustive()
    }
 }

@@ -541,7 +612,7 @@ pub(crate) fn token_source_for(
    settings: &VaultConnectionSettings,
 ) -> Result<Box<dyn TokenSource>> {
    match auth_method {
-        VaultAuthMethod::Token { token } => Ok(Box::new(StaticToken::new(token.clone()))),
+        VaultAuthMethod::Token { token } => Ok(Box::new(StaticToken::new(settings, token.clone())?)),
        VaultAuthMethod::AppRole {
            role_id,
            secret_id,
@@ -1201,14 +1272,22 @@ mod tests {
        (Arc::new(provider), state)
    }

+    /// A provider whose token reports no expiry, which is what `lookup-self`
+    /// answers for a root or periodic-root token. Scripted rather than backed
+    /// by [`StaticToken`] because the real source now asks Vault what it holds.
    async fn static_provider() -> VaultCredentialProvider {
        VaultCredentialProvider::new(
            test_settings(),
-            Box::new(StaticToken::new(TEST_TOKEN.to_string())),
+            Box::new(ScriptedSource {
+                state: Arc::new(ScriptedState::default()),
+                ttl: Duration::ZERO,
+                renewable: false,
+                login_delay: Duration::ZERO,
+            }),
            test_policy(Duration::from_secs(10), Duration::from_secs(5)),
        )
        .await
-        .expect("static provider must build without a live Vault")
+        .expect("a token without an expiry must build without a live Vault")
    }

    #[tokio::test]
@@ -1231,20 +1310,57 @@ mod tests {
        assert!(provider.spawn_renewal_task().is_none(), "a token without a lease has nothing to renew");
    }

-    #[tokio::test]
-    async fn test_static_token_source_yields_configured_token() {
-        let settings = test_settings();
-        let source = token_source_for(
+    #[test]
+    fn test_static_token_source_builds_without_contacting_vault() {
+        token_source_for(
            &VaultAuthMethod::Token {
                token: TEST_TOKEN.to_string(),
            },
-            &settings,
+            &test_settings(),
        )
        .expect("token auth must map to a source");
+    }

-        let lease = source.acquire().await.expect("static acquire cannot fail");
-        assert_eq!(lease.expose(), TEST_TOKEN);
-        assert!(lease.lease_info().is_none(), "static tokens must not carry a lease");
+    /// backlog#2369 P3: `vault token create` defaults to a 768-hour TTL, so
+    /// hard-coding "no lease" for token auth left the renewal task unstarted
+    /// and turned a healthy cluster into one that answers 403 a month later.
+    /// The lease now comes from what Vault reports.
+    #[test]
+    fn static_token_lease_follows_what_vault_reports() {
+        assert_eq!(
+            static_token_lease(0, Some(true)),
+            None,
+            "a token Vault reports as non-expiring must keep behaving as one"
+        );
+        assert_eq!(
+            static_token_lease(0, None),
+            None,
+            "a non-expiring token stays non-expiring whatever renewable says"
+        );
+        assert_eq!(
+            static_token_lease(2_764_800, Some(true)),
+            Some(LeaseInfo {
+                ttl: Duration::from_secs(2_764_800),
+                renewable: true,
+            }),
+            "the default 768-hour token must be tracked and renewed"
+        );
+        assert_eq!(
+            static_token_lease(3_600, Some(false)),
+            Some(LeaseInfo {
+                ttl: Duration::from_secs(3_600),
+                renewable: false,
+            }),
+            "an expiring token that cannot be renewed still needs its expiry tracked"
+        );
+        assert_eq!(
+            static_token_lease(3_600, None),
+            Some(LeaseInfo {
+                ttl: Duration::from_secs(3_600),
+                renewable: false,
+            }),
+            "an omitted renewable flag must not be read as renewable"
+        );
    }

    #[tokio::test]
@@ -1759,7 +1875,7 @@ mod tests {
                renewable: true,
            }),
        );
-        let static_source = StaticToken::new(TEST_TOKEN.to_string());
+        let static_source = StaticToken::new(&test_settings(), TEST_TOKEN.to_string()).expect("static source");
        let approle_source = AppRoleLogin::new(
            &test_settings(),
            "approle".to_string(),
@@ -90,6 +90,7 @@ s3s = { workspace = true, features = ["minio"] }
 hex-simd.workspace = true

 [dev-dependencies]
+temp-env = { workspace = true, features = ["async_closure"] }
 tokio = { workspace = true, features = ["test-util"] }
 tokio-test = { workspace = true }
 criterion = { workspace = true, features = ["html_reports"] }
@@ -327,6 +327,55 @@ where
    }
 }

+/// Read-side switch for the pre-`1.0.0-alpha.91` nonce layout, in which a whole
+/// v1 segment reused the part nonce for every block.
+///
+/// On by default, because turning it off refuses to decrypt objects written
+/// before that release. Block zero's derived nonce equals that base nonce, so
+/// the layout also lets a frame encrypted at index zero authenticate anywhere
+/// in its segment; the in-segment layout lock catches that as soon as a later
+/// frame disagrees, but a stream that is nothing but repeats of frame zero has
+/// no such later frame. A deployment with no pre-alpha.91 objects should set
+/// this to `false` to remove that surface outright (backlog#2369 P2).
+///
+// RUSTFS_COMPAT_TODO(backlog-2369-legacy-nonce-fallback): Remove after the
+// minimum supported direct-upgrade release and after migration tooling has
+// rewritten every pre-alpha.91 encrypted object.
+pub const ENV_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK: &str = "RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK";
+const DEFAULT_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK: bool = true;
+
+fn legacy_nonce_fallback_enabled() -> bool {
+    #[cfg(test)]
+    {
+        rustfs_utils::get_env_bool(
+            ENV_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK,
+            DEFAULT_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK,
+        )
+    }
+    #[cfg(not(test))]
+    {
+        static CACHED: std::sync::OnceLock<bool> = std::sync::OnceLock::new();
+        *CACHED.get_or_init(|| {
+            rustfs_utils::get_env_bool(
+                ENV_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK,
+                DEFAULT_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK,
+            )
+        })
+    }
+}
+
+/// The nonce layout selected while decoding a legacy v1 segment.
+///
+/// A historical writer used one of these layouts consistently for every
+/// block in a segment. Once a non-zero block identifies that layout, accepting
+/// another layout would let an attacker replay a block encrypted at index zero.
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum V1NonceLayout {
+    Current,
+    LegacyBlock,
+    ReusedPart,
+}
+
 pin_project! {
    /// A reader wrapper that decrypts data on the fly using AES-256-GCM.
    /// This is a demonstration. For production, use a secure and audited crypto library.
@@ -358,6 +407,8 @@ pin_project! {
        segment_frames: usize,
        stream_saw_v2: bool,
        segments_completed: usize,
+        v1_nonce_layout: Option<V1NonceLayout>,
+        legacy_nonce_fallback: bool,
    }
 }

@@ -391,6 +442,8 @@ where
            segment_frames: 0,
            stream_saw_v2: false,
            segments_completed: 0,
+            v1_nonce_layout: None,
+            legacy_nonce_fallback: legacy_nonce_fallback_enabled(),
        }
    }

@@ -441,6 +494,8 @@ where
            segment_frames: 0,
            stream_saw_v2: false,
            segments_completed: 0,
+            v1_nonce_layout: None,
+            legacy_nonce_fallback: legacy_nonce_fallback_enabled(),
        }
    }
 }
@@ -547,6 +602,7 @@ where
                    *this.segment_frame_version = None;
                    *this.saw_final_frame = false;
                    *this.segment_frames = 0;
+                    *this.v1_nonce_layout = None;

                    if *this.multipart_mode {
                        let next_part = if *this.current_part_index + 1 < this.multipart_parts.len() {
@@ -696,26 +752,46 @@ where
                    *this.base_nonce
                };
                let legacy_block_nonce = derive_block_nonce(&legacy_part_nonce, *this.block_index);
-                match this.cipher.decrypt(&nonce, ciphertext) {
-                    Ok(plaintext) => plaintext,
-                    Err(primary_err) => {
-                        let legacy_nonce =
-                            Nonce::try_from(legacy_block_nonce.as_slice()).map_err(|_| Error::other("invalid nonce length"))?;
-
-                        match this.cipher.decrypt(&legacy_nonce, ciphertext) {
-                            Ok(plaintext) => plaintext,
-                            Err(_) => {
-                                // Accept previously written streams that reused the part nonce
-                                // for every block inside a segment.
-                                let legacy_part_nonce = Nonce::try_from(legacy_part_nonce.as_slice())
-                                    .map_err(|_| Error::other("invalid nonce length"))?;
-                                this.cipher
-                                    .decrypt(&legacy_part_nonce, ciphertext)
-                                    .map_err(|_| Error::other(format!("decrypt error: {primary_err}")))?
-                            }
+                let legacy_part_nonce =
+                    Nonce::try_from(legacy_part_nonce.as_slice()).map_err(|_| Error::other("invalid nonce length"))?;
+                let legacy_block_nonce =
+                    Nonce::try_from(legacy_block_nonce.as_slice()).map_err(|_| Error::other("invalid nonce length"))?;
+                let layouts = [
+                    (V1NonceLayout::Current, &nonce),
+                    (V1NonceLayout::LegacyBlock, &legacy_block_nonce),
+                    (V1NonceLayout::ReusedPart, &legacy_part_nonce),
+                ];
+                let selected = if *this.block_index == 0 { None } else { *this.v1_nonce_layout };
+                let mut plaintext = None;
+                let mut last_error = None;
+                for (layout, candidate_nonce) in layouts {
+                    if selected.is_some_and(|expected| expected != layout) {
+                        continue;
+                    }
+                    if layout == V1NonceLayout::ReusedPart && !*this.legacy_nonce_fallback {
+                        continue;
+                    }
+                    match this.cipher.decrypt(candidate_nonce, ciphertext) {
+                        Ok(value) => {
+                            plaintext = Some((value, layout));
+                            break;
                        }
+                        Err(error) => last_error = Some(error),
                    }
                }
+                let (plaintext, layout) = plaintext.ok_or_else(|| {
+                    Error::new(
+                        std::io::ErrorKind::InvalidData,
+                        format!(
+                            "decrypt error: {}",
+                            last_error.map_or_else(|| "nonce layout rejected".to_string(), |error| error.to_string())
+                        ),
+                    )
+                })?;
+                if *this.block_index > 0 && this.v1_nonce_layout.is_none() {
+                    *this.v1_nonce_layout = Some(layout);
+                }
+                plaintext
            };
            if *this.current_frame_type == FRAME_TYPE_V2_FINAL {
                *this.saw_final_frame = true;
@@ -1003,6 +1079,154 @@ mod tests {
        assert_eq!(&decrypted, data);
    }

+    /// Encrypts `block_count` full v1 blocks, then overwrites frame one with a
+    /// verbatim copy of frame zero. Every frame is the same length, so the
+    /// stream keeps its original size and the forgery is invisible to any
+    /// length check.
+    async fn v1_stream_with_frame_zero_replayed_at_index_one(key: [u8; 32], nonce: [u8; 12], block_count: usize) -> Vec<u8> {
+        assert!(block_count >= 2, "a replay needs at least two frames");
+        let mut data = Vec::with_capacity(ENCRYPTION_BLOCK_SIZE * block_count);
+        for index in 0..block_count {
+            data.extend(std::iter::repeat_n(0xA1u8.wrapping_add(index as u8 * 17), ENCRYPTION_BLOCK_SIZE));
+        }
+
+        let mut encrypt_reader = EncryptReader::new(Cursor::new(data), key, nonce);
+        let mut encrypted = Vec::new();
+        encrypt_reader.read_to_end(&mut encrypted).await.expect("encrypt v1 frames");
+
+        // Header layout: [type][len:24][crc:32]; `len` counts the payload plus
+        // its own 4-byte CRC field, so the frame occupies 8 + (len - 4) bytes.
+        let declared_len = (encrypted[1] as usize) | ((encrypted[2] as usize) << 8) | ((encrypted[3] as usize) << 16);
+        let frame_len = 8 + declared_len - 4;
+        let replayed_first = encrypted[..frame_len].to_vec();
+        encrypted[frame_len..frame_len * 2].copy_from_slice(&replayed_first);
+        encrypted
+    }
+
+    #[tokio::test]
+    async fn decrypt_reader_rejects_a_replayed_first_v1_frame() {
+        let key = [0x11; 32];
+        let nonce = [0x22; 12];
+        let encrypted = v1_stream_with_frame_zero_replayed_at_index_one(key, nonce, 3).await;
+
+        let mut decrypt_reader = DecryptReader::new(Cursor::new(encrypted), key, nonce);
+        let error = decrypt_reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("a repeated index-zero frame must not authenticate at index one");
+        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+    }
+
+    /// The in-segment layout lock must not cost compatibility: every legacy v1
+    /// shape the fallback chain exists for still decrypts under the default.
+    #[tokio::test]
+    async fn legacy_v1_streams_still_decrypt_under_the_default_fallback() {
+        temp_env::async_with_vars([(ENV_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK, None::<&str>)], async {
+            assert!(legacy_nonce_fallback_enabled(), "the legacy nonce fallback must stay on by default");
+
+            let mut key = [0u8; 32];
+            let mut nonce = [0u8; 12];
+            rand::rng().fill_bytes(&mut key);
+            rand::rng().fill_bytes(&mut nonce);
+            let mut data = vec![0u8; ENCRYPTION_BLOCK_SIZE * 3 + 17];
+            rand::rng().fill(&mut data[..]);
+
+            // Modern single-part stream.
+            let mut encrypted = Vec::new();
+            EncryptReader::new(Cursor::new(data.clone()), key, nonce)
+                .read_to_end(&mut encrypted)
+                .await
+                .expect("modern v1 stream should encrypt");
+            let mut decrypted = Vec::new();
+            DecryptReader::new(Cursor::new(encrypted), key, nonce)
+                .read_to_end(&mut decrypted)
+                .await
+                .expect("modern v1 stream should decrypt");
+            assert_eq!(decrypted, data);
+
+            // Pre-alpha.91 stream that reused the part nonce for every block.
+            let legacy = encrypt_with_legacy_nonce_reuse(&data, key, nonce);
+            let mut decrypted = Vec::new();
+            DecryptReader::new(Cursor::new(legacy), key, nonce)
+                .read_to_end(&mut decrypted)
+                .await
+                .expect("a reused-nonce legacy stream should still decrypt");
+            assert_eq!(decrypted, data);
+        })
+        .await;
+    }
+
+    /// The residual after the layout lock: a stream that is nothing but repeats
+    /// of frame zero has no later frame to disagree with the reused-part
+    /// layout, so only turning the fallback off rejects it.
+    #[tokio::test]
+    async fn a_two_frame_replay_is_closed_only_by_disabling_the_legacy_fallback() {
+        let key = [0x33; 32];
+        let nonce = [0x44; 12];
+        let encrypted = v1_stream_with_frame_zero_replayed_at_index_one(key, nonce, 2).await;
+
+        temp_env::async_with_vars([(ENV_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK, None::<&str>)], async {
+            let mut forged = Vec::new();
+            DecryptReader::new(Cursor::new(encrypted.clone()), key, nonce)
+                .read_to_end(&mut forged)
+                .await
+                .expect("with the fallback on this forgery is still accepted");
+            assert_eq!(forged.len(), ENCRYPTION_BLOCK_SIZE * 2);
+            assert_eq!(
+                &forged[..ENCRYPTION_BLOCK_SIZE],
+                &forged[ENCRYPTION_BLOCK_SIZE..],
+                "the accepted forgery is frame zero's plaintext twice over"
+            );
+        })
+        .await;
+
+        temp_env::async_with_vars([(ENV_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK, Some("false"))], async {
+            let error = DecryptReader::new(Cursor::new(encrypted.clone()), key, nonce)
+                .read_to_end(&mut Vec::new())
+                .await
+                .expect_err("with the fallback off the replayed frame must not authenticate");
+            assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+        })
+        .await;
+    }
+
+    /// Turning the fallback off removes exactly the third layout: modern v1
+    /// streams keep decrypting, pre-alpha.91 reused-nonce streams stop.
+    #[tokio::test]
+    async fn disabling_the_legacy_nonce_fallback_refuses_only_reused_part_nonces() {
+        let mut key = [0u8; 32];
+        let mut nonce = [0u8; 12];
+        rand::rng().fill_bytes(&mut key);
+        rand::rng().fill_bytes(&mut nonce);
+        let mut data = vec![0u8; ENCRYPTION_BLOCK_SIZE * 3 + 17];
+        rand::rng().fill(&mut data[..]);
+
+        let mut modern = Vec::new();
+        EncryptReader::new(Cursor::new(data.clone()), key, nonce)
+            .read_to_end(&mut modern)
+            .await
+            .expect("modern v1 stream should encrypt");
+        let legacy = encrypt_with_legacy_nonce_reuse(&data, key, nonce);
+
+        temp_env::async_with_vars([(ENV_RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK, Some("false"))], async {
+            assert!(!legacy_nonce_fallback_enabled(), "the switch must be observed");
+
+            let mut decrypted = Vec::new();
+            DecryptReader::new(Cursor::new(modern), key, nonce)
+                .read_to_end(&mut decrypted)
+                .await
+                .expect("modern v1 streams must keep decrypting with the fallback off");
+            assert_eq!(decrypted, data);
+
+            let error = DecryptReader::new(Cursor::new(legacy), key, nonce)
+                .read_to_end(&mut Vec::new())
+                .await
+                .expect_err("the third layout must be gone when the fallback is off");
+            assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
+        })
+        .await;
+    }
+
    #[tokio::test]
    async fn test_decrypt_reader_only() {
        // Encrypt some data first
@@ -38,9 +38,10 @@
 - `put-file-auth-epoch-strict` internode put_file epoch compatibility: rc.2 peers can cache a remote put_file capability before that remote node restarts, then continue sending v1 authenticated uploads with the old server epoch; those peers cannot recover from the 409 conflict used by newer clients to trigger a re-probe. Servers temporarily accept signed, non-nil stale put_file epochs while legacy put_file auth remains non-strict so mixed-version rolling upgrades can finish multipart/object writes. Remove the stale-epoch fallback after the minimum supported RustFS peer version re-probes put_file capability after server-epoch conflicts and legacy put_file auth is no longer accepted.
 - `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.
 - `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus.
- `backlog-1316` legacy encrypted multipart range seek: the feature remains opt-in until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently.
+- `backlog-1316` legacy encrypted multipart range seek: the feature is on by default (RUSTFS_ENCRYPTED_RANGE_SEEK, default true) and the switch remains only as a kill switch until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently.
 - `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
 - `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.
+- `backlog-2369-legacy-nonce-fallback` pre-alpha.91 v1 segment nonce layout: releases before `1.0.0-alpha.91` reused a segment's part nonce for every block inside it, so the decrypt reader keeps that layout as its third and last v1 fallback. Because block zero's derived nonce equals that base nonce, the layout also lets a frame encrypted at index zero authenticate anywhere in its segment. Reads now lock a segment to whichever layout decoded its first non-zero block, which rejects a replay as soon as a later frame disagrees; a stream that is nothing but repeats of frame zero has no such later frame, so a deployment holding no pre-alpha.91 objects should set RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK=false to drop the layout outright. Remove the fallback (and the switch) after the minimum supported direct-upgrade release, and after migration tooling has rewritten every pre-alpha.91 encrypted object.
 - `sse-local-dek-json-v1` legacy local SSE DEK decoding: releases before the JSON envelope wrote wrapped DEKs as `base64(nonce):base64(ciphertext)`, so readers retain that decoder while all new writes use the versioned JSON envelope. Remove the colon decoder after the minimum supported direct-upgrade release writes JSON envelopes and migration tooling has rewritten every retained legacy object.
 - `not-initialized-error-code-v1` typed control-plane not-initialized wire code: control-plane RPC responses historically signaled an uninitialized peer only through the literal error_info string "errServerNotInitialized" (one drift site says "storage layer not initialized"). Responses now dual-carry a typed ControlPlaneErrorCode beside the legacy string, and clients prefer the code; the string stays populated and the client substring fallback (is_err_not_initialized, control_plane_failure) stays in place so mixed-version clusters keep classifying older peers' responses. Remove the substring fallback (and stop populating error_info for this case) after the minimum supported RustFS peer version always sends error_code.
 - `backlog-2097-tier-mutation-v4-error-text` tier-mutation Prepare rejection classification: a v3 server rejects a v4 request before store/runtime dispatch with the FailedPrecondition status and an authenticated, byte-exact unsupported-version message. A v4 coordinator recognizes only that exact code/message/requested-version tuple as definitely not persisted and fails the mutation without sending that peer an incompatible Abort; Unimplemented, near-text, missing/unknown failure classes, timeouts, and every other transport outcome remain ambiguous and stay in identity-bound Abort fanout. There is no automatic v3 retry. New servers retain v3 request/proof decoding for older coordinators, while operators must pause tier edit/remove/clear during a mixed v3/v4 rollout. Remove the text classifier after the minimum supported RustFS peer version returns the signed v4 PreDispatchRejected failure class.
@@ -15,6 +15,16 @@ RustFS ships several KMS backends. They differ not only in deployment effort but
 | Vault Transit | `VaultTransit` | Key-encryption keys never leave Vault; only Transit ciphertext is visible outside | Vault Transit engine (cryptographic isolation) | Delegated to Vault storage | Via Vault Transit key versioning | Deployments that need key material to be unreadable through storage APIs |
 | AWS KMS | `AWS` (alias `AwsKms`) | Key material never leaves AWS KMS; RustFS mirrors no key state | AWS KMS (cryptographic isolation) + IAM | Delegated to AWS | On-demand `RotateKeyOnDemand`; prior backing keys stay usable for decryption | Deployments rooted in AWS IAM — read [AWS KMS: deviations from the shared backend contract](#aws-kms-deviations-from-the-shared-backend-contract) first |

+## No KMS configured: the SSE-S3 local master key
+
+A deployment that never configures a KMS can still serve **SSE-S3** by setting `RUSTFS_SSE_S3_MASTER_KEY` to a base64-encoded 32-byte key. Data keys are then wrapped with that key using AES-256-GCM, on the node that serves the write. Understand three consequences before relying on it:
+
+- **The key is the whole confidentiality boundary.** It lives in the process environment of every node, with no ACL, no audit trail and no policy engine in front of it.
+- **Objects written this way can never be rotated.** There is no key record to rotate and no rewrap path; changing the value makes every object written under the old one unreadable. Migrating to a KMS later means rewriting those objects (for example with CopyObject), not reconfiguring.
+- **It does not serve SSE-KMS.** A request for `x-amz-server-side-encryption: aws:kms` on a node with no running KMS is refused — `400 InvalidRequest` when KMS was never configured, `503` when a configured service is not running. Earlier releases silently wrapped the data key with the local master key while still stamping `aws:kms` and the requested key id into the object metadata; that metadata claimed a KMS protection the object never had. If a deployment depended on that, either configure a KMS or ask for `AES256`.
+
+The value is unset by default, and a deployment that neither configures a KMS nor sets it simply cannot serve SSE-S3 (the write is refused, never silently downgraded to plaintext).
+
 ## Migrating from MinIO: encrypted objects do not carry over

 > **Warning: default RustFS builds fail closed on objects that MinIO encrypted.** This applies to SSE-S3, SSE-KMS, and SSE-C, whichever KMS backend you configure; configuring `Static` with MinIO's key material does not make them readable. Such objects list and HEAD normally (their `xl.meta` parses), and only the payload read fails — with S3 `InvalidObjectState`, never plaintext. Read a sample of encrypted objects, not just their listings, before decommissioning the MinIO deployment.
@@ -117,6 +127,37 @@ Decryption loads exactly the version recorded in the envelope and fails closed w

 Do not rotate any key until **every** RustFS node runs a build that understands the `master_key_version` envelope field. Older binaries ignore the field and always decrypt with the current material: harmless while nothing has been rotated, but after a rotation they fail to decrypt every object wrapped by an earlier key version. Complete the rolling upgrade of the entire cluster first, then rotate. The rest of this constraint class is collected in [Mixed-version clusters during a rolling upgrade](#mixed-version-clusters-during-a-rolling-upgrade).

+## SSE-C requires a secure transport
+
+An SSE-C request carries the customer's AES key in a request header, so AWS S3 and MinIO both refuse one that did not arrive over TLS. A plaintext hop hands that key to anyone on the path, and because the object cannot be read without the same key, the exposure lasts as long as the object does.
+
+This release reports rather than refuses, because flipping straight to a rejection would break every plaintext staging and test deployment inside a release window:
+
+- Every SSE-C request on a plaintext transport increments `rustfs_ssec_plaintext_requests_total` and logs one `ssec_request_without_tls` warning per process.
+- `RUSTFS_SSE_C_REQUIRE_TLS=true` (default `false`) refuses those requests now, with the same `400 InvalidRequest` wording AWS uses. Confirm the counter reads zero before enabling it.
+- The default is expected to flip in a later release.
+
+The verdict is per connection: a listener that terminates TLS satisfies it, and so does an `https` protocol forwarded by a proxy the trusted-proxy configuration accepts. A direct plaintext client asserts nothing, and a forwarded protocol from an untrusted peer is not consulted.
+
+## Object ciphertext format: what the v1 frame layout does and does not authenticate
+
+Every object RustFS writes today uses the **v1** frame layout (the v2 layout exists and is read automatically, but its write switch `RUSTFS_ENCRYPTION_FRAME_V2` is off by default). Each frame is authenticated with AES-256-GCM under a nonce derived from the object's base nonce and the frame's index. Three properties do **not** follow from that, and an operator's threat model has to account for them:
+
+- **No frame-index binding.** A frame's index is not part of its associated data. Authentication proves a frame was produced under this object's key; it does not by itself prove the frame belongs at the position it occupies.
+- **No final-frame authentication.** Nothing in a v1 stream marks the last frame, so a stream that has been cut short is not distinguishable from a shorter object by cryptographic means.
+- **Truncation is not detected server-side.** A full GET is cut off by the length gate mid-stream and surfaces as `IncompleteBody` — after the response headers have already gone out. A ranged GET that ends early looks like an ordinary EOF and is not reported at all. A client that needs a truncation signal must compare the delivered length against `Content-Length` itself.
+
+These matter only to an attacker who can already rewrite the underlying shards. Shard integrity uses a keyed-hash-free checksum (HighwayHash), which such an attacker can recompute, so it is not a barrier.
+
+Two historical shapes additionally reuse a GCM nonce and cannot be repaired by any read-side change:
+
+| Shape | Written by | Consequence | Migration |
+| --- | --- | --- | --- |
+| Multipart objects written before `1.0.0-alpha.91` | The pre-alpha.91 writer reused a segment's part nonce for every block in it | The whole segment shares one nonce; a frame from index zero authenticates anywhere in that segment | Rewrite in place with CopyObject; then set `RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK=false` |
+| SSE-C objects written before `1.0.0-beta.9` that carry no stored IV | The nonce was derived deterministically from bucket and key | Historical versions of the same key share a nonce, which affects confidentiality as well as forgeability | Rewrite in place with CopyObject |
+
+The decrypt reader locks each segment to whichever nonce layout decoded its first non-zero-index frame, so a replayed frame is rejected as soon as any later frame disagrees. A stream that is nothing but repeats of frame zero has no later frame to disagree, so a deployment that holds no pre-alpha.91 objects should set `RUSTFS_ENCRYPTION_LEGACY_NONCE_FALLBACK=false` (default `true`) to drop that layout entirely. Turning it off refuses to decrypt pre-alpha.91 objects, so migrate first.
+
 ## Mixed-version clusters during a rolling upgrade

 During a rolling upgrade KMS state is shared three ways: **Vault** holds key records and Transit metadata, **cluster storage** holds the persisted KMS configuration, and **each node's process memory** holds caches and the live backend instance. Nodes on different builds agree on the first, may disagree on the third, and can disagree on configuration for as long as the operator leaves them running, because the reload broadcast that converges configuration is one of the things an older build rejects. This section is written for the KV2 and Transit backends; the Local backend is unsupported for multi-node deployments regardless of version (see the [deployment support matrix](#deployment-support-matrix)).
@@ -237,6 +278,7 @@ The Local backend stores one JSON record per key (`<key_id>.key`) plus an Argon2
 - `Local` is the default backend (`kms_backend` defaults to `local`) and is a development, testing and demo backend; it is not supported for production. Activating a backend whose capabilities report `production_supported: false` logs a `kms_backend_positioning` warning on every start, restart and reconfigure, and the `kms/status` capability matrix carries the same flag. The positioning is a warning, not a gate.
 - Configuration validation enforces stricter rules outside explicit development mode: a master key is required and `key_dir` must not live under the process temp directory.
 - The RustFS Kubernetes operator places the key directory on a PersistentVolumeClaim, so keys survive pod rescheduling.
+- **A multi-node deployment cannot share it.** Key material lives on each node's own disk and the Argon2id salt is generated per node, so two nodes derive different keys from the same `master_key`. An object encrypted on node A cannot be decrypted on node B; behind a load balancer that appears as intermittent 500s on reads that succeeded a moment earlier. Configuring `Local` while the deployment is distributed logs `kms_node_local_backend_in_distributed_deployment` and appends the same warning to the `kms/configure` response. This stays a warning, not a gate.
 - Production multi-node deployments should use the Vault Transit backend.

 ### Deployment support matrix
@@ -9,12 +9,14 @@ The drill rehearses the complete loop — back up, lose the persistence layer, p

 The drill covers the **Local** backend, the only backend RustFS produces a full-material bundle for. The responsibility split is described in `crates/kms/src/backup/capability.rs`:

-| Backend | What a RustFS bundle carries | What restores it |
+| Backend | RustFS bundle export | What restores it |
 | --- | --- | --- |
 | Local | Key records, all stored versions, the KDF salt, sanitized configuration | The RustFS restore in this runbook |
-| Static | Non-sensitive references only | The operator re-supplies the secret out of band |
-| Vault KV2 + Transit | KV metadata and Transit ciphertext references | Vault's native snapshot restore, then the RustFS orchestration |
-| Vault Transit | Metadata, configuration references, verification data | Vault's native snapshot restore, then the RustFS orchestration |
+| Static | Refused with `501`; RustFS holds no material to export | The operator re-supplies the secret out of band |
+| Vault KV2 | Refused with `501` | Vault's native snapshot restore, then the RustFS orchestration |
+| Vault Transit | Refused with `501` | Vault's native snapshot restore, then the RustFS orchestration |
+
+The `501` is not a gap in this runbook: `POST /rustfs/admin/v3/kms/backup` refuses any backend other than `Local` (`rustfs/src/admin/handlers/kms_backup.rs`, `execute_backup`), so no RustFS bundle exists to plan around for the other three. Note that `capability.rs` still *declares* `FullMaterial` responsibility for Vault KV2 in storage-only mode; no export path implements it, so treat the declaration as a reservation, not a capability.

 For the Vault backends there is no RustFS-side export: the cryptographic root is non-exportable and comes back through Vault's own disaster-recovery flow. RustFS owns the refusal to proceed before that has happened and the ordering of everything after it — see the Vault section below.

@@ -225,6 +225,10 @@ KMS configured through the admin API is persisted to cluster storage and restore

 To recover from `load_failed` — or from any state where the server runs but its in-memory KMS lags the persisted configuration — call `POST /rustfs/admin/v3/kms/reload` (`kms:ServiceControl`). It re-reads the persisted configuration from cluster storage and reconfigures the service without resubmitting secrets, then broadcasts the reload to peer nodes. If reload keeps failing, check cluster storage health first (the read needs quorum), then `RUSTFS_KMS_CONFIG_SECRET`: an unseal error means the secret is missing or differs from the one that sealed the persisted copy — it must be identical on every node.

+Reload short-circuits only when this node is **already running** the persisted configuration. A node whose KMS failed to start keeps that configuration and sits in `Error`, so reload reconfigures it — which starts the service — rather than reporting success while the node stays down. The same holds on every peer, which reaches the same path through the reload broadcast.
+
+**Cluster-wide versus node-local routes.** `configure`, `reconfigure` and `reload` are cluster operations: the node that serves the request broadcasts to its peers. `start` and `stop` are node-local and are **not** broadcast. Calling `POST /rustfs/admin/v3/kms/stop` through a load balancer therefore stops whichever node answered and leaves the cluster in a mixed state; address a specific node directly when you mean node-local semantics, and expect a later cluster-wide `reload` to start a stopped node again.
+
 A separate event, `kms_config_load_skipped` with `reason="storage_uninitialized"`, comes from the ambient loader used by the peer-reload RPC path; seeing it outside a peer reload indicates a request arrived before storage initialization finished.

 ## Threshold calibration
@@ -8,11 +8,21 @@

 | Method | Config tag | Credential lifetime | Background renewal | Recommended for |
 | --- | --- | --- | --- | --- |
-| Static token | `Token` | Whatever the operator provisioned; RustFS never renews it | None | Development; short-lived experiments |
+| Static token | `Token` | Whatever the operator provisioned; read from Vault at login | Renewed at half TTL when Vault reports the token as renewable | Development; short-lived experiments |
 | AppRole | `AppRole` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production without a Vault Agent sidecar |
 | Kubernetes | `Kubernetes` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production on Kubernetes, with no credential to distribute |
 | Agent token file | `TokenFile` | Owned by Vault Agent; RustFS only re-reads the sink file | File re-read once per poll interval | Production with a Vault Agent (or equivalent) managing auth |

+### Static token: what RustFS now knows about it
+
+`vault token create` grants a 768-hour TTL by default, so a static token normally *does* expire. At login RustFS calls `auth/token/lookup-self` and adopts whatever Vault reports:
+
+- **No expiry** (a root or periodic-root token, `ttl` 0): unchanged — no lease is tracked, no renewal task runs, and the token is never refused locally.
+- **Expiring and renewable:** the ordinary renewal loop takes over, renewing at half the remaining TTL and publishing the remaining-TTL gauge.
+- **Expiring but not renewable:** a `vault_static_token_not_renewable` warning is logged with the remaining TTL, the gauge is published, and requests fail closed inside the safety window rather than lapsing mid-flight against Vault. Rotate to a fresh token, or move to AppRole, Kubernetes, or an agent-managed token file.
+
+The probe never fails the login. A token whose policy omits `lookup-self` (Vault's `default` policy grants it), or a Vault that is unreachable at that moment, logs `vault_static_token_lookup_failed` and falls back to the previous behaviour — no lease tracked, no renewal — so a deployment that works today keeps working. That fallback lasts for the life of the client generation, so treat the warning as something to fix rather than tolerate.
+
 Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with any other method, or `RUSTFS_KMS_VAULT_KUBERNETES_ROLE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID`, is rejected at startup with a configuration error because the effective identity would be ambiguous. A leftover `RUSTFS_KMS_VAULT_TOKEN` alongside a configured login method is tolerated and ignored, so a stale variable cannot silently downgrade the identity.

 All of these are read the same way whether the service is started with `RUSTFS_KMS_ENABLE=true` or configured later through `POST /rustfs/admin/v3/kms/configure`.
@@ -160,7 +160,7 @@ behavior.
 | config.rustfs.kms.vault.vault_backend | string | `""`| The vault backend, `vault-kv2` or `vault-transit`. |
 | config.rustfs.kms.vault.vault_address | string | `""`| The vault address. |
 | config.rustfs.kms.vault.vault_token | string | `""`| The vault token. Rendered into a dedicated Secret (`<fullname>-kms-secret`), never into the ConfigMap. |
-| config.rustfs.kms.vault.vault_mount_path | string | `"transit"`| The vault mount path, only works if `vault_backend` equals `vault-transit` . |
+| config.rustfs.kms.vault.vault_mount_path | string | `"transit"`| The vault mount path. Rendered as `RUSTFS_KMS_VAULT_MOUNT_PATH` for `vault-transit`, and as `RUSTFS_KMS_VAULT_KV_MOUNT` for `vault-kv2` (only when set; unset keeps the `secret` default). |
 | config.rustfs.kms.vault.default_key | string | `"transit"`| The master key id for RustFS. |
 | extraEnv | list | `[]` | Extra environment variables for the RustFS container. An explicit `RUSTFS_LOCAL_ENDPOINT_HOST` or `RUSTFS_VOLUMES`, or a bounded, dynamic, or unrecognized startup mode, disables generated anchor injection. `POD_NAME` and `RUSTFS_ADDRESS` remain independent overrides. |
 | extraVolumes | list | `[]` | Extra volumes to add to the pod spec. Supported in both standalone (Deployment) and distributed (StatefulSet) modes. |
@@ -23,6 +23,7 @@ use crate::admin::runtime_sources::{
    current_or_init_kms_runtime_service_manager,
 };
 use crate::admin::storage_api::config::{read_admin_config, save_admin_config};
+use crate::admin::storage_api::ecstore_topology::is_dist_erasure;
 use crate::admin::storage_api::error::StorageError;
 use crate::admin::storage_api::runtime::ECStore;
 use crate::admin::storage_api::s3::{S3ErrorCode, error as admin_s3_error};
@@ -507,6 +508,21 @@ pub async fn reload_persisted_kms_config() -> Result<(), String> {
    reload_persisted_kms_config_from_store(store, kms_service_manager_from_context(), "peer_reload").await
 }

+/// Whether a reload may return early because this node is already serving
+/// exactly the persisted configuration.
+///
+/// Byte-identical configuration is not sufficient on its own. A node whose KMS
+/// failed to start keeps its configuration and sits in `Error`, so comparing
+/// only the bytes turned the documented recovery call
+/// (`POST /rustfs/admin/v3/kms/reload`) into a no-op that reported success and
+/// left the node down — including on every peer, which reaches this same
+/// function through the reload broadcast (backlog#2369 P1). Any state other
+/// than `Running` falls through to `reconfigure`, which starts the service when
+/// none is running.
+fn kms_reload_is_already_current(status: rustfs_kms::KmsServiceStatus, config_is_unchanged: bool) -> bool {
+    matches!(status, rustfs_kms::KmsServiceStatus::Running) && config_is_unchanged
+}
+
 async fn reload_persisted_kms_config_from_store(
    store: Arc<ECStore>,
    service_manager: Arc<rustfs_kms::KmsServiceManager>,
@@ -525,11 +541,11 @@ async fn reload_persisted_kms_config_from_store(
        return Err("no persisted KMS configuration is available".to_string());
    };

-    if service_manager
+    let config_is_unchanged = service_manager
        .get_config()
        .await
-        .is_some_and(|current| kms_config_is_unchanged(&current, &config))
-    {
+        .is_some_and(|current| kms_config_is_unchanged(&current, &config));
+    if kms_reload_is_already_current(service_manager.get_status().await, config_is_unchanged) {
        info!(
            event = "kms_service_state",
            component = LOG_COMPONENT_ADMIN,
@@ -624,6 +640,51 @@ fn local_success_with_peer_report(message: &str, unconverged: &[String]) -> (boo
    )
 }

+/// What a node-local KMS backend means for a multi-node deployment
+/// (backlog#2369 P7.4).
+///
+/// The Local backend keeps key material on each node's own disk and generates
+/// its KDF salt per node, so two nodes derive different keys from the same
+/// `master_key`. An object encrypted on node A cannot be decrypted on node B:
+/// behind a load balancer that shows up as intermittent 500s on reads that
+/// worked a moment earlier. The product decision to warn rather than refuse
+/// stands; the generic "development only" warning simply never said what
+/// actually goes wrong, so an operator had no way to connect the symptom to
+/// the cause.
+///
+/// Returns the sentence to append to the configure response, or `None` when the
+/// combination does not apply.
+async fn node_local_backend_warning(backend: &rustfs_kms::KmsBackend) -> Option<&'static str> {
+    if !matches!(backend, rustfs_kms::KmsBackend::Local) || !is_dist_erasure().await {
+        return None;
+    }
+
+    warn!(
+        component = LOG_COMPONENT_ADMIN,
+        subsystem = LOG_SUBSYSTEM_KMS,
+        event = "kms_node_local_backend_in_distributed_deployment",
+        backend = rustfs_kms::KmsBackend::Local.as_str(),
+        "The Local KMS backend stores key material on each node's own disk with a per-node salt, so objects \
+         encrypted on one node cannot be decrypted on another. In a distributed deployment this surfaces as \
+         intermittent 500s on reads behind a load balancer. Use Vault Transit, Vault KV2 or AWS KMS for a \
+         multi-node deployment"
+    );
+
+    Some(
+        "Warning: the Local KMS backend is node-local. Key material and its salt live on each node's own disk, so \
+         objects encrypted on one node cannot be decrypted on another and reads behind a load balancer will fail \
+         intermittently. Use Vault Transit, Vault KV2 or AWS KMS for a distributed deployment",
+    )
+}
+
+/// Append the node-local backend warning to a successful configure message.
+fn with_node_local_backend_warning(message: String, warning: Option<&'static str>) -> String {
+    match warning {
+        Some(warning) => format!("{message}. {warning}"),
+        None => message,
+    }
+}
+
 pub fn register_kms_dynamic_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
    r.insert(
        Method::POST,
@@ -746,6 +807,7 @@ impl Operation for ConfigureKmsHandler {
        let kms_config = configure_request.to_kms_config();

        let persisted_config = kms_config.clone();
+        let node_local_warning = node_local_backend_warning(&kms_config.backend).await;
        let (success, message, status) = match service_manager
            .configure_with_persistence(kms_config, || async move {
                save_kms_config(&persisted_config)
@@ -768,7 +830,7 @@ impl Operation for ConfigureKmsHandler {
                let unconverged = broadcast_kms_config_reload().await;
                let (success, message) = local_success_with_peer_report("KMS configured successfully", &unconverged);
                audit.finish(KmsAdminOperation::Configure, None, None);
-                (success, message, status)
+                (success, with_node_local_backend_warning(message, node_local_warning), status)
            }
            Err(e) => {
                let error_msg = format!("Failed to configure KMS: {e}");
@@ -1345,6 +1407,7 @@ impl Operation for ReconfigureKmsHandler {
        let kms_config = configure_request.to_kms_config();

        let persisted_config = kms_config.clone();
+        let node_local_warning = node_local_backend_warning(&kms_config.backend).await;
        let (success, message, status) = match service_manager
            .reconfigure_with_persistence(kms_config, || async move {
                save_kms_config(&persisted_config)
@@ -1368,7 +1431,7 @@ impl Operation for ReconfigureKmsHandler {
                let (success, message) =
                    local_success_with_peer_report("KMS reconfigured and restarted successfully", &unconverged);
                audit.finish(KmsAdminOperation::Reconfigure, None, None);
-                (success, message, status)
+                (success, with_node_local_backend_warning(message, node_local_warning), status)
            }
            Err(e) => {
                let error_msg = format!("Failed to reconfigure KMS: {e}");
@@ -1420,9 +1483,10 @@ impl Operation for ReconfigureKmsHandler {
 mod tests {
    use super::{
        KmsConfigLoadError, decode_persisted_kms_config, ensure_kms_config_persistable, ensure_kms_request_persistable,
-        kms_config_fingerprint, kms_config_is_unchanged, kms_configure_actions, kms_service_control_actions,
-        load_kms_config_with, local_success_with_peer_report, normalize_configure_request_secrets, open_persisted_kms_config,
-        redacted_canonical_config, register_kms_dynamic_route, seal_persisted_kms_config,
+        kms_config_fingerprint, kms_config_is_unchanged, kms_configure_actions, kms_reload_is_already_current,
+        kms_service_control_actions, load_kms_config_with, local_success_with_peer_report, normalize_configure_request_secrets,
+        open_persisted_kms_config, redacted_canonical_config, register_kms_dynamic_route, seal_persisted_kms_config,
+        with_node_local_backend_warning,
    };
    use crate::admin::router::{AdminOperation, S3Router};
    use crate::admin::storage_api::error::StorageError;
@@ -1432,6 +1496,49 @@ mod tests {
    use std::path::PathBuf;
    use tempfile::TempDir;

+    /// backlog#2369 P1: a node whose KMS failed to start keeps its persisted
+    /// configuration, so an unchanged-bytes comparison made the documented
+    /// recovery call a no-op that still reported success.
+    #[test]
+    fn kms_reload_only_short_circuits_for_a_running_service() {
+        use rustfs_kms::KmsServiceStatus;
+
+        assert!(
+            kms_reload_is_already_current(KmsServiceStatus::Running, true),
+            "a running service on identical configuration has nothing to apply"
+        );
+        assert!(
+            !kms_reload_is_already_current(KmsServiceStatus::Running, false),
+            "changed configuration must always be applied"
+        );
+
+        for status in [
+            KmsServiceStatus::NotConfigured,
+            KmsServiceStatus::Configured,
+            KmsServiceStatus::Error("vault unreachable at startup".to_string()),
+        ] {
+            assert!(
+                !kms_reload_is_already_current(status.clone(), true),
+                "reload must reconfigure instead of reporting success from {status:?}"
+            );
+        }
+    }
+
+    /// backlog#2369 P7.4: the operator has to learn the consequence from the
+    /// response, not just from a log line the configuring client never sees.
+    #[test]
+    fn a_node_local_backend_warning_reaches_the_configure_response() {
+        let plain = with_node_local_backend_warning("KMS configured successfully".to_string(), None);
+        assert_eq!(plain, "KMS configured successfully");
+
+        let warned = with_node_local_backend_warning(
+            "KMS configured successfully".to_string(),
+            Some("Warning: the Local KMS backend is node-local"),
+        );
+        assert!(warned.starts_with("KMS configured successfully."), "{warned}");
+        assert!(warned.contains("node-local"), "{warned}");
+    }
+
    fn assert_has_action(actions: &[Action], action: Action) {
        assert!(actions.contains(&action), "expected action list to contain {action:?}");
    }
@@ -41,6 +41,12 @@ pub(crate) mod ecstore_cluster {
    };
 }

+/// Deployment topology. The KMS configure path needs it to tell an operator
+/// that a node-local backend cannot serve a multi-node deployment.
+pub(crate) mod ecstore_topology {
+    pub(crate) use crate::storage::storage_api::is_dist_erasure;
+}
+
 mod ecstore_config {
    pub(crate) use crate::storage::storage_api::ecstore_config::{com, init, storageclass};
 }
@@ -2042,7 +2042,7 @@ impl DefaultObjectUsecase {
            bucket_sse_config.as_ref().map(|(config, _timestamp)| config),
            original_sse,
            ssekms_key_id,
-            false,
+            sse_customer_algorithm.is_some() || sse_customer_key.is_some() || sse_customer_key_md5.is_some(),
        );
        if effective_sse
            .as_ref()
@@ -1515,7 +1515,7 @@ impl DefaultObjectUsecase {
            bucket_sse_config.as_ref().map(|(config, _timestamp)| config),
            server_side_encryption,
            ssekms_key_id,
-            false,
+            sse_customer_algorithm.is_some() || sse_customer_key.is_some() || sse_customer_key_md5.is_some(),
        );
        debug!(
            target: "rustfs::app::object_usecase",
@@ -3524,6 +3524,104 @@ mod tests {
        );
    }

+    async fn install_bucket_default_sse_for_test(bucket: &str, algorithm: &'static str, kms_key_id: Option<&str>) {
+        use crate::app::storage_api::test::bucket::utils::serialize;
+        use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};
+
+        let sys = get_global_bucket_metadata_sys().expect("bucket metadata system");
+        let metadata = {
+            let sys = sys.read().await;
+            sys.get(bucket).await.expect("bucket metadata cached")
+        };
+        let mut metadata = (*metadata).clone();
+        let config = ServerSideEncryptionConfiguration {
+            rules: vec![ServerSideEncryptionRule {
+                apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
+                    sse_algorithm: ServerSideEncryption::from_static(algorithm),
+                    kms_master_key_id: kms_key_id.map(|id| id.to_string()),
+                }),
+                blocked_encryption_types: None,
+                bucket_key_enabled: None,
+            }],
+        };
+        metadata.encryption_config_xml = serialize(&config).expect("sse config serializes");
+        metadata.sse_config = Some(config);
+        set_bucket_metadata(bucket.to_string(), metadata)
+            .await
+            .expect("install bucket default SSE");
+    }
+
+    /// backlog#2368 B1: an SSE-C request suppresses the bucket default, the way
+    /// COPY already did. PUT passed a hard-coded `has_explicit_ssec = false`,
+    /// so the default filled in a managed algorithm and the request then failed
+    /// its own mutual-exclusion check — every bucket with default encryption
+    /// refused SSE-C outright.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn sse_c_put_is_accepted_on_a_bucket_with_default_encryption() {
+        use md5::{Digest as _, Md5};
+
+        for (algorithm, kms_key_id, prefix) in [
+            (ServerSideEncryption::AES256, None, "ssec-over-aes256-default"),
+            (ServerSideEncryption::AWS_KMS, Some("bucket-key"), "ssec-over-kms-default"),
+        ] {
+            let (store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket(prefix, 1 << 20).await;
+            install_bucket_default_sse_for_test(&bucket, algorithm, kms_key_id).await;
+
+            let customer_key = [0x2a_u8; 32];
+            let key_md5 = {
+                let mut hasher = Md5::new();
+                hasher.update(customer_key);
+                base64_simd::STANDARD.encode_to_string(hasher.finalize())
+            };
+            let payload = Bytes::from_static(b"customer-key protected payload");
+            let input = PutObjectInput::builder()
+                .bucket(bucket.clone())
+                .key("ledger.csv".to_string())
+                .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone()))))
+                .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64")))
+                .sse_customer_algorithm(Some("AES256".to_string()))
+                .sse_customer_key(Some(base64_simd::STANDARD.encode_to_string(customer_key)))
+                .sse_customer_key_md5(Some(key_md5))
+                .build()
+                .expect("SSE-C PUT input must build");
+
+            DefaultObjectUsecase::from_global()
+                .execute_put_object(&FS::new(), build_request(input, Method::PUT))
+                .await
+                .unwrap_or_else(|err| panic!("a {algorithm} default bucket must accept an SSE-C PUT: {err:?}"));
+
+            let stored = store
+                .get_object_info(&bucket, "ledger.csv", &ObjectOptions::default())
+                .await
+                .expect("the SSE-C object should be readable");
+            assert!(
+                stored
+                    .user_defined
+                    .keys()
+                    .any(|key| key.eq_ignore_ascii_case("x-amz-server-side-encryption-customer-algorithm")),
+                "the object must be stored as SSE-C: {:?}",
+                stored.user_defined
+            );
+            assert!(
+                !stored
+                    .user_defined
+                    .keys()
+                    .any(|key| key.eq_ignore_ascii_case("x-amz-server-side-encryption-aws-kms-key-id")),
+                "the bucket default must not attach a KMS key to an SSE-C object: {:?}",
+                stored.user_defined
+            );
+            assert!(
+                !stored
+                    .user_defined
+                    .values()
+                    .any(|value| value == ServerSideEncryption::AWS_KMS),
+                "the bucket default must not claim managed encryption on an SSE-C object: {:?}",
+                stored.user_defined
+            );
+        }
+    }
+
    #[tokio::test]
    #[serial_test::serial]
    async fn quota_rejects_ciphertext_replication_before_polling_the_body() {
@@ -240,10 +240,8 @@ pub(super) fn has_put_sse_request_headers(headers: &HeaderMap) -> bool {
 /// A request-level value always wins; the bucket default only fills a gap, and
 /// the unknown-algorithm fallback lives once in [`bucket_default_write_sse`].
 ///
-/// `has_explicit_ssec` suppresses the default entirely. Only COPY passes `true`
-/// today: its destination may carry SSE-C, which must not also be given managed
-/// encryption. PUT and extract pass `false`, matching their current behaviour —
-/// see backlog#1826 for the divergence that leaves.
+/// `has_explicit_ssec` suppresses the default entirely: an SSE-C destination
+/// must not also be given managed encryption.
 ///
 /// Callers layering further overrides (PUT's `ciphertext_passthrough`) apply
 /// them to the returned pair.
@@ -263,7 +261,14 @@ pub(super) fn resolve_bucket_default_sse(
    };

    let effective_sse = requested_sse.or_else(|| bucket_default().map(bucket_default_write_sse));
-    let effective_kms_key_id = requested_kms_key_id.or_else(|| bucket_default().and_then(|sse| sse.kms_master_key_id.clone()));
+    let effective_kms_key_id = if effective_sse
+        .as_ref()
+        .is_some_and(|sse| sse.as_str() == ServerSideEncryption::AWS_KMS)
+    {
+        requested_kms_key_id.or_else(|| bucket_default().and_then(|sse| sse.kms_master_key_id.clone()))
+    } else {
+        requested_kms_key_id
+    };
    (effective_sse, effective_kms_key_id)
 }

@@ -1187,6 +1192,21 @@ mod tests {
        assert_eq!(kms_key_id.as_deref(), Some("request-key"));
    }

+    #[test]
+    fn resolve_bucket_default_sse_does_not_inherit_a_kms_key_for_an_explicit_sse_s3_request() {
+        let config = bucket_sse_config_with(ServerSideEncryption::AWS_KMS, Some("bucket-key"));
+
+        let (sse, kms_key_id) = resolve_bucket_default_sse(
+            Some(&config),
+            Some(ServerSideEncryption::from_static(ServerSideEncryption::AES256)),
+            None,
+            false,
+        );
+
+        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AES256));
+        assert!(kms_key_id.is_none(), "an SSE-S3 request must not inherit the bucket KMS key");
+    }
+
    #[test]
    fn resolve_bucket_default_sse_fills_gaps_from_the_bucket_default() {
        let config = bucket_sse_config_with(ServerSideEncryption::AWS_KMS, Some("bucket-key"));
@@ -24,6 +24,77 @@ const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the nu
 /// S3 error code for a request that names a KMS key the KMS does not hold.
 pub const KMS_KEY_NOT_FOUND_ERROR_CODE: &str = "KMS.NotFoundException";

+/// Map a KMS failure that surfaced on the S3 data path to its S3 error code.
+///
+/// The contract deliberately differs from the admin lifecycle handlers
+/// (`kms_key_lifecycle::lifecycle_error_status`): there a key id is the
+/// resource being addressed, so a missing key is `404`. Here the key id
+/// arrives inside a request header or a bucket default, so a key that is
+/// missing, disabled, or otherwise unusable is configuration the caller has to
+/// correct — AWS answers `400`, and reporting `500` instead both misfiles the
+/// failure as a server fault and makes SDKs back off and retry a request that
+/// cannot succeed.
+///
+/// `None` keeps the caller's fallthrough, which is the `500` that integrity
+/// faults — damaged, unreadable, or unknown-format key material — must keep.
+///
+/// Messages either name what the caller asked for or stay generic; detail that
+/// belongs to the deployment rather than the request stays in `source`, which
+/// the caller attaches and the logs retain.
+fn data_plane_kms_error(error: &rustfs_kms::KmsError) -> Option<(S3ErrorCode, String)> {
+    use rustfs_kms::KmsError as Kms;
+
+    let generic = |code: S3ErrorCode| {
+        let message = ApiError::error_code_to_message(&code);
+        Some((code, message))
+    };
+
+    match error {
+        Kms::KeyNotFound { key_id } => Some((
+            S3ErrorCode::Custom(KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+            format!("KMS key not found: {key_id}"),
+        )),
+        Kms::KeyVersionNotFound { key_id, version } => Some((
+            S3ErrorCode::Custom(KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+            format!("KMS key version {version} not found for key {key_id}"),
+        )),
+        // The key exists but its state forbids the operation (disabled,
+        // pending deletion). AWS treats an invalid key state as a request
+        // error, not a server fault.
+        Kms::InvalidOperation { .. } => Some((S3ErrorCode::InvalidRequest, error.to_string())),
+        // A policy decision needs a human, so it must be distinguishable from
+        // the transient classes an SDK retries.
+        Kms::AccessDenied { .. } => generic(S3ErrorCode::AccessDenied),
+        // Request-side faults: what was asked for cannot be served as asked.
+        Kms::ContextMismatch { .. }
+        | Kms::InvalidKey { .. }
+        | Kms::ValidationError { .. }
+        | Kms::UnsupportedAlgorithm { .. }
+        | Kms::InvalidKeySize { .. } => Some((S3ErrorCode::InvalidRequest, error.to_string())),
+        // A deployment whose KMS configuration cannot serve the request (for
+        // example no key named and no default). Actionable, but the detail
+        // describes the deployment, so it stays out of the response body.
+        Kms::ConfigurationError { .. } => {
+            Some((S3ErrorCode::InvalidRequest, "The KMS configuration cannot serve this request".to_string()))
+        }
+        // Transient: worth retrying, and must be counted against availability
+        // rather than against the caller. `IoError` belongs here because it is
+        // how a backend reports that its key store itself was unreachable —
+        // rustfs/rustfs#7470 separated that from a missing key precisely so the
+        // two stop looking alike, and leaving it on the 500 fallthrough would
+        // erase that distinction again at the S3 boundary.
+        Kms::BackendError { .. }
+        | Kms::IoError { .. }
+        | Kms::OperationTimedOut { .. }
+        | Kms::OperationCancelled { .. }
+        | Kms::CredentialsUnavailable { .. }
+        | Kms::CacheError { .. } => generic(S3ErrorCode::ServiceUnavailable),
+        // A permanent gap in the configured backend, never a missing resource.
+        Kms::UnsupportedCapability { .. } => Some((S3ErrorCode::NotImplemented, error.to_string())),
+        _ => None,
+    }
+}
+
 /// HTTP status of the error codes s3s cannot derive on its own.
 ///
 /// s3s answers `None` for every `Custom` code, which the response layer turns
@@ -472,12 +543,7 @@ impl From<StorageError> for ApiError {
                };
            }

-            if inner.downcast_ref::<KmsUnavailableError>().is_some()
-                || matches!(
-                    inner.downcast_ref::<rustfs_kms::KmsError>(),
-                    Some(rustfs_kms::KmsError::BackendError { .. })
-                )
-            {
+            if inner.downcast_ref::<KmsUnavailableError>().is_some() {
                return ApiError {
                    code: S3ErrorCode::ServiceUnavailable,
                    message: ApiError::error_code_to_message(&S3ErrorCode::ServiceUnavailable),
@@ -485,14 +551,11 @@ impl From<StorageError> for ApiError {
                };
            }

-            // A request header or bucket default naming a key the KMS does not
-            // hold is the caller's mistake to correct, and S3 reports it as
-            // 400 `KMS.NotFoundException`. Left to the fallthrough it became a
-            // 500 whose generic message hid which key was missing.
-            if let Some(rustfs_kms::KmsError::KeyNotFound { key_id }) = inner.downcast_ref::<rustfs_kms::KmsError>() {
-                let message = format!("KMS key not found: {key_id}");
+            if let Some(kms_error) = inner.downcast_ref::<rustfs_kms::KmsError>()
+                && let Some((code, message)) = data_plane_kms_error(kms_error)
+            {
                return ApiError {
-                    code: S3ErrorCode::Custom(KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+                    code,
                    message,
                    source: Some(Box::new(err)),
                };
@@ -1060,6 +1123,91 @@ mod tests {
        assert_eq!(s3_error.status_code(), Some(StatusCode::BAD_REQUEST));
    }

+    /// backlog#2368 B6: every KMS failure class that is not an integrity fault
+    /// carries a status that says whether retrying, fixing the request, or
+    /// calling a human is the right response. Collapsing them onto 500 made
+    /// SDKs back off on unfixable configuration errors and filed every one of
+    /// them as a server fault.
+    #[test]
+    fn kms_data_plane_errors_are_classified_by_what_the_caller_should_do() {
+        let cases: Vec<(rustfs_kms::KmsError, S3ErrorCode)> = vec![
+            // Names something the KMS does not hold.
+            (
+                rustfs_kms::KmsError::key_version_not_found("finance-key", 7),
+                S3ErrorCode::Custom(KMS_KEY_NOT_FOUND_ERROR_CODE.into()),
+            ),
+            // The key exists but its state forbids the operation.
+            (rustfs_kms::KmsError::invalid_key_state("disabled"), S3ErrorCode::InvalidRequest),
+            // Request-side faults.
+            (rustfs_kms::KmsError::context_mismatch("bucket differs"), S3ErrorCode::InvalidRequest),
+            (rustfs_kms::KmsError::invalid_key("malformed key id"), S3ErrorCode::InvalidRequest),
+            (rustfs_kms::KmsError::validation_error("empty key id"), S3ErrorCode::InvalidRequest),
+            (rustfs_kms::KmsError::unsupported_algorithm("aes-999"), S3ErrorCode::InvalidRequest),
+            (rustfs_kms::KmsError::invalid_key_size(32, 16), S3ErrorCode::InvalidRequest),
+            (
+                rustfs_kms::KmsError::configuration_error("no default key configured"),
+                S3ErrorCode::InvalidRequest,
+            ),
+            // A policy decision that needs a human, never retried by an SDK.
+            (rustfs_kms::KmsError::access_denied("no kms:Decrypt grant"), S3ErrorCode::AccessDenied),
+            // Transient, worth retrying, counted against availability.
+            (rustfs_kms::KmsError::backend_error("vault refused"), S3ErrorCode::ServiceUnavailable),
+            (
+                rustfs_kms::KmsError::operation_timed_out("attempt deadline"),
+                S3ErrorCode::ServiceUnavailable,
+            ),
+            (
+                rustfs_kms::KmsError::operation_cancelled("shutting down"),
+                S3ErrorCode::ServiceUnavailable,
+            ),
+            (
+                rustfs_kms::KmsError::credentials_unavailable("approle login failed"),
+                S3ErrorCode::ServiceUnavailable,
+            ),
+            (rustfs_kms::KmsError::cache_error("poisoned"), S3ErrorCode::ServiceUnavailable),
+            // A key store that cannot be read is an outage, not a missing key:
+            // rustfs/rustfs#7470 made the backend say so, and the S3 boundary
+            // has to keep the two apart.
+            (
+                rustfs_kms::KmsError::io_error("No such file or directory (os error 2)"),
+                S3ErrorCode::ServiceUnavailable,
+            ),
+            // A permanent gap in the configured backend, never a missing resource.
+            (
+                rustfs_kms::KmsError::unsupported_capability("local", "rewrap"),
+                S3ErrorCode::NotImplemented,
+            ),
+        ];
+
+        for (error, expected) in cases {
+            let description = error.to_string();
+            let api_error = ApiError::from(StorageError::other(error));
+            assert_eq!(api_error.code, expected, "wrong code for: {description}");
+            assert_ne!(api_error.code, S3ErrorCode::InternalError, "must not be a server fault: {description}");
+        }
+    }
+
+    /// A deployment-side configuration message describes the server, not the
+    /// request, so it stays in `source` the way the storage-IO mapping does.
+    #[test]
+    fn kms_configuration_errors_do_not_echo_deployment_detail() {
+        let detail = "vault mount /secret/rustfs-prod has no default key";
+        let api_error = ApiError::from(StorageError::other(rustfs_kms::KmsError::configuration_error(detail)));
+
+        assert_eq!(api_error.code, S3ErrorCode::InvalidRequest);
+        assert!(
+            !api_error.message.contains(detail),
+            "message leaked deployment detail: {}",
+            api_error.message
+        );
+        let source = api_error
+            .source
+            .as_deref()
+            .and_then(|source| source.downcast_ref::<StorageError>())
+            .expect("API error should retain the storage error source");
+        assert!(source.to_string().contains(detail), "the detail must survive on the source");
+    }
+
    #[test]
    fn test_generated_error_codes_keep_their_own_status() {
        let s3_error = S3Error::from(ApiError::from(StorageError::other(rustfs_kms::KmsError::backend_error("down"))));
@@ -28,6 +28,7 @@ use crate::server::{
        StsQueryApiCompatLayer, VirtualHostStyleHintLayer, redact_sensitive_uri_query,
    },
    rate_limit::{RateLimitLayer, api_rate_limit_layer_from_env},
+    ssec_transport::SsecTransportLayer,
    strip_valid_port_suffix,
    tls_material::{
        TlsAcceptFailure, TlsAcceptorHolder, TlsHandshakeFailureKind, accept_tls_with_deadline, build_acceptor_from_loaded,
@@ -1846,6 +1847,11 @@ fn process_connection(
            request_body_idle_timeout,
        } = context;

+        // Whether this listener terminated TLS for this connection; the SSE-C
+        // transport policy needs the connection's own answer, not a
+        // deployment-wide setting.
+        let connection_is_tls = tls_acceptor.is_some();
+
        // Build the hybrid service per-connection.
        // Note: NodeService is not Clone (holds LocalPeerS3Client), and the SwiftService
        // type is feature-gated, so we cannot pre-build the full hybrid service.
@@ -1964,6 +1970,14 @@ fn process_connection(
                // a spoof-proof client IP. Absent (None) unless enabled via
                // RUSTFS_API_RATE_LIMIT_ENABLE with a non-zero RPM.
                .option_layer(rate_limit_layer.clone())
+                // backlog#2369 P7.2: an SSE-C request carries the customer key
+                // in a header, so a plaintext hop leaks it permanently. Sits
+                // beside the rate limiter: after the trusted-proxy layer, which
+                // is what makes a forwarded `https` protocol trustworthy, and
+                // after the request context so a rejection can echo the request
+                // id. Reports by default; refuses only under
+                // RUSTFS_SSE_C_REQUIRE_TLS.
+                .layer(SsecTransportLayer::new(connection_is_tls))
                // CRITICAL: Insert ReadinessGateLayer before business logic
                // This stops requests from hitting IAMAuth or Storage if they are not ready.
                .layer(ReadinessGateLayer::new(readiness.clone()))
@@ -27,6 +27,7 @@ mod readiness;
 mod runtime;
 pub(crate) mod runtime_sources;
 mod service_state;
+mod ssec_transport;
 pub mod tls_material;

 use tracing::warn;
@@ -0,0 +1,276 @@
+// 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-C transport policy (backlog#2369 P7.2).
+//!
+//! An SSE-C request carries the customer's AES key in a request header, so AWS
+//! S3 and MinIO both refuse one that did not arrive over TLS. RustFS accepted
+//! them on any transport, which means a plaintext hop hands the key to anyone
+//! on the path — and the object is then unreadable without that same key, so
+//! the exposure is permanent for as long as the object lives.
+//!
+//! Refusing outright is the correct end state but not a safe default to adopt
+//! inside a release window: the project's own s3-tests and e2e lanes, and most
+//! staging deployments, speak plain HTTP. This release therefore reports:
+//! every SSE-C request on a plaintext transport increments
+//! `rustfs_ssec_plaintext_requests_total` and logs one warning per process, so
+//! an operator can see whether anything would break before the default flips.
+//! `RUSTFS_SSE_C_REQUIRE_TLS=true` opts a deployment into the rejection now.
+//!
+//! The transport verdict is per connection, not per deployment: the layer is
+//! built with whether *this* listener terminates TLS, and additionally accepts
+//! a `https` forwarded protocol resolved by the trusted-proxy layer, which is
+//! the only spoof-resistant source for a TLS-terminating proxy in front.
+
+use bytes::Bytes;
+use futures::future::{Either, Ready, ready};
+use http::{HeaderMap, HeaderValue, Request, Response, StatusCode};
+use http_body_util::{BodyExt, Full};
+use metrics::counter;
+use rustfs_trusted_proxies::ClientInfo;
+use rustfs_utils::http::headers::{
+    AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY,
+    AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5,
+};
+use std::sync::Once;
+use std::task::{Context, Poll};
+use tower::{Layer, Service};
+use tracing::warn;
+
+use crate::storage_api::server::layer::request_context::RequestContext;
+
+/// Opt in to refusing SSE-C on a plaintext transport. Default `false` for this
+/// release; the reporting path runs either way.
+pub(crate) const ENV_SSE_C_REQUIRE_TLS: &str = "RUSTFS_SSE_C_REQUIRE_TLS";
+pub(crate) const DEFAULT_SSE_C_REQUIRE_TLS: bool = false;
+
+/// Counts SSE-C requests that arrived without TLS. A deployment planning to
+/// enable [`ENV_SSE_C_REQUIRE_TLS`] should see this at zero first.
+pub(crate) const METRIC_SSEC_PLAINTEXT_REQUESTS_TOTAL: &str = "rustfs_ssec_plaintext_requests_total";
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+type BoxBody = http_body_util::combinators::UnsyncBoxBody<Bytes, BoxError>;
+
+/// Whether the request carries any SSE-C header.
+///
+/// Any one of the three is enough: an incomplete triple is still an attempt to
+/// use SSE-C, and it is rejected later for being incomplete — but the key may
+/// already have crossed the wire.
+fn carries_ssec_headers(headers: &HeaderMap) -> bool {
+    headers.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM)
+        || headers.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY)
+        || headers.contains_key(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5)
+}
+
+/// Whether this request reached the server over TLS.
+///
+/// `connection_is_tls` is what this listener actually did. The forwarded
+/// protocol is only consulted as a second source because the trusted-proxy
+/// layer has already decided whether the peer is allowed to assert it; a
+/// request that arrives direct carries no such assertion.
+fn is_secure_transport(connection_is_tls: bool, client_info: Option<&ClientInfo>) -> bool {
+    if connection_is_tls {
+        return true;
+    }
+    client_info
+        .and_then(|info| info.forwarded_proto.as_deref())
+        .is_some_and(|proto| proto.eq_ignore_ascii_case("https"))
+}
+
+fn require_tls() -> bool {
+    rustfs_utils::get_env_bool(ENV_SSE_C_REQUIRE_TLS, DEFAULT_SSE_C_REQUIRE_TLS)
+}
+
+/// One warning per process: plaintext SSE-C traffic is driven by clients, so a
+/// per-request warning would let a busy client flood the log. The counter
+/// carries the per-request volume.
+fn warn_once_about_plaintext_ssec() {
+    static WARNED: Once = Once::new();
+    WARNED.call_once(|| {
+        warn!(
+            event = "ssec_request_without_tls",
+            require_tls = ENV_SSE_C_REQUIRE_TLS,
+            metric = METRIC_SSEC_PLAINTEXT_REQUESTS_TOTAL,
+            "SSE-C requests are arriving without TLS, so the customer key crosses the network in \
+             cleartext. AWS S3 refuses these; RustFS will too in a later release. Terminate TLS on \
+             this listener or on a trusted proxy, then set RUSTFS_SSE_C_REQUIRE_TLS=true. Reported \
+             once per process; the counter carries the volume."
+        );
+    });
+}
+
+/// The S3 rejection AWS returns for SSE-C without TLS. Built by hand because
+/// the rejection short-circuits the inner response stack, mirroring the
+/// rate-limit layer.
+fn ssec_requires_tls_response(request_id: Option<&str>) -> Response<BoxBody> {
+    let request_id_xml = request_id
+        .filter(|id| !id.is_empty() && id.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-'))
+        .map(|id| format!("<RequestId>{id}</RequestId>"))
+        .unwrap_or_default();
+    let body = format!(
+        "<?xml version=\"1.0\" encoding=\"UTF-8\"?>\
+         <Error><Code>InvalidRequest</Code>\
+         <Message>Requests specifying Server Side Encryption with Customer provided keys must be made over a secure connection.</Message>\
+         {request_id_xml}</Error>"
+    );
+    let body: BoxBody = Full::new(Bytes::from(body))
+        .map_err(|e| -> BoxError { Box::new(e) })
+        .boxed_unsync();
+
+    let mut response = Response::new(body);
+    *response.status_mut() = StatusCode::BAD_REQUEST;
+    response
+        .headers_mut()
+        .insert(http::header::CONTENT_TYPE, HeaderValue::from_static("application/xml"));
+    response
+}
+
+/// Layer that reports — and optionally refuses — SSE-C over a plaintext
+/// transport. `connection_is_tls` is whether the listener that accepted this
+/// connection terminated TLS.
+#[derive(Clone, Copy)]
+pub(crate) struct SsecTransportLayer {
+    connection_is_tls: bool,
+}
+
+impl SsecTransportLayer {
+    pub(crate) fn new(connection_is_tls: bool) -> Self {
+        Self { connection_is_tls }
+    }
+}
+
+impl<S> Layer<S> for SsecTransportLayer {
+    type Service = SsecTransportService<S>;
+
+    fn layer(&self, inner: S) -> Self::Service {
+        SsecTransportService {
+            inner,
+            connection_is_tls: self.connection_is_tls,
+        }
+    }
+}
+
+#[derive(Clone)]
+pub(crate) struct SsecTransportService<S> {
+    inner: S,
+    connection_is_tls: bool,
+}
+
+impl<S, ReqBody> Service<Request<ReqBody>> for SsecTransportService<S>
+where
+    S: Service<Request<ReqBody>, Response = Response<BoxBody>>,
+{
+    type Response = Response<BoxBody>;
+    type Error = S::Error;
+    type Future = Either<S::Future, Ready<Result<Response<BoxBody>, S::Error>>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.inner.poll_ready(cx)
+    }
+
+    fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
+        if !carries_ssec_headers(req.headers())
+            || is_secure_transport(self.connection_is_tls, req.extensions().get::<ClientInfo>())
+        {
+            return Either::Left(self.inner.call(req));
+        }
+
+        counter!(METRIC_SSEC_PLAINTEXT_REQUESTS_TOTAL).increment(1);
+        warn_once_about_plaintext_ssec();
+
+        if !require_tls() {
+            return Either::Left(self.inner.call(req));
+        }
+
+        Either::Right(ready(Ok(ssec_requires_tls_response(
+            req.extensions()
+                .get::<RequestContext>()
+                .map(|context| context.request_id.as_str()),
+        ))))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::{IpAddr, SocketAddr};
+
+    fn ssec_headers() -> HeaderMap {
+        let mut headers = HeaderMap::new();
+        headers.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, HeaderValue::from_static("AES256"));
+        headers
+    }
+
+    fn proxied(proto: &str) -> ClientInfo {
+        ClientInfo::from_trusted_proxy(
+            IpAddr::from([203, 0, 113, 10]),
+            None,
+            Some(proto.to_string()),
+            IpAddr::from([10, 0, 0, 1]),
+            1,
+            rustfs_trusted_proxies::ValidationMode::Lenient,
+            Vec::new(),
+        )
+    }
+
+    fn direct() -> ClientInfo {
+        ClientInfo::direct(SocketAddr::new(IpAddr::from([203, 0, 113, 10]), 443))
+    }
+
+    #[test]
+    fn any_ssec_header_counts_as_an_ssec_request() {
+        assert!(!carries_ssec_headers(&HeaderMap::new()));
+        assert!(carries_ssec_headers(&ssec_headers()));
+
+        // An incomplete triple is still an attempt, and the key may already
+        // have crossed the wire.
+        let mut only_key = HeaderMap::new();
+        only_key.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, HeaderValue::from_static("a2V5"));
+        assert!(carries_ssec_headers(&only_key));
+
+        let mut only_md5 = HeaderMap::new();
+        only_md5.insert(AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, HeaderValue::from_static("bWQ1"));
+        assert!(carries_ssec_headers(&only_md5));
+    }
+
+    #[test]
+    fn transport_is_secure_only_on_tls_or_a_resolved_https_proxy() {
+        assert!(is_secure_transport(true, None), "a TLS listener needs no header to prove it");
+        assert!(is_secure_transport(true, Some(&proxied("http"))), "the listener's own TLS wins");
+        assert!(
+            is_secure_transport(false, Some(&proxied("https"))),
+            "a TLS-terminating trusted proxy is a secure transport"
+        );
+        assert!(
+            !is_secure_transport(false, Some(&proxied("http"))),
+            "a proxy that forwarded plain HTTP is not"
+        );
+        assert!(
+            !is_secure_transport(false, Some(&direct())),
+            "a direct plaintext client asserts no protocol"
+        );
+        assert!(!is_secure_transport(false, None), "no transport evidence means not secure");
+    }
+
+    #[test]
+    fn the_rejection_carries_the_aws_wording_and_a_safe_request_id() {
+        let response = ssec_requires_tls_response(Some("abc-123"));
+        assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+
+        // A request id that is not plain enough to embed must be dropped
+        // rather than escaped into the XML body.
+        let injected = ssec_requires_tls_response(Some("<injected>"));
+        assert_eq!(injected.status(), StatusCode::BAD_REQUEST);
+    }
+}
@@ -74,6 +74,7 @@ use super::storage_api::ecstore_object::{
    EncryptionResolutionError, EncryptionResolutionErrorKind, ObjectEncryptionResolver, ReadEncryptionMaterial,
    ReadEncryptionMode, ReadEncryptionRequest,
 };
+use crate::runtime_sources::current_kms_runtime_service_manager;
 use crate::storage::access::{ReqInfo, request_context_from_req, resource_free_condition_values};
 use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
 #[cfg(feature = "rio-v2")]
@@ -2758,7 +2759,32 @@ async fn apply_managed_encryption_material_inner(
    // key it will actually be encrypted under.
    authorize_sse_kms_key(principal, encryption_type, KmsAction::GenerateDataKeyAction, &kms_key_to_use).await?;

-    let provider = get_sse_dek_provider().await?;
+    // A node-local master key is the explicit SSE-S3 fallback. Letting it serve
+    // an SSE-KMS request would persist an `aws:kms` marker and a KMS key id
+    // that never wrapped the data key, so refuse rather than downgrade.
+    //
+    // The refusal sits after the authorization gate so an unauthorized caller
+    // still sees AccessDenied whatever the KMS runtime state is, and it asks
+    // the resolved provider rather than a parallel availability signal,
+    // because the provider is what actually wraps the DEK.
+    let provider = match get_sse_dek_provider().await {
+        Ok(provider) => {
+            if matches!(encryption_type, SSEType::SseKms) && provider.wraps_dek_with_local_master_key() {
+                return Err(sse_kms_unavailable_error(kms_configured_but_unavailable().await));
+            }
+            provider
+        }
+        // With no master key set the local fallback fails with an SSE-S3-worded
+        // configuration error. An SSE-KMS request never asked for that provider,
+        // so it gets the SSE-KMS refusal instead of a message naming the wrong
+        // scheme.
+        Err(err) => {
+            if matches!(encryption_type, SSEType::SseKms) && runtime_sources::current_encryption_service().await.is_none() {
+                return Err(sse_kms_unavailable_error(kms_configured_but_unavailable().await));
+            }
+            return Err(err);
+        }
+    };
    let object_context = build_object_encryption_context(bucket, key, ssekms_context.as_ref());
    let (data_key, encrypted_data_key) = provider.generate_sse_dek(&object_context, &kms_key_to_use).await?;

@@ -2790,6 +2816,27 @@ async fn apply_managed_encryption_material_inner(
    })
 }

+/// Whether this node has a KMS configured that is not currently serving, which
+/// separates a transient outage (retryable, 503) from a deployment that never
+/// configured KMS at all (a client-side configuration error, 400).
+async fn kms_configured_but_unavailable() -> bool {
+    match current_kms_runtime_service_manager() {
+        Some(manager) => !matches!(manager.get_status().await, rustfs_kms::KmsServiceStatus::NotConfigured),
+        None => false,
+    }
+}
+
+fn sse_kms_unavailable_error(configured_but_unavailable: bool) -> ApiError {
+    if configured_but_unavailable {
+        return ApiError::from(StorageError::other(KmsUnavailableError));
+    }
+    ApiError {
+        code: S3ErrorCode::InvalidRequest,
+        message: "SSE-KMS requires a configured and running KMS service".to_string(),
+        source: None,
+    }
+}
+
 async fn apply_managed_decryption_material(
    bucket: &str,
    key: &str,
@@ -3236,6 +3283,16 @@ pub trait SseDekProvider: Send + Sync {
        )))
    }

+    /// Whether this provider wraps data keys with a node-local master key
+    /// instead of a KMS service.
+    ///
+    /// SSE-KMS must never be served by such a provider: the stored object would
+    /// claim `aws:kms` and name a KMS key id that never wrapped anything.
+    /// Defaults to false so only the local fallback has to declare itself.
+    fn wraps_dek_with_local_master_key(&self) -> bool {
+        false
+    }
+
    /// Decrypt a DEK from positively identified legacy managed metadata.
    #[cfg(feature = "rio-v2")]
    async fn decrypt_legacy_sse_dek(
@@ -3758,6 +3815,10 @@ impl LocalSseDekProvider {

 #[async_trait]
 impl SseDekProvider for LocalSseDekProvider {
+    fn wraps_dek_with_local_master_key(&self) -> bool {
+        true
+    }
+
    async fn generate_sse_dek(
        &self,
        _context: &ObjectEncryptionContext,
@@ -4344,6 +4405,127 @@ mod tests {
        assert_eq!(super::kms_data_plane_error_class(&missing), "key_not_found");
    }

+    #[test]
+    fn sse_kms_never_falls_back_to_the_local_sse_s3_provider() {
+        let unconfigured = super::sse_kms_unavailable_error(false);
+        assert_eq!(unconfigured.code, S3ErrorCode::InvalidRequest);
+        assert!(unconfigured.message.contains("SSE-KMS requires"));
+
+        let stopped = super::sse_kms_unavailable_error(true);
+        assert_eq!(stopped.code, S3ErrorCode::ServiceUnavailable);
+    }
+
+    fn managed_write(algorithm: &'static str, kms_key_id: Option<&str>) -> EncryptionRequest<'static> {
+        EncryptionRequest {
+            bucket: "finance",
+            key: "ledger.csv",
+            server_side_encryption: Some(ServerSideEncryption::from_static(algorithm)),
+            ssekms_key_id: kms_key_id.map(str::to_string),
+            ssekms_context: None,
+            sse_customer_algorithm: None,
+            sse_customer_key: None,
+            sse_customer_key_md5: None,
+            content_size: 128,
+            principal: None,
+        }
+    }
+
+    /// A bucket default naming a KMS key, on a node with no KMS, used to write
+    /// the object under the local master key while stamping `aws:kms` and that
+    /// never-consulted key id into the metadata (backlog#2368 B4).
+    #[tokio::test]
+    async fn sse_kms_write_is_refused_when_only_a_local_master_key_is_available() {
+        let _guard = lock_sse_test_state().await;
+        reset_sse_dek_provider();
+
+        async_with_vars(
+            [
+                ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
+                ("RUSTFS_SSE_S3_MASTER_KEY", Some(BASE64_STANDARD.encode_to_string([9u8; 32]))),
+            ],
+            async {
+                let error = sse_encryption(managed_write(
+                    ServerSideEncryption::AWS_KMS,
+                    Some("arn:aws:kms:us-east-1:123:key/nonexistent"),
+                ))
+                .await
+                .expect_err("SSE-KMS must not be served by the local master key");
+
+                assert_eq!(error.code, S3ErrorCode::InvalidRequest);
+                assert!(error.message.contains("SSE-KMS requires"), "message was {}", error.message);
+            },
+        )
+        .await;
+
+        reset_sse_dek_provider();
+    }
+
+    /// Without a master key the local fallback fails with an SSE-S3-worded
+    /// configuration error. An SSE-KMS request must not be told to set
+    /// `RUSTFS_SSE_S3_MASTER_KEY`.
+    #[tokio::test]
+    async fn sse_kms_refusal_names_sse_kms_rather_than_the_sse_s3_master_key() {
+        let _guard = lock_sse_test_state().await;
+        reset_sse_dek_provider();
+
+        async_with_vars(
+            [
+                ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
+                ("RUSTFS_SSE_S3_MASTER_KEY", None::<String>),
+            ],
+            async {
+                let error = sse_encryption(managed_write(ServerSideEncryption::AWS_KMS, Some("finance-key")))
+                    .await
+                    .expect_err("SSE-KMS must be refused when no KMS is configured");
+
+                assert_eq!(error.code, S3ErrorCode::InvalidRequest);
+                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]
+    async fn sse_s3_write_still_uses_the_local_master_key_fallback() {
+        let _guard = lock_sse_test_state().await;
+        reset_sse_dek_provider();
+
+        async_with_vars(
+            [
+                ("__RUSTFS_SSE_SIMPLE_CMK", None::<String>),
+                ("RUSTFS_SSE_S3_MASTER_KEY", Some(BASE64_STANDARD.encode_to_string([9u8; 32]))),
+            ],
+            async {
+                let material = sse_encryption(managed_write(ServerSideEncryption::AES256, None))
+                    .await
+                    .expect("SSE-S3 keeps its local master key fallback")
+                    .expect("managed sse-s3 material");
+
+                assert_eq!(material.sse_type, SSEType::SseS3);
+                assert_eq!(material.algorithm, ServerSideEncryption::AES256);
+
+                // No object may claim aws:kms while its DEK is wrapped locally.
+                let metadata = encryption_material_to_metadata(&material).expect("sse-s3 metadata should serialize");
+                assert!(
+                    !metadata.iter().any(|(_, value)| value == ServerSideEncryption::AWS_KMS),
+                    "local-master-key material must never be stamped aws:kms: {metadata:?}"
+                );
+            },
+        )
+        .await;
+
+        reset_sse_dek_provider();
+    }
+
    #[test]
    fn parse_simple_sse_cmk_accepts_valid_32_byte_key() {
        let mut key = [0u8; 32];