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

use crate::runtime_sources::{AppContext, ServerContextSlot, current_action_credentials, current_ready_iam_handle};
use http::HeaderMap;
use http::Uri;
use rustfs_credentials::Credentials;
use rustfs_iam::error::Error as IamError;
use rustfs_iam::sys::{
    SESSION_POLICY_NAME, get_claims_from_token_with_secret, get_claims_from_token_with_secret_allow_missing_exp,
};
use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive, is_server_derived_condition_key};
use rustfs_trusted_proxies::ClientInfo;
use rustfs_utils::MaskedAccessKey;
use rustfs_utils::http::{AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER};
use s3s::S3Error;
use s3s::S3ErrorCode;
use s3s::S3Result;
use s3s::auth::S3Auth;
use s3s::auth::SecretKey;
use s3s::auth::SimpleAuth;
use s3s::s3_error;
use serde_json::Value;
use std::collections::HashMap;
use std::net::SocketAddr;
use subtle::ConstantTimeEq;
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tracing::{debug, trace, warn};
use url::form_urlencoded;

const LOG_COMPONENT_AUTH: &str = "auth";
const LOG_SUBSYSTEM_CREDENTIALS: &str = "credentials";
const LOG_SUBSYSTEM_KEYSTONE: &str = "keystone";
const LOG_SUBSYSTEM_REQUEST: &str = "request";
const EVENT_SECRET_KEY_LOOKUP_FAILED: &str = "secret_key_lookup_failed";
const EVENT_ACCESS_KEY_VALIDATION_STARTED: &str = "access_key_validation_started";
const EVENT_KEYSTONE_CREDENTIALS_DETECTED: &str = "keystone_credentials_detected";
const EVENT_KEYSTONE_CREDENTIALS_VALIDATED: &str = "keystone_credentials_validated";
const EVENT_KEYSTONE_CONTEXT_MISSING: &str = "keystone_context_missing";
const EVENT_SESSION_TOKEN_EXTRACTION: &str = "session_token_extraction";

/// RustFS-specific query capability for a single presigned PutObject request.
pub(crate) const RUSTFS_MAX_CONTENT_LENGTH_QUERY: &str = "x-rustfs-max-content-length";
pub(crate) const RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY: &str = "x-rustfs-max-total-object-size";

/// Inserted by the S3 access boundary after the upstream verifier accepts a
/// request as SigV4 presigned. Downstream capability parsing must require this
/// marker instead of treating query syntax as proof of authentication.
#[derive(Debug, Clone, Copy)]
pub(crate) struct VerifiedPresignedRequest;

#[derive(Debug, Clone, Copy)]
pub(crate) struct VerifiedSigV4Request;

/// Performs constant-time string comparison to prevent timing attacks.
///
/// This function should be used when comparing sensitive values like passwords,
/// API keys, or authentication tokens. It ensures the comparison time is
/// independent of the position where strings differ and handles length differences
/// securely.
///
/// # Security Note
/// This implementation uses the `subtle` crate to provide cryptographically
/// sound constant-time guarantees. The function is resistant to timing side-channel
/// attacks and suitable for security-critical comparisons.
///
/// # Example
/// ```
/// use rustfs::auth::constant_time_eq;
///
/// let secret1 = "my-secret-key";
/// let secret2 = "my-secret-key";
/// let secret3 = "wrong-secret";
///
/// assert!(constant_time_eq(secret1, secret2));
/// assert!(!constant_time_eq(secret1, secret3));
/// ```
pub fn constant_time_eq(a: &str, b: &str) -> bool {
    a.as_bytes().ct_eq(b.as_bytes()).into()
}

// Authentication type constants
const JWT_ALGORITHM: &str = "Bearer ";
const SIGN_V2_ALGORITHM: &str = "AWS ";
const SIGN_V4_ALGORITHM: &str = "AWS4-HMAC-SHA256";
const STREAMING_CONTENT_SHA256: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD";
const STREAMING_CONTENT_SHA256_TRAILER: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD-TRAILER";
pub(crate) const UNSIGNED_PAYLOAD_TRAILER: &str = "STREAMING-UNSIGNED-PAYLOAD-TRAILER";
const ACTION_HEADER: &str = "Action";
const AMZ_CREDENTIAL: &str = "X-Amz-Credential";
const AMZ_ACCESS_KEY_ID: &str = "AWSAccessKeyId";
pub const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";

// Authentication type enum
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub enum AuthType {
    #[default]
    Unknown,
    Anonymous,
    Presigned,
    PresignedV2,
    PostPolicy,
    StreamingSigned,
    Signed,
    SignedV2,
    #[allow(clippy::upper_case_acronyms)]
    JWT,
    #[allow(clippy::upper_case_acronyms)]
    STS,
    StreamingSignedTrailer,
    StreamingUnsignedTrailer,
}

pub struct IAMAuth {
    simple_auth: SimpleAuth,
    access_key: String,
    secret_key: SecretKey,
    server_ctx: Option<std::sync::Arc<ServerContextSlot>>,
}

impl Clone for IAMAuth {
    fn clone(&self) -> Self {
        Self {
            simple_auth: SimpleAuth::from_single(self.access_key.clone(), self.secret_key.clone()),
            access_key: self.access_key.clone(),
            secret_key: self.secret_key.clone(),
            server_ctx: self.server_ctx.clone(),
        }
    }
}

impl IAMAuth {
    pub fn new(ak: impl Into<String>, sk: impl Into<SecretKey>) -> Self {
        let access_key = ak.into();
        let secret_key = sk.into();
        let simple_auth = SimpleAuth::from_single(access_key.clone(), secret_key.clone());
        Self {
            simple_auth,
            access_key,
            secret_key,
            server_ctx: None,
        }
    }

    pub(crate) fn with_server_context(
        ak: impl Into<String>,
        sk: impl Into<SecretKey>,
        server_ctx: std::sync::Arc<ServerContextSlot>,
    ) -> Self {
        let mut auth = Self::new(ak, sk);
        auth.server_ctx = Some(server_ctx);
        auth
    }
}

#[async_trait::async_trait]
impl S3Auth for IAMAuth {
    async fn get_secret_key(&self, access_key: &str) -> S3Result<SecretKey> {
        // NEW: Check if Keystone credentials are present in task-local storage
        // This handles pure X-Auth-Token requests without Authorization header
        use rustfs_keystone::KEYSTONE_CREDENTIALS;

        if let Ok(Some(creds)) = KEYSTONE_CREDENTIALS.try_with(|c| c.clone()) {
            debug!(
                event = EVENT_KEYSTONE_CREDENTIALS_DETECTED,
                component = LOG_COMPONENT_AUTH,
                subsystem = LOG_SUBSYSTEM_KEYSTONE,
                principal = %MaskedAccessKey(&creds.parent_user),
                result = "token_auth",
                "Keystone task-local credentials detected"
            );
            // Return empty secret key - Keystone uses token validation, not AWS signatures
            return Ok(SecretKey::from(String::new()));
        }

        if access_key.is_empty() {
            return Err(s3_error!(UnauthorizedAccess, "Your account is not signed up"));
        }

        // Check if this is a Keystone access key (from mixed auth scenario)
        // Keystone credentials use token authentication, not signature verification
        if access_key.starts_with("keystone:") {
            debug!(
                event = EVENT_KEYSTONE_CREDENTIALS_DETECTED,
                component = LOG_COMPONENT_AUTH,
                subsystem = LOG_SUBSYSTEM_KEYSTONE,
                access_key = %MaskedAccessKey(access_key),
                result = "token_auth",
                "Keystone token-auth access key detected"
            );
            // Return empty secret key - Keystone uses token validation, not AWS signatures
            // The actual credentials are stored in task-local storage by KeystoneAuthMiddleware
            return Ok(SecretKey::from(String::new()));
        }

        if access_key == self.access_key {
            return Ok(self.secret_key.clone());
        }

        if let Ok(key) = self.simple_auth.get_secret_key(access_key).await {
            return Ok(key);
        }

        let iam_store = match &self.server_ctx {
            Some(server_ctx) => server_ctx
                .installed_app_context()
                .filter(|context| context.iam().is_ready())
                .map(|context| context.iam().handle())
                .ok_or(()),
            None => current_ready_iam_handle().map_err(|_| ()),
        };
        if let Ok(iam_store) = iam_store {
            // Use check_key instead of get_user to ensure user is loaded from disk if not in cache
            // This is important for newly created users that may not be in cache yet.
            // check_key will automatically attempt to load the user from disk if not found in cache.
            match iam_store.check_key(access_key).await {
                Ok((Some(id), _valid)) => {
                    // Return secret key for signature verification regardless of user status.
                    // Authorization will be checked separately in the authorization phase.
                    return Ok(SecretKey::from(id.credentials.secret_key));
                }
                Ok((None, _)) => {
                    warn!(
                        event = EVENT_SECRET_KEY_LOOKUP_FAILED,
                        component = LOG_COMPONENT_AUTH,
                        subsystem = LOG_SUBSYSTEM_CREDENTIALS,
                        access_key = %MaskedAccessKey(access_key),
                        reason = "no_such_user",
                        "Secret key lookup rejected"
                    );
                }
                Err(e) => {
                    warn!(
                        event = EVENT_SECRET_KEY_LOOKUP_FAILED,
                        component = LOG_COMPONENT_AUTH,
                        subsystem = LOG_SUBSYSTEM_CREDENTIALS,
                        access_key = %MaskedAccessKey(access_key),
                        error = ?e,
                        reason = "check_key_error",
                        "Secret key lookup errored"
                    );
                    return Err(iam_lookup_error_to_s3_error(&e));
                }
            }
        } else {
            warn!(
                event = EVENT_SECRET_KEY_LOOKUP_FAILED,
                component = LOG_COMPONENT_AUTH,
                subsystem = LOG_SUBSYSTEM_CREDENTIALS,
                access_key = %MaskedAccessKey(access_key),
                reason = "iam_not_initialized",
                "Secret key lookup skipped"
            );
        }

        Err(s3_error!(
            InvalidAccessKeyId,
            "The Access Key Id you provided does not exist in our records."
        ))
    }
}

fn iam_lookup_error_to_s3_error(_err: &IamError) -> S3Error {
    s3_error!(InternalError, "IAM user lookup failed")
}

// check_key_valid checks the key is valid or not. return the user's credentials and if the user is the owner.
pub async fn check_key_valid(session_token: &str, access_key: &str) -> S3Result<(Credentials, bool)> {
    check_key_valid_with_context(session_token, access_key, None).await
}

fn has_root_access(sys_cred: &Credentials, cred: &Credentials) -> bool {
    (constant_time_eq(&sys_cred.access_key, &cred.access_key) || constant_time_eq(&cred.parent_user, &sys_cred.access_key))
        && !cred
            .claims
            .as_ref()
            .is_some_and(|claims| claims.contains_key(SESSION_POLICY_NAME) || rustfs_iam::sys::is_rustfs_oidc_claims(claims))
}

/// Validate an access key, resolving the root credentials and IAM system from
/// an explicit application context when one is given (backlog#1052 S6).
///
/// A per-server request path passes its own context so a second embedded
/// server authenticates against its own root identity and IAM domain instead
/// of the process defaults; `None` falls back to the ambient globals — the
/// single-instance legacy behavior that all existing callers keep.
pub async fn check_key_valid_with_context(
    session_token: &str,
    access_key: &str,
    ctx: Option<&AppContext>,
) -> S3Result<(Credentials, bool)> {
    // KEYSTONE INTEGRATION: Check if Keystone credentials are present in task-local storage
    // This handles both:
    // 1. Pure X-Auth-Token requests (access_key may be empty)
    // 2. Keystone access keys formatted as "keystone:user_id"
    use crate::auth_keystone;
    use rustfs_keystone::KEYSTONE_CREDENTIALS;

    // Try to get Keystone credentials from task-local storage first
    // Add debug logging for UI authentication tracking
    debug!(
        event = EVENT_ACCESS_KEY_VALIDATION_STARTED,
        component = LOG_COMPONENT_AUTH,
        subsystem = LOG_SUBSYSTEM_CREDENTIALS,
        access_key = %MaskedAccessKey(access_key),
        has_session_token = !session_token.is_empty(),
        "Access key validation started"
    );
    if let Ok(Some(credentials)) = KEYSTONE_CREDENTIALS.try_with(|creds| creds.clone()) {
        debug!(
            event = EVENT_KEYSTONE_CREDENTIALS_DETECTED,
            component = LOG_COMPONENT_AUTH,
            subsystem = LOG_SUBSYSTEM_KEYSTONE,
            result = "task_local",
            "Keystone task-local credentials detected"
        );

        if !auth_keystone::is_keystone_enabled() {
            return Err(s3_error!(InvalidAccessKeyId, "Keystone authentication is not enabled"));
        }

        debug!(
            event = EVENT_KEYSTONE_CREDENTIALS_VALIDATED,
            component = LOG_COMPONENT_AUTH,
            subsystem = LOG_SUBSYSTEM_KEYSTONE,
            principal = %MaskedAccessKey(&credentials.parent_user),
            has_project_name = credentials
                .claims
                .as_ref()
                .and_then(|c| c.get("keystone_project_name"))
                .and_then(|v| v.as_str())
                .is_some(),
            "Keystone task-local credentials validated"
        );

        // Determine if user is admin (owner-level access)
        // Users with "admin" or "reseller_admin" role have owner permissions
        // Roles are stored in claims["keystone_roles"] by the middleware
        let is_owner = credentials
            .claims
            .as_ref()
            .and_then(|claims| claims.get("keystone_roles"))
            .and_then(|roles| roles.as_array())
            .map(|roles| {
                roles
                    .iter()
                    .any(|role| role.as_str().map(|r| r == "admin" || r == "reseller_admin").unwrap_or(false))
            })
            .unwrap_or(false);

        debug!(
            event = EVENT_KEYSTONE_CREDENTIALS_VALIDATED,
            component = LOG_COMPONENT_AUTH,
            subsystem = LOG_SUBSYSTEM_KEYSTONE,
            principal = %MaskedAccessKey(&credentials.parent_user),
            is_owner,
            "Keystone owner permissions evaluated"
        );

        return Ok((credentials, is_owner));
    }

    // Legacy check for explicit "keystone:" prefix (for backwards compatibility)
    if access_key.starts_with("keystone:") {
        warn!(
            event = EVENT_KEYSTONE_CONTEXT_MISSING,
            component = LOG_COMPONENT_AUTH,
            subsystem = LOG_SUBSYSTEM_KEYSTONE,
            access_key = %MaskedAccessKey(access_key),
            "Keystone context missing for access key"
        );

        if !auth_keystone::is_keystone_enabled() {
            return Err(s3_error!(InvalidAccessKeyId, "Keystone authentication is not enabled"));
        }

        return Err(s3_error!(InvalidAccessKeyId, "Keystone authentication requires X-Auth-Token header"));
    }

    // Prefer this server's context (backlog#1052 S6); fall back to the ambient
    // process credentials when no context was threaded in.
    let root_cred = match ctx {
        Some(context) => context.action_credentials().get(),
        None => current_action_credentials(),
    };
    let Some(mut cred) = root_cred else {
        return Err(S3Error::with_message(
            S3ErrorCode::InternalError,
            format!("get_global_action_cred {:?}", IamError::IamSysNotInitialized),
        ));
    };

    let sys_cred = cred.clone();

    if !constant_time_eq(&cred.access_key, access_key) {
        let iam_store = match ctx {
            Some(context) if context.iam().is_ready() => Ok(context.iam().handle()),
            Some(_) => Err(()),
            None => current_ready_iam_handle().map_err(|_| ()),
        };
        let Ok(iam_store) = iam_store else {
            return Err(S3Error::with_message(
                S3ErrorCode::InternalError,
                format!("check_key_valid {:?}", IamError::IamSysNotInitialized),
            ));
        };

        let (u, ok) = iam_store
            .check_key(access_key)
            .await
            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed1 {e}")))?;

        if !ok {
            let Some(ref u) = u else {
                warn!(
                    event = EVENT_SECRET_KEY_LOOKUP_FAILED,
                    component = LOG_COMPONENT_AUTH,
                    subsystem = LOG_SUBSYSTEM_CREDENTIALS,
                    access_key = %MaskedAccessKey(access_key),
                    reason = "user_not_found",
                    "Access key validation rejected"
                );
                return Err(s3_error!(InvalidAccessKeyId, "check key failed"));
            };

            if u.credentials.status == "off" {
                warn!(
                    event = EVENT_SECRET_KEY_LOOKUP_FAILED,
                    component = LOG_COMPONENT_AUTH,
                    subsystem = LOG_SUBSYSTEM_CREDENTIALS,
                    access_key = %MaskedAccessKey(access_key),
                    reason = "account_disabled",
                    "Access key validation rejected"
                );
                return Err(s3_error!(InvalidAccessKeyId, "check key failed"));
            }

            warn!(
                event = EVENT_SECRET_KEY_LOOKUP_FAILED,
                component = LOG_COMPONENT_AUTH,
                subsystem = LOG_SUBSYSTEM_CREDENTIALS,
                access_key = %MaskedAccessKey(access_key),
                reason = "validation_failed",
                "Access key validation rejected"
            );
            return Err(s3_error!(InvalidRequest, "check key failed"));
        }

        let Some(u) = u else {
            return Err(s3_error!(InvalidAccessKeyId, "check key failed"));
        };

        cred = u.credentials;
    }

    let claims = check_claims_from_token_with_context(session_token, &cred, ctx)
        .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("check claims failed {e}")))?;

    cred.claims = if !claims.is_empty() { Some(claims) } else { None };

    let owner = has_root_access(&sys_cred, &cred);
    Ok((cred, owner))
}

pub fn check_claims_from_token(token: &str, cred: &Credentials) -> S3Result<HashMap<String, Value>> {
    check_claims_from_token_with_context(token, cred, None)
}

fn check_claims_from_token_with_context(
    token: &str,
    cred: &Credentials,
    ctx: Option<&AppContext>,
) -> S3Result<HashMap<String, Value>> {
    if !token.is_empty() && cred.access_key.is_empty() {
        return Err(s3_error!(InvalidRequest, "no access key"));
    }

    if token.is_empty() && cred.is_temp() && !cred.is_service_account() {
        return Err(s3_error!(InvalidRequest, "invalid token1"));
    }

    if !token.is_empty() && !cred.is_temp() {
        return Err(s3_error!(InvalidRequest, "invalid token2"));
    }

    if !cred.is_service_account() && cred.is_temp() && !constant_time_eq(token, &cred.session_token) {
        return Err(s3_error!(InvalidRequest, "invalid token3"));
    }

    if cred.is_temp() && cred.is_expired() {
        return Err(s3_error!(InvalidRequest, "invalid access key is temp and expired"));
    }

    let sys_cred = match ctx {
        Some(context) => context.action_credentials().get(),
        None => current_action_credentials(),
    };
    let Some(sys_cred) = sys_cred else {
        return Err(s3_error!(InternalError, "action cred not init"));
    };

    let (token, secret) = if cred.is_service_account() {
        (cred.session_token.as_str(), cred.secret_key.as_str())
    } else {
        (token, sys_cred.secret_key.as_str())
    };

    if !token.is_empty() {
        let claims: HashMap<String, Value> = if cred.is_service_account() {
            get_claims_from_token_with_secret_allow_missing_exp(token, secret)
                .map_err(|_e| s3_error!(InvalidRequest, "invalid token"))?
        } else {
            get_claims_from_token_with_secret(token, secret).map_err(|_e| s3_error!(InvalidRequest, "invalid token"))?
        };
        return Ok(claims);
    }

    Ok(HashMap::new())
}

pub fn get_session_token<'a>(uri: &'a Uri, hds: &'a HeaderMap) -> Option<&'a str> {
    let token = hds
        .get("x-amz-security-token")
        .map(|v| v.to_str().unwrap_or_default())
        .or_else(|| get_query_param(uri.query().unwrap_or_default(), "x-amz-security-token"));

    trace!(
        event = EVENT_SESSION_TOKEN_EXTRACTION,
        component = LOG_COMPONENT_AUTH,
        subsystem = LOG_SUBSYSTEM_REQUEST,
        has_session_token = token.is_some(),
        "Completed session token extraction"
    );

    token
}

pub(crate) fn extract_string_list_claim(claims: &HashMap<String, Value>, claim_name: &str) -> Vec<String> {
    match get_claim_case_insensitive(claims, claim_name) {
        ClaimLookup::Found(Value::Array(values)) => values.iter().filter_map(|v| v.as_str().map(ToOwned::to_owned)).collect(),
        ClaimLookup::Found(Value::String(value)) => value
            .split(',')
            .map(str::trim)
            .filter(|v| !v.is_empty())
            .map(ToOwned::to_owned)
            .collect(),
        ClaimLookup::Missing | ClaimLookup::Ambiguous | ClaimLookup::Found(_) => Vec::new(),
    }
}

fn policy_source_ip(remote_addr: Option<SocketAddr>, client_info: Option<&ClientInfo>) -> String {
    client_info
        .map(|info| info.real_ip.to_string())
        .or_else(|| remote_addr.map(|addr| addr.ip().to_string()))
        .unwrap_or_default()
}

fn policy_secure_transport(client_info: Option<&ClientInfo>) -> bool {
    client_info
        .and_then(|info| info.forwarded_proto.as_deref())
        .map(|proto| proto.eq_ignore_ascii_case("https"))
        .unwrap_or(false)
}

/// Get condition values for policy evaluation
///
/// # Arguments
/// * `header` - HTTP headers of the request
/// * `cred` - User credentials
/// * `version_id` - Optional version ID of the object
/// * `region` - Optional region/location constraint
/// * `remote_addr` - Optional remote address of the connection
///
/// # Returns
/// * `HashMap<String, Vec<String>>` - Condition values for policy evaluation
///
pub fn get_condition_values(
    header: &HeaderMap,
    cred: &Credentials,
    version_id: Option<&str>,
    region: Option<s3s::region::Region>,
    remote_addr: Option<SocketAddr>,
) -> HashMap<String, Vec<String>> {
    get_condition_values_with_client_info(header, cred, version_id, region, remote_addr, None)
}

/// Get condition values for policy evaluation with verified client information.
pub fn get_condition_values_with_client_info(
    header: &HeaderMap,
    cred: &Credentials,
    version_id: Option<&str>,
    region: Option<s3s::region::Region>,
    remote_addr: Option<SocketAddr>,
    client_info: Option<&ClientInfo>,
) -> HashMap<String, Vec<String>> {
    get_condition_values_with_query_and_client_info(header, cred, version_id, region, remote_addr, None, client_info)
}

/// Get condition values for policy evaluation with optional query-string values.
///
/// # Arguments
/// * `header` - HTTP headers of the request
/// * `cred` - User credentials
/// * `version_id` - Optional version ID of the object
/// * `region` - Optional region/location constraint
/// * `remote_addr` - Optional remote address of the connection
/// * `query` - Optional request query string
///
/// # Returns
/// * `HashMap<String, Vec<String>>` - Condition values for policy evaluation
pub fn get_condition_values_with_query(
    header: &HeaderMap,
    cred: &Credentials,
    version_id: Option<&str>,
    region: Option<s3s::region::Region>,
    remote_addr: Option<SocketAddr>,
    query: Option<&str>,
) -> HashMap<String, Vec<String>> {
    get_condition_values_with_query_and_client_info(header, cred, version_id, region, remote_addr, query, None)
}

/// Get condition values for policy evaluation with optional query-string values
/// and verified client information from trusted proxy middleware.
pub fn get_condition_values_with_query_and_client_info(
    header: &HeaderMap,
    cred: &Credentials,
    version_id: Option<&str>,
    region: Option<s3s::region::Region>,
    remote_addr: Option<SocketAddr>,
    query: Option<&str>,
    client_info: Option<&ClientInfo>,
) -> HashMap<String, Vec<String>> {
    let username = if cred.is_temp() || cred.is_service_account() {
        cred.parent_user.clone()
    } else {
        cred.access_key.clone()
    };

    let sys_cred = current_action_credentials().unwrap_or_default();

    let claims = &cred.claims;

    let principal_type = if !username.is_empty() {
        if claims.is_some() {
            "AssumedRole"
        } else if constant_time_eq(&sys_cred.access_key, &username) {
            "Account"
        } else {
            "User"
        }
    } else {
        "Anonymous"
    };

    // Get current time
    let curr_time = OffsetDateTime::now_utc();
    let epoch_time = curr_time.unix_timestamp();

    // Use provided version ID or empty string
    let vid = version_id.unwrap_or("");

    // Determine auth type and signature version from headers and query
    let (auth_type, signature_version) = determine_auth_type_and_version_with_query(header, query);

    let is_tls = policy_secure_transport(client_info);
    let source_ip = policy_source_ip(remote_addr, client_info);

    let mut args = HashMap::new();

    // Add basic time and security info
    args.insert("CurrentTime".to_owned(), vec![curr_time.format(&Rfc3339).unwrap_or_default()]);
    args.insert("EpochTime".to_owned(), vec![epoch_time.to_string()]);
    args.insert("SecureTransport".to_owned(), vec![is_tls.to_string()]);
    args.insert("SourceIp".to_owned(), vec![source_ip]);

    // Add user agent and referer
    if let Some(user_agent) = header.get("user-agent") {
        args.insert("UserAgent".to_owned(), vec![user_agent.to_str().unwrap_or("").to_string()]);
    }
    if let Some(referer) = header.get("referer") {
        args.insert("Referer".to_owned(), vec![referer.to_str().unwrap_or("").to_string()]);
    }

    // Add user and principal info
    args.insert("userid".to_owned(), vec![username.clone()]);
    args.insert("username".to_owned(), vec![username]);
    args.insert("principaltype".to_owned(), vec![principal_type.to_string()]);

    // Add version ID
    if !vid.is_empty() {
        args.insert("versionid".to_owned(), vec![vid.to_string()]);
    }

    // Add signature version and auth type
    if !signature_version.is_empty() {
        args.insert("signatureversion".to_owned(), vec![signature_version]);
    }
    if !auth_type.is_empty() {
        args.insert("authType".to_owned(), vec![auth_type]);
    }

    if let Some(lc) = region
        && !lc.as_str().is_empty()
    {
        args.insert("LocationConstraint".to_owned(), vec![lc.to_string()]);
    }

    let mut clone_header = header.clone();
    if let Some(v) = clone_header.get("x-amz-signature-age") {
        args.insert("signatureAge".to_string(), vec![v.to_str().unwrap_or("").to_string()]);
        clone_header.remove("x-amz-signature-age");
    }

    for obj_lock in &[
        AMZ_OBJECT_LOCK_MODE_LOWER,
        AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER,
        AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
    ] {
        let values = clone_header
            .get_all(*obj_lock)
            .iter()
            .map(|v| v.to_str().unwrap_or("").to_string())
            .collect::<Vec<String>>();
        if !values.is_empty() {
            args.insert(obj_lock.trim_start_matches("x-amz-").to_string(), values);
        }
        clone_header.remove(*obj_lock);
    }

    // S3 policy condition keys use "x-amz-grant-*" (policy key s3:x-amz-grant-* -> name() returns x-amz-grant-*)
    for grant_header in &[
        "x-amz-grant-full-control",
        "x-amz-grant-read",
        "x-amz-grant-write",
        "x-amz-grant-read-acp",
        "x-amz-grant-write-acp",
    ] {
        let values = clone_header
            .get_all(*grant_header)
            .iter()
            .map(|v| v.to_str().unwrap_or("").to_string())
            .collect::<Vec<String>>();
        if !values.is_empty() {
            args.insert((*grant_header).to_string(), values);
        }
        clone_header.remove(*grant_header);
    }

    // Claims and group membership are part of the verified identity, so they are
    // resolved before request headers are merged in below.
    if let Some(claims) = &cred.claims {
        for (k, v) in claims {
            if let Some(v_str) = v.as_str() {
                args.insert(k.trim_start_matches("ldap").to_lowercase(), vec![v_str.to_string()]);
            }
        }

        let grps = extract_string_list_claim(claims, "groups");
        if !grps.is_empty() {
            args.insert("groups".to_string(), grps);
        }

        let roles = extract_string_list_claim(claims, "roles");
        if !roles.is_empty() {
            args.insert("roles".to_string(), roles);
        }
    }

    if let Some(groups) = &cred.groups
        && !args.contains_key("groups")
    {
        args.insert("groups".to_string(), groups.clone());
    }

    // Every remaining header is attacker-controlled. A header must never contribute
    // to a condition key that describes the caller's own identity or the connection,
    // otherwise sending `userid: admin` (or any `jwt:`/`ldap:` claim name) would let a
    // request satisfy a policy condition about itself. Reject those names outright --
    // both the ones already populated above and the well-known identity keys that are
    // absent for this credential, since an absent key is exactly what a spoofed header
    // would fill in.
    for key in clone_header.keys() {
        if key.as_str().eq_ignore_ascii_case("x-amz-tagging") {
            continue;
        }
        if is_reserved_condition_key(key.as_str(), &args) {
            continue;
        }
        args.insert(
            key.as_str().to_string(),
            header
                .get_all(key)
                .iter()
                .map(|v| v.to_str().unwrap_or("").to_string())
                .collect(),
        );
    }

    args
}

/// Whether a request header is forbidden from contributing to policy condition key
/// `key`, either because the server already derived that key from verified state or
/// because it is a well-known identity/context key that only the server may populate.
fn is_reserved_condition_key(key: &str, server_derived: &HashMap<String, Vec<String>>) -> bool {
    server_derived.contains_key(key)
        || [
            AMZ_OBJECT_LOCK_MODE_LOWER,
            AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER,
            AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
        ]
        .iter()
        .any(|header| key.eq_ignore_ascii_case(header.trim_start_matches("x-amz-")))
        || is_server_derived_condition_key(key)
}

/// Get request authentication type
///
/// # Arguments
/// * `header` - HTTP headers of the request
///
/// # Returns
/// * `AuthType` - The determined authentication type
///
pub fn get_request_auth_type(header: &HeaderMap) -> AuthType {
    get_request_auth_type_with_query(header, None)
}

pub(crate) fn get_request_auth_type_with_query(header: &HeaderMap, query: Option<&str>) -> AuthType {
    if is_request_signature_v2(header) {
        AuthType::SignedV2
    } else if is_request_presigned_signature_v2(header, query) {
        AuthType::PresignedV2
    } else if is_request_sign_streaming_v4(header) {
        AuthType::StreamingSigned
    } else if is_request_sign_streaming_trailer_v4(header) {
        AuthType::StreamingSignedTrailer
    } else if is_request_unsigned_trailer_v4(header) {
        AuthType::StreamingUnsignedTrailer
    } else if is_request_signature_v4(header) {
        AuthType::Signed
    } else if is_request_presigned_signature_v4_with_query(header, query) {
        AuthType::Presigned
    } else if is_request_jwt(header) {
        AuthType::JWT
    } else if is_request_post_policy_signature_v4(header) {
        AuthType::PostPolicy
    } else if is_request_sts(header) {
        AuthType::STS
    } else if is_request_anonymous(header) {
        AuthType::Anonymous
    } else {
        AuthType::Unknown
    }
}

fn determine_auth_type_and_version_with_query(header: &HeaderMap, query: Option<&str>) -> (String, String) {
    match get_request_auth_type_with_query(header, query) {
        AuthType::JWT => ("JWT".to_string(), String::new()),
        AuthType::SignedV2 => ("REST-HEADER".to_string(), "AWS2".to_string()),
        AuthType::PresignedV2 => ("REST-QUERY-STRING".to_string(), "AWS2".to_string()),
        AuthType::StreamingSigned | AuthType::StreamingSignedTrailer | AuthType::StreamingUnsignedTrailer => {
            ("REST-HEADER".to_string(), "AWS4-HMAC-SHA256".to_string())
        }
        AuthType::Signed => ("REST-HEADER".to_string(), "AWS4-HMAC-SHA256".to_string()),
        AuthType::Presigned => ("REST-QUERY-STRING".to_string(), "AWS4-HMAC-SHA256".to_string()),
        AuthType::PostPolicy => ("POST".to_string(), String::new()),
        AuthType::STS => ("STS".to_string(), String::new()),
        AuthType::Anonymous => ("Anonymous".to_string(), String::new()),
        AuthType::Unknown => (String::new(), String::new()),
    }
}

/// Verify if request has JWT
///
/// # Arguments
/// * `header` - HTTP headers of the request
///
/// # Returns
/// * `bool` - True if request has JWT, false otherwise
fn is_request_jwt(header: &HeaderMap) -> bool {
    if let Some(auth) = header.get("authorization")
        && let Ok(auth_str) = auth.to_str()
    {
        return auth_str.starts_with(JWT_ALGORITHM);
    }
    false
}

/// Verify if request has AWS Signature Version '4'
///
/// # Arguments
/// * `header` - HTTP headers of the request
///
/// # Returns
/// * `bool` - True if request has AWS Signature Version '4', false otherwise
fn is_request_signature_v4(header: &HeaderMap) -> bool {
    if let Some(auth) = header.get("authorization")
        && let Ok(auth_str) = auth.to_str()
    {
        return auth_str.starts_with(SIGN_V4_ALGORITHM);
    }
    false
}

/// Verify if request has AWS Signature Version '2'
///
/// # Arguments
/// * `header` - HTTP headers of the request
///
/// # Returns
/// * `bool` - True if request has AWS Signature Version '2', false otherwise
fn is_request_signature_v2(header: &HeaderMap) -> bool {
    if let Some(auth) = header.get("authorization")
        && let Ok(auth_str) = auth.to_str()
    {
        return !auth_str.starts_with(SIGN_V4_ALGORITHM) && auth_str.starts_with(SIGN_V2_ALGORITHM);
    }
    false
}

pub(crate) fn is_request_presigned_signature_v4_with_query(header: &HeaderMap, query: Option<&str>) -> bool {
    if let Some(credential) = header.get(AMZ_CREDENTIAL) {
        return !credential.to_str().unwrap_or("").is_empty();
    }
    query.is_some_and(|query| {
        form_urlencoded::parse(query.as_bytes())
            .find(|(name, _)| name.eq_ignore_ascii_case("x-amz-credential"))
            .is_some_and(|(_, credential)| !credential.is_empty())
    })
}

/// Verify request has AWS PreSign Version '2'
///
/// # Arguments
/// * `header` - HTTP headers of the request
///
/// # Returns
/// * `bool` - True if request has AWS PreSign Version '2', false otherwise
fn is_request_presigned_signature_v2(header: &HeaderMap, query: Option<&str>) -> bool {
    if let Some(access_key) = header.get(AMZ_ACCESS_KEY_ID) {
        return !access_key.to_str().unwrap_or("").is_empty();
    }
    query
        .and_then(|query| get_query_param(query, "awsaccesskeyid"))
        .is_some_and(|access_key| !access_key.is_empty())
}

/// Verify if request has AWS Post policy Signature Version '4'
///
/// # Arguments
/// * `header` - HTTP headers of the request
///
/// # Returns
/// * `bool` - True if request has AWS Post policy Signature Version '4', false otherwise
fn is_request_post_policy_signature_v4(header: &HeaderMap) -> bool {
    if let Some(content_type) = header.get("content-type")
        && let Ok(ct) = content_type.to_str()
    {
        return ct.contains("multipart/form-data");
    }
    false
}

/// Verify if the request has AWS Streaming Signature Version '4'
fn is_request_sign_streaming_v4(header: &HeaderMap) -> bool {
    if let Some(content_sha256) = header.get("x-amz-content-sha256")
        && let Ok(sha256_str) = content_sha256.to_str()
    {
        return sha256_str == STREAMING_CONTENT_SHA256;
    }
    false
}

// Verify if the request has AWS Streaming Signature Version '4' with trailer
fn is_request_sign_streaming_trailer_v4(header: &HeaderMap) -> bool {
    if let Some(content_sha256) = header.get("x-amz-content-sha256")
        && let Ok(sha256_str) = content_sha256.to_str()
    {
        return sha256_str == STREAMING_CONTENT_SHA256_TRAILER;
    }
    false
}

// Verify if the request has AWS Streaming Signature Version '4' with unsigned content and trailer
fn is_request_unsigned_trailer_v4(header: &HeaderMap) -> bool {
    if let Some(content_sha256) = header.get("x-amz-content-sha256")
        && let Ok(sha256_str) = content_sha256.to_str()
    {
        return sha256_str == UNSIGNED_PAYLOAD_TRAILER;
    }
    false
}

// Verify if request is STS (Security Token Service)
fn is_request_sts(header: &HeaderMap) -> bool {
    if let Some(action) = header.get(ACTION_HEADER) {
        return !action.to_str().unwrap_or("").is_empty();
    }
    false
}

// Verify if request is anonymous
fn is_request_anonymous(header: &HeaderMap) -> bool {
    header.get("authorization").is_none()
}

pub fn get_query_param<'a>(query: &'a str, param_name: &str) -> Option<&'a str> {
    let param_name = param_name.to_lowercase();

    for pair in query.split('&') {
        let mut parts = pair.splitn(2, '=');
        if let (Some(key), Some(value)) = (parts.next(), parts.next())
            && key.to_lowercase() == param_name
        {
            return Some(value);
        }
    }
    None
}

/// Parse the RustFS presigned PutObject size capability after authentication.
///
/// The query value is covered by SigV4 when it is present before presigning, but
/// the signature does not assign any semantics to the extension. Keep parsing
/// strict and only enable the capability for a verified SigV4 presigned request.
pub(crate) fn parse_presigned_put_max_content_length(
    header: &HeaderMap,
    query: Option<&str>,
    verified_presigned: bool,
) -> S3Result<Option<u64>> {
    let Some(query) = query else {
        return Ok(None);
    };

    let mut value = None;
    let mut decoded_query = Vec::new();
    for (name, candidate) in form_urlencoded::parse(query.as_bytes()) {
        decoded_query.push((name.to_string(), candidate.to_string()));
        if name == RUSTFS_MAX_CONTENT_LENGTH_QUERY {
            if value.is_some() {
                return Err(S3Error::with_message(
                    S3ErrorCode::InvalidRequest,
                    format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must appear exactly once"),
                ));
            }
            value = Some(candidate.into_owned());
        } else if name.eq_ignore_ascii_case(RUSTFS_MAX_CONTENT_LENGTH_QUERY) {
            return Err(S3Error::with_message(
                S3ErrorCode::InvalidRequest,
                format!("query parameter name must be exactly {RUSTFS_MAX_CONTENT_LENGTH_QUERY}"),
            ));
        }
    }

    let Some(value) = value else {
        return Ok(None);
    };

    let query_value = |wanted: &str| {
        decoded_query
            .iter()
            .find(|(name, _)| name.eq_ignore_ascii_case(wanted))
            .map(|(_, value)| value.as_str())
    };
    let is_complete_sigv4_query = [
        ("x-amz-algorithm", "AWS4-HMAC-SHA256"),
        ("x-amz-date", ""),
        ("x-amz-expires", ""),
        ("x-amz-signedheaders", ""),
        ("x-amz-credential", ""),
        ("x-amz-signature", ""),
    ]
    .into_iter()
    .all(|(name, expected)| {
        query_value(name).is_some_and(|value| !value.is_empty() && (expected.is_empty() || value == expected))
    });
    if !verified_presigned
        || !is_complete_sigv4_query
        || !matches!(get_request_auth_type_with_query(header, Some(query)), AuthType::Presigned)
    {
        return Err(S3Error::with_message(
            S3ErrorCode::InvalidRequest,
            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} requires a SigV4 presigned request"),
        ));
    }

    let limit = value.parse::<u64>().map_err(|_| {
        S3Error::with_message(
            S3ErrorCode::InvalidRequest,
            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must be a non-negative 64-bit integer"),
        )
    })?;

    Ok(Some(limit))
}

/// Reject the PutObject-only size capability when it appears on another
/// operation. Callers must invoke this after request authentication has run.
pub(crate) fn reject_presigned_put_max_content_length_for_other_operation(
    header: &HeaderMap,
    query: Option<&str>,
    verified_presigned: bool,
) -> S3Result<()> {
    if parse_presigned_put_max_content_length(header, query, verified_presigned)?.is_some() {
        return Err(S3Error::with_message(
            S3ErrorCode::InvalidRequest,
            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
        ));
    }
    Ok(())
}

/// Parse the V2 multipart total-size capability after SigV4 authentication.
/// Header-authenticated CreateMultipartUpload requests are accepted because the
/// custom query is covered by the SigV4 canonical request; later multipart
/// operations read the immutable value from the upload session metadata.
pub(crate) fn parse_presigned_multipart_max_total_object_size(
    header: &HeaderMap,
    query: Option<&str>,
    verified_sigv4: bool,
) -> S3Result<Option<u64>> {
    let Some(query) = query else {
        return Ok(None);
    };

    let mut value = None;
    let mut decoded_query = Vec::new();
    for (name, candidate) in form_urlencoded::parse(query.as_bytes()) {
        decoded_query.push((name.to_string(), candidate.to_string()));
        if name == RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY {
            if value.is_some() {
                return Err(S3Error::with_message(
                    S3ErrorCode::InvalidRequest,
                    format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} must appear exactly once"),
                ));
            }
            value = Some(candidate.into_owned());
        } else if name.eq_ignore_ascii_case(RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY) {
            return Err(S3Error::with_message(
                S3ErrorCode::InvalidRequest,
                format!("query parameter name must be exactly {RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY}"),
            ));
        }
    }

    let Some(value) = value else {
        return Ok(None);
    };

    let auth_type = get_request_auth_type_with_query(header, Some(query));
    let is_presigned = matches!(auth_type, AuthType::Presigned);
    let is_header_signed = matches!(auth_type, AuthType::Signed);
    let complete_presigned_query = [
        ("x-amz-algorithm", "AWS4-HMAC-SHA256"),
        ("x-amz-date", ""),
        ("x-amz-expires", ""),
        ("x-amz-signedheaders", ""),
        ("x-amz-credential", ""),
        ("x-amz-signature", ""),
    ]
    .into_iter()
    .all(|(name, expected)| {
        decoded_query
            .iter()
            .find(|(candidate, _)| candidate.eq_ignore_ascii_case(name))
            .is_some_and(|(_, candidate)| !candidate.is_empty() && (expected.is_empty() || candidate == expected))
    });

    let authenticated = verified_sigv4 && (is_header_signed || (is_presigned && complete_presigned_query));
    if !authenticated {
        return Err(S3Error::with_message(
            S3ErrorCode::InvalidRequest,
            format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} requires a verified SigV4 request"),
        ));
    }

    value.parse::<u64>().map(Some).map_err(|_| {
        S3Error::with_message(
            S3ErrorCode::InvalidRequest,
            format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} must be a non-negative 64-bit integer"),
        )
    })
}

pub(crate) fn reject_presigned_multipart_max_total_object_size_for_other_operation(
    header: &HeaderMap,
    query: Option<&str>,
    verified_sigv4: bool,
) -> S3Result<()> {
    if parse_presigned_multipart_max_total_object_size(header, query, verified_sigv4)?.is_some() {
        return Err(S3Error::with_message(
            S3ErrorCode::InvalidRequest,
            format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} is only supported for CreateMultipartUpload"),
        ));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::runtime_sources::{IamInterface, KmsInterface};
    use http::{HeaderMap, HeaderValue, Uri};
    use rustfs_credentials::Credentials;
    use rustfs_iam::{
        store::{
            Store as _,
            object::{IAM_CONFIG_PREFIX, ObjectStore},
        },
        sys::IamSys,
    };
    use rustfs_kms::KmsServiceManager;
    use rustfs_policy::auth::get_new_credentials_with_metadata;
    use rustfs_trusted_proxies::ValidationMode;
    use s3s::auth::SecretKey;
    use serde_json::json;
    use std::collections::HashMap;
    use std::sync::Arc;
    use time::OffsetDateTime;

    struct ContextIam {
        handle: Arc<IamSys<ObjectStore>>,
    }

    impl IamInterface for ContextIam {
        fn handle(&self) -> Arc<IamSys<ObjectStore>> {
            self.handle.clone()
        }

        fn is_ready(&self) -> bool {
            true
        }
    }

    struct TestKms;

    impl KmsInterface for TestKms {
        fn handle(&self) -> Arc<KmsServiceManager> {
            Arc::new(KmsServiceManager::new())
        }
    }

    fn create_test_credentials() -> Credentials {
        Credentials {
            access_key: "test-access-key".to_string(),
            secret_key: "test-secret-key".to_string(),
            session_token: "".to_string(),
            expiration: None,
            status: "on".to_string(),
            parent_user: "".to_string(),
            groups: None,
            claims: None,
            name: Some("test-user".to_string()),
            description: Some("test user for auth tests".to_string()),
        }
    }

    fn create_temp_credentials() -> Credentials {
        Credentials {
            access_key: "temp-access-key".to_string(),
            secret_key: "temp-secret-key".to_string(),
            session_token: "temp-session-token".to_string(),
            expiration: Some(OffsetDateTime::now_utc() + time::Duration::hours(1)),
            status: "on".to_string(),
            parent_user: "parent-user".to_string(),
            groups: Some(vec!["test-group".to_string()]),
            claims: None,
            name: Some("temp-user".to_string()),
            description: Some("temporary user for auth tests".to_string()),
        }
    }

    fn create_service_account_credentials() -> Credentials {
        let mut claims = HashMap::new();
        claims.insert(rustfs_credentials::IAM_POLICY_CLAIM_NAME_SA.to_string(), json!("test-policy"));

        Credentials {
            access_key: "service-access-key".to_string(),
            secret_key: "service-secret-key".to_string(),
            session_token: "service-session-token".to_string(),
            expiration: None,
            status: "on".to_string(),
            parent_user: "service-parent".to_string(),
            groups: None,
            claims: Some(claims),
            name: Some("service-account".to_string()),
            description: Some("service account for auth tests".to_string()),
        }
    }

    #[test]
    fn oidc_session_cannot_inherit_root_access_from_display_name() {
        let sys_cred = Credentials {
            access_key: "root-access-key".to_string(),
            ..Default::default()
        };
        let oidc_cred = Credentials {
            access_key: "temporary-access-key".to_string(),
            parent_user: sys_cred.access_key.clone(),
            claims: Some(HashMap::from([
                ("iss".to_string(), json!("rustfs-oidc")),
                ("oidc_provider".to_string(), json!("default")),
                ("sub".to_string(), json!("subject-123")),
            ])),
            ..Default::default()
        };

        assert!(!has_root_access(&sys_cred, &oidc_cred));
    }

    #[test]
    fn test_iam_auth_creation() {
        let access_key = "test-access-key";
        let secret_key = SecretKey::from("test-secret-key");

        let iam_auth = IAMAuth::new(access_key, secret_key);

        // The struct should be created successfully
        // We can't easily test internal state without exposing it,
        // but we can test it doesn't panic on creation
        assert_eq!(size_of_val(&iam_auth), size_of::<IAMAuth>());
    }

    #[tokio::test]
    async fn test_iam_auth_clone_preserves_bootstrap_secret() {
        let iam_auth = IAMAuth::new("test-ak", SecretKey::from("test-sk"));
        let cloned = iam_auth.clone();

        let secret = cloned.get_secret_key("test-ak").await;
        assert!(secret.is_ok());
    }

    #[tokio::test]
    async fn test_iam_auth_get_secret_key_empty_access_key() {
        let iam_auth = IAMAuth::new("test-ak", SecretKey::from("test-sk"));

        let result = iam_auth.get_secret_key("").await;

        assert!(result.is_err());
        let error = result.unwrap_err();
        assert_eq!(error.code(), &S3ErrorCode::UnauthorizedAccess);
        assert!(error.message().unwrap_or("").contains("Your account is not signed up"));
    }

    #[test]
    fn test_iam_lookup_error_maps_to_internal_error() {
        let result = iam_lookup_error_to_s3_error(&IamError::Io(std::io::Error::other("load user failed")));

        assert_eq!(result.code(), &S3ErrorCode::InternalError);
        assert_eq!(result.message(), Some("IAM user lookup failed"));
    }

    #[test]
    fn test_check_claims_from_token_empty_token_and_access_key() {
        let mut cred = create_test_credentials();
        cred.access_key = "".to_string();

        let result = check_claims_from_token("test-token", &cred);

        assert!(result.is_err());
        let error = result.unwrap_err();
        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
        assert!(error.message().unwrap_or("").contains("no access key"));
    }

    #[test]
    fn test_check_claims_from_token_temp_credentials_without_token() {
        let mut cred = create_temp_credentials();
        // Make it non-service account
        cred.claims = None;

        let result = check_claims_from_token("", &cred);

        assert!(result.is_err());
        let error = result.unwrap_err();
        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
        assert!(error.message().unwrap_or("").contains("invalid token1"));
    }

    #[test]
    fn test_check_claims_from_token_non_temp_with_token() {
        let mut cred = create_test_credentials();
        cred.session_token = "".to_string(); // Make it non-temp

        let result = check_claims_from_token("some-token", &cred);

        assert!(result.is_err());
        let error = result.unwrap_err();
        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
        assert!(error.message().unwrap_or("").contains("invalid token2"));
    }

    #[test]
    fn test_check_claims_from_token_mismatched_session_token() {
        let mut cred = create_temp_credentials();
        // Make sure it's not a service account
        cred.claims = None;

        let result = check_claims_from_token("wrong-session-token", &cred);

        assert!(result.is_err());
        let error = result.unwrap_err();
        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
        assert!(error.message().unwrap_or("").contains("invalid token3"));
    }

    #[test]
    fn test_check_claims_from_token_expired_credentials() {
        let mut cred = create_temp_credentials();
        cred.expiration = Some(OffsetDateTime::now_utc() - time::Duration::hours(1)); // Expired
        cred.claims = None; // Make sure it's not a service account

        let result = check_claims_from_token(&cred.session_token, &cred);

        assert!(result.is_err());
        let error = result.unwrap_err();
        assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);

        // The function checks various conditions in order. An expired temp credential
        // might trigger other validation errors first (like token mismatch)
        let msg = error.message().unwrap_or("");
        let is_valid_error = msg.contains("invalid access key is temp and expired")
            || msg.contains("invalid token")
            || msg.contains("action cred not init");
        assert!(is_valid_error, "Unexpected error message: '{msg}'");
    }

    #[test]
    fn test_check_claims_from_token_valid_non_temp_credentials() {
        let mut cred = create_test_credentials();
        cred.session_token = "".to_string(); // Make it non-temp

        let result = check_claims_from_token("", &cred);

        // This might fail due to global state dependencies, but should return error about global cred init
        if let Ok(claims) = result {
            assert!(claims.is_empty());
        } else if let Err(error) = result {
            assert_eq!(error.code(), &S3ErrorCode::InternalError);
            assert!(error.message().unwrap_or("").contains("action cred not init"));
        }
    }

    #[tokio::test]
    async fn check_claims_uses_the_explicit_context_signing_secret() {
        let (_temp_dir, _disk_paths, store) = crate::app::gating_test_env::isolated_multi_pool_ecstore().await;
        ObjectStore::new(store.clone())
            .save_iam_config(serde_json::json!({"version": 1}), format!("{}/format.json", *IAM_CONFIG_PREFIX))
            .await
            .expect("seed request IAM format");
        let iam = rustfs_iam::build_iam_sys(store.clone())
            .await
            .expect("request IAM should initialize");
        let matching = AppContext::new(store.clone(), Arc::new(ContextIam { handle: iam.clone() }), Arc::new(TestKms));
        let mismatching = AppContext::new(store, Arc::new(ContextIam { handle: iam.clone() }), Arc::new(TestKms));
        assert!(matching.publish_action_credentials(Credentials {
            access_key: "matching-root".to_string(),
            secret_key: "matching-signing-secret".to_string(),
            status: "on".to_string(),
            ..Default::default()
        }));
        assert!(mismatching.publish_action_credentials(Credentials {
            access_key: "mismatching-root".to_string(),
            secret_key: "mismatching-signing-secret".to_string(),
            status: "on".to_string(),
            ..Default::default()
        }));
        let claims = HashMap::from([
            (
                "exp".to_string(),
                json!((OffsetDateTime::now_utc() + time::Duration::minutes(5)).unix_timestamp()),
            ),
            ("context".to_string(), json!("matching")),
        ]);
        let mut credential = get_new_credentials_with_metadata(&claims, "matching-signing-secret")
            .expect("temporary credentials should be generated");
        credential.parent_user = "matching-root".to_string();
        iam.set_temp_user(&credential.access_key, &credential, None)
            .await
            .expect("temporary credentials should be stored in request IAM");

        let (verified, _) = check_key_valid_with_context(&credential.session_token, &credential.access_key, Some(&matching))
            .await
            .expect("the matching request context should verify the token");
        assert_eq!(
            verified.claims.as_ref().and_then(|claims| claims.get("context")),
            Some(&json!("matching"))
        );
        assert!(
            check_key_valid_with_context(&credential.session_token, &credential.access_key, Some(&mismatching))
                .await
                .is_err(),
            "a different server context must not validate the token"
        );
    }

    #[test]
    fn test_get_session_token_from_header() {
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-security-token", HeaderValue::from_static("test-session-token"));

        let uri: Uri = "https://example.com/".parse().unwrap();

        let token = get_session_token(&uri, &headers);

        assert_eq!(token, Some("test-session-token"));
    }

    #[test]
    fn test_get_session_token_from_query_param() {
        let headers = HeaderMap::new();
        let uri: Uri = "https://example.com/?x-amz-security-token=query-session-token"
            .parse()
            .unwrap();

        let token = get_session_token(&uri, &headers);

        assert_eq!(token, Some("query-session-token"));
    }

    #[test]
    fn test_get_session_token_header_takes_precedence() {
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-security-token", HeaderValue::from_static("header-token"));

        let uri: Uri = "https://example.com/?x-amz-security-token=query-token".parse().unwrap();

        let token = get_session_token(&uri, &headers);

        assert_eq!(token, Some("header-token"));
    }

    #[test]
    fn test_get_session_token_no_token() {
        let headers = HeaderMap::new();
        let uri: Uri = "https://example.com/".parse().unwrap();

        let token = get_session_token(&uri, &headers);

        assert_eq!(token, None);
    }

    #[test]
    fn test_get_condition_values_regular_user() {
        let cred = create_test_credentials();
        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("userid"), Some(&vec!["test-access-key".to_string()]));
        assert_eq!(conditions.get("username"), Some(&vec!["test-access-key".to_string()]));
        assert_eq!(conditions.get("principaltype"), Some(&vec!["User".to_string()]));
    }

    #[test]
    fn test_get_condition_values_with_presigned_query() {
        let cred = create_test_credentials();
        let headers = HeaderMap::new();
        let uri: Uri = "https://example.com/?X-Amz-Credential=AKIAIOSFODNN7EXAMPLE%2F20130524%2Fus-east-1%2Fs3%2Faws4_request"
            .parse()
            .unwrap();

        let conditions = get_condition_values_with_query(&headers, &cred, None, None, None, uri.query());

        assert_eq!(conditions.get("signatureversion"), Some(&vec!["AWS4-HMAC-SHA256".to_string()]));
        assert_eq!(conditions.get("authType"), Some(&vec!["REST-QUERY-STRING".to_string()]));
    }

    #[test]
    fn test_get_condition_values_temp_user() {
        let cred = create_temp_credentials();
        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("userid"), Some(&vec!["parent-user".to_string()]));
        assert_eq!(conditions.get("username"), Some(&vec!["parent-user".to_string()]));
        assert_eq!(conditions.get("principaltype"), Some(&vec!["User".to_string()]));
    }

    #[test]
    fn test_get_condition_values_service_account() {
        let cred = create_service_account_credentials();
        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("userid"), Some(&vec!["service-parent".to_string()]));
        assert_eq!(conditions.get("username"), Some(&vec!["service-parent".to_string()]));
        // Service accounts with claims should be "AssumedRole" type
        assert_eq!(conditions.get("principaltype"), Some(&vec!["AssumedRole".to_string()]));
    }

    #[test]
    fn ghsa_6r96_identity_condition_keys_ignore_spoofed_headers() {
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        // A caller naming its headers after identity condition keys must not be able
        // to add or replace values the server derives from the credential.
        headers.insert("userid", "admin".parse().unwrap());
        headers.insert("username", "admin".parse().unwrap());
        headers.insert("principaltype", "Account".parse().unwrap());
        headers.insert("signatureversion", "AWS4-HMAC-SHA256".parse().unwrap());

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("userid"), Some(&vec!["test-access-key".to_string()]));
        assert_eq!(conditions.get("username"), Some(&vec!["test-access-key".to_string()]));
        assert_eq!(conditions.get("principaltype"), Some(&vec!["User".to_string()]));
        assert!(
            !conditions
                .get("signatureversion")
                .is_some_and(|v| v.iter().any(|s| s == "AWS4-HMAC-SHA256")),
            "an unsigned request must not gain a signatureversion from a header"
        );
    }

    #[test]
    fn version_id_condition_ignores_spoofed_headers() {
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        headers.insert("versionid", "spoofed-version".parse().unwrap());

        let conditions = get_condition_values(&headers, &cred, None, None, None);
        assert_eq!(conditions.get("versionid"), None);

        let conditions = get_condition_values(&headers, &cred, Some("server-version"), None, None);
        assert_eq!(conditions.get("versionid"), Some(&vec!["server-version".to_string()]));
    }

    #[test]
    fn ghsa_6r96_claim_condition_keys_ignore_spoofed_headers() {
        // The credential carries no groups/roles claims, so these keys are absent --
        // precisely the case a spoofed header would otherwise fill in.
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        headers.insert("groups", "admins".parse().unwrap());
        headers.insert("roles", "RustFS.ConsoleAdmin".parse().unwrap());
        headers.insert("sub", "someone-else".parse().unwrap());

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("groups"), None, "groups must come from the credential only");
        assert_eq!(conditions.get("roles"), None, "roles must come from claims only");
        assert_eq!(conditions.get("sub"), None, "jwt claim keys must come from claims only");
    }

    #[test]
    fn test_request_headers_still_reach_conditions() {
        // The reserved list must stay narrow: ordinary request headers, including the
        // `s3:x-amz-*` condition keys, are still expected to be available to policies.
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-content-sha256", "UNSIGNED-PAYLOAD".parse().unwrap());
        headers.insert("x-amz-server-side-encryption", "AES256".parse().unwrap());

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("x-amz-content-sha256"), Some(&vec!["UNSIGNED-PAYLOAD".to_string()]));
        assert_eq!(conditions.get("x-amz-server-side-encryption"), Some(&vec!["AES256".to_string()]));
    }

    #[test]
    fn test_get_condition_values_with_object_lock_headers() {
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        headers.insert(AMZ_OBJECT_LOCK_MODE_LOWER, HeaderValue::from_static("GOVERNANCE"));
        headers.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, HeaderValue::from_static("OFF"));
        headers.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, HeaderValue::from_static("2024-12-31T23:59:59Z"));
        headers.insert("object-lock-mode", HeaderValue::from_static("COMPLIANCE"));
        headers.insert("object-lock-legal-hold", HeaderValue::from_static("ON"));
        headers.insert("object-lock-retain-until-date", HeaderValue::from_static("2099-12-31T23:59:59Z"));

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("object-lock-mode"), Some(&vec!["GOVERNANCE".to_string()]));
        assert_eq!(conditions.get("object-lock-legal-hold"), Some(&vec!["OFF".to_string()]));
        assert_eq!(
            conditions.get("object-lock-retain-until-date"),
            Some(&vec!["2024-12-31T23:59:59Z".to_string()])
        );
    }

    #[test]
    fn object_lock_condition_aliases_cannot_spoof_canonical_headers() {
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        headers.insert("object-lock-mode", HeaderValue::from_static("COMPLIANCE"));
        headers.insert("object-lock-legal-hold", HeaderValue::from_static("ON"));
        headers.insert("object-lock-retain-until-date", HeaderValue::from_static("2099-12-31T23:59:59Z"));

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("object-lock-mode"), None);
        assert_eq!(conditions.get("object-lock-legal-hold"), None);
        assert_eq!(conditions.get("object-lock-retain-until-date"), None);
    }

    #[test]
    fn test_get_condition_values_with_grant_headers() {
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-grant-full-control", HeaderValue::from_static("id=owner-123"));
        headers.insert(
            "x-amz-grant-read",
            HeaderValue::from_static("uri=http://acs.amazonaws.com/groups/global/AllUsers"),
        );

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("x-amz-grant-full-control"), Some(&vec!["id=owner-123".to_string()]));
        assert_eq!(
            conditions.get("x-amz-grant-read"),
            Some(&vec!["uri=http://acs.amazonaws.com/groups/global/AllUsers".to_string()])
        );
    }

    #[test]
    fn test_get_condition_values_with_signature_age() {
        let cred = create_test_credentials();
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-signature-age", HeaderValue::from_static("300"));

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("signatureAge"), Some(&vec!["300".to_string()]));
        // Verify the header is removed after processing
        // (we can't directly test this without changing the function signature)
    }

    #[test]
    fn test_get_condition_values_with_claims() {
        let mut cred = create_service_account_credentials();
        let mut claims = HashMap::new();
        claims.insert("ldapUsername".to_string(), json!("ldap-user"));
        claims.insert("groups".to_string(), json!(["group1", "group2"]));
        cred.claims = Some(claims);

        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("username"), Some(&vec!["ldap-user".to_string()]));
        assert_eq!(conditions.get("groups"), Some(&vec!["group1".to_string(), "group2".to_string()]));
    }

    #[test]
    fn test_get_condition_values_with_roles_claim_array() {
        let mut cred = create_service_account_credentials();
        let mut claims = HashMap::new();
        claims.insert("roles".to_string(), json!(["role1", "role2"]));
        cred.claims = Some(claims);

        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("roles"), Some(&vec!["role1".to_string(), "role2".to_string()]));
    }

    #[test]
    fn test_get_condition_values_with_roles_claim_csv_and_case_insensitive() {
        let mut cred = create_service_account_credentials();
        let mut claims = HashMap::new();
        claims.insert("Roles".to_string(), json!("role1, role2"));
        cred.claims = Some(claims);

        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("roles"), Some(&vec!["role1".to_string(), "role2".to_string()]));
    }

    #[test]
    fn test_get_condition_values_with_roles_claim_ambiguous_case_insensitive_match_returns_empty() {
        let mut cred = create_service_account_credentials();
        let mut claims = HashMap::new();
        claims.insert("Roles".to_string(), json!(["role1"]));
        claims.insert("ROLES".to_string(), json!(["role2"]));
        cred.claims = Some(claims);

        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("roles"), None);
    }

    #[test]
    fn test_get_condition_values_with_credential_groups() {
        let mut cred = create_test_credentials();
        cred.groups = Some(vec!["cred-group1".to_string(), "cred-group2".to_string()]);

        let headers = HeaderMap::new();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(
            conditions.get("groups"),
            Some(&vec!["cred-group1".to_string(), "cred-group2".to_string()])
        );
    }

    #[test]
    fn test_get_query_param_found() {
        let query = "param1=value1&param2=value2&param3=value3";

        let result = get_query_param(query, "param2");

        assert_eq!(result, Some("value2"));
    }

    #[test]
    fn test_get_query_param_case_insensitive() {
        let query = "Param1=value1&PARAM2=value2&param3=value3";

        let result = get_query_param(query, "param2");

        assert_eq!(result, Some("value2"));
    }

    #[test]
    fn test_get_query_param_not_found() {
        let query = "param1=value1&param2=value2&param3=value3";

        let result = get_query_param(query, "param4");

        assert_eq!(result, None);
    }

    #[test]
    fn test_get_query_param_empty_query() {
        let query = "";

        let result = get_query_param(query, "param1");

        assert_eq!(result, None);
    }

    #[test]
    fn test_get_query_param_malformed_query() {
        let query = "param1&param2=value2&param3";

        let result = get_query_param(query, "param2");

        assert_eq!(result, Some("value2"));

        let result = get_query_param(query, "param1");

        assert_eq!(result, None);
    }

    #[test]
    fn test_get_query_param_with_equals_in_value() {
        let query = "param1=value=with=equals&param2=value2";

        let result = get_query_param(query, "param1");

        assert_eq!(result, Some("value=with=equals"));
    }

    #[test]
    fn presigned_put_max_content_length_requires_exactly_one_signed_query_value() {
        let headers = HeaderMap::new();
        let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";

        let query = format!("{signed_prefix}&x-rustfs-max-content-length=104857600");
        assert_eq!(
            parse_presigned_put_max_content_length(&headers, Some(&query), true).unwrap(),
            Some(104_857_600)
        );

        let encoded_credential = query.replacen("X-Amz-Credential", "X%2DAmz-Credential", 1);
        assert_eq!(
            parse_presigned_put_max_content_length(&headers, Some(&encoded_credential), true).unwrap(),
            Some(104_857_600)
        );

        let duplicate = format!("{query}&x-rustfs-max-content-length=1");
        assert_eq!(
            parse_presigned_put_max_content_length(&headers, Some(&duplicate), true)
                .unwrap_err()
                .code(),
            &S3ErrorCode::InvalidRequest
        );

        let wrong_case = format!("{signed_prefix}&X-RustFS-Max-Content-Length=1");
        assert_eq!(
            parse_presigned_put_max_content_length(&headers, Some(&wrong_case), true)
                .unwrap_err()
                .code(),
            &S3ErrorCode::InvalidRequest
        );

        assert_eq!(
            reject_presigned_put_max_content_length_for_other_operation(&headers, Some(&query), true)
                .unwrap_err()
                .code(),
            &S3ErrorCode::InvalidRequest
        );
    }

    #[test]
    fn presigned_put_max_content_length_rejects_unsigned_or_invalid_values() {
        let headers = HeaderMap::new();
        let forged = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/credential&X-Amz-Signature=fake&x-rustfs-max-content-length=1";
        assert_eq!(
            parse_presigned_put_max_content_length(&headers, Some(forged), false)
                .unwrap_err()
                .code(),
            &S3ErrorCode::InvalidRequest
        );
        for query in [
            "x-rustfs-max-content-length=1",
            "X-Amz-Credential=test/credential&x-rustfs-max-content-length=-1",
            "X-Amz-Credential=test/credential&x-rustfs-max-content-length=18446744073709551616",
        ] {
            let error = parse_presigned_put_max_content_length(&headers, Some(query), true).unwrap_err();
            assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
        }
    }

    #[test]
    fn multipart_max_total_object_size_requires_signed_create_request() {
        let headers = HeaderMap::new();
        let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";
        let query = format!("{signed_prefix}&x-rustfs-max-total-object-size=104857600");

        assert_eq!(
            parse_presigned_multipart_max_total_object_size(&headers, Some(&query), true).unwrap(),
            Some(104_857_600)
        );
        assert_eq!(
            reject_presigned_multipart_max_total_object_size_for_other_operation(&headers, Some(&query), true)
                .unwrap_err()
                .code(),
            &S3ErrorCode::InvalidRequest
        );
    }

    #[test]
    fn multipart_max_total_object_size_rejects_tampering_and_invalid_values() {
        let headers = HeaderMap::new();
        let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";
        for query in [
            "x-rustfs-max-total-object-size=1",
            "X-RustFS-Max-Total-Object-Size=1",
            "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=1&x-rustfs-max-total-object-size=2",
            "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=-1",
            "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=18446744073709551616",
        ] {
            assert_eq!(
                parse_presigned_multipart_max_total_object_size(&headers, Some(query), true)
                    .unwrap_err()
                    .code(),
                &S3ErrorCode::InvalidRequest
            );
        }

        let forged = format!("{signed_prefix}&x-rustfs-max-total-object-size=1");
        assert_eq!(
            parse_presigned_multipart_max_total_object_size(&headers, Some(&forged), false)
                .unwrap_err()
                .code(),
            &S3ErrorCode::InvalidRequest
        );
    }

    #[test]
    fn test_credentials_is_expired() {
        let mut cred = create_test_credentials();
        cred.expiration = Some(OffsetDateTime::now_utc() - time::Duration::hours(1));

        assert!(cred.is_expired());
    }

    #[test]
    fn test_credentials_is_not_expired() {
        let mut cred = create_test_credentials();
        cred.expiration = Some(OffsetDateTime::now_utc() + time::Duration::hours(1));

        assert!(!cred.is_expired());
    }

    #[test]
    fn test_credentials_no_expiration() {
        let cred = create_test_credentials();

        assert!(!cred.is_expired());
    }

    #[test]
    fn test_credentials_is_temp() {
        let cred = create_temp_credentials();

        assert!(cred.is_temp());
    }

    #[test]
    fn test_credentials_is_not_temp_no_session_token() {
        let mut cred = create_test_credentials();
        cred.session_token = "".to_string();

        assert!(!cred.is_temp());
    }

    #[test]
    fn test_credentials_is_not_temp_expired() {
        let mut cred = create_temp_credentials();
        cred.expiration = Some(OffsetDateTime::now_utc() - time::Duration::hours(1));

        assert!(!cred.is_temp());
    }

    #[test]
    fn test_credentials_is_service_account() {
        let cred = create_service_account_credentials();

        assert!(cred.is_service_account());
    }

    #[test]
    fn test_credentials_is_not_service_account() {
        let cred = create_test_credentials();

        assert!(!cred.is_service_account());
    }

    #[test]
    fn test_get_request_auth_type_jwt() {
        let mut headers = HeaderMap::new();
        headers.insert("authorization", HeaderValue::from_static("Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"));

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::JWT);
    }

    #[test]
    fn test_get_request_auth_type_signature_v2() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "authorization",
            HeaderValue::from_static("AWS AKIAIOSFODNN7EXAMPLE:frJIUN8DYpKDtOLCwo//bqJZQ1iY="),
        );

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::SignedV2);
    }

    #[test]
    fn test_get_request_auth_type_signature_v4() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "authorization",
            HeaderValue::from_static("AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request"),
        );

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::Signed);
    }

    #[test]
    fn test_get_request_auth_type_presigned_v2() {
        let mut headers = HeaderMap::new();
        headers.insert("AWSAccessKeyId", HeaderValue::from_static("AKIAIOSFODNN7EXAMPLE"));

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::PresignedV2);
    }

    #[test]
    fn test_get_request_auth_type_presigned_v2_from_query() {
        let headers = HeaderMap::new();
        let uri: Uri = "https://example.com/?AWSAccessKeyId=AKIAIOSFODNN7EXAMPLE&Signature=example&Expires=1672531200"
            .parse()
            .unwrap();

        let auth_type = get_request_auth_type_with_query(&headers, uri.query());

        assert_eq!(auth_type, AuthType::PresignedV2);
    }

    #[test]
    fn test_get_request_auth_type_presigned_v4() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "X-Amz-Credential",
            HeaderValue::from_static("AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request"),
        );

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::Presigned);
    }

    #[test]
    fn test_get_request_auth_type_presigned_v4_from_query() {
        let headers = HeaderMap::new();
        let uri: Uri = "https://example.com/?X-Amz-Credential=AKIAIOSFODNN7EXAMPLE%2F20130524%2Fus-east-1%2Fs3%2Faws4_request"
            .parse()
            .unwrap();

        let auth_type = get_request_auth_type_with_query(&headers, uri.query());

        assert_eq!(auth_type, AuthType::Presigned);
    }

    #[test]
    fn test_get_request_auth_type_post_policy() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "content-type",
            HeaderValue::from_static("multipart/form-data; boundary=----WebKitFormBoundary7MA4YWxkTrZu0gW"),
        );

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::PostPolicy);
    }

    #[test]
    fn test_get_request_auth_type_streaming_signed() {
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-content-sha256", HeaderValue::from_static("STREAMING-AWS4-HMAC-SHA256-PAYLOAD"));

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::StreamingSigned);
    }

    #[test]
    fn test_get_request_auth_type_streaming_signed_trailer() {
        let mut headers = HeaderMap::new();
        headers.insert(
            "x-amz-content-sha256",
            HeaderValue::from_static("STREAMING-AWS4-HMAC-SHA256-PAYLOAD-TRAILER"),
        );

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::StreamingSignedTrailer);
    }

    #[test]
    fn test_get_request_auth_type_streaming_unsigned_trailer() {
        let mut headers = HeaderMap::new();
        headers.insert("x-amz-content-sha256", HeaderValue::from_static("STREAMING-UNSIGNED-PAYLOAD-TRAILER"));

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::StreamingUnsignedTrailer);
    }

    #[test]
    fn test_get_request_auth_type_sts() {
        let mut headers = HeaderMap::new();
        headers.insert("Action", HeaderValue::from_static("AssumeRole"));

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::STS);
    }

    #[test]
    fn test_get_request_auth_type_anonymous() {
        let headers = HeaderMap::new();

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::Anonymous);
    }

    #[test]
    fn test_get_request_auth_type_unknown() {
        let mut headers = HeaderMap::new();
        headers.insert("authorization", HeaderValue::from_static("CustomAuth token123"));

        let auth_type = get_request_auth_type(&headers);

        assert_eq!(auth_type, AuthType::Unknown);
    }

    #[test]
    fn test_constant_time_eq() {
        assert!(constant_time_eq("test", "test"));
        assert!(!constant_time_eq("Test", "test"), "first-byte mismatch must fail");
        assert!(!constant_time_eq("tesu", "test"), "last-byte mismatch must fail");
        assert!(!constant_time_eq("test", "test1"), "longer candidate must fail");
        assert!(!constant_time_eq("test1", "test"), "shorter candidate must fail");
        assert!(!constant_time_eq("", "test"));
        assert!(constant_time_eq("", ""));

        // Test with credentials-like strings
        let key1 = "AKIAIOSFODNN7EXAMPLE";
        let key2 = "AKIAIOSFODNN7EXAMPLE";
        let key3 = "AKIAIOSFODNN7EXAMPLF";
        assert!(constant_time_eq(key1, key2));
        assert!(!constant_time_eq(key1, key3));
    }

    #[test]
    fn session_token_comparison_uses_constant_time_helper() {
        let source = include_str!("auth.rs");
        let production = source.split_once("#[cfg(test)]").map_or(source, |(production, _)| production);
        let ordinary_comparison = ["token ", "!=", " cred.session_token"].concat();
        assert!(
            !production.contains(&ordinary_comparison),
            "temporary session tokens must not use ordinary string comparison"
        );
        assert!(production.contains("!constant_time_eq(token, &cred.session_token)"));
    }

    #[test]
    fn test_get_condition_values_source_ip() {
        let mut headers = HeaderMap::new();
        let cred = Credentials::default();

        // Case 1: No headers, no remote addr -> empty string
        let conditions = get_condition_values(&headers, &cred, None, None, None);
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "");

        // Case 2: No headers, with remote addr -> remote addr
        let remote_addr: std::net::SocketAddr = "192.168.0.10:12345".parse().unwrap();
        let conditions = get_condition_values(&headers, &cred, None, None, Some(remote_addr));
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "192.168.0.10");

        // Case 3: X-Forwarded-For is ignored without verified proxy context
        headers.insert("x-forwarded-for", HeaderValue::from_static("10.0.0.1"));
        let conditions = get_condition_values(&headers, &cred, None, None, Some(remote_addr));
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "192.168.0.10");

        // Case 4: X-Forwarded-For with multiple IPs is ignored without verified proxy context
        headers.insert("x-forwarded-for", HeaderValue::from_static("10.0.0.3, 10.0.0.4"));
        let conditions = get_condition_values(&headers, &cred, None, None, Some(remote_addr));
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "192.168.0.10");

        // Case 5: X-Real-IP is ignored without verified proxy context
        headers.remove("x-forwarded-for");
        headers.insert("x-real-ip", HeaderValue::from_static("10.0.0.2"));
        let conditions = get_condition_values(&headers, &cred, None, None, Some(remote_addr));
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "192.168.0.10");

        // Case 6: Forwarded is ignored without verified proxy context
        headers.remove("x-real-ip");
        headers.insert("forwarded", HeaderValue::from_static("for=10.0.0.5;proto=http"));
        let conditions = get_condition_values(&headers, &cred, None, None, Some(remote_addr));
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "192.168.0.10");

        // Case 7: Forwarded with quotes and multiple values is ignored without verified proxy context
        headers.insert("forwarded", HeaderValue::from_static("for=\"10.0.0.6\", for=10.0.0.7"));
        let conditions = get_condition_values(&headers, &cred, None, None, Some(remote_addr));
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "192.168.0.10");

        // Case 8: IPv6 Remote Addr
        let remote_addr_v6: std::net::SocketAddr = "[2001:db8::1]:8080".parse().unwrap();
        headers.clear();
        let conditions = get_condition_values(&headers, &cred, None, None, Some(remote_addr_v6));
        assert_eq!(conditions.get("SourceIp").unwrap()[0], "2001:db8::1");
    }

    #[test]
    fn test_get_condition_values_uses_verified_client_info() {
        let mut headers = HeaderMap::new();
        headers.insert("x-forwarded-for", HeaderValue::from_static("10.0.0.1"));
        headers.insert("x-forwarded-proto", HeaderValue::from_static("https"));
        let cred = Credentials::default();
        let remote_addr: std::net::SocketAddr = "192.168.0.10:12345".parse().unwrap();
        let client_info = ClientInfo::from_trusted_proxy(
            "10.0.0.1".parse().unwrap(),
            None,
            Some("https".to_string()),
            "192.168.0.10".parse().unwrap(),
            1,
            ValidationMode::Lenient,
            Vec::new(),
        );

        let conditions =
            get_condition_values_with_client_info(&headers, &cred, None, None, Some(remote_addr), Some(&client_info));

        assert_eq!(conditions.get("SourceIp").unwrap()[0], "10.0.0.1");
        assert_eq!(conditions.get("SecureTransport").unwrap()[0], "true");
    }

    #[test]
    fn test_get_condition_values_ignores_unverified_secure_transport_header() {
        let mut headers = HeaderMap::new();
        headers.insert("x-forwarded-proto", HeaderValue::from_static("https"));
        let cred = Credentials::default();

        let conditions = get_condition_values(&headers, &cred, None, None, None);

        assert_eq!(conditions.get("SecureTransport").unwrap()[0], "false");
    }

    // ========== KEYSTONE AUTHENTICATION TESTS ==========

    #[tokio::test]
    async fn test_check_key_valid_keystone_not_enabled() {
        // Test that keystone: access key fails when Keystone is not enabled
        let result = check_key_valid("dummy-token", "keystone:user123").await;

        // Should fail with InvalidAccessKeyId because Keystone is not enabled
        assert!(result.is_err());
        let err = result.unwrap_err();
        assert_eq!(*err.code(), s3s::S3ErrorCode::InvalidAccessKeyId);
    }

    #[tokio::test]
    async fn test_check_key_valid_keystone_no_credentials() {
        use rustfs_keystone::KEYSTONE_CREDENTIALS;

        // Test behavior when Keystone would be enabled but no credentials in task-local
        // This simulates a request that bypassed middleware
        KEYSTONE_CREDENTIALS
            .scope(None, async {
                // Call function that checks for keystone: prefix
                // In real scenario, would check is_keystone_enabled() first
                let access_key = "keystone:user123";
                if access_key.starts_with("keystone:") {
                    // Without credentials in task-local, this should fail
                    let creds_result = KEYSTONE_CREDENTIALS.try_with(|c: &Option<Credentials>| c.clone());
                    assert!(creds_result.is_ok()); // try_with succeeds
                    assert!(creds_result.unwrap().is_none()); // but value is None
                }
            })
            .await;
    }

    #[test]
    fn test_keystone_role_detection_admin() {
        // Test role detection logic for admin role
        let mut claims: HashMap<String, serde_json::Value> = HashMap::new();
        claims.insert("roles".to_string(), json!(["admin", "member"]));

        let is_owner = claims
            .get("roles")
            .and_then(|roles| roles.as_array())
            .map(|roles| {
                roles
                    .iter()
                    .any(|role| role.as_str().map(|r| r == "admin" || r == "reseller_admin").unwrap_or(false))
            })
            .unwrap_or(false);

        assert!(is_owner);
    }

    #[test]
    fn test_keystone_role_detection_reseller_admin() {
        // Test role detection logic for reseller_admin role
        let mut claims: HashMap<String, serde_json::Value> = HashMap::new();
        claims.insert("roles".to_string(), json!(["reseller_admin"]));

        let is_owner = claims
            .get("roles")
            .and_then(|roles| roles.as_array())
            .map(|roles| {
                roles
                    .iter()
                    .any(|role| role.as_str().map(|r| r == "admin" || r == "reseller_admin").unwrap_or(false))
            })
            .unwrap_or(false);

        assert!(is_owner);
    }

    #[test]
    fn test_keystone_role_detection_non_admin() {
        // Test role detection logic for non-admin roles
        let mut claims: HashMap<String, serde_json::Value> = HashMap::new();
        claims.insert("roles".to_string(), json!(["member", "reader"]));

        let is_owner = claims
            .get("roles")
            .and_then(|roles| roles.as_array())
            .map(|roles| {
                roles
                    .iter()
                    .any(|role| role.as_str().map(|r| r == "admin" || r == "reseller_admin").unwrap_or(false))
            })
            .unwrap_or(false);

        assert!(!is_owner);
    }

    #[test]
    fn test_keystone_role_detection_empty() {
        // Test role detection logic for empty roles
        let mut claims: HashMap<String, serde_json::Value> = HashMap::new();
        claims.insert("roles".to_string(), json!([]));

        let is_owner = claims
            .get("roles")
            .and_then(|roles| roles.as_array())
            .map(|roles| {
                roles
                    .iter()
                    .any(|role| role.as_str().map(|r| r == "admin" || r == "reseller_admin").unwrap_or(false))
            })
            .unwrap_or(false);

        assert!(!is_owner);
    }

    #[test]
    fn test_keystone_role_detection_no_claim() {
        // Test role detection logic when roles claim is missing
        let claims: HashMap<String, serde_json::Value> = HashMap::new();

        let is_owner = claims
            .get("roles")
            .and_then(|roles| roles.as_array())
            .map(|roles| {
                roles
                    .iter()
                    .any(|role| role.as_str().map(|r| r == "admin" || r == "reseller_admin").unwrap_or(false))
            })
            .unwrap_or(false);

        assert!(!is_owner);
    }

    #[tokio::test]
    async fn test_keystone_task_local_storage() {
        use rustfs_keystone::KEYSTONE_CREDENTIALS;

        // Test that task-local storage properly stores and retrieves credentials
        let mut claims = HashMap::new();
        claims.insert("project_id".to_string(), json!("project123"));
        claims.insert("roles".to_string(), json!(["member"]));

        let test_creds = Credentials {
            access_key: "keystone:testuser".to_string(),
            secret_key: String::new(),
            session_token: String::new(),
            expiration: None,
            status: "on".to_string(),
            parent_user: "testuser".to_string(),
            groups: None,
            claims: Some(claims),
            name: Some("Test User".to_string()),
            description: None,
        };

        // Outside scope, should fail
        let result = KEYSTONE_CREDENTIALS.try_with(|c: &Option<Credentials>| c.clone());
        assert!(result.is_err());

        // Inside scope, should succeed
        KEYSTONE_CREDENTIALS
            .scope(Some(test_creds.clone()), async {
                let result = KEYSTONE_CREDENTIALS.try_with(|c: &Option<Credentials>| c.clone());
                assert!(result.is_ok());
                let creds = result.unwrap();
                assert!(creds.is_some());
                assert_eq!(creds.unwrap().access_key, "keystone:testuser");
            })
            .await;

        // After scope, should fail again
        let result = KEYSTONE_CREDENTIALS.try_with(|c: &Option<Credentials>| c.clone());
        assert!(result.is_err());
    }
}

#[cfg(test)]
mod tests_policy {
    use rustfs_policy::policy::action::{Action, S3Action};
    use rustfs_policy::policy::{Args, BucketPolicy, BucketPolicyArgs, Policy};
    use std::collections::HashMap;

    #[tokio::test]
    async fn test_iam_policy_source_ip() {
        let policy_json = r#"{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": ["s3:GetObject"],
                    "Resource": ["arn:aws:s3:::mybucket/*"],
                    "Condition": {
                        "IpAddress": {
                            "aws:SourceIp": "192.168.1.0/24"
                        }
                    }
                }
            ]
        }"#;

        let policy: Policy = serde_json::from_str(policy_json).expect("Failed to parse IAM policy");

        // Case 1: Matching IP
        let mut conditions = HashMap::new();
        conditions.insert("SourceIp".to_string(), vec!["192.168.1.10".to_string()]);

        let claims = HashMap::new();
        let args = Args {
            account: "test-account",
            groups: &None,
            action: Action::S3Action(S3Action::GetObjectAction),
            bucket: "mybucket",
            conditions: &conditions,
            is_owner: false,
            object: "myobject",
            claims: &claims,
            deny_only: false,
        };

        assert!(policy.is_allowed(&args).await, "IAM Policy should allow matching IP");

        // Case 2: Non-matching IP
        let mut conditions_fail = HashMap::new();
        conditions_fail.insert("SourceIp".to_string(), vec!["10.0.0.1".to_string()]);

        let args_fail = Args {
            conditions: &conditions_fail,
            ..args
        };

        assert!(!policy.is_allowed(&args_fail).await, "IAM Policy should deny non-matching IP");
    }

    /// The failure this issue is about: when `remote_addr` is dropped the
    /// `aws:SourceIp` key never reaches the condition map, and `AddrFunc::evaluate`
    /// returns `false` for an absent key. That flips two policy shapes in
    /// opposite directions, and only one of them looks like a failure
    /// (rustfs/backlog#1885).
    #[tokio::test]
    async fn source_ip_policies_break_in_both_directions_when_the_key_is_missing() {
        let allow_from_office = |effect: &str| {
            format!(
                r#"{{
                    "Version": "2012-10-17",
                    "Statement": [
                        {{"Effect": "Allow", "Action": ["admin:ConfigUpdate"], "Resource": ["arn:aws:s3:::*"]}},
                        {{
                            "Effect": "{effect}",
                            "Action": ["admin:ConfigUpdate"],
                            "Resource": ["arn:aws:s3:::*"],
                            "Condition": {{"IpAddress": {{"aws:SourceIp": "192.168.1.0/24"}}}}
                        }}
                    ]
                }}"#
            )
        };

        let claims = HashMap::new();
        let groups = None;
        let mut with_ip = HashMap::new();
        with_ip.insert("SourceIp".to_string(), vec!["192.168.1.10".to_string()]);
        let without_ip: HashMap<String, Vec<String>> = HashMap::new();

        let args_with_ip = Args {
            account: "test-account",
            groups: &groups,
            action: Action::AdminAction(rustfs_policy::policy::action::AdminAction::ConfigUpdateAdminAction),
            bucket: "",
            conditions: &with_ip,
            is_owner: false,
            object: "",
            claims: &claims,
            deny_only: false,
        };
        let args_without_ip = Args {
            conditions: &without_ip,
            ..args_with_ip
        };

        // Deny + blacklist: the bypass shape. With the key present the deny
        // matches and the request is refused; drop the key and the deny stops
        // matching, so a source that policy means to block gets through.
        let deny_policy: Policy = serde_json::from_str(&allow_from_office("Deny")).expect("deny policy parses");
        assert!(
            !deny_policy.is_allowed(&args_with_ip).await,
            "a blacklisted source must be refused while aws:SourceIp is present"
        );
        assert!(
            deny_policy.is_allowed(&args_without_ip).await,
            "dropping remote_addr makes the Deny statement unreachable — this is the bypass"
        );

        // Allow + whitelist: the availability shape, and the only one an
        // operator would notice, which is why the bypass above went unseen.
        let allow_policy: Policy = serde_json::from_str(
            r#"{
                "Version": "2012-10-17",
                "Statement": [{
                    "Effect": "Allow",
                    "Action": ["admin:ConfigUpdate"],
                    "Resource": ["arn:aws:s3:::*"],
                    "Condition": {"IpAddress": {"aws:SourceIp": "192.168.1.0/24"}}
                }]
            }"#,
        )
        .expect("allow policy parses");
        assert!(
            allow_policy.is_allowed(&args_with_ip).await,
            "a whitelisted source must be allowed while aws:SourceIp is present"
        );
        assert!(
            !allow_policy.is_allowed(&args_without_ip).await,
            "dropping remote_addr locks out a legitimate admin"
        );
    }

    #[tokio::test]
    async fn test_bucket_policy_source_ip() {
        let policy_json = r#"{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {"AWS": ["*"]},
                    "Action": ["s3:GetObject"],
                    "Resource": ["arn:aws:s3:::mybucket/*"],
                    "Condition": {
                        "IpAddress": {
                            "aws:SourceIp": "192.168.1.0/24"
                        }
                    }
                }
            ]
        }"#;

        let policy: BucketPolicy = serde_json::from_str(policy_json).expect("Failed to parse Bucket policy");

        // Case 1: Matching IP
        let mut conditions = HashMap::new();
        conditions.insert("SourceIp".to_string(), vec!["192.168.1.10".to_string()]);

        let args = BucketPolicyArgs {
            account: "test-account",
            groups: &None,
            action: Action::S3Action(S3Action::GetObjectAction),
            bucket: "mybucket",
            conditions: &conditions,
            is_owner: false,
            object: "myobject",
        };

        assert!(policy.is_allowed(&args).await, "Bucket Policy should allow matching IP");

        // Case 2: Non-matching IP
        let mut conditions_fail = HashMap::new();
        conditions_fail.insert("SourceIp".to_string(), vec!["10.0.0.1".to_string()]);

        let args_fail = BucketPolicyArgs {
            conditions: &conditions_fail,
            ..args
        };

        assert!(!policy.is_allowed(&args_fail).await, "Bucket Policy should deny non-matching IP");
    }
}
