feat(storage): carry native values through shared SSE

Keep SSE configuration, requests, materials and read response fields on
native strings. Project legacy enums at the DTO boundaries and retain the
existing validation order, bucket defaults and KMS authorization behavior.
Carrier diagnostics omit customer keys and encryption key material.

Production lines: +186/-166. Add 5 negative diagnostic tests, 0 positives.
Do not implement or register an owned PutObject operation.

Validation: fmt, default and gateway-stack all-target Clippy, 777 affected
unit tests, script-tests with aligned Python, test wiring and prepush-quick.
Tests ran on 9ea8e8d37f9670f8da3dc5cb6ed5fe6e6e9bb592; rebased onto
a9dd5f0fd1 with no overlapping files or pin change.

Mutation witnesses (8 sites, 9 failing cases): restore derived Debug for
SseTypeV2, EncryptionRequest, DecryptionRequest, EncryptionMaterial and
DecryptionMaterial -> all five native_*debug_withholds* tests fail; change
unknown default mapping -> shared and copy unknown-default tests fail;
admit an unsupported algorithm -> rejects_invalid_sse_algorithm fails;
project AES reads as KMS -> classification_withholds_the_wrapping_key_for_sse_s3_reads fails.
Restore exact source hashes, touch the SSE source and rebuild: 777/777 pass.

Not verified: full workspace, optional rio-v2/FIPS builds, or CI for this commit.
No intended legacy protocol change; diagnostics intentionally omit key fields.

rustfs/src/app/multipart_usecase.rs

@@ -120,7 +120,6 @@ use rustfs_utils::http::{
 };
 use s3s::{S3Error, S3ErrorCode, S3Result, s3_error};
 use std::collections::{HashMap, HashSet};
-use std::str::FromStr;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
@@ -348,7 +347,7 @@ mod tests {
                 let prepare_request = PrepareEncryptionRequest {
                     bucket: "bucket",
                     key: "object",
-                    server_side_encryption: Some(ServerSideEncryption::from_static(ServerSideEncryption::AES256)),
+                    server_side_encryption: Some(ServerSideEncryption::AES256.to_owned()),
                     ssekms_key_id: None,
                     ssekms_context: None,
                     sse_customer_algorithm: None,

rustfs/src/app/multipart_usecase/create.rs

@@ -102,7 +102,9 @@ impl DefaultMultipartUsecase {
             ..
         } = req.input.clone();
 
-        let server_side_encryption = server_side_encryption.or(extract_server_side_encryption_from_headers(&req.headers)?);
+        let server_side_encryption = server_side_encryption
+            .map(|value| value.as_str().to_owned())
+            .or(extract_server_side_encryption_from_headers(&req.headers)?);
         let ssekms_key_id = ssekms_key_id.or_else(|| {
             req.headers
                 .get("x-amz-server-side-encryption-aws-kms-key-id")
@@ -301,7 +303,7 @@ impl DefaultMultipartUsecase {
             bucket: Some(bucket),
             key: Some(key),
             upload_id: Some(upload_id),
-            server_side_encryption: effective_sse,
+            server_side_encryption: effective_sse.map(Into::into),
             sse_customer_algorithm,
             sse_customer_key_md5,
             ssekms_key_id: effective_kms_key_id,

rustfs/src/app/multipart_usecase/upload_part.rs

@@ -281,14 +281,7 @@ impl DefaultMultipartUsecase {
         let (server_side_encryption, ssekms_key_id) = if has_ssec || preserve_ciphertext {
             (None, None)
         } else {
-            let sse = fi
-                .user_defined
-                .get("x-amz-server-side-encryption")
-                .map(|s| {
-                    ServerSideEncryption::from_str(s)
-                        .map_err(|e| ApiError::from(StorageError::other(format!("Invalid server-side encryption: {e}"))))
-                })
-                .transpose()?;
+            let sse = fi.user_defined.get("x-amz-server-side-encryption").cloned();
             let key_id = match sse.as_ref() {
                 Some(sse) if sse.as_str() == ServerSideEncryption::AWS_KMS => fi
                     .user_defined
@@ -422,7 +415,7 @@ impl DefaultMultipartUsecase {
         let upload_part_extra_checksum_headers = crate::app::object_usecase::additional_checksum_echo_pairs(&opts.want_checksum);
 
         let output = UploadPartOutput {
-            server_side_encryption: requested_sse,
+            server_side_encryption: requested_sse.map(ServerSideEncryption::from),
             ssekms_key_id: requested_kms_key_id,
             sse_customer_algorithm,
             sse_customer_key_md5,

rustfs/src/app/multipart_usecase/upload_part_copy.rs

@@ -322,14 +322,7 @@ impl DefaultMultipartUsecase {
             HashReader::from_stream(src_stream, length, actual_size, None, None, false).map_err(ApiError::from)?
         };
 
-        let server_side_encryption = mp_info
-            .user_defined
-            .get("x-amz-server-side-encryption")
-            .map(|s| {
-                ServerSideEncryption::from_str(s)
-                    .map_err(|e| ApiError::from(StorageError::other(format!("Invalid server-side encryption: {e}"))))
-            })
-            .transpose()?;
+        let server_side_encryption = mp_info.user_defined.get("x-amz-server-side-encryption").cloned();
         let has_ssec = mp_info
             .user_defined
             .contains_key("x-amz-server-side-encryption-customer-algorithm");
@@ -438,7 +431,7 @@ impl DefaultMultipartUsecase {
         let output = UploadPartCopyOutput {
             copy_part_result: Some(copy_part_result),
             copy_source_version_id: resolved_src_version_id,
-            server_side_encryption: requested_sse,
+            server_side_encryption: requested_sse.map(ServerSideEncryption::from),
             ssekms_key_id: requested_kms_key_id,
             sse_customer_algorithm,
             sse_customer_key_md5,

rustfs/src/app/object/copy.rs

@@ -232,7 +232,7 @@ impl DefaultObjectUsecase {
         }
         let ssekms_context = extract_ssekms_context_from_headers(&req.headers)?;
         validate_sse_headers_for_write(
-            requested_sse.as_ref(),
+            requested_sse.as_ref().map(|value| value.as_str()),
             requested_kms_key_id.as_ref(),
             ssekms_context.as_ref(),
             sse_customer_algorithm.as_ref(),
@@ -456,7 +456,7 @@ impl DefaultObjectUsecase {
 
         let (mut effective_sse, mut effective_kms_key_id) = resolve_bucket_default_sse(
             bucket_sse_config.as_ref().map(|(config, _)| config),
-            requested_sse,
+            requested_sse.map(|value| value.as_str().to_owned()),
             requested_kms_key_id,
             has_explicit_ssec,
         );
@@ -918,7 +918,7 @@ impl DefaultObjectUsecase {
         let output = CopyObjectOutput {
             copy_object_result: Some(copy_object_result),
             copy_source_version_id,
-            server_side_encryption: effective_sse,
+            server_side_encryption: effective_sse.map(Into::into),
             ssekms_key_id: effective_kms_key_id,
             sse_customer_algorithm,
             sse_customer_key_md5,
@@ -1019,7 +1019,7 @@ mod tests {
             .and_then(|rule| rule.apply_server_side_encryption_by_default.as_ref())
             .map(bucket_default_write_sse);
 
-        assert_eq!(effective_sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AES256));
+        assert_eq!(effective_sse.as_deref(), Some(ServerSideEncryption::AES256));
 
         // Valid algorithms map to themselves, byte-identical to the PUT path.
         for (configured, expected) in [

rustfs/src/app/object/extract.rs

@@ -2069,7 +2069,9 @@ impl DefaultObjectUsecase {
         let sse_customer_key = sse_customer_key.or(h_key);
         let sse_customer_key_md5 = sse_customer_key_md5.or(h_md5);
 
-        let original_sse = server_side_encryption.or(extract_server_side_encryption_from_headers(&req.headers)?);
+        let original_sse = server_side_encryption
+            .map(|value| value.as_str().to_owned())
+            .or(extract_server_side_encryption_from_headers(&req.headers)?);
         let bucket_sse_config = load_bucket_default_sse_config(&bucket).await?;
         let (mut effective_sse, mut effective_kms_key_id) = resolve_bucket_default_sse(
             bucket_sse_config.as_ref().map(|(config, _timestamp)| config),
@@ -2084,7 +2086,7 @@ impl DefaultObjectUsecase {
             return Err(s3_error!(NotImplemented, "SSE-KMS is not supported for extract uploads"));
         }
         validate_sse_headers_for_write(
-            effective_sse.as_ref(),
+            effective_sse.as_deref(),
             effective_kms_key_id.as_ref(),
             extract_ssekms_context_from_headers(&req.headers)?.as_ref(),
             sse_customer_algorithm.as_ref(),

rustfs/src/app/object/get.rs

@@ -3364,7 +3364,7 @@ impl DefaultObjectUsecase {
                 headers
                     .sse_customer_algorithm
                     .is_none()
-                    .then_some(headers.server_side_encryption),
+                    .then(|| ServerSideEncryption::from(headers.server_side_encryption)),
                 headers.sse_customer_algorithm,
                 headers.sse_customer_key_md5,
                 headers.ssekms_key_id,

rustfs/src/app/object/put.rs

@@ -920,7 +920,7 @@ pub(super) struct PutObjectContentInput {
 
 /// Encryption inputs of a PUT after the S3 input/header merge.
 pub(super) struct PutObjectSseInput {
-    pub(super) server_side_encryption: Option<ServerSideEncryption>,
+    pub(super) server_side_encryption: Option<String>,
     pub(super) ssekms_key_id: Option<SSEKMSKeyId>,
     pub(super) sse_customer_algorithm: Option<SSECustomerAlgorithm>,
     pub(super) sse_customer_key: Option<s3s::dto::SSECustomerKey>,
@@ -1102,7 +1102,7 @@ fn fail_put_object(completion: PutObjectCompletion, err: S3Error) -> S3Error {
 pub(super) struct PutObjectCommitted {
     pub(super) obj_info: ObjectInfo,
     pub(super) put_versioned: bool,
-    pub(super) effective_sse: Option<ServerSideEncryption>,
+    pub(super) effective_sse: Option<String>,
     pub(super) effective_kms_key_id: Option<SSEKMSKeyId>,
     pub(super) sse_customer_algorithm: Option<SSECustomerAlgorithm>,
     pub(super) sse_customer_key_md5: Option<SSECustomerKeyMD5>,
@@ -1281,7 +1281,9 @@ impl DefaultObjectUsecase {
         let sse_customer_key_md5 = sse_customer_key_md5.or(h_md5);
 
         // Merge server_side_encryption from headers (fallback when S3 layer does not populate input)
-        let server_side_encryption = server_side_encryption.or(extract_server_side_encryption_from_headers(&req.headers)?);
+        let server_side_encryption = server_side_encryption
+            .map(|value| value.as_str().to_owned())
+            .or(extract_server_side_encryption_from_headers(&req.headers)?);
 
         // Validate object key
         validate_new_object_key(&key, request_method_name)?;
@@ -1389,7 +1391,8 @@ impl DefaultObjectUsecase {
             server_side_encryption: committed
                 .effective_sse
                 .clone()
-                .filter(|_| committed.sse_customer_algorithm.is_none()),
+                .filter(|_| committed.sse_customer_algorithm.is_none())
+                .map(ServerSideEncryption::from),
             sse_customer_algorithm: committed.sse_customer_algorithm.clone(),
             sse_customer_key_md5: committed.sse_customer_key_md5.clone(),
             ssekms_key_id: committed.effective_kms_key_id.clone(),
@@ -1633,7 +1636,7 @@ impl DefaultObjectUsecase {
 
         // Validate SSE-C headers early: reject partial/invalid combinations per S3 spec
         validate_sse_headers_for_write(
-            effective_sse.as_ref(),
+            effective_sse.as_deref(),
             effective_kms_key_id.as_ref(),
             extract_ssekms_context_from_headers(headers)?.as_ref(),
             sse_customer_algorithm.as_ref(),

rustfs/src/app/object/shared.rs

@@ -17,7 +17,7 @@
 use super::*;
 use crate::app::storage_api::s3::dto::{
     ChecksumAlgorithm, ChecksumType, ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockMode, ObjectLockRetention,
-    ObjectLockRetentionMode, Range, SSEKMSKeyId, ServerSideEncryption, StorageClass, Timestamp, TimestampFormat,
+    ObjectLockRetentionMode, Range, ServerSideEncryption, StorageClass, Timestamp, TimestampFormat,
 };
 use crate::app::storage_api::s3::header::{X_AMZ_DELETE_MARKER, X_AMZ_VERSION_ID};
 use crate::app::storage_api::s3::{S3Error, S3ErrorCode, S3Result, StdError, s3_error};
@@ -347,10 +347,10 @@ pub(super) fn has_put_sse_request_headers(headers: &HeaderMap) -> bool {
 /// them to the returned pair.
 pub(super) fn resolve_bucket_default_sse(
     bucket_sse_config: Option<&ServerSideEncryptionConfiguration>,
-    requested_sse: Option<ServerSideEncryption>,
-    requested_kms_key_id: Option<SSEKMSKeyId>,
+    requested_sse: Option<String>,
+    requested_kms_key_id: Option<String>,
     has_explicit_ssec: bool,
-) -> (Option<ServerSideEncryption>, Option<SSEKMSKeyId>) {
+) -> (Option<String>, Option<String>) {
     let bucket_default = || {
         if has_explicit_ssec {
             return None;
@@ -1503,12 +1503,12 @@ mod tests {
 
         let (sse, kms_key_id) = resolve_bucket_default_sse(
             Some(&config),
-            Some(ServerSideEncryption::from_static(ServerSideEncryption::AES256)),
-            Some(SSEKMSKeyId::from("request-key".to_string())),
+            Some(String::from(ServerSideEncryption::AES256)),
+            Some("request-key".to_string()),
             false,
         );
 
-        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AES256));
+        assert_eq!(sse.as_deref(), Some(ServerSideEncryption::AES256));
         assert_eq!(kms_key_id.as_deref(), Some("request-key"));
     }
 
@@ -1516,14 +1516,10 @@ mod tests {
     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,
-        );
+        let (sse, kms_key_id) =
+            resolve_bucket_default_sse(Some(&config), Some(String::from(ServerSideEncryption::AES256)), None, false);
 
-        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AES256));
+        assert_eq!(sse.as_deref(), Some(ServerSideEncryption::AES256));
         assert!(kms_key_id.is_none(), "an SSE-S3 request must not inherit the bucket KMS key");
     }
 
@@ -1533,7 +1529,7 @@ mod tests {
 
         let (sse, kms_key_id) = resolve_bucket_default_sse(Some(&config), None, None, false);
 
-        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AWS_KMS));
+        assert_eq!(sse.as_deref(), Some(ServerSideEncryption::AWS_KMS));
         assert_eq!(kms_key_id.as_deref(), Some("bucket-key"));
     }
 
@@ -1547,7 +1543,7 @@ mod tests {
 
         let (sse, kms_key_id) = resolve_bucket_default_sse(Some(&config), None, None, false);
 
-        assert_eq!(sse.as_ref().map(|sse| sse.as_str()), Some(ServerSideEncryption::AES256));
+        assert_eq!(sse.as_deref(), Some(ServerSideEncryption::AES256));
         assert!(kms_key_id.is_none());
     }