// 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.

//! Server-Side Encryption (SSE) utilities
//!
//! This module provides reusable components for handling S3 Server-Side Encryption:
//! - SSE-S3 (AES256): Server-managed encryption with S3-managed keys
//! - SSE-KMS (aws:kms): Server-managed encryption with KMS-managed keys
//! - SSE-C (AES256): Customer-provided encryption keys
//!
//! ## Architecture
//!
//! ### Unified API
//! The module provides two core functions that automatically route to the correct encryption method:
//! - `sse_encryption()` - Unified encryption entry point
//! - `sse_decryption()` - Unified decryption entry point
//!
//! ### Managed SSE (SSE-S3 / SSE-KMS)
//! - Keys are managed by the server-side KMS service
//! - Data keys are generated and encrypted by KMS
//! - Encryption metadata is stored in object metadata
//!
//! ### Customer-Provided Keys (SSE-C)
//! - Keys are provided by the client on every request
//! - Server validates key using MD5 hash
//! - Keys are NEVER stored on the server
//!
//! ## Usage Example
//!
//! ```rust,ignore
//! // Unified encryption API
//! let request = EncryptionRequest {
//!     bucket: &bucket,
//!     key: &key,
//!     server_side_encryption: effective_sse.as_ref(),
//!     ssekms_key_id: effective_kms_key_id.as_deref(),
//!     sse_customer_algorithm: sse_customer_algorithm.as_ref(),
//!     sse_customer_key: sse_customer_key.as_deref(),
//!     sse_customer_key_md5: sse_customer_key_md5.as_deref(),
//!     content_size: actual_size,
//!     part_number: None,
//! };
//!
//! if let Some(material) = sse_encryption(request).await? {
//!     reader = material.wrap_reader(reader);
//!     metadata.extend(material.metadata);
//! }
//!
//! // Unified decryption API
//! let request = DecryptionRequest {
//!     bucket: &bucket,
//!     key: &key,
//!     metadata: &metadata,
//!     sse_customer_key: sse_customer_key.as_deref(),
//!     sse_customer_key_md5: sse_customer_key_md5.as_deref(),
//!     part_number: None,
//! };
//!
//! if let Some(material) = sse_decryption(request).await? {
//!     let (decrypted_reader, plaintext_size) = material.wrap_reader(reader, actual_size).await?;
//!     reader = decrypted_reader;
//!     content_size = plaintext_size;
//! }
//! ```

use aes_gcm::{
    Aes256Gcm, Key, Nonce,
    aead::{Aead, KeyInit},
};
use async_trait::async_trait;
use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
use http::HeaderMap;
use rand::Rng;
use rustfs_ecstore::error::StorageError;
use rustfs_filemeta::ObjectPartInfo;
use rustfs_kms::{
    DataKey,
    service_manager::get_global_encryption_service,
    types::{EncryptionMetadata, ObjectEncryptionContext},
};
use rustfs_rio::{DecryptReader, DynReader, EncryptReader, HardLimitReader, ReadStream, boxed_reader, wrap_reader};
use rustfs_utils::get_env_opt_str;
use s3s::S3ErrorCode;
use s3s::dto::ServerSideEncryption;
use std::collections::HashMap;
use std::sync::{Arc, OnceLock};
use tracing::{debug, error};

const INTERNAL_ENCRYPTION_KEY_ID_HEADER: &str = "x-rustfs-encryption-key-id";

use crate::error::ApiError;
use rustfs_ecstore::bucket::metadata_sys;
use rustfs_ecstore::error::Error;
use s3s::dto::{SSECustomerAlgorithm, SSECustomerKey, SSECustomerKeyMD5, SSEKMSKeyId};

// ============================================================================
// High-Level SSE Configuration
// ============================================================================

const DEFAULT_SSE_ALGORITHM: &str = "AES256";

const SUPPORT_SSE_ALGORITHMS: &[&str] = &[DEFAULT_SSE_ALGORITHM];

// check sse type
#[allow(unused)]
pub fn get_sse_type(
    server_side_encryption: Option<&ServerSideEncryption>,
    customer_algorithm: Option<&SSECustomerAlgorithm>,
    customer_key: Option<&SSECustomerKey>,
    customer_key_md5: Option<&SSECustomerKeyMD5>,
) -> Option<SSEType> {
    if customer_algorithm.is_some() && customer_key.is_some() && customer_key_md5.is_some() {
        return Some(SSEType::SseC);
    }

    let sse = server_side_encryption?;
    match sse.as_str() {
        ServerSideEncryption::AES256 => Some(SSEType::SseS3),
        ServerSideEncryption::AWS_KMS => Some(SSEType::SseKms),
        _ => None,
    }
}

/// SSE configuration resolved from request and bucket defaults
#[derive(Debug)]
pub struct SseConfiguration {
    /// Effective server-side encryption algorithm (after considering bucket defaults)
    pub effective_sse: ServerSideEncryption,
    /// Effective KMS key ID (after considering bucket defaults)
    pub effective_kms_key_id: Option<SSEKMSKeyId>,
}

/// Prepare SSE configuration by resolving request parameters with bucket defaults
///
/// This function:
/// 1. Queries bucket default encryption configuration
/// 2. Resolves effective encryption (request overrides bucket default)
/// 3. Prepares metadata headers for managed SSE
///
/// # Arguments
/// * `bucket` - Bucket name
/// * `server_side_encryption` - SSE algorithm from request (SSE-S3 or SSE-KMS)
/// * `ssekms_key_id` - KMS key ID from request
/// * `sse_customer_algorithm` - SSE-C algorithm from request
///
/// # Returns
/// `SseConfiguration` with resolved encryption parameters and metadata headers
async fn prepare_sse_configuration(
    bucket: &str,
    server_side_encryption: Option<ServerSideEncryption>,
    ssekms_key_id: Option<SSEKMSKeyId>,
) -> Result<Option<SseConfiguration>, ApiError> {
    if let Some(server_side_encryption) = server_side_encryption.clone()
        && server_side_encryption.as_str() == ServerSideEncryption::AES256
    {
        return Ok(Some(SseConfiguration {
            effective_sse: server_side_encryption,
            effective_kms_key_id: None,
        }));
    }

    if let Some(server_side_encryption) = server_side_encryption.clone()
        && let Some(ssekms_key_id) = ssekms_key_id
    {
        return Ok(Some(SseConfiguration {
            effective_sse: server_side_encryption,
            effective_kms_key_id: Some(ssekms_key_id),
        }));
    }

    // Get bucket default encryption configuration.
    let bucket_sse_config_result = metadata_sys::get_sse_config(bucket).await;
    debug!("bucket_sse_config_result={:?}", bucket_sse_config_result);

    if let Ok((bucket_sse_config, _timestamp)) = bucket_sse_config_result {
        let effective_sse = server_side_encryption.clone().or_else(|| {
            bucket_sse_config.rules.first().and_then(|rule| {
                debug!("Processing SSE rule: {:?}", rule);
                rule.apply_server_side_encryption_by_default.as_ref().map(|sse| {
                    debug!("Found SSE default: {:?}", sse);
                    match sse.sse_algorithm.as_str() {
                        "AES256" => ServerSideEncryption::from_static(ServerSideEncryption::AES256),
                        "aws:kms" => ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS),
                        _ => ServerSideEncryption::from_static(ServerSideEncryption::AES256), // fallback
                    }
                })
            })
        });
        if effective_sse.is_none() {
            return Ok(None);
        }

        debug!("effective_sse={:?} (original={:?})", effective_sse, server_side_encryption);

        let effective_kms_key_id = resolve_effective_kms_key_id(effective_sse.as_ref(), ssekms_key_id, || {
            bucket_sse_config.rules.first().and_then(|rule| {
                rule.apply_server_side_encryption_by_default
                    .as_ref()
                    .and_then(|sse| sse.kms_master_key_id.clone())
            })
        });

        Ok(Some(SseConfiguration {
            effective_sse: effective_sse.unwrap(),
            effective_kms_key_id,
        }))
    } else if let Err(e) = bucket_sse_config_result {
        match e {
            Error::ConfigNotFound => {
                // The bucket has no SSE config. If the user explicitly requested
                // aws:kms, we must honor that — return the explicit SSE header so
                // downstream logic can try (and fail if KMS is unavailable).
                if let Some(sse) = server_side_encryption {
                    Ok(Some(SseConfiguration {
                        effective_sse: sse,
                        effective_kms_key_id: ssekms_key_id,
                    }))
                } else {
                    Ok(None)
                }
            }
            _ => Err(ApiError::from(e)),
        }
    } else {
        Ok(None)
    }
}

fn resolve_effective_kms_key_id<F>(
    effective_sse: Option<&ServerSideEncryption>,
    requested_kms_key_id: Option<SSEKMSKeyId>,
    bucket_default_kms_key_id: F,
) -> Option<SSEKMSKeyId>
where
    F: FnOnce() -> Option<SSEKMSKeyId>,
{
    if effective_sse.is_none_or(|sse| sse.as_str() != ServerSideEncryption::AWS_KMS) {
        return requested_kms_key_id;
    }

    requested_kms_key_id.or_else(bucket_default_kms_key_id)
}

#[derive(Debug, Clone)]
pub enum SseTypeV2 {
    SseS3(ServerSideEncryption),
    SseKms(ServerSideEncryption, Option<SSEKMSKeyId>),
    SseC(SSECustomerAlgorithm, SSECustomerKey, SSECustomerKeyMD5),
}

impl SseTypeV2 {
    #[allow(unused)]
    pub fn to_metadata(&self) -> HashMap<String, String> {
        sse_configuration_to_metadata(self)
    }
}

pub async fn prepare_sse_configuration_v2(
    bucket: &str,
    server_side_encryption: Option<ServerSideEncryption>,
    customer_algorithm: Option<SSECustomerAlgorithm>,
    customer_key: Option<SSECustomerKey>,
    customer_key_md5: Option<SSECustomerKeyMD5>,
    ssekms_key_id: Option<SSEKMSKeyId>,
) -> Result<Option<SseTypeV2>, ApiError> {
    if let Some(customer_algorithm) = customer_algorithm
        && let Some(customer_key_md5) = customer_key_md5
    {
        // if create_multipart_upload request, customer_key is not provided
        let customer_key = customer_key.unwrap_or_default();

        return Ok(Some(SseTypeV2::SseC(customer_algorithm, customer_key, customer_key_md5)));
    }

    let sse_config = prepare_sse_configuration(bucket, server_side_encryption, ssekms_key_id).await?;

    if let Some(sse_config) = sse_config {
        return match sse_config.effective_sse.as_str() {
            ServerSideEncryption::AES256 => Ok(Some(SseTypeV2::SseS3(sse_config.effective_sse))),
            ServerSideEncryption::AWS_KMS => {
                Ok(Some(SseTypeV2::SseKms(sse_config.effective_sse.clone(), sse_config.effective_kms_key_id)))
            }
            _ => Ok(None),
        };
    }

    Ok(None)
}

#[allow(unused)]
pub fn sse_configuration_to_metadata(sse_configuration: &SseTypeV2) -> HashMap<String, String> {
    let mut metadata = HashMap::new();
    match sse_configuration {
        SseTypeV2::SseS3(sse) => {
            metadata.insert("x-amz-server-side-encryption".to_string(), sse.as_str().to_string());
        }
        SseTypeV2::SseKms(sse, kms_key_id) => {
            metadata.insert("x-amz-server-side-encryption".to_string(), sse.as_str().to_string());
            if let Some(kms_key_id) = kms_key_id {
                metadata.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), kms_key_id.to_string());
            }
        }
        SseTypeV2::SseC(algorithm, _key, key_md5) => {
            metadata.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
            metadata.insert(
                "x-amz-server-side-encryption-customer-algorithm".to_string(),
                algorithm.as_str().to_string(),
            );
            metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), key_md5.to_string());
        }
    }

    metadata
}

// ============================================================================
// Core Types - Unified Encryption/Decryption API
// ============================================================================

/// Request parameters for unified encryption
#[derive(Debug, Clone)]
pub struct EncryptionRequest<'a> {
    /// Bucket name
    pub bucket: &'a str,
    /// Object key
    pub key: &'a str,
    /// Server-side encryption algorithm (SSE-S3 or SSE-KMS)
    pub server_side_encryption: Option<ServerSideEncryption>,
    /// KMS key ID (for SSE-KMS)
    pub ssekms_key_id: Option<SSEKMSKeyId>,
    /// SSE-C algorithm (customer-provided key)
    pub sse_customer_algorithm: Option<SSECustomerAlgorithm>,
    /// SSE-C key (Base64-encoded)
    pub sse_customer_key: Option<SSECustomerKey>,
    /// SSE-C key MD5 (Base64-encoded)
    pub sse_customer_key_md5: Option<SSECustomerKeyMD5>,
    /// Content size (for metadata)
    pub content_size: i64,

    /// Part number (for multipart upload, None for single-part)
    pub part_number: Option<usize>,
    pub part_key: Option<String>,
    pub part_nonce: Option<String>,
}

impl EncryptionRequest<'_> {
    pub fn check_upload_part_customer_key_md5(
        &self,
        user_defined: &HashMap<String, String>,
        customer_key_md5: Option<SSECustomerKeyMD5>,
    ) -> Result<(), ApiError> {
        if let Some(customer_key_md5) = customer_key_md5 {
            // if customer_key_md5 is provided, check if it matches the metadata
            let customer_key_md5_from_metadata = user_defined.get("x-amz-server-side-encryption-customer-key-md5");
            if let Some(customer_key_md5_from_metadata) = customer_key_md5_from_metadata
                && customer_key_md5_from_metadata != customer_key_md5.as_str()
            {
                return Err(ApiError::from(StorageError::other("Customer key MD5 mismatch")));
            }
        }

        Ok(())
    }
}

#[inline]
fn sse_invalid_argument(message: &str) -> ApiError {
    ApiError {
        code: S3ErrorCode::InvalidArgument,
        message: message.to_string(),
        source: None,
    }
}

/// SSE-C parameters extracted from headers (algorithm, key, key MD5).
pub(crate) type SsecParamsFromHeaders = (Option<SSECustomerAlgorithm>, Option<SSECustomerKey>, Option<SSECustomerKeyMD5>);

/// Extract SSE-C parameters from request headers.
/// Used as fallback when the S3 layer does not populate them in the input struct.
///
/// Returns an error if an SSE-C header is present but cannot be parsed as valid UTF-8,
/// ensuring malformed headers do not bypass validation.
pub(crate) fn extract_ssec_params_from_headers(headers: &HeaderMap) -> Result<SsecParamsFromHeaders, ApiError> {
    let algorithm = match headers.get("x-amz-server-side-encryption-customer-algorithm") {
        None => Ok(None),
        Some(v) => v
            .to_str()
            .map(|s| Some(SSECustomerAlgorithm::from(s.to_string())))
            .map_err(|_| sse_invalid_argument("The x-amz-server-side-encryption-customer-algorithm header must be valid UTF-8.")),
    }?;

    let key = match headers.get("x-amz-server-side-encryption-customer-key") {
        None => Ok(None),
        Some(v) => v
            .to_str()
            .map(|s| Some(SSECustomerKey::from(s.to_string())))
            .map_err(|_| sse_invalid_argument("The x-amz-server-side-encryption-customer-key header must be valid UTF-8.")),
    }?;

    let key_md5 = match headers.get("x-amz-server-side-encryption-customer-key-md5") {
        None => Ok(None),
        Some(v) => v
            .to_str()
            .map(|s| Some(SSECustomerKeyMD5::from(s.to_string())))
            .map_err(|_| sse_invalid_argument("The x-amz-server-side-encryption-customer-key-md5 header must be valid UTF-8.")),
    }?;

    Ok((algorithm, key, key_md5))
}

/// Extract x-amz-server-side-encryption from request headers.
/// Used as fallback when the S3 layer does not populate it in the input struct.
///
/// Returns an error if the header is present but cannot be parsed as valid UTF-8,
/// ensuring malformed headers do not bypass validation.
pub(crate) fn extract_server_side_encryption_from_headers(headers: &HeaderMap) -> Result<Option<ServerSideEncryption>, ApiError> {
    match headers.get("x-amz-server-side-encryption") {
        None => Ok(None),
        Some(v) => v
            .to_str()
            .map(|s| Some(ServerSideEncryption::from(s.to_string())))
            .map_err(|_| sse_invalid_argument("The x-amz-server-side-encryption header must be valid UTF-8.")),
    }
}

#[inline]
pub(crate) fn validate_sse_headers_for_write(
    server_side_encryption: Option<&ServerSideEncryption>,
    ssekms_key_id: Option<&SSEKMSKeyId>,
    sse_customer_algorithm: Option<&SSECustomerAlgorithm>,
    sse_customer_key: Option<&SSECustomerKey>,
    sse_customer_key_md5: Option<&SSECustomerKeyMD5>,
    require_sse_customer_key: bool,
) -> Result<(), ApiError> {
    if let Some(sse) = server_side_encryption {
        let s = sse.as_str();
        if s != ServerSideEncryption::AES256 && s != ServerSideEncryption::AWS_KMS {
            return Err(sse_invalid_argument(
                "The SSE algorithm specified is not supported. The valid values are AES256 or aws:kms.",
            ));
        }
    }

    let has_ssec_headers = sse_customer_algorithm.is_some() || sse_customer_key.is_some() || sse_customer_key_md5.is_some();
    let has_managed_headers = server_side_encryption.is_some() || ssekms_key_id.is_some();

    if has_ssec_headers {
        if has_managed_headers {
            return Err(sse_invalid_argument(
                "The SSE-C and managed server-side encryption headers cannot be used together.",
            ));
        }

        let has_valid_ssec_headers = if require_sse_customer_key {
            matches!(
                (sse_customer_algorithm, sse_customer_key, sse_customer_key_md5),
                (Some(_), Some(_), Some(_))
            )
        } else {
            matches!((sse_customer_algorithm, sse_customer_key_md5), (Some(_), Some(_)))
        };

        if !has_valid_ssec_headers {
            let message = if require_sse_customer_key {
                "Missing SSE-C parameters. Algorithm, customer key and customer key MD5 are all required."
            } else {
                "Missing SSE-C parameters. Algorithm and customer key MD5 are required."
            };

            return Err(ssec_invalid_request(message));
        }
    }

    if ssekms_key_id.is_some() && server_side_encryption.is_none_or(|sse| sse.as_str() != ServerSideEncryption::AWS_KMS) {
        return Err(sse_invalid_argument(
            "The SSE-KMS key ID header can only be used when x-amz-server-side-encryption is set to aws:kms.",
        ));
    }

    Ok(())
}

#[inline]
pub(crate) fn validate_sse_headers_for_read(metadata: &HashMap<String, String>, headers: &HeaderMap) -> Result<(), ApiError> {
    let has_req_ssec = headers.contains_key("x-amz-server-side-encryption-customer-algorithm")
        || headers.contains_key("x-amz-server-side-encryption-customer-key")
        || headers.contains_key("x-amz-server-side-encryption-customer-key-md5");

    let has_req_sse = headers.contains_key("x-amz-server-side-encryption")
        || headers.contains_key("x-amz-server-side-encryption-aws-kms-key-id");

    let is_object_ssec = metadata.contains_key("x-amz-server-side-encryption-customer-algorithm");
    let is_object_sse = metadata.contains_key("x-amz-server-side-encryption");

    if is_object_ssec {
        if has_req_sse {
            return Err(sse_invalid_argument(
                "Server-side encryption headers cannot be used with an object encrypted using SSE-C.",
            ));
        }
        return Ok(());
    }

    if is_object_sse && has_req_ssec {
        return Err(sse_invalid_argument(
            "SSE-C headers cannot be used with an object encrypted using server-side managed encryption.",
        ));
    }

    if has_req_ssec {
        return Err(ssec_invalid_request(
            "The object was stored without SSE-C. The correct SSE-C parameters must not be provided.",
        ));
    }

    if has_req_sse {
        return Err(sse_invalid_argument(
            "The object is not encrypted with server-side encryption. Do not provide server-side encryption headers.",
        ));
    }

    Ok(())
}

/// Request parameters for unified decryption
#[derive(Debug)]
pub struct DecryptionRequest<'a> {
    /// Bucket name
    pub bucket: &'a str,
    /// Object key
    pub key: &'a str,
    /// Object metadata containing encryption headers
    pub metadata: &'a HashMap<String, String>,
    /// SSE-C key (Base64-encoded) - required if object was encrypted with SSE-C
    pub sse_customer_key: Option<&'a SSECustomerKey>,
    /// SSE-C key MD5 (Base64-encoded) - required if object was encrypted with SSE-C
    pub sse_customer_key_md5: Option<&'a SSECustomerKeyMD5>,
    /// Part number (for multipart upload, None for single-part)
    pub part_number: Option<usize>, // Unused Fields
    /// Parts information for multipart objects
    pub parts: &'a [ObjectPartInfo],
    /// Object-level ETag, used to distinguish multipart objects from single-part objects.
    pub etag: Option<&'a str>,
}

/// Unified encryption material returned by `apply_encryption()`
#[derive(Debug)]
pub struct EncryptionMaterial {
    #[allow(unused)]
    pub sse_type: SSEType,
    pub server_side_encryption: ServerSideEncryption,
    pub kms_key_id: Option<SSEKMSKeyId>,

    #[allow(unused)]
    pub algorithm: SSECustomerAlgorithm,

    /// Encryption key bytes
    pub key_bytes: [u8; 32],
    /// Nonce/IV for encryption
    pub nonce: [u8; 12],
    /// Metadata to store with the object
    pub metadata: HashMap<String, String>,
}

/// Unified decryption material returned by `apply_decryption()`
#[derive(Debug)]
pub struct DecryptionMaterial {
    #[allow(unused)]
    pub sse_type: SSEType,
    pub server_side_encryption: ServerSideEncryption,
    pub kms_key_id: Option<SSEKMSKeyId>,
    pub algorithm: SSECustomerAlgorithm,
    pub customer_key_md5: Option<SSECustomerKeyMD5>, // if use SSE-C, check key md5

    /// Decryption key bytes
    pub key_bytes: [u8; 32],
    /// Nonce/IV for decryption
    pub nonce: [u8; 12],
    /// Original unencrypted size (if available)
    pub original_size: Option<i64>,

    /// Whether this is a multipart object
    pub is_multipart: bool,
    /// Part information for multipart objects
    pub parts: Vec<ObjectPartInfo>,
}

fn is_multipart_object(etag: Option<&str>, parts: &[ObjectPartInfo]) -> bool {
    if parts.len() > 1 {
        return true;
    }

    etag.map(|etag| etag.trim_matches('"').len() != 32).unwrap_or(false)
}

/// Type of encryption used
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SSEType {
    /// SSE-S3 (AES256)
    SseS3,
    /// SSE-KMS (aws:kms)
    SseKms,
    /// SSE-C (customer-provided key)
    SseC,
}

impl EncryptionMaterial {
    /// Wrap a reader with encryption
    pub fn wrap_reader<R>(&self, reader: R) -> Box<EncryptReader<R>>
    where
        R: rustfs_rio::ReadStream + 'static,
    {
        Box::new(EncryptReader::new(reader, self.key_bytes, self.nonce))
    }
}

impl DecryptionMaterial {
    /// Wrap a reader with decryption
    /// For multipart objects, use `wrap_multipart_stream` instead
    pub fn wrap_single_reader<R>(&self, reader: R) -> Box<DecryptReader<R>>
    where
        R: rustfs_rio::ReadStream + 'static,
    {
        Box::new(DecryptReader::new(reader, self.key_bytes, self.nonce))
    }

    /// Wrap a stream with multipart decryption
    /// Returns the decrypted reader and the total plaintext size
    pub async fn wrap_multipart_stream<R>(&self, encrypted_stream: R) -> Result<(DynReader, i64), StorageError>
    where
        R: ReadStream + 'static,
    {
        decrypt_multipart_managed_stream(encrypted_stream, &self.parts, self.key_bytes, self.nonce).await
    }

    /// Unified method to wrap stream with decryption and hard limit
    /// Handles both single-part and multipart objects, applies decryption and size limiting
    /// Accepts a readable stream (from object storage) and returns (decrypted_reader, plaintext_size)
    pub async fn wrap_reader<R>(self, stream: R, actual_size: i64) -> Result<(DynReader, i64), StorageError>
    where
        R: ReadStream + 'static,
    {
        let (mut final_stream, response_content_length): (DynReader, i64) = if self.is_multipart {
            // Multipart decryption
            let (decrypted_reader, plain_size) = self.wrap_multipart_stream(stream).await?;
            (decrypted_reader, plain_size)
        } else {
            // Single-part decryption keeps Reader capabilities via the generic wrapper helper.
            let decrypt_reader = self.wrap_single_reader(wrap_reader(stream));
            let plain_size = self.original_size.unwrap_or(actual_size);
            (decrypt_reader, plain_size)
        };

        // Add hard limit reader to prevent over-reading
        // final_stream is already a DynReader, no need to wrap with WarpReader
        let limit_reader = HardLimitReader::new(final_stream, response_content_length);
        final_stream = Box::new(limit_reader);

        debug!(
            "{:?} decryption applied: plaintext_size={}, encrypted_size={}",
            self.sse_type, response_content_length, actual_size
        );

        Ok((final_stream, response_content_length))
    }
}

// ============================================================================
// Core API - Unified Encryption/Decryption Entry Points
// ============================================================================

/// **Core API**: Apply encryption based on request parameters
///
/// This function automatically routes to the appropriate encryption method:
/// - SSE-C if customer key is provided
/// - SSE-S3/SSE-KMS if server-side encryption is requested
/// - None if no encryption is requested
///
/// # Arguments
/// * `request` - Encryption request with all possible encryption parameters
///
/// # Returns
/// * `Ok(Some(material))` - Encryption should be applied with the returned material
/// * `Ok(None)` - No encryption requested
/// * `Err` - Encryption configuration error
///
/// # Example
/// ```rust,ignore
/// let request = EncryptionRequest {
///     bucket: &bucket,
///     key: &key,
///     server_side_encryption: effective_sse.as_ref(),
///     ssekms_key_id: effective_kms_key_id.as_deref(),
///     sse_customer_algorithm: sse_customer_algorithm.as_ref(),
///     sse_customer_key: sse_customer_key.as_deref(),
///     sse_customer_key_md5: sse_customer_key_md5.as_deref(),
///     content_size: actual_size,
///     part_number: None,
/// };
///
/// if let Some(material) = sse_encryption(request).await? {
///     reader = material.wrap_reader(reader);
///     metadata.extend(material.metadata);
/// }
/// ```
pub async fn sse_encryption(request: EncryptionRequest<'_>) -> Result<Option<EncryptionMaterial>, ApiError> {
    validate_sse_headers_for_write(
        request.server_side_encryption.as_ref(),
        request.ssekms_key_id.as_ref(),
        request.sse_customer_algorithm.as_ref(),
        request.sse_customer_key.as_ref(),
        request.sse_customer_key_md5.as_ref(),
        true,
    )?;

    // Priority 1: SSE-C (customer-provided key)
    if let (Some(algorithm), Some(key), Some(key_md5)) =
        (request.sse_customer_algorithm, request.sse_customer_key, request.sse_customer_key_md5)
    {
        return apply_ssec_encryption_material(
            request.bucket,
            request.key,
            algorithm,
            key,
            key_md5,
            request.content_size,
            request.part_number,
        )
        .await
        .map(Some);
    }

    // Priority 2: Managed SSE (SSE-S3 or SSE-KMS)
    let sse_config = prepare_sse_configuration(request.bucket, request.server_side_encryption, request.ssekms_key_id).await?;

    if let Some(sse_config) = sse_config
        && is_managed_sse(&sse_config.effective_sse)
    {
        return apply_managed_encryption_material(
            request.bucket,
            request.key,
            sse_config.effective_sse,
            sse_config.effective_kms_key_id,
            request.content_size,
            request.part_number,
            request.part_key,
            request.part_nonce,
        )
        .await
        .map(Some);
    }

    // No encryption requested
    Ok(None)
}

/// **Core API**: Apply encryption based on request parameters
///
/// sse_prepare_encryption, support SSE-C, SSE-S3, SSE-KMS
pub struct PrepareEncryptionRequest<'a> {
    /// Bucket name
    pub bucket: &'a str,
    /// Object key
    pub key: &'a str,
    /// Server-side encryption algorithm (SSE-S3 or SSE-KMS)
    pub server_side_encryption: Option<ServerSideEncryption>,
    /// KMS key ID (for SSE-KMS)
    pub ssekms_key_id: Option<SSEKMSKeyId>,
    /// SSE-C algorithm (customer-provided key)
    pub sse_customer_algorithm: Option<SSECustomerAlgorithm>,
    /// SSE-C key MD5 (Base64-encoded)
    pub sse_customer_key_md5: Option<SSECustomerKeyMD5>,
}

pub async fn sse_prepare_encryption(request: PrepareEncryptionRequest<'_>) -> Result<Option<EncryptionMaterial>, ApiError> {
    validate_sse_headers_for_write(
        request.server_side_encryption.as_ref(),
        request.ssekms_key_id.as_ref(),
        request.sse_customer_algorithm.as_ref(),
        None,
        request.sse_customer_key_md5.as_ref(),
        false,
    )?;

    let sse_type = prepare_sse_configuration_v2(
        request.bucket,
        request.server_side_encryption,
        request.sse_customer_algorithm,
        None,
        request.sse_customer_key_md5,
        request.ssekms_key_id,
    )
    .await?;

    // apply encryption material
    let material = match sse_type {
        Some(SseTypeV2::SseS3(sse)) => {
            apply_managed_encryption_material(request.bucket, request.key, sse, None, 0, None, None, None).await?
        }
        Some(SseTypeV2::SseKms(sse, kms_key_id)) => {
            apply_managed_encryption_material(request.bucket, request.key, sse, kms_key_id, 0, None, None, None).await?
        }
        Some(SseTypeV2::SseC(algorithm, _, key_md5)) => apply_ssec_prepare_encryption_material(algorithm, key_md5).await?,
        None => return Ok(None),
    };

    Ok(Some(material))
}

/// **Core API**: Apply decryption based on stored metadata
///
/// This function automatically detects the encryption type from metadata:
/// - SSE-C if customer key is provided
/// - SSE-S3/SSE-KMS if managed encryption metadata is found
/// - None if object is not encrypted
///
/// # Arguments
/// * `request` - Decryption request with metadata and optional customer key
///
/// # Returns
/// * `Ok(Some(material))` - Decryption should be applied with the returned material
/// * `Ok(None)` - Object is not encrypted
/// * `Err` - Decryption configuration error or key mismatch
///
/// # Example
/// ```rust,ignore
/// let request = DecryptionRequest {
///     bucket: &bucket,
///     key: &key,
///     metadata: &metadata,
///     sse_customer_key: sse_customer_key.as_deref(),
///     sse_customer_key_md5: sse_customer_key_md5.as_deref(),
///     part_number: None,
/// };
///
/// if let Some(material) = sse_decryption(request).await? {
///     let (decrypted_reader, plaintext_size) = material.wrap_reader(reader, actual_size).await?;
///     reader = decrypted_reader;
///     content_size = plaintext_size;
/// }
/// ```
pub async fn sse_decryption(request: DecryptionRequest<'_>) -> Result<Option<DecryptionMaterial>, ApiError> {
    let is_multipart = is_multipart_object(request.etag, request.parts);

    // Check for SSE-C encryption
    if request
        .metadata
        .contains_key("x-amz-server-side-encryption-customer-algorithm")
    {
        let (key, key_md5) = match (request.sse_customer_key, request.sse_customer_key_md5) {
            (Some(k), Some(md5)) => (k, md5),
            _ => {
                return Err(ssec_invalid_request(
                    "The object was stored using a form of Server Side Encryption. \
                     The correct parameters must be provided to retrieve the object.",
                ));
            }
        };

        // Verify that the provided key MD5 matches the stored MD5 for security
        let stored_md5 = request.metadata.get("x-amz-server-side-encryption-customer-key-md5");
        verify_ssec_key_match(key_md5, stored_md5)?;

        let mut material =
            apply_ssec_decryption_material(request.bucket, request.key, request.metadata, key, key_md5, request.part_number)
                .await?;
        material.is_multipart = is_multipart;
        material.parts = request.parts.to_vec();
        material.customer_key_md5 = Some(key_md5.clone());

        return Ok(Some(material));
    }

    // Check for managed SSE encryption
    if request.metadata.contains_key("x-rustfs-encryption-key") {
        let mut material_opt =
            apply_managed_decryption_material(request.bucket, request.key, request.metadata, request.part_number).await?;
        if let Some(ref mut material) = material_opt {
            material.is_multipart = is_multipart;
            material.parts = request.parts.to_vec();
        }
        return Ok(material_opt);
    }

    // No encryption detected
    Ok(None)
}

// ============================================================================
// Internal Implementation - SSE-C
// ============================================================================

async fn apply_ssec_prepare_encryption_material(
    algorithm: SSECustomerAlgorithm,
    sse_key_md5: SSECustomerKeyMD5,
) -> Result<EncryptionMaterial, ApiError> {
    // Build metadata
    let mut metadata = HashMap::new();

    metadata.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
    metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), algorithm.clone());
    metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), sse_key_md5);

    Ok(EncryptionMaterial {
        sse_type: SSEType::SseC,
        server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AES256),
        kms_key_id: None,
        algorithm,
        key_bytes: [0; 32],
        nonce: [0; 12],
        metadata,
    })
}

async fn apply_ssec_encryption_material(
    bucket: &str,
    key: &str,
    algorithm: SSECustomerAlgorithm,
    sse_key: SSECustomerKey,
    sse_key_md5: SSECustomerKeyMD5,
    content_size: i64,
    part_number: Option<usize>,
) -> Result<EncryptionMaterial, ApiError> {
    let params = SsecParams {
        algorithm,
        key: sse_key,
        key_md5: sse_key_md5,
    };

    let validated = validate_ssec_params(params)?;

    // Generate nonce (deterministic for SSE-C)
    let base_nonce = generate_ssec_nonce(bucket, key);
    let nonce = if let Some(part_num) = part_number {
        derive_part_nonce(base_nonce, part_num)
    } else {
        base_nonce
    };

    // Build metadata
    let mut metadata = HashMap::new();

    metadata.insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
    metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), validated.algorithm.clone());
    metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), validated.key_md5.clone());
    metadata.insert(
        "x-amz-server-side-encryption-customer-original-size".to_string(),
        content_size.to_string(),
    );

    Ok(EncryptionMaterial {
        sse_type: SSEType::SseC,
        server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AES256),
        kms_key_id: None,
        algorithm: validated.algorithm,
        key_bytes: validated.key_bytes,
        nonce,
        metadata,
    })
}

async fn apply_ssec_decryption_material(
    bucket: &str,
    key: &str,
    metadata: &HashMap<String, String>,
    sse_key: &str,
    sse_key_md5: &str,
    part_number: Option<usize>,
) -> Result<DecryptionMaterial, ApiError> {
    // Validate provided key
    let algorithm = metadata
        .get("x-amz-server-side-encryption-customer-algorithm")
        .map(|s| s.as_str())
        .unwrap_or("AES256");

    let params = SsecParams {
        algorithm: algorithm.to_string(),
        key: sse_key.to_string(),
        key_md5: sse_key_md5.to_string(),
    };

    let validated = validate_ssec_params(params)?;

    // Generate nonce (same as encryption)
    let base_nonce = generate_ssec_nonce(bucket, key);
    let nonce = if let Some(part_num) = part_number {
        derive_part_nonce(base_nonce, part_num)
    } else {
        base_nonce
    };

    let original_size = metadata
        .get("x-amz-server-side-encryption-customer-original-size")
        .and_then(|s| s.parse::<i64>().ok());

    Ok(DecryptionMaterial {
        sse_type: SSEType::SseC,
        server_side_encryption: ServerSideEncryption::from_static(ServerSideEncryption::AES256), // const
        kms_key_id: None,
        algorithm: SSECustomerAlgorithm::from(algorithm),

        customer_key_md5: None,
        key_bytes: validated.key_bytes,
        nonce,
        original_size,

        is_multipart: false,
        parts: Vec::new(),
    })
}

// ============================================================================
// Internal Implementation - Managed SSE (SSE-S3 / SSE-KMS)
// ============================================================================

#[allow(clippy::too_many_arguments)]
async fn apply_managed_encryption_material(
    bucket: &str,
    key: &str,
    server_side_encryption: ServerSideEncryption,
    kms_key_id: Option<SSEKMSKeyId>,
    content_size: i64,
    part_number: Option<usize>,
    part_key: Option<String>,
    part_nonce: Option<String>,
) -> Result<EncryptionMaterial, ApiError> {
    // For multipart, we only generate keys at CompleteMultipartUpload
    // During UploadPart, we use the same base nonce with incremented counter
    // This is handled externally, so here we just generate the base material

    if !is_managed_sse(&server_side_encryption) {
        return Err(ApiError::from(StorageError::other(format!(
            "Unsupported server-side encryption: {}",
            server_side_encryption.as_str()
        ))));
    }

    let encryption_type = match server_side_encryption.as_str() {
        "AES256" => SSEType::SseS3,
        "aws:kms" => SSEType::SseKms,
        _ => SSEType::SseS3,
    };

    let mut context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string());
    if content_size >= 0 {
        context = context.with_size(content_size as u64);
    }

    // Determine KMS key ID to use for internal key wrapping.
    let mut kms_key_candidate = kms_key_id.clone();
    if kms_key_candidate.is_none() {
        // Try to get default key from KMS service (if available)
        if let Some(service) = get_global_encryption_service().await {
            kms_key_candidate = service.get_default_key_id().cloned();
        }
    }

    let kms_key_to_use = match (encryption_type, kms_key_candidate.clone()) {
        (SSEType::SseS3, Some(kms_key_id)) => kms_key_id,
        (SSEType::SseS3, None) => "default".to_string(),
        (SSEType::SseKms, Some(kms_key_id)) => kms_key_id,
        (SSEType::SseKms, None) => {
            return Err(ApiError::from(StorageError::other(
                "No KMS key available for managed server-side encryption (required for SSE-KMS)",
            )));
        }
        _ => unreachable!("managed SSE branch only supports SSE-S3 or SSE-KMS"),
    };

    let provider = get_sse_dek_provider().await?;

    let (data_key, encrypted_data_key) = if let Some(part_number) = part_number
        && let Some(part_nonce) = part_nonce
        && let Some(part_key) = part_key
        && part_number >= 1
    // upload_part mode, dek generate by create_multipart_upload
    {
        let _base_nonce = BASE64_STANDARD
            .decode(part_nonce.as_bytes())
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decode nonce: {e}"))))?;
        if _base_nonce.len() != 12 {
            return Err(ApiError::from(StorageError::other("Invalid encryption nonce length; expected 12 bytes")));
        }
        let mut base_nonce_array = [0u8; 12];
        base_nonce_array.copy_from_slice(&_base_nonce[..12]);
        let encrypted_data_key = BASE64_STANDARD
            .decode(part_key.as_bytes())
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decode data key: {e}"))))?;
        let _data_key = provider
            .decrypt_sse_dek(encrypted_data_key.as_slice(), &kms_key_to_use)
            .await?;
        let data_key = DataKey {
            plaintext_key: _data_key,
            nonce: derive_part_nonce(base_nonce_array, part_number),
        };

        // load original data key from metadata
        (data_key, encrypted_data_key)
    } else {
        // Use factory pattern to get provider (test or production mode)
        let (data_key, encrypted_data_key) = provider
            .generate_sse_dek(bucket, key, &kms_key_to_use)
            .await
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {e}"))))?;
        (data_key, encrypted_data_key)
    };

    let algorithm = server_side_encryption.as_str().to_string();

    let encryption_metadata = EncryptionMetadata {
        algorithm: algorithm.clone(),
        key_id: kms_key_to_use.clone(),
        key_version: 1,
        iv: data_key.nonce.to_vec(),
        tag: None,
        encryption_context: context.encryption_context.clone(),
        encrypted_at: jiff::Zoned::now(),
        original_size: if content_size >= 0 { content_size as u64 } else { 0 },
        encrypted_data_key,
    };

    // Build metadata headers
    let mut metadata = HashMap::new();

    // Try to use service for metadata formatting if available, otherwise build manually
    if let Some(service) = get_global_encryption_service().await {
        metadata = service.metadata_to_headers(&encryption_metadata);
    } else {
        // Manual metadata building for test mode
        metadata.insert(
            "x-rustfs-encryption-key".to_string(),
            BASE64_STANDARD.encode(&encryption_metadata.encrypted_data_key),
        );
        metadata.insert("x-rustfs-encryption-iv".to_string(), BASE64_STANDARD.encode(&encryption_metadata.iv));
        metadata.insert("x-rustfs-encryption-algorithm".to_string(), encryption_metadata.algorithm.clone());
        metadata.insert("x-amz-server-side-encryption".to_string(), server_side_encryption.as_str().to_string());
    }

    if matches!(encryption_type, SSEType::SseKms) {
        metadata.insert("x-amz-server-side-encryption-aws-kms-key-id".to_string(), kms_key_to_use.clone());
    } else {
        metadata.remove("x-amz-server-side-encryption-aws-kms-key-id");
    }
    metadata.insert(INTERNAL_ENCRYPTION_KEY_ID_HEADER.to_string(), kms_key_to_use.clone());

    metadata.insert(
        "x-rustfs-encryption-original-size".to_string(),
        encryption_metadata.original_size.to_string(),
    );

    Ok(EncryptionMaterial {
        sse_type: encryption_type,
        server_side_encryption,
        kms_key_id: matches!(encryption_type, SSEType::SseKms).then_some(kms_key_to_use),
        algorithm,

        key_bytes: data_key.plaintext_key,
        nonce: data_key.nonce,
        metadata,
    })
}

async fn apply_managed_decryption_material(
    _bucket: &str,
    _key: &str,
    metadata: &HashMap<String, String>,
    part_number: Option<usize>,
) -> Result<Option<DecryptionMaterial>, ApiError> {
    if !metadata.contains_key("x-rustfs-encryption-key") || !metadata.contains_key("x-amz-server-side-encryption") {
        return Ok(None);
    }

    // Safe: presence is guaranteed by the contains_key check above.
    let server_side_encryption = metadata.get("x-amz-server-side-encryption").cloned().unwrap_or_default();

    // Parse metadata - try using service if available, otherwise parse manually
    let (encrypted_data_key, iv, algorithm) = if let Some(service) = get_global_encryption_service().await {
        // Production mode: use service for metadata parsing
        let parsed = service
            .headers_to_metadata(metadata)
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to parse encryption metadata: {e}"))))?;

        if parsed.iv.len() != 12 {
            return Err(ApiError::from(StorageError::other("Invalid encryption nonce length; expected 12 bytes")));
        }

        (parsed.encrypted_data_key, parsed.iv, parsed.algorithm)
    } else {
        // Test mode: parse metadata manually
        let encrypted_key_b64 = metadata
            .get("x-rustfs-encryption-key")
            .ok_or_else(|| ApiError::from(StorageError::other("Missing encrypted key in metadata")))?;
        let encrypted_data_key = BASE64_STANDARD
            .decode(encrypted_key_b64)
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decode encrypted key: {e}"))))?;

        let iv_b64 = metadata
            .get("x-rustfs-encryption-iv")
            .ok_or_else(|| ApiError::from(StorageError::other("Missing IV in metadata")))?;
        let iv = BASE64_STANDARD
            .decode(iv_b64)
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decode IV: {e}"))))?;

        if iv.len() != 12 {
            return Err(ApiError::from(StorageError::other("Invalid encryption nonce length; expected 12 bytes")));
        }

        let algorithm = metadata
            .get("x-rustfs-encryption-algorithm")
            .cloned()
            .unwrap_or_else(|| "AES256".to_string());

        (encrypted_data_key, iv, algorithm)
    };

    // Extract KMS key ID from metadata (optional, used for provider context)
    let kms_key_id = metadata
        .get(INTERNAL_ENCRYPTION_KEY_ID_HEADER)
        .or_else(|| metadata.get("x-amz-server-side-encryption-aws-kms-key-id"))
        .cloned()
        .unwrap_or_else(|| "default".to_string());

    // Use factory pattern to get provider (test or production mode)
    let provider = get_sse_dek_provider().await?;
    let key_bytes = provider
        .decrypt_sse_dek(&encrypted_data_key, &kms_key_id)
        .await
        .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {e}"))))?;

    let mut base_nonce = [0u8; 12];
    base_nonce.copy_from_slice(&iv[..12]);
    let nonce = if let Some(part_num) = part_number {
        derive_part_nonce(base_nonce, part_num)
    } else {
        base_nonce
    };

    let original_size = metadata
        .get("x-rustfs-encryption-original-size")
        .and_then(|s| s.parse::<i64>().ok());

    let encryption_type = match server_side_encryption.as_str() {
        ServerSideEncryption::AES256 => SSEType::SseS3,
        ServerSideEncryption::AWS_KMS => SSEType::SseKms,
        _ => SSEType::SseS3,
    };

    Ok(Some(DecryptionMaterial {
        sse_type: encryption_type,
        server_side_encryption: ServerSideEncryption::from(server_side_encryption),
        kms_key_id: Some(SSEKMSKeyId::from(kms_key_id)),
        algorithm,
        customer_key_md5: None,

        key_bytes,
        nonce,
        original_size,

        is_multipart: false,
        parts: Vec::new(),
    }))
}

// ============================================================================
// Legacy Types (for backward compatibility)
// ============================================================================

/// Validated SSE-C parameters
#[derive(Debug, Clone)]
pub struct ValidatedSsecParams {
    /// Encryption algorithm (always "AES256" for SSE-C)
    pub algorithm: SSECustomerAlgorithm,
    /// Decoded encryption key bytes (32 bytes for AES-256)
    pub key_bytes: [u8; 32],
    /// Base64-encoded MD5 of the key
    pub key_md5: SSECustomerKeyMD5,
}

/// SSE-C parameters from client request
#[derive(Debug, Clone)]
pub struct SsecParams {
    /// Encryption algorithm
    pub algorithm: SSECustomerAlgorithm,
    /// Base64-encoded encryption key
    pub key: SSECustomerKey,
    /// Base64-encoded MD5 of the key
    pub key_md5: SSECustomerKeyMD5,
}

// ============================================================================
// SSE DEK Provider Abstraction (Factory Pattern)
// ============================================================================

/// Trait for SSE data encryption key management
/// Abstracts the source of encryption keys (KMS, test provider, etc.)
#[async_trait]
pub trait SseDekProvider: Send + Sync {
    /// Generate an SSE data encryption key
    async fn generate_sse_dek(&self, bucket: &str, key: &str, kms_key_id: &str) -> Result<(DataKey, Vec<u8>), ApiError>;

    /// Decrypt an SSE data encryption key (returns only plaintext key, nonce should be read from metadata)
    async fn decrypt_sse_dek(&self, encrypted_dek: &[u8], kms_key_id: &str) -> Result<[u8; 32], ApiError>;
}

// ============================================================================
// Production KMS-backed DEK Provider
// ============================================================================

/// Production KMS-backed DEK provider
/// Resolves the latest global ObjectEncryptionService on each call.
struct KmsSseDekProvider;

impl KmsSseDekProvider {
    /// Create a new KMS-backed provider
    pub async fn new() -> Result<Self, ApiError> {
        Self::current_service()
            .await
            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
        Ok(Self)
    }

    async fn current_service() -> Option<Arc<rustfs_kms::service::ObjectEncryptionService>> {
        get_global_encryption_service().await
    }
}

#[async_trait]
impl SseDekProvider for KmsSseDekProvider {
    async fn generate_sse_dek(&self, bucket: &str, key: &str, kms_key_id: &str) -> Result<(DataKey, Vec<u8>), ApiError> {
        let context = ObjectEncryptionContext::new(bucket.to_string(), key.to_string());

        let kms_key_option = Some(kms_key_id.to_string());
        let service = Self::current_service()
            .await
            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
        let (data_key, encrypted_data_key) = service
            .create_data_key(&kms_key_option, &context)
            .await
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to create data key: {}", e))))?;

        Ok((data_key, encrypted_data_key))
    }

    async fn decrypt_sse_dek(&self, encrypted_dek: &[u8], _kms_key_id: &str) -> Result<[u8; 32], ApiError> {
        // Create a minimal context for decryption
        let context = ObjectEncryptionContext::new("".to_string(), "".to_string());
        let service = Self::current_service()
            .await
            .ok_or_else(|| ApiError::from(StorageError::other("KMS encryption service is not initialized")))?;
        let data_key = service
            .decrypt_data_key(encrypted_dek, &context)
            .await
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt data key: {}", e))))?;

        Ok(data_key.plaintext_key)
    }
}

// ============================================================================
// Test/Simple DEK Provider
// ============================================================================

/// Simple SSE DEK provider for testing purposes
///
/// This provider reads a single 32-byte customer master key (CMK) from the
/// `__RUSTFS_SSE_SIMPLE_CMK` environment variable. The key must be base64-encoded.
///
/// # Environment Variable Format
///
/// ```text
/// __RUSTFS_SSE_SIMPLE_CMK=<base64_encoded_32_byte_key>
/// ```
///
/// Example:
/// ```bash
/// export __RUSTFS_SSE_SIMPLE_CMK="AKHul86TBMMJ3+VrGlh9X3dHJsOtSXOXHOODPwmAnOo="
/// ```
///
/// # Key Generation
///
/// Use the provided script to generate a valid key:
/// ```bash
/// # Windows
/// .\scripts\generate-sse-keys.ps1
///
/// # Linux/Unix/macOS
/// ./scripts/generate-sse-keys.sh
/// ```
pub(crate) struct TestSseDekProvider {
    master_key: [u8; 32],
}

impl TestSseDekProvider {
    /// Create a SimpleSseDekProvider with a predefined key (for testing)
    #[cfg(test)]
    pub fn new_with_key(master_key: [u8; 32]) -> Self {
        Self { master_key }
    }

    pub fn new() -> Self {
        let cmk_value = std::env::var("__RUSTFS_SSE_SIMPLE_CMK").unwrap_or_else(|_| "".to_string());

        let master_key = if !cmk_value.is_empty() {
            match BASE64_STANDARD.decode(cmk_value.trim()) {
                Ok(v) => {
                    let decoded_len = v.len();
                    match v.try_into() {
                        Ok(arr) => {
                            tracing::info!("Successfully loaded SSE master key (32 bytes)");
                            arr
                        }
                        Err(_) => {
                            tracing::error!("Failed to load master key: decoded key is not 32 bytes (got {decoded_len} bytes)");
                            [0u8; 32]
                        }
                    }
                }
                Err(e) => {
                    tracing::error!("Failed to load master key: invalid base64 encoding: {e}");
                    [0u8; 32]
                }
            }
        } else {
            [0u8; 32]
        };

        if master_key == [0u8; 32] {
            tracing::error!(
                "No valid SSE master key loaded. Set __RUSTFS_SSE_SIMPLE_CMK environment variable to a base64-encoded 32-byte key."
            );
            std::process::exit(1);
        }

        Self { master_key }
    }

    /// Create a local SSE DEK provider for SSE-S3 when KMS is not configured.
    /// Uses RUSTFS_SSE_S3_MASTER_KEY (base64 32-byte) if set; otherwise a built-in default.
    /// Allows PUT/GET to work without KMS (backward compatible).
    pub fn new_for_local_sse() -> Self {
        let master_key = match get_env_opt_str("RUSTFS_SSE_S3_MASTER_KEY") {
            Some(v) if !v.trim().is_empty() => match BASE64_STANDARD.decode(v.trim()) {
                Ok(decoded) if decoded.len() == 32 => {
                    let mut arr = [0u8; 32];
                    arr.copy_from_slice(&decoded[..32]);
                    tracing::info!("Using RUSTFS_SSE_S3_MASTER_KEY for SSE-S3 (KMS not configured)");
                    arr
                }
                _ => {
                    tracing::warn!("RUSTFS_SSE_S3_MASTER_KEY invalid (expected base64 32 bytes); using default for SSE-S3");
                    [0u8; 32]
                }
            },
            _ => {
                tracing::debug!(
                    "KMS not configured; using built-in default key for SSE-S3 (set RUSTFS_SSE_S3_MASTER_KEY for production)"
                );
                [0u8; 32]
            }
        };
        Self { master_key }
    }

    // Simple encryption of DEK
    pub(crate) fn encrypt_dek(dek: [u8; 32], cmk_value: [u8; 32]) -> Result<String, ApiError> {
        // Use AES-256-GCM to encrypt DEK
        let key = Key::<Aes256Gcm>::from(cmk_value);

        let cipher = Aes256Gcm::new(&key);
        let nonce = Nonce::from([0u8; 12]);
        let ciphertext = cipher
            .encrypt(&nonce, dek.as_slice())
            .map_err(|_| ApiError::from(StorageError::other("Failed to encrypt DEK")))?;

        // nonce:ciphertext
        Ok(format!("{}:{}", BASE64_STANDARD.encode(nonce), BASE64_STANDARD.encode(ciphertext)))
    }

    // Simple decryption of DEK
    pub(crate) fn decrypt_dek(encrypted_dek: &str, cmk_value: [u8; 32]) -> Result<[u8; 32], ApiError> {
        let parts: Vec<&str> = encrypted_dek.split(':').collect();
        if parts.len() != 2 {
            return Err(ApiError::from(StorageError::other("Invalid encrypted DEK format")));
        }

        let nonce_vec = BASE64_STANDARD
            .decode(parts[0])
            .map_err(|_| ApiError::from(StorageError::other("Invalid nonce format")))?;
        let ciphertext = BASE64_STANDARD
            .decode(parts[1])
            .map_err(|_| ApiError::from(StorageError::other("Invalid ciphertext format")))?;

        let key = Key::<Aes256Gcm>::from(cmk_value);
        let cipher = Aes256Gcm::new(&key);

        let nonce_array: [u8; 12] = nonce_vec
            .try_into()
            .map_err(|_| ApiError::from(StorageError::other("Invalid nonce length")))?;
        let nonce = Nonce::from(nonce_array);

        let plaintext = cipher
            .decrypt(&nonce, ciphertext.as_slice())
            .map_err(|e| ApiError::from(StorageError::other(format!("Failed to decrypt DEK: {e}"))))?;

        let dek: [u8; 32] = plaintext
            .try_into()
            .map_err(|_| ApiError::from(StorageError::other("Decrypted DEK has invalid length")))?;

        Ok(dek)
    }
}

#[async_trait]
impl SseDekProvider for TestSseDekProvider {
    async fn generate_sse_dek(&self, _bucket: &str, _key: &str, _kms_key_id: &str) -> Result<(DataKey, Vec<u8>), ApiError> {
        // Generate a 32-byte array as data key
        let mut dek = [0u8; 32];
        rand::rng().fill_bytes(&mut dek);

        // Generate a 12-byte array as IV
        let mut nonce = [0u8; 12];
        rand::rng().fill_bytes(&mut nonce);

        // Encrypt data key with master key
        let encrypted_dek = Self::encrypt_dek(dek, self.master_key)?;

        // Return data key and IV
        Ok((
            DataKey {
                plaintext_key: dek,
                nonce,
            },
            encrypted_dek.into_bytes(),
        ))
    }

    async fn decrypt_sse_dek(&self, encrypted_dek: &[u8], _kms_key_id: &str) -> Result<[u8; 32], ApiError> {
        // Decrypt data key with master key
        let encrypted_dek_str = std::str::from_utf8(encrypted_dek)
            .map_err(|_| ApiError::from(StorageError::other("Invalid UTF-8 in encrypted DEK")))?;
        let dek = Self::decrypt_dek(encrypted_dek_str, self.master_key)?;
        Ok(dek)
    }
}

// ============================================================================
// Factory Function for SSE DEK Provider
// ============================================================================

/// Global SSE DEK provider cache
static GLOBAL_SSE_DEK_PROVIDER: OnceLock<Arc<dyn SseDekProvider>> = OnceLock::new();

/// Get or initialize the global SSE DEK provider
///
/// Factory function that automatically selects the appropriate provider:
/// - If `__RUSTFS_SSE_SIMPLE_CMK` environment variable exists: use SimpleSseDekProvider (test mode)
/// - Otherwise: use KmsSseDekProvider (production mode with real KMS)
///
/// # Returns
/// Arc to the global SSE DEK provider instance
///
/// # Example
/// ```rust,ignore
/// let provider = get_sse_dek_provider().await?;
/// let (data_key, encrypted_dek) = provider
///     .generate_sse_dek("bucket", "key", "kms-key-id")
///     .await?;
/// ```
pub async fn get_sse_dek_provider() -> Result<Arc<dyn SseDekProvider>, ApiError> {
    // Check if already initialized
    if let Some(provider) = GLOBAL_SSE_DEK_PROVIDER.get() {
        return Ok(provider.clone());
    }

    // Determine provider: KMS when available, else test env, else local SSE-S3 fallback (no KMS)
    let provider: Arc<dyn SseDekProvider> = if get_global_encryption_service().await.is_some() {
        debug!("Using KmsSseDekProvider (KMS configured)");
        Arc::new(KmsSseDekProvider::new().await?)
    } else if std::env::var("__RUSTFS_SSE_SIMPLE_CMK").is_ok() {
        debug!("Using SimpleSseDekProvider (test mode) based on __RUSTFS_SSE_SIMPLE_CMK");
        Arc::new(TestSseDekProvider::new())
    } else {
        debug!("Using local SSE-S3 provider (KMS not configured)");
        Arc::new(TestSseDekProvider::new_for_local_sse())
    };

    // Store in global cache
    GLOBAL_SSE_DEK_PROVIDER
        .set(provider.clone())
        .map_err(|_| ApiError::from(StorageError::other("Failed to initialize global SSE DEK provider (already set)")))?;

    Ok(provider)
}

/// Reset the global SSE DEK provider (for testing only)
///
/// Note: OnceLock doesn't support reset in stable Rust.
/// Tests should set environment variables before first call to `get_sse_dek_provider()`.
#[cfg(test)]
#[allow(dead_code)]
pub fn reset_sse_dek_provider() {
    // OnceLock doesn't support reset - this is a documentation placeholder
    // Consider using arc_swap::ArcSwap if runtime reset is needed
}

// ============================================================================
// Legacy Functions (SSE-S3 / SSE-KMS)
// ============================================================================

/// Check if the server_side_encryption is a managed SSE type (SSE-S3 or SSE-KMS)
#[inline]
pub fn is_managed_sse(server_side_encryption: &ServerSideEncryption) -> bool {
    matches!(server_side_encryption.as_str(), "AES256" | "aws:kms")
}

/// Strip managed encryption metadata from object metadata
///
/// Removes all managed SSE-related headers before returning object metadata to client.
/// This is necessary because encryption is transparent to S3 clients.
pub fn strip_managed_encryption_metadata(metadata: &mut HashMap<String, String>) {
    const KEYS: [&str; 7] = [
        "x-amz-server-side-encryption",
        "x-amz-server-side-encryption-aws-kms-key-id",
        "x-rustfs-encryption-iv",
        "x-rustfs-encryption-tag",
        "x-rustfs-encryption-key",
        "x-rustfs-encryption-context",
        "x-rustfs-encryption-original-size",
    ];

    for key in KEYS.iter() {
        metadata.remove(*key);
    }
}

// ============================================================================
// Multipart Encryption Support
// ============================================================================

pub fn derive_part_nonce(base: [u8; 12], part_number: usize) -> [u8; 12] {
    derive_nonce_offset(base, 4, part_number)
}

#[cfg(test)]
fn derive_legacy_part_nonce(base: [u8; 12], part_number: usize) -> [u8; 12] {
    derive_nonce_offset(base, 8, part_number)
}

fn derive_nonce_offset(mut base: [u8; 12], start: usize, offset: usize) -> [u8; 12] {
    let current = u32::from_be_bytes([base[start], base[start + 1], base[start + 2], base[start + 3]]);
    let incremented = current.wrapping_add(offset as u32);
    base[start..start + 4].copy_from_slice(&incremented.to_be_bytes());
    base
}

pub(crate) async fn decrypt_multipart_managed_stream<R>(
    encrypted_stream: R,
    parts: &[ObjectPartInfo],
    key_bytes: [u8; 32],
    base_nonce: [u8; 12],
) -> Result<(DynReader, i64), StorageError>
where
    R: ReadStream + 'static,
{
    let total_plain_size = parts
        .iter()
        .map(|part| {
            if part.actual_size > 0 {
                part.actual_size
            } else {
                part.size as i64
            }
        })
        .sum();

    let multipart_parts = parts.iter().map(|part| part.number).collect();
    let reader = boxed_reader(DecryptReader::new_multipart(
        wrap_reader(encrypted_stream),
        key_bytes,
        base_nonce,
        multipart_parts,
    ));

    Ok((reader, total_plain_size))
}

// ============================================================================
// SSE-C Functions
// ============================================================================

/// Validate SSE-C parameters from client request
///
/// Validates:
/// 1. Algorithm is "AES256"
/// 2. Key is valid Base64 and exactly 32 bytes
/// 3. MD5 hash matches the key
///
/// # Returns
/// `ValidatedSsecParams` with decoded key bytes
pub fn validate_ssec_params(params: SsecParams) -> Result<ValidatedSsecParams, ApiError> {
    if !SUPPORT_SSE_ALGORITHMS.contains(&params.algorithm.as_str()) {
        return Err(ssec_invalid_request(&format!(
            "Unsupported SSE-C algorithm: {}. Only {} is supported.",
            params.algorithm, DEFAULT_SSE_ALGORITHM
        )));
    }

    let key_bytes = BASE64_STANDARD.decode(&params.key).map_err(|e| {
        error!("Failed to decode SSE-C key: {}", e);
        ssec_invalid_request("Invalid SSE-C key: not valid Base64.")
    })?;

    if key_bytes.len() != 32 {
        return Err(ssec_invalid_request(&format!(
            "SSE-C key must be 32 bytes (256 bits), got {} bytes.",
            key_bytes.len()
        )));
    }

    let computed_md5 = BASE64_STANDARD.encode(md5::compute(&key_bytes).0);
    if computed_md5 != params.key_md5 {
        error!("SSE-C key MD5 mismatch: expected '{}', got '{}'", params.key_md5, computed_md5);
        return Err(ssec_invalid_request(
            "The calculated MD5 hash of the key did not match the hash that was provided.",
        ));
    }

    let key_array: [u8; 32] = key_bytes
        .try_into()
        .map_err(|_| ssec_invalid_request("SSE-C key must be exactly 32 bytes."))?;

    Ok(ValidatedSsecParams {
        algorithm: params.algorithm,
        key_bytes: key_array,
        key_md5: params.key_md5,
    })
}

/// Generate deterministic nonce for SSE-C encryption
///
/// The nonce is derived from the bucket and key to ensure:
/// 1. Same object always gets the same nonce (required for SSE-C)
/// 2. Different objects get different nonces
pub fn generate_ssec_nonce(bucket: &str, key: &str) -> [u8; 12] {
    let nonce_source = format!("{bucket}-{key}");
    let nonce_hash = md5::compute(nonce_source.as_bytes());
    let mut nonce = [0u8; 12];
    nonce.copy_from_slice(&nonce_hash.0[..12]);
    nonce
}

/// Verify SSE-C key matches the stored metadata.
///
/// Used during GetObject/HeadObject to ensure the client provided the correct key.
/// Returns 400 InvalidRequest on mismatch, consistent with AWS S3 behavior.
pub fn verify_ssec_key_match(provided_md5: &str, stored_md5: Option<&String>) -> Result<(), ApiError> {
    match stored_md5 {
        Some(stored) if stored == provided_md5 => Ok(()),
        Some(_) => Err(ssec_invalid_request(
            "The provided encryption parameters did not match the ones used originally to encrypt the object.",
        )),
        None => Err(ssec_invalid_request("Object has no stored SSE-C key metadata.")),
    }
}

/// Validate that the SSE-C headers required for reading an SSE-C encrypted object
/// are present in the request. This is used by HeadObject which does not decrypt
/// the data but still must verify the caller holds the correct key.
///
/// Performs full validation: decodes the customer key, recomputes its MD5,
/// verifies the client-provided MD5 header matches the key, then compares
/// the computed MD5 against the stored metadata. This prevents a client from
/// bypassing validation by guessing/obtaining only the stored MD5 without
/// possessing the actual encryption key.
///
/// Returns `Ok(())` if either the object is not SSE-C encrypted, or valid SSE-C
/// headers are provided and the key matches. Returns 400 InvalidRequest otherwise.
pub fn validate_ssec_for_read(
    metadata: &HashMap<String, String>,
    sse_customer_key: Option<&SSECustomerKey>,
    sse_customer_key_md5: Option<&SSECustomerKeyMD5>,
) -> Result<(), ApiError> {
    let stored_algorithm = metadata.get("x-amz-server-side-encryption-customer-algorithm");
    if stored_algorithm.is_none() {
        return Ok(());
    }

    let (key, key_md5) = match (sse_customer_key, sse_customer_key_md5) {
        (Some(k), Some(md5)) => (k, md5),
        _ => {
            return Err(ssec_invalid_request(
                "The object was stored using a form of Server Side Encryption. \
                 The correct parameters must be provided to retrieve the object.",
            ));
        }
    };

    // Full param validation: decode key, verify 32 bytes, recompute MD5
    // from actual key bytes and compare to the client-provided MD5 header.
    let algorithm = stored_algorithm.cloned().unwrap_or_else(|| DEFAULT_SSE_ALGORITHM.to_string());
    let validated = validate_ssec_params(SsecParams {
        algorithm,
        key: key.to_string(),
        key_md5: key_md5.clone(),
    })?;

    let stored_md5 = metadata.get("x-amz-server-side-encryption-customer-key-md5");
    verify_ssec_key_match(&validated.key_md5, stored_md5)
}

/// Build an `ApiError` with `InvalidRequest` (HTTP 400) for SSE-C related errors.
fn ssec_invalid_request(message: &str) -> ApiError {
    ApiError {
        code: S3ErrorCode::InvalidRequest,
        message: message.to_string(),
        source: None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use http::HeaderValue;

    #[test]
    fn test_extract_ssec_params_from_headers() {
        let mut headers = http::HeaderMap::new();
        let (algo, key, md5) = extract_ssec_params_from_headers(&headers).unwrap();
        assert!(algo.is_none());
        assert!(key.is_none());
        assert!(md5.is_none());

        headers.insert("x-amz-server-side-encryption-customer-algorithm", HeaderValue::from_static("AES256"));
        let (algo, key, md5) = extract_ssec_params_from_headers(&headers).unwrap();
        assert_eq!(algo.as_deref(), Some("AES256"));
        assert!(key.is_none());
        assert!(md5.is_none());

        headers.insert(
            "x-amz-server-side-encryption-customer-key",
            HeaderValue::from_static("pO3upElrwuEXSoFwCfnZPdSsmt/xWeFa0N9KgDijwVs="),
        );
        headers.insert(
            "x-amz-server-side-encryption-customer-key-md5",
            HeaderValue::from_static("DWygnHRtgiJ77HCm+1rvHw=="),
        );
        let (algo, key, md5) = extract_ssec_params_from_headers(&headers).unwrap();
        assert_eq!(algo.as_deref(), Some("AES256"));
        assert!(key.is_some());
        assert!(md5.is_some());
    }

    #[test]
    fn test_extract_ssec_params_from_headers_rejects_invalid_utf8() {
        let mut headers = http::HeaderMap::new();
        // Header value with invalid UTF-8; to_str() will fail
        let invalid_utf8 = HeaderValue::from_bytes(b"invalid-\x80-utf8").unwrap();
        headers.insert("x-amz-server-side-encryption-customer-algorithm", invalid_utf8);
        let result = extract_ssec_params_from_headers(&headers);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_extract_server_side_encryption_from_headers_rejects_invalid_utf8() {
        let mut headers = http::HeaderMap::new();
        let invalid_utf8 = HeaderValue::from_bytes(b"aes:kms-\x80-invalid").unwrap();
        headers.insert("x-amz-server-side-encryption", invalid_utf8);
        let result = extract_server_side_encryption_from_headers(&headers);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_validate_sse_headers_for_write_rejects_algorithm_without_key() {
        let algorithm = SSECustomerAlgorithm::from("AES256".to_string());
        let result = validate_sse_headers_for_write(
            None,
            None,
            Some(&algorithm),
            None,
            None,
            true, // PutObject requires all three
        );
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_sse_headers_for_write_rejects_algorithm_and_key_without_md5() {
        let algorithm = SSECustomerAlgorithm::from("AES256".to_string());
        let key = SSECustomerKey::from("pO3upElrwuEXSoFwCfnZPdSsmt/xWeFa0N9KgDijwVs=".to_string());
        let result = validate_sse_headers_for_write(
            None,
            None,
            Some(&algorithm),
            Some(&key),
            None,
            true, // PutObject requires all three
        );
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_sse_headers_for_write_rejects_invalid_sse_algorithm() {
        let bad_sse = ServerSideEncryption::from_static("aes:kms");
        let result = validate_sse_headers_for_write(Some(&bad_sse), None, None, None, None, true);
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_validate_sse_headers_for_write_rejects_ssec_with_managed_sse() {
        let algorithm = SSECustomerAlgorithm::from("AES256".to_string());
        let key = SSECustomerKey::from("pO3upElrwuEXSoFwCfnZPdSsmt/xWeFa0N9KgDijwVs=".to_string());
        let key_md5 = SSECustomerKeyMD5::from("DWygnHRtgiJ77HCm+1rvHw==".to_string());
        let server_side_encryption = ServerSideEncryption::from_static(ServerSideEncryption::AES256);
        let result = validate_sse_headers_for_write(
            Some(&server_side_encryption),
            None,
            Some(&algorithm),
            Some(&key),
            Some(&key_md5),
            true,
        );
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_is_managed_sse() {
        assert!(is_managed_sse(&ServerSideEncryption::from_static("AES256")));
        assert!(is_managed_sse(&ServerSideEncryption::from_static("aws:kms")));
    }

    #[test]
    fn test_derive_part_nonce() {
        let base = [1, 2, 3, 4, 5, 6, 7, 8, 0, 0, 0, 10];
        let part1 = derive_part_nonce(base, 1);
        let part2 = derive_part_nonce(base, 2);

        assert_eq!(&base[..4], &part1[..4]);
        assert_eq!(&base[8..], &part1[8..]);
        assert_ne!(&base[4..8], &part1[4..8]);
        assert_ne!(&part1[4..8], &part2[4..8]);
    }

    #[tokio::test]
    async fn test_decrypt_multipart_managed_stream_accepts_legacy_part_nonce_layout() {
        use std::io::Cursor;
        use tokio::io::AsyncReadExt;

        let key_bytes = [7u8; 32];
        let base_nonce = [3u8; 12];

        let part_one_plaintext = vec![0x11; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 19];
        let part_two_plaintext = vec![0x22; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 37];

        let part_one_nonce = derive_legacy_part_nonce(base_nonce, 1);
        let part_two_nonce = derive_legacy_part_nonce(base_nonce, 2);

        let first_part = {
            let mut buf = Vec::new();
            EncryptReader::new(Cursor::new(part_one_plaintext.clone()), key_bytes, part_one_nonce)
                .read_to_end(&mut buf)
                .await
                .unwrap();
            buf
        };
        let second_part = {
            let mut buf = Vec::new();
            EncryptReader::new(Cursor::new(part_two_plaintext.clone()), key_bytes, part_two_nonce)
                .read_to_end(&mut buf)
                .await
                .unwrap();
            buf
        };

        let mut encrypted_stream = Vec::with_capacity(first_part.len() + second_part.len());
        encrypted_stream.extend_from_slice(&first_part);
        encrypted_stream.extend_from_slice(&second_part);

        let parts = vec![
            ObjectPartInfo {
                number: 1,
                size: first_part.len(),
                actual_size: part_one_plaintext.len() as i64,
                ..Default::default()
            },
            ObjectPartInfo {
                number: 2,
                size: second_part.len(),
                actual_size: part_two_plaintext.len() as i64,
                ..Default::default()
            },
        ];

        let (mut decrypted_reader, plaintext_size) =
            decrypt_multipart_managed_stream(Cursor::new(encrypted_stream), &parts, key_bytes, base_nonce)
                .await
                .unwrap();

        let mut decrypted = Vec::new();
        decrypted_reader.read_to_end(&mut decrypted).await.unwrap();

        let mut expected = part_one_plaintext;
        expected.extend_from_slice(&part_two_plaintext);

        assert_eq!(plaintext_size, expected.len() as i64);
        assert_eq!(decrypted, expected);
    }

    #[tokio::test]
    async fn test_decrypt_multipart_managed_stream_supports_current_nonce_layout() {
        use std::io::Cursor;
        use tokio::io::AsyncReadExt;

        let key_bytes = [9u8; 32];
        let base_nonce = [5u8; 12];

        let part_one_plaintext = vec![0x33; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 11];
        let part_two_plaintext = vec![0x44; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE * 2 + 7];
        let part_one_nonce = derive_part_nonce(base_nonce, 1);
        let part_two_nonce = derive_part_nonce(base_nonce, 2);

        let first_part = {
            let mut buf = Vec::new();
            EncryptReader::new(Cursor::new(part_one_plaintext.clone()), key_bytes, part_one_nonce)
                .read_to_end(&mut buf)
                .await
                .unwrap();
            buf
        };
        let second_part = {
            let mut buf = Vec::new();
            EncryptReader::new(Cursor::new(part_two_plaintext.clone()), key_bytes, part_two_nonce)
                .read_to_end(&mut buf)
                .await
                .unwrap();
            buf
        };

        let mut encrypted_stream = Vec::with_capacity(first_part.len() + second_part.len());
        encrypted_stream.extend_from_slice(&first_part);
        encrypted_stream.extend_from_slice(&second_part);

        let parts = vec![
            ObjectPartInfo {
                number: 1,
                size: first_part.len(),
                actual_size: part_one_plaintext.len() as i64,
                ..Default::default()
            },
            ObjectPartInfo {
                number: 2,
                size: second_part.len(),
                actual_size: part_two_plaintext.len() as i64,
                ..Default::default()
            },
        ];

        let (mut decrypted_reader, plaintext_size) =
            decrypt_multipart_managed_stream(Cursor::new(encrypted_stream), &parts, key_bytes, base_nonce)
                .await
                .unwrap();

        let mut decrypted = Vec::new();
        decrypted_reader.read_to_end(&mut decrypted).await.unwrap();

        let mut expected = part_one_plaintext;
        expected.extend_from_slice(&part_two_plaintext);

        assert_eq!(plaintext_size, expected.len() as i64);
        assert_eq!(decrypted, expected);
    }

    #[tokio::test]
    async fn test_decrypt_multipart_managed_stream_uses_actual_part_numbers_for_nonce_derivation() {
        use std::io::Cursor;
        use tokio::io::AsyncReadExt;

        let key_bytes = [0xAu8; 32];
        let base_nonce = [0xBu8; 12];

        let part_three_plaintext = vec![0x55; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 13];
        let part_five_plaintext = vec![0x66; rustfs_rio::DEFAULT_ENCRYPTION_BLOCK_SIZE + 29];

        let part_three_nonce = derive_part_nonce(base_nonce, 3);
        let part_five_nonce = derive_part_nonce(base_nonce, 5);

        let encrypted_three = {
            let mut buf = Vec::new();
            EncryptReader::new(Cursor::new(part_three_plaintext.clone()), key_bytes, part_three_nonce)
                .read_to_end(&mut buf)
                .await
                .unwrap();
            buf
        };
        let encrypted_five = {
            let mut buf = Vec::new();
            EncryptReader::new(Cursor::new(part_five_plaintext.clone()), key_bytes, part_five_nonce)
                .read_to_end(&mut buf)
                .await
                .unwrap();
            buf
        };

        let mut encrypted_stream = Vec::with_capacity(encrypted_three.len() + encrypted_five.len());
        encrypted_stream.extend_from_slice(&encrypted_three);
        encrypted_stream.extend_from_slice(&encrypted_five);

        let parts = vec![
            ObjectPartInfo {
                number: 3,
                size: encrypted_three.len(),
                actual_size: part_three_plaintext.len() as i64,
                ..Default::default()
            },
            ObjectPartInfo {
                number: 5,
                size: encrypted_five.len(),
                actual_size: part_five_plaintext.len() as i64,
                ..Default::default()
            },
        ];

        let (mut decrypted_reader, plaintext_size) =
            decrypt_multipart_managed_stream(Cursor::new(encrypted_stream), &parts, key_bytes, base_nonce)
                .await
                .unwrap();

        let mut decrypted = Vec::new();
        decrypted_reader.read_to_end(&mut decrypted).await.unwrap();

        let mut expected = part_three_plaintext;
        expected.extend_from_slice(&part_five_plaintext);

        assert_eq!(plaintext_size, expected.len() as i64);
        assert_eq!(decrypted, expected);
    }

    #[test]
    fn test_generate_ssec_nonce() {
        let nonce1 = generate_ssec_nonce("bucket1", "key1");
        let nonce2 = generate_ssec_nonce("bucket1", "key1");
        let nonce3 = generate_ssec_nonce("bucket1", "key2");

        // Same inputs should produce same nonce
        assert_eq!(nonce1, nonce2);

        // Different inputs should produce different nonce
        assert_ne!(nonce1, nonce3);

        // Nonce should be exactly 12 bytes
        assert_eq!(nonce1.len(), 12);
    }

    #[test]
    fn test_validate_ssec_params_success() {
        let key = BASE64_STANDARD.encode([42u8; 32]);
        let key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 32]).0);

        let params = SsecParams {
            algorithm: "AES256".to_string(),
            key,
            key_md5,
        };

        let result = validate_ssec_params(params);
        assert!(result.is_ok());
        let validated = result.unwrap();
        assert_eq!(validated.key_bytes, [42u8; 32]);
    }

    #[test]
    fn test_validate_ssec_params_wrong_algorithm() {
        let key = BASE64_STANDARD.encode([42u8; 32]);
        let key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 32]).0);

        let params = SsecParams {
            algorithm: "AES128".to_string(), // Wrong algorithm
            key,
            key_md5,
        };

        let result = validate_ssec_params(params);
        assert!(result.is_err());
    }

    #[test]
    fn test_validate_ssec_params_wrong_key_length() {
        let key = BASE64_STANDARD.encode([42u8; 16]); // Only 16 bytes
        let key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 16]).0);

        let params = SsecParams {
            algorithm: "AES256".to_string(),
            key,
            key_md5,
        };

        let result = validate_ssec_params(params);
        assert!(result.is_err());
    }

    #[test]
    fn test_validate_ssec_params_wrong_md5() {
        let key = BASE64_STANDARD.encode([42u8; 32]);
        let key_md5 = BASE64_STANDARD.encode([99u8; 16]); // Wrong MD5

        let params = SsecParams {
            algorithm: "AES256".to_string(),
            key,
            key_md5,
        };

        let result = validate_ssec_params(params);
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_sse_encryption_rejects_partial_ssec_headers() {
        let bucket = "test-bucket";
        let key = "test-key";
        let sse_key = BASE64_STANDARD.encode([42u8; 32]);
        let sse_key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 32]).0);
        let content_size = 1024;

        let request_missing_md5 = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: Some("AES256".to_string()),
            sse_customer_key: Some(sse_key.clone()),
            sse_customer_key_md5: None,
            content_size,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let err = sse_encryption(request_missing_md5).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);

        let request_missing_key = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: Some("AES256".to_string()),
            sse_customer_key: None,
            sse_customer_key_md5: Some(sse_key_md5.clone()),
            content_size,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let err = sse_encryption(request_missing_key).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);

        let request_missing_algorithm = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: None,
            sse_customer_key: Some(sse_key),
            sse_customer_key_md5: Some(sse_key_md5),
            content_size,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let err = sse_encryption(request_missing_algorithm).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[tokio::test]
    async fn test_sse_prepare_encryption_rejects_partial_ssec_headers() {
        let bucket = "test-bucket";
        let key = "test-key";
        let sse_key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 32]).0);

        let request_missing_algorithm = PrepareEncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: None,
            sse_customer_key_md5: Some(sse_key_md5),
        };

        let err = sse_prepare_encryption(request_missing_algorithm).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_sse_headers_for_write_allows_aws_kms_without_key_id() {
        let server_side_encryption: ServerSideEncryption = "aws:kms".to_string().into();

        let result = validate_sse_headers_for_write(Some(&server_side_encryption), None, None, None, None, true);

        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_sse_prepare_encryption_allows_ssec_headers_without_customer_key() {
        let bucket = "test-bucket";
        let key = "test-key";
        let sse_key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 32]).0);

        let request = PrepareEncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: Some("AES256".to_string()),
            sse_customer_key_md5: Some(sse_key_md5),
        };

        let material = sse_prepare_encryption(request)
            .await
            .expect("prepare should accept ssec headers");
        assert!(material.is_some());
        let metadata = &material.expect("ssec metadata should be generated").metadata;
        assert_eq!(metadata.get("x-amz-server-side-encryption").unwrap(), "AES256");
        assert_eq!(metadata.get("x-amz-server-side-encryption-customer-algorithm").unwrap(), "AES256");
    }

    #[tokio::test]
    async fn test_sse_encryption_rejects_kms_key_with_invalid_algorithm() {
        let bucket = "test-bucket";
        let key = "test-key";
        let content_size = 1024;

        let request = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: Some("AES256".to_string().into()),
            ssekms_key_id: Some("test-key".to_string()),
            sse_customer_algorithm: None,
            sse_customer_key: None,
            sse_customer_key_md5: None,
            content_size,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let err = sse_encryption(request).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[tokio::test]
    async fn test_sse_encryption_rejects_kms_key_without_algorithm() {
        let bucket = "test-bucket";
        let key = "test-key";
        let content_size = 1024;

        let request = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: Some("test-key".to_string()),
            sse_customer_algorithm: None,
            sse_customer_key: None,
            sse_customer_key_md5: None,
            content_size,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let err = sse_encryption(request).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[tokio::test]
    async fn test_sse_encryption_rejects_conflict_between_kms_and_ssec() {
        let bucket = "test-bucket";
        let key = "test-key";
        let content_size = 1024;
        let sse_key = BASE64_STANDARD.encode([42u8; 32]);
        let sse_key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 32]).0);

        let request = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: Some("aws:kms".to_string().into()),
            ssekms_key_id: Some("test-key".to_string()),
            sse_customer_algorithm: Some("AES256".to_string()),
            sse_customer_key: Some(sse_key),
            sse_customer_key_md5: Some(sse_key_md5),
            content_size,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let err = sse_encryption(request).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_resolve_effective_kms_key_id_ignores_bucket_default_for_explicit_sse_s3() {
        let effective_sse = ServerSideEncryption::from_static(ServerSideEncryption::AES256);

        let kms_key_id = resolve_effective_kms_key_id(Some(&effective_sse), None, || Some("bucket-default".to_string()));

        assert_eq!(kms_key_id, None);
    }

    #[test]
    fn test_resolve_effective_kms_key_id_uses_bucket_default_for_sse_kms() {
        let effective_sse = ServerSideEncryption::from_static(ServerSideEncryption::AWS_KMS);

        let kms_key_id = resolve_effective_kms_key_id(Some(&effective_sse), None, || Some("bucket-default".to_string()));

        assert_eq!(kms_key_id.as_deref(), Some("bucket-default"));
    }

    #[tokio::test]
    async fn test_sse_encryption_persists_aws_kms_header_for_kms_objects() {
        let request = EncryptionRequest {
            bucket: "test-bucket",
            key: "test-key",
            server_side_encryption: Some("aws:kms".to_string().into()),
            ssekms_key_id: Some("test-key".to_string()),
            sse_customer_algorithm: None,
            sse_customer_key: None,
            sse_customer_key_md5: None,
            content_size: 1024,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let material = sse_encryption(request).await.expect("kms encryption should succeed");
        let metadata = material.expect("managed kms encryption should return material").metadata;

        assert_eq!(metadata.get("x-amz-server-side-encryption").map(String::as_str), Some("aws:kms"));
        assert_eq!(
            metadata
                .get("x-amz-server-side-encryption-aws-kms-key-id")
                .map(String::as_str),
            Some("test-key")
        );
    }

    #[tokio::test]
    async fn test_sse_encryption_omits_kms_header_for_sse_s3_objects() {
        let request = EncryptionRequest {
            bucket: "test-bucket",
            key: "test-key",
            server_side_encryption: Some(ServerSideEncryption::from_static(ServerSideEncryption::AES256)),
            ssekms_key_id: None,
            sse_customer_algorithm: None,
            sse_customer_key: None,
            sse_customer_key_md5: None,
            content_size: 1024,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };

        let material = sse_encryption(request).await.expect("sse-s3 encryption should succeed");
        let material = material.expect("managed sse-s3 encryption should return material");

        assert_eq!(material.kms_key_id, None);
        assert_eq!(material.metadata.get("x-amz-server-side-encryption").map(String::as_str), Some("AES256"));
        assert!(!material.metadata.contains_key("x-amz-server-side-encryption-aws-kms-key-id"));
        assert_eq!(
            material.metadata.get(INTERNAL_ENCRYPTION_KEY_ID_HEADER).map(String::as_str),
            Some("default")
        );
    }

    #[test]
    fn test_strip_managed_encryption_metadata() {
        let mut metadata = HashMap::new();
        metadata.insert("x-amz-server-side-encryption".to_string(), "aws:kms".to_string());
        metadata.insert("x-rustfs-encryption-key".to_string(), "encrypted_key".to_string());
        metadata.insert("content-type".to_string(), "text/plain".to_string());

        strip_managed_encryption_metadata(&mut metadata);

        assert!(!metadata.contains_key("x-amz-server-side-encryption"));
        assert!(!metadata.contains_key("x-rustfs-encryption-key"));
        assert!(metadata.contains_key("content-type"));
    }

    #[test]
    fn test_is_multipart_object_treats_single_part_multipart_etag_as_multipart() {
        let metadata = HashMap::from([("etag".to_string(), "0123456789abcdef0123456789abcdef-1".to_string())]);
        let parts = vec![ObjectPartInfo {
            number: 1,
            size: 128,
            actual_size: 64,
            ..Default::default()
        }];

        assert!(is_multipart_object(metadata.get("etag").map(String::as_str), &parts));
    }

    #[test]
    fn test_is_multipart_object_keeps_regular_single_part_object_as_non_multipart() {
        let metadata = HashMap::from([("etag".to_string(), "0123456789abcdef0123456789abcdef".to_string())]);
        let parts = vec![ObjectPartInfo {
            number: 1,
            size: 128,
            actual_size: 64,
            ..Default::default()
        }];

        assert!(!is_multipart_object(metadata.get("etag").map(String::as_str), &parts));
    }

    #[test]
    fn test_verify_ssec_key_match_success() {
        let md5 = "test_md5".to_string();
        let result = verify_ssec_key_match("test_md5", Some(&md5));
        assert!(result.is_ok());
    }

    #[test]
    fn test_verify_ssec_key_match_mismatch() {
        let md5 = "stored_md5".to_string();
        let result = verify_ssec_key_match("provided_md5", Some(&md5));
        assert!(result.is_err());
    }

    #[test]
    fn test_verify_ssec_key_match_no_stored() {
        let result = verify_ssec_key_match("provided_md5", None);
        assert!(result.is_err());
    }

    #[test]
    fn test_upload_part_customer_key_md5_comparison_is_case_sensitive() {
        let mut metadata = HashMap::new();
        metadata.insert(
            "x-amz-server-side-encryption-customer-key-md5".to_string(),
            "AbCdEfGhIjKlMnOpQrStUvWxYz0123456789+/==".to_string(),
        );

        let request = EncryptionRequest {
            bucket: "bucket",
            key: "object",
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: None,
            sse_customer_key: None,
            sse_customer_key_md5: None,
            content_size: 1,
            part_number: Some(1),
            part_key: None,
            part_nonce: None,
        };

        let mismatch = "aBcDeFgHiJkLmNoPqRsTuVwXyZ0123456789+/==".to_string();
        let result = request.check_upload_part_customer_key_md5(&metadata, Some(mismatch));
        assert!(result.is_err());
    }

    #[test]
    fn test_upload_part_customer_key_md5_exact_match() {
        let mut metadata = HashMap::new();
        let md5 = "AbCdEfGhIjKlMnOpQrStUvWxYz0123456789+/==".to_string();
        metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), md5.clone());

        let request = EncryptionRequest {
            bucket: "bucket",
            key: "object",
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: None,
            sse_customer_key: None,
            sse_customer_key_md5: None,
            content_size: 1,
            part_number: Some(1),
            part_key: None,
            part_nonce: None,
        };

        let result = request.check_upload_part_customer_key_md5(&metadata, Some(md5));
        assert!(result.is_ok());
    }

    // ============================================================================
    // Integration Tests - Encrypt/Decrypt with SimpleSseDekProvider
    // ============================================================================

    #[tokio::test]
    async fn test_simple_sse_dek_provider_encrypt_decrypt() {
        use std::io::Cursor;
        use tokio::io::AsyncReadExt;

        // 1. Setup: Create SimpleSseDekProvider with test master key
        let provider = TestSseDekProvider::new_with_key([42u8; 32]);

        // 2. Generate a data encryption key
        let bucket = "test-bucket";
        let key = "test-key";
        let kms_key_id = "default"; // Key ID is ignored in simple provider

        let (data_key, _encrypted_dek) = provider
            .generate_sse_dek(bucket, key, kms_key_id)
            .await
            .expect("Failed to generate DEK");

        // 3. Prepare test data (plaintext)
        let plaintext = b"Hello, World! This is a test message for encryption and decryption.";
        println!("Original plaintext: {:?}", String::from_utf8_lossy(plaintext));
        println!("Plaintext length: {} bytes", plaintext.len());

        // 4. Encrypt with EncryptReader.
        let plaintext_reader = Cursor::new(plaintext.to_vec());
        let mut encrypt_reader = EncryptReader::new(plaintext_reader, data_key.plaintext_key, data_key.nonce);

        // Read encrypted data
        let mut encrypted_data = Vec::new();
        encrypt_reader
            .read_to_end(&mut encrypted_data)
            .await
            .expect("Failed to read encrypted data");

        println!("Encrypted data length: {} bytes", encrypted_data.len());
        println!(
            "First 16 bytes of encrypted data: {:02x?}",
            &encrypted_data[..16.min(encrypted_data.len())]
        );

        // Verify encrypted data is different from plaintext
        assert_ne!(
            &encrypted_data[..plaintext.len()],
            plaintext,
            "Encrypted data should be different from plaintext"
        );

        // 5. Decrypt with DecryptReader.
        let encrypted_reader = Cursor::new(encrypted_data);
        let mut decrypt_reader = DecryptReader::new(encrypted_reader, data_key.plaintext_key, data_key.nonce);

        // Read decrypted data
        let mut decrypted_data = Vec::new();
        decrypt_reader
            .read_to_end(&mut decrypted_data)
            .await
            .expect("Failed to read decrypted data");

        println!("Decrypted data: {:?}", String::from_utf8_lossy(&decrypted_data));
        println!("Decrypted length: {} bytes", decrypted_data.len());

        // 6. Verify decrypted data matches original plaintext
        assert_eq!(decrypted_data, plaintext, "Decrypted data should match original plaintext");

        println!("✅ Encryption/Decryption test passed!");
    }

    #[tokio::test]
    async fn test_simple_sse_dek_provider_encrypt_decrypt_large_data() {
        use std::io::Cursor;
        use tokio::io::AsyncReadExt;

        // 1. Setup: Create SimpleSseDekProvider with test master key
        let provider = TestSseDekProvider::new_with_key([42u8; 32]);

        let bucket = "test-bucket";
        let key = "test-key-large";
        let kms_key_id = "default";

        let (data_key, _encrypted_dek) = provider
            .generate_sse_dek(bucket, key, kms_key_id)
            .await
            .expect("Failed to generate DEK");

        // Create 1MB of test data
        let plaintext_size = 1024 * 1024; // 1MB
        let plaintext: Vec<u8> = (0..plaintext_size).map(|i| (i % 256) as u8).collect();
        println!("Testing with {} bytes of data", plaintext.len());

        // Encrypt.
        let plaintext_reader = Cursor::new(plaintext.clone());
        let mut encrypt_reader = EncryptReader::new(plaintext_reader, data_key.plaintext_key, data_key.nonce);

        let mut encrypted_data = Vec::new();
        encrypt_reader
            .read_to_end(&mut encrypted_data)
            .await
            .expect("Failed to encrypt large data");

        println!("Encrypted {} bytes to {} bytes", plaintext.len(), encrypted_data.len());

        // Decrypt.
        let encrypted_reader = Cursor::new(encrypted_data);
        let mut decrypt_reader = DecryptReader::new(encrypted_reader, data_key.plaintext_key, data_key.nonce);

        let mut decrypted_data = Vec::new();
        decrypt_reader
            .read_to_end(&mut decrypted_data)
            .await
            .expect("Failed to decrypt large data");

        // Verify
        assert_eq!(decrypted_data.len(), plaintext.len(), "Decrypted size should match original");
        assert_eq!(decrypted_data, plaintext, "Decrypted data should match original plaintext");

        println!("✅ Large data encryption/decryption test passed!");
    }

    #[tokio::test]
    async fn test_simple_sse_dek_provider_different_nonces() {
        use std::io::Cursor;
        use tokio::io::AsyncReadExt;

        // 1. Setup: Create SimpleSseDekProvider with test master key
        let provider = TestSseDekProvider::new_with_key([42u8; 32]);

        let bucket = "test-bucket";
        let key = "test-key";
        let kms_key_id = "default";

        // Generate two different keys (with different nonces)
        let (data_key1, _) = provider
            .generate_sse_dek(bucket, key, kms_key_id)
            .await
            .expect("Failed to generate DEK 1");

        let (data_key2, _) = provider
            .generate_sse_dek(bucket, key, kms_key_id)
            .await
            .expect("Failed to generate DEK 2");

        // Verify nonces are different
        assert_ne!(data_key1.nonce, data_key2.nonce, "Different keys should have different nonces");

        // Same plaintext
        let plaintext = b"Same plaintext";

        // Encrypt with first key.
        let reader1 = Cursor::new(plaintext.to_vec());
        let mut encrypt_reader1 = EncryptReader::new(reader1, data_key1.plaintext_key, data_key1.nonce);
        let mut encrypted1 = Vec::new();
        encrypt_reader1.read_to_end(&mut encrypted1).await.unwrap();

        // Encrypt with second key.
        let reader2 = Cursor::new(plaintext.to_vec());
        let mut encrypt_reader2 = EncryptReader::new(reader2, data_key2.plaintext_key, data_key2.nonce);
        let mut encrypted2 = Vec::new();
        encrypt_reader2.read_to_end(&mut encrypted2).await.unwrap();

        // Verify ciphertexts are different (due to different nonces/keys)
        assert_ne!(
            encrypted1, encrypted2,
            "Same plaintext with different nonces should produce different ciphertext"
        );

        println!("✅ Different nonces produce different ciphertext - test passed!");
    }

    #[tokio::test]
    async fn test_simple_sse_dek_provider_decrypt_with_encrypted_dek() {
        use std::io::Cursor;
        use tokio::io::AsyncReadExt;

        // 1. Setup: Create SimpleSseDekProvider with test master key
        let provider = TestSseDekProvider::new_with_key([42u8; 32]);

        let bucket = "test-bucket";
        let key = "test-key";
        let kms_key_id = "default";

        // 1. Generate DEK and get encrypted DEK
        let (data_key, encrypted_dek) = provider
            .generate_sse_dek(bucket, key, kms_key_id)
            .await
            .expect("Failed to generate DEK");

        let original_plaintext_key = data_key.plaintext_key;
        let original_nonce = data_key.nonce;

        // 2. Simulate storing encrypted_dek and nonce in metadata
        // In real scenario, nonce would be stored separately in metadata

        // 3. Later, decrypt the DEK
        let decrypted_plaintext_key = provider
            .decrypt_sse_dek(&encrypted_dek, kms_key_id)
            .await
            .expect("Failed to decrypt DEK");

        // 4. Verify decrypted key matches original
        assert_eq!(
            decrypted_plaintext_key, original_plaintext_key,
            "Decrypted DEK should match original plaintext key"
        );

        // 5. Use decrypted key to encrypt/decrypt data
        let plaintext = b"Test data with decrypted DEK";

        // Encrypt with original key.
        let reader = Cursor::new(plaintext.to_vec());
        let mut encrypt_reader = EncryptReader::new(reader, original_plaintext_key, original_nonce);
        let mut encrypted_data = Vec::new();
        encrypt_reader.read_to_end(&mut encrypted_data).await.unwrap();

        // Decrypt with recovered key (simulating GET operation).
        let reader = Cursor::new(encrypted_data);
        let mut decrypt_reader = DecryptReader::new(
            reader,
            decrypted_plaintext_key,
            original_nonce, // In real scenario, read from metadata
        );
        let mut decrypted_data = Vec::new();
        decrypt_reader.read_to_end(&mut decrypted_data).await.unwrap();

        // Verify
        assert_eq!(decrypted_data, plaintext, "Data decrypted with recovered key should match original");

        println!("✅ Full cycle (generate -> encrypt DEK -> decrypt DEK -> decrypt data) test passed!");
    }

    #[tokio::test]
    async fn test_kms_sse_dek_provider_uses_latest_reconfigured_service() {
        use rustfs_kms::config::KmsConfig;
        use rustfs_kms::types::{CreateKeyRequest, KeyUsage};
        use std::sync::OnceLock;
        use tempfile::TempDir;
        use tokio::sync::Mutex;

        static KMS_TEST_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
        let _guard = KMS_TEST_LOCK.get_or_init(|| Mutex::new(())).lock().await;

        let manager = rustfs_kms::init_global_kms_service_manager();

        let first_dir = TempDir::new().expect("first temp dir");
        manager
            .reconfigure(KmsConfig::local(first_dir.path().to_path_buf()))
            .await
            .expect("first KMS reconfigure should succeed");
        manager
            .get_encryption_service()
            .await
            .expect("first encryption service should exist")
            .create_key(CreateKeyRequest {
                key_name: Some("first-key".to_string()),
                key_usage: KeyUsage::EncryptDecrypt,
                description: None,
                policy: None,
                tags: HashMap::new(),
                origin: None,
            })
            .await
            .expect("first key should be created");

        let provider = KmsSseDekProvider::new().await.expect("provider should initialize");
        provider
            .generate_sse_dek("bucket", "object", "first-key")
            .await
            .expect("provider should use the initial service");

        let second_dir = TempDir::new().expect("second temp dir");
        manager
            .reconfigure(KmsConfig::local(second_dir.path().to_path_buf()))
            .await
            .expect("second KMS reconfigure should succeed");
        manager
            .get_encryption_service()
            .await
            .expect("second encryption service should exist")
            .create_key(CreateKeyRequest {
                key_name: Some("second-key".to_string()),
                key_usage: KeyUsage::EncryptDecrypt,
                description: None,
                policy: None,
                tags: HashMap::new(),
                origin: None,
            })
            .await
            .expect("second key should be created");

        provider
            .generate_sse_dek("bucket", "object", "second-key")
            .await
            .expect("provider should resolve the latest reconfigured service");

        manager.stop().await.expect("kms service should stop cleanly");
    }

    #[test]
    fn test_encryption_type_enum() {
        // Test EncryptionType enum
        assert_eq!(SSEType::SseS3, SSEType::SseS3);
        assert_eq!(SSEType::SseKms, SSEType::SseKms);
        assert_eq!(SSEType::SseC, SSEType::SseC);
        assert_ne!(SSEType::SseS3, SSEType::SseKms);

        // Test Debug format
        let debug_str = format!("{:?}", SSEType::SseKms);
        assert!(debug_str.contains("SseKms"));
    }

    #[test]
    fn test_verify_ssec_key_match_returns_invalid_request() {
        let stored = "stored_md5".to_string();
        let err = verify_ssec_key_match("wrong_md5", Some(&stored)).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_verify_ssec_key_match_no_stored_returns_invalid_request() {
        let err = verify_ssec_key_match("any_md5", None).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_ssec_for_read_non_encrypted_object() {
        let metadata = HashMap::new();
        let result = validate_ssec_for_read(&metadata, None, None);
        assert!(result.is_ok());
    }

    #[test]
    fn test_validate_ssec_for_read_missing_customer_key() {
        let mut metadata = HashMap::new();
        metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string());
        metadata.insert(
            "x-amz-server-side-encryption-customer-key-md5".to_string(),
            "DWygnHRtgiJ77HCm+1rvHw==".to_string(),
        );

        let err = validate_ssec_for_read(&metadata, None, None).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_ssec_for_read_wrong_key() {
        // Key A is used to "encrypt" the object (stored MD5 is from key A).
        let key_a = [42u8; 32];
        let stored_md5 = BASE64_STANDARD.encode(md5::compute(key_a).0);

        let mut metadata = HashMap::new();
        metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string());
        metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), stored_md5);

        // Key B is a different key; its MD5 won't match stored MD5.
        let key_b = [99u8; 32];
        let key_b_b64 = BASE64_STANDARD.encode(key_b);
        let key_b_md5 = BASE64_STANDARD.encode(md5::compute(key_b).0);

        let err = validate_ssec_for_read(&metadata, Some(&key_b_b64), Some(&key_b_md5)).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_ssec_for_read_correct_key() {
        let key_bytes = [42u8; 32];
        let key_b64 = BASE64_STANDARD.encode(key_bytes);
        let key_md5 = BASE64_STANDARD.encode(md5::compute(key_bytes).0);

        let mut metadata = HashMap::new();
        metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string());
        metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), key_md5.clone());

        let result = validate_ssec_for_read(&metadata, Some(&key_b64), Some(&key_md5));
        assert!(result.is_ok());
    }

    #[test]
    fn test_validate_ssec_for_read_spoofed_md5() {
        // A client provides the correct stored MD5 in the header but with a
        // DIFFERENT key. The server must recompute MD5 from the key bytes and
        // reject the request because the recomputed MD5 won't match the header.
        let real_key = [42u8; 32];
        let stored_md5 = BASE64_STANDARD.encode(md5::compute(real_key).0);

        let mut metadata = HashMap::new();
        metadata.insert("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string());
        metadata.insert("x-amz-server-side-encryption-customer-key-md5".to_string(), stored_md5.clone());

        // Attacker has a different key but tries to pass the stored MD5 as their header
        let fake_key = [99u8; 32];
        let fake_key_b64 = BASE64_STANDARD.encode(fake_key);

        let err = validate_ssec_for_read(&metadata, Some(&fake_key_b64), Some(&stored_md5)).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_sse_headers_for_read_rejects_kms_on_plain_object() {
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-server-side-encryption", http::HeaderValue::from_static("aws:kms"));
        headers.insert("x-amz-server-side-encryption-aws-kms-key-id", http::HeaderValue::from_static("test-key"));

        let metadata = HashMap::new();
        let err = validate_sse_headers_for_read(&metadata, &headers).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_validate_sse_headers_for_read_rejects_ssec_on_plain_object() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "x-amz-server-side-encryption-customer-algorithm",
            http::HeaderValue::from_static("AES256"),
        );
        headers.insert("x-amz-server-side-encryption-customer-key", http::HeaderValue::from_static("test-key"));
        headers.insert(
            "x-amz-server-side-encryption-customer-key-md5",
            http::HeaderValue::from_static("test-key-md5"),
        );

        let metadata = HashMap::new();
        let err = validate_sse_headers_for_read(&metadata, &headers).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_sse_headers_for_read_rejects_ssec_on_managed_object() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "x-amz-server-side-encryption-customer-algorithm",
            http::HeaderValue::from_static("AES256"),
        );
        headers.insert("x-amz-server-side-encryption-customer-key", http::HeaderValue::from_static("test-key"));
        headers.insert(
            "x-amz-server-side-encryption-customer-key-md5",
            http::HeaderValue::from_static("test-key-md5"),
        );

        let metadata = HashMap::from([("x-amz-server-side-encryption".to_string(), "aws:kms".to_string())]);
        let err = validate_sse_headers_for_read(&metadata, &headers).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_validate_sse_headers_for_read_allows_encrypted_object_without_request_headers() {
        let metadata = HashMap::from([
            ("x-amz-server-side-encryption".to_string(), "aws:kms".to_string()),
            ("x-rustfs-encryption-key".to_string(), "encrypted-key".to_string()),
        ]);
        let headers = HeaderMap::new();
        assert!(validate_sse_headers_for_read(&metadata, &headers).is_ok());
    }

    #[test]
    fn test_validate_sse_headers_for_read_rejects_sse_on_ssec_object() {
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-server-side-encryption", http::HeaderValue::from_static("aws:kms"));
        headers.insert("x-amz-server-side-encryption-aws-kms-key-id", http::HeaderValue::from_static("test-key"));

        let metadata = HashMap::from([("x-amz-server-side-encryption-customer-algorithm".to_string(), "AES256".to_string())]);
        let err = validate_sse_headers_for_read(&metadata, &headers).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidArgument);
    }

    #[test]
    fn test_validate_ssec_params_returns_invalid_request_on_bad_algorithm() {
        let key = BASE64_STANDARD.encode([42u8; 32]);
        let key_md5 = BASE64_STANDARD.encode(md5::compute([42u8; 32]).0);
        let params = SsecParams {
            algorithm: "AES128".to_string(),
            key,
            key_md5,
        };
        let err = validate_ssec_params(params).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    #[test]
    fn test_validate_ssec_params_returns_invalid_request_on_bad_md5() {
        let key = BASE64_STANDARD.encode([42u8; 32]);
        let params = SsecParams {
            algorithm: "AES256".to_string(),
            key,
            key_md5: BASE64_STANDARD.encode([99u8; 16]),
        };
        let err = validate_ssec_params(params).unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);
    }

    // ========================================================================
    // Unit tests for issue #2041: no mandatory KMS when encryption not used
    // ========================================================================

    /// When SSE-C params are not present and no managed SSE is requested,
    /// encryption should be skipped (Ok(None)). Ensures we do not require KMS
    /// when the client sends no encryption headers.
    #[tokio::test]
    async fn test_sse_encryption_skip_when_no_ssec_and_no_managed_sse_requested() {
        let request = EncryptionRequest {
            bucket: "test-bucket",
            key: "test-key",
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: None,
            sse_customer_key: None,
            sse_customer_key_md5: None,
            content_size: 1024,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };
        let result = sse_encryption(request).await;
        match &result {
            Ok(None) => {}
            Ok(Some(_)) => panic!("expected no encryption material when no SSE params provided"),
            Err(e) => {
                assert!(
                    !e.message.contains("No KMS key"),
                    "must not require KMS when no encryption requested; got: {}",
                    e.message
                );
            }
        }
    }

    /// When SSE-C params are partial or invalid, sse_encryption must return an error.
    #[tokio::test]
    async fn test_sse_encryption_errors_on_invalid_ssec_params() {
        let bucket = "test-bucket";
        let key = "test-key";
        let sse_key = BASE64_STANDARD.encode([42u8; 32]);
        let wrong_md5 = BASE64_STANDARD.encode([99u8; 16]);

        let request_wrong_md5 = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: Some("AES256".to_string()),
            sse_customer_key: Some(sse_key.clone()),
            sse_customer_key_md5: Some(wrong_md5),
            content_size: 1024,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };
        let err = sse_encryption(request_wrong_md5).await.unwrap_err();
        assert_eq!(err.code, S3ErrorCode::InvalidRequest);

        let request_unsupported_algorithm = EncryptionRequest {
            bucket,
            key,
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: Some("unsupported-algo".to_string()),
            sse_customer_key: Some(sse_key),
            sse_customer_key_md5: Some(BASE64_STANDARD.encode(md5::compute([42u8; 32]).0)),
            content_size: 1024,
            part_number: None,
            part_key: None,
            part_nonce: None,
        };
        let err = sse_encryption(request_unsupported_algorithm).await.unwrap_err();
        assert!(err.code == S3ErrorCode::InvalidRequest || err.code == S3ErrorCode::InvalidArgument);
    }

    /// When bucket has no SSE-S3/aws:kms setting and request has no SSE headers,
    /// encryption should be skipped (Ok(None)). Ensures no mandatory bucket default SSE.
    #[tokio::test]
    async fn test_sse_prepare_encryption_skip_when_no_params_and_no_bucket_sse() {
        let request = PrepareEncryptionRequest {
            bucket: "test-bucket-no-sse-config",
            key: "test-key",
            server_side_encryption: None,
            ssekms_key_id: None,
            sse_customer_algorithm: None,
            sse_customer_key_md5: None,
        };
        let result = sse_prepare_encryption(request).await;
        match &result {
            Ok(None) => {}
            Ok(Some(_)) => panic!("expected no encryption when bucket has no SSE config and no request SSE"),
            Err(e) => {
                assert!(
                    !e.message.contains("No KMS key"),
                    "must not require KMS when no bucket SSE and no request SSE; got: {}",
                    e.message
                );
            }
        }
    }
}
