// 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 super::iam_error::iam_error_to_s3_error;
use crate::admin::access_key_identity;
use crate::admin::handlers::site_replication::{encode_service_account_replication_policy, site_replication_iam_change_hook};
use crate::admin::runtime_sources::current_action_credentials;
use crate::admin::utils::{encode_compatible_admin_payload, has_space_be, is_compat_admin_request, read_compatible_admin_body};
use crate::auth::{constant_time_eq, get_condition_values, get_session_token};
use crate::server::{ADMIN_PREFIX, RemoteAddr};
use crate::{
    admin::router::{AdminOperation, Operation, S3Router},
    auth::check_key_valid,
};
use http::HeaderMap;
use hyper::{Method, StatusCode};
use matchit::Params;
use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE;
use rustfs_credentials::Credentials as StoredCredentials;
use rustfs_iam::error::{is_err_no_such_service_account, is_err_no_such_temp_account};
use rustfs_iam::store::Store as IamStore;
use rustfs_iam::sys::{NewServiceAccountOpts, UpdateServiceAccountOpts};
use rustfs_madmin::{
    ACCESS_KEY_LIST_ALL, ACCESS_KEY_LIST_STS_ONLY, ACCESS_KEY_LIST_SVCACC_ONLY, ACCESS_KEY_LIST_USERS_ONLY, AddServiceAccountReq,
    AddServiceAccountResp, Credentials, InfoServiceAccountResp, ListAccessKeysResp, ListServiceAccountsResp,
    SITE_REPL_API_VERSION, SRIAMItem, SRSessionPolicy, SRSvcAccChange, SRSvcAccCreate, SRSvcAccDelete, SRSvcAccUpdate,
    ServiceAccountInfo, TemporaryAccountInfoResp, UpdateServiceAccountReq,
};
use rustfs_policy::policy::action::{Action, AdminAction};
use rustfs_policy::policy::{Args, DEFAULT_VERSION, Policy};
use s3s::S3ErrorCode::InvalidRequest;
use s3s::header::CONTENT_LENGTH;
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::Deserialize;
use serde_urlencoded::from_bytes;
use std::collections::HashMap;
use time::OffsetDateTime;
use tracing::{debug, warn};
use url::form_urlencoded;

const LOG_COMPONENT_ADMIN: &str = "admin";
const LOG_SUBSYSTEM_SERVICE_ACCOUNT: &str = "service_account";
const EVENT_ADMIN_SERVICE_ACCOUNT_STATE: &str = "admin_service_account_state";
const EMPTY_EXPLICIT_SERVICE_ACCOUNT_POLICY: &str =
    r#"{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::*"]}]}"#;

fn sr_session_policy_from_policy(policy: Option<&Policy>) -> S3Result<SRSessionPolicy> {
    let Some(policy) = policy else {
        return Ok(SRSessionPolicy::default());
    };

    let raw = serde_json::to_string(policy).map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?;
    SRSessionPolicy::from_json(&raw).map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))
}

fn normalize_service_account_policy(mut policy: Policy) -> S3Result<Policy> {
    if policy.statements.is_empty() && (!policy.id.is_empty() || !policy.version.is_empty()) {
        let id = policy.id;
        policy = Policy::parse_config(EMPTY_EXPLICIT_SERVICE_ACCOUNT_POLICY.as_bytes())
            .map_err(|e| s3_error!(InternalError, "parse empty service account policy failed: {:?}", e))?;
        policy.id = id;
    } else if policy.version.is_empty() && !policy.statements.is_empty() {
        policy.version = DEFAULT_VERSION.to_string();
    }
    Ok(policy)
}

fn parse_new_service_account_policy(policy: Option<&serde_json::Value>) -> S3Result<Option<Policy>> {
    let Some(policy) = policy else {
        return Ok(None);
    };
    let policy = normalize_service_account_policy(parse_service_account_policy(policy)?)?;
    Ok((!policy.id.is_empty() || !policy.version.is_empty() || !policy.statements.is_empty()).then_some(policy))
}

fn compat_time_sentinel() -> OffsetDateTime {
    OffsetDateTime::UNIX_EPOCH
}

fn list_expiration_or_sentinel(expiration: Option<OffsetDateTime>) -> Option<OffsetDateTime> {
    Some(expiration.unwrap_or_else(compat_time_sentinel))
}

fn expiration_for_admin_path(path: &str, expiration: Option<OffsetDateTime>) -> Option<OffsetDateTime> {
    if is_compat_admin_request(path) {
        list_expiration_or_sentinel(expiration)
    } else {
        expiration
    }
}

fn delete_service_account_success_status(path: &str) -> StatusCode {
    if is_compat_admin_request(path) {
        StatusCode::NO_CONTENT
    } else {
        StatusCode::OK
    }
}

fn is_service_account_owner_of(caller: &StoredCredentials, target_parent_user: &str) -> bool {
    let caller_parent = if caller.parent_user.is_empty() {
        caller.access_key.as_str()
    } else {
        caller.parent_user.as_str()
    };

    caller_parent == target_parent_user
}

/// GHSA-5354: whether a caller resolving to `owner` may create a service account
/// parented to `target_user`, given the caller's own key (`req_user`) and its
/// effective parent (`req_parent_user`, which equals `req_user` for a top-level
/// user or the parent for a derived credential).
///
/// The `CreateServiceAccountAdminAction` permission gates *whether* a caller may
/// create service accounts, not *for whom*. A non-owner is therefore confined to
/// its own scope; only an owner may target another user — including the root
/// credential, whose existence check is deliberately lenient. Without this a
/// caller holding only that action could pass `target_user = <root access key>`
/// and mint a root-parented service account that authenticates as owner via
/// `prepare_service_account_auth`, a full-takeover privilege escalation. Mirrors
/// `imported_service_account_parent_allowed` on the ImportIam path (GHSA-566f).
fn add_service_account_parent_within_scope(target_user: &str, req_user: &str, req_parent_user: &str, owner: bool) -> bool {
    owner || target_user == req_user || target_user == req_parent_user
}

/// Fail-closed ownership check used by DeleteServiceAccount for a caller that
/// lacks the RemoveServiceAccount admin action.
///
/// Returns `true` only when the target service account was successfully loaded
/// AND is owned by the caller. A `None` target — meaning `get_service_account`
/// failed with a non-not-found error (e.g. a stored-credential decode error) so
/// ownership could not be verified — denies, as does an owner mismatch. The
/// previous `svc_account.is_some_and(|v| v.parent_user != user)` form returned
/// `false` for a `None` target and thus fell through to the delete, an authz
/// bypass (backlog#806).
fn non_admin_may_delete_service_account(caller_user: &str, target_parent_user: Option<&str>) -> bool {
    matches!(target_parent_user, Some(parent) if parent == caller_user)
}

fn map_service_account_lookup_error(err: rustfs_iam::error::Error, action: &str) -> S3Error {
    debug!(
        component = LOG_COMPONENT_ADMIN,
        subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
        event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
        action,
        result = "lookup_failed",
        error = ?err,
        "admin service account state"
    );
    if is_err_no_such_service_account(&err) {
        iam_error_to_s3_error(err)
    } else {
        s3_error!(InternalError, "{action}")
    }
}

fn map_temp_account_lookup_error(err: rustfs_iam::error::Error, action: &str) -> S3Error {
    debug!(
        component = LOG_COMPONENT_ADMIN,
        subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
        event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
        action,
        result = "temporary_lookup_failed",
        error = ?err,
        "admin service account state"
    );
    if is_err_no_such_temp_account(&err) {
        iam_error_to_s3_error(err)
    } else {
        s3_error!(InternalError, "{action}")
    }
}

fn parse_service_account_policy(policy: &serde_json::Value) -> S3Result<Policy> {
    let policy_bytes = serde_json::to_vec(policy).map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?;
    Policy::parse_config(&policy_bytes).map_err(|e| {
        debug!(
            component = LOG_COMPONENT_ADMIN,
            subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
            event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
            result = "policy_parse_failed",
            error = ?e,
            "admin service account state"
        );
        match e {
            rustfs_policy::error::Error::PolicyError(rustfs_policy::policy::Error::NonResource) => {
                s3_error!(InvalidArgument, "invalid service account policy: Resource is empty")
            }
            rustfs_policy::error::Error::PolicyError(err) => {
                s3_error!(InvalidArgument, "invalid service account policy: {}", err)
            }
            rustfs_policy::error::Error::Io(err) if err.get_ref().is_some_and(|source| source.is::<serde_json::Error>()) => {
                s3_error!(
                    InvalidArgument,
                    "Policy format is invalid. Please check the JSON structure and ensure it follows the IAM policy format."
                )
            }
            err => s3_error!(InvalidArgument, "invalid service account policy: {}", err),
        }
    })
}

fn parse_update_service_account_policy(new_policy: Option<serde_json::Value>) -> S3Result<Option<Policy>> {
    let Some(policy) = new_policy else {
        return Ok(None);
    };

    Ok(Some(parse_service_account_policy(&policy)?))
}

fn service_account_update_replication_change(
    access_key: &str,
    update: &UpdateServiceAccountReq,
    session_policy: Option<&Policy>,
) -> S3Result<SRSvcAccChange> {
    Ok(SRSvcAccChange {
        update: Some(SRSvcAccUpdate {
            access_key: access_key.to_string(),
            secret_key: update.new_secret_key.clone().unwrap_or_default(),
            status: update.new_status.clone().unwrap_or_default(),
            name: update.new_name.clone().unwrap_or_default(),
            description: update.new_description.clone().unwrap_or_default(),
            session_policy: sr_session_policy_from_policy(session_policy)?,
            expiration: update.new_expiration,
            api_version: Some(SITE_REPL_API_VERSION.to_string()),
        }),
        api_version: Some(SITE_REPL_API_VERSION.to_string()),
        ..Default::default()
    })
}

pub fn register_service_account_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
    r.insert(
        Method::POST,
        format!("{}{}", ADMIN_PREFIX, "/v3/update-service-account").as_str(),
        AdminOperation(&UpdateServiceAccount {}),
    )?;

    r.insert(
        Method::GET,
        format!("{}{}", ADMIN_PREFIX, "/v3/info-service-account").as_str(),
        AdminOperation(&InfoServiceAccount {}),
    )?;
    r.insert(
        Method::GET,
        format!("{}{}", ADMIN_PREFIX, "/v3/temporary-account-info").as_str(),
        AdminOperation(&TemporaryAccountInfo {}),
    )?;
    r.insert(
        Method::GET,
        format!("{}{}", ADMIN_PREFIX, "/v3/info-access-key").as_str(),
        AdminOperation(&InfoAccessKey {}),
    )?;

    r.insert(
        Method::GET,
        format!("{}{}", ADMIN_PREFIX, "/v3/list-service-accounts").as_str(),
        AdminOperation(&ListServiceAccount {}),
    )?;
    r.insert(
        Method::GET,
        format!("{}{}", ADMIN_PREFIX, "/v3/list-access-keys-bulk").as_str(),
        AdminOperation(&ListAccessKeysBulk {}),
    )?;

    r.insert(
        Method::DELETE,
        format!("{}{}", ADMIN_PREFIX, "/v3/delete-service-accounts").as_str(),
        AdminOperation(&DeleteServiceAccount {}),
    )?;
    r.insert(
        Method::DELETE,
        format!("{}{}", ADMIN_PREFIX, "/v3/delete-service-account").as_str(),
        AdminOperation(&DeleteServiceAccount {}),
    )?;

    r.insert(
        Method::PUT,
        format!("{}{}", ADMIN_PREFIX, "/v3/add-service-accounts").as_str(),
        AdminOperation(&AddServiceAccount {}),
    )?;
    r.insert(
        Method::PUT,
        format!("{}{}", ADMIN_PREFIX, "/v3/add-service-account").as_str(),
        AdminOperation(&AddServiceAccount {}),
    )?;

    Ok(())
}

pub struct AddServiceAccount {}
#[async_trait::async_trait]
impl Operation for AddServiceAccount {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let Some(req_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &req_cred.access_key).await?;

        let body = read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), &cred.secret_key).await?;

        let create_req: AddServiceAccountReq =
            serde_json::from_slice(&body[..]).map_err(|e| s3_error!(InvalidRequest, "unmarshal body failed, e: {:?}", e))?;

        // create_req.expiration = create_req.expiration.and_then(|expire| expire.replace_millisecond(0).ok());

        if has_space_be(&create_req.access_key) {
            return Err(s3_error!(InvalidRequest, "access key has spaces"));
        }

        create_req.validate().map_err(|e| S3Error::with_message(InvalidRequest, e))?;

        let session_policy = parse_new_service_account_policy(create_req.policy.as_ref())?;
        let replication_policy = session_policy
            .as_ref()
            .map(serde_json::to_string)
            .transpose()
            .map_err(|e| s3_error!(InvalidArgument, "marshal policy failed: {:?}", e))?;

        let Some(sys_cred) = current_action_credentials() else {
            return Err(s3_error!(InvalidRequest, "get sys cred failed"));
        };

        if constant_time_eq(&sys_cred.access_key, &create_req.access_key) {
            return Err(s3_error!(InvalidArgument, "can't create user with system access key"));
        }

        let mut target_user = if let Some(u) = create_req.target_user {
            u
        } else {
            cred.access_key.clone()
        };

        let req_user = cred.access_key.clone();
        let mut req_parent_user = cred.access_key.clone();
        let req_groups = cred.groups.clone();
        let mut req_is_derived_cred = false;

        if cred.is_service_account() || cred.is_temp() {
            req_parent_user = cred.parent_user.clone();
            req_is_derived_cred = true;
        }

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        // This family deliberately calls `is_allowed` directly instead of going
        // through `validate_admin_request`, and must keep doing so
        // (backlog#1886). The shared helper returns as soon as *any* candidate
        // action is allowed — an OR. The checks here are an AND: each one must
        // pass, and a later stage additionally needs `owner` for the GHSA-5354
        // parent-scope guard and drives `deny_only` dynamically. Replacing these
        // with the shared gate would widen authorization.
        if !iam_store
            .is_allowed(&Args {
                account: &cred.access_key,
                groups: &cred.groups,
                action: Action::AdminAction(AdminAction::CreateServiceAccountAdminAction),
                bucket: "",
                conditions: &get_condition_values(
                    &req.headers,
                    &cred,
                    None,
                    None,
                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                ),
                is_owner: owner,
                object: "",
                claims: cred.claims_or_empty(),
                deny_only: false, // Always require explicit Allow permission
            })
            .await
        {
            return Err(s3_error!(AccessDenied, "access denied"));
        }

        if target_user != cred.access_key {
            let has_user = iam_store.get_user(&target_user).await;
            if has_user.is_none() && target_user != sys_cred.access_key {
                return Err(s3_error!(InvalidRequest, "target user not exist"));
            }
        }

        let is_svc_acc = target_user == req_user || target_user == req_parent_user;
        // GHSA-5354: confine a non-owner caller to its own scope so it cannot mint
        // a root-parented (owner) service account. Evaluated on the original
        // `target_user` the caller submitted, before the derived-credential rewrite
        // to `req_parent_user` below, so the guard is bound to the parent actually
        // requested. Its allow set is exactly `owner || is_svc_acc`. See the helper's
        // doc comment.
        if !add_service_account_parent_within_scope(&target_user, &req_user, &req_parent_user, owner) {
            return Err(s3_error!(AccessDenied, "service account parent is outside requester scope"));
        }

        let mut target_groups = None;
        let opts = NewServiceAccountOpts {
            access_key: create_req.access_key,
            secret_key: create_req.secret_key,
            name: create_req.name,
            description: create_req.description.or(create_req.comment),
            expiration: create_req.expiration,
            session_policy,
            ..Default::default()
        };

        if is_svc_acc {
            if req_is_derived_cred {
                if req_parent_user.is_empty() {
                    return Err(s3_error!(AccessDenied, "only derived cred can create service account"));
                }
                target_user = req_parent_user;
            }

            target_groups = req_groups;
        }

        let replication_groups = target_groups.clone().unwrap_or_default();
        let replication_name = opts.name.clone().unwrap_or_default();
        let replication_description = opts.description.clone().unwrap_or_default();
        let replication_expiration = opts.expiration;

        let create_result = if is_svc_acc {
            iam_store
                .new_service_account_from_caller(&target_user, target_groups, opts, &cred)
                .await
        } else {
            let replication_claims = opts.claims.clone().unwrap_or_default();
            iam_store
                .new_service_account(&target_user, target_groups, opts)
                .await
                .map(|(credentials, updated_at)| (credentials, updated_at, replication_claims))
        };
        let (new_cred, updated_at, replication_claims) = create_result.map_err(|e| {
            debug!(
                component = LOG_COMPONENT_ADMIN,
                subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
                target_user = %target_user,
                result = "create_failed",
                error = ?e,
                "admin service account state"
            );
            match e {
                rustfs_iam::error::Error::InvalidAccessKeyLength
                | rustfs_iam::error::Error::InvalidSecretKeyLength
                | rustfs_iam::error::Error::AccessKeyAlreadyExists => iam_error_to_s3_error(e),
                rustfs_iam::error::Error::IAMActionNotAllowed => s3_error!(AccessDenied, "access denied"),
                err => s3_error!(InternalError, "create service account failed, e: {:?}", err),
            }
        })?;
        let (replication_session_policy, oidc_service_account_envelope) =
            encode_service_account_replication_policy(&replication_claims, replication_policy.as_deref())?;

        if let Err(err) = site_replication_iam_change_hook(SRIAMItem {
            r#type: "service-account".to_string(),
            svc_acc_change: Some(SRSvcAccChange {
                create: Some(SRSvcAccCreate {
                    parent: target_user.clone(),
                    access_key: new_cred.access_key.clone(),
                    secret_key: new_cred.secret_key.clone(),
                    groups: replication_groups,
                    claims: replication_claims,
                    session_policy: replication_session_policy,
                    status: String::new(),
                    name: replication_name,
                    description: replication_description,
                    expiration: replication_expiration,
                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
                }),
                oidc_service_account_envelope,
                api_version: Some(SITE_REPL_API_VERSION.to_string()),
                ..Default::default()
            }),
            updated_at: Some(updated_at),
            api_version: Some(SITE_REPL_API_VERSION.to_string()),
            ..Default::default()
        })
        .await
        {
            warn!(
                component = LOG_COMPONENT_ADMIN,
                subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
                access_key = %new_cred.access_key,
                action = "create",
                result = "site_replication_hook_failed",
                error = ?err,
                "admin service account state"
            );
        }

        let resp = AddServiceAccountResp {
            credentials: Credentials {
                access_key: &new_cred.access_key,
                secret_key: &new_cred.secret_key,
                session_token: None,
                expiration: new_cred.expiration,
            },
        };

        let body = serde_json::to_vec(&resp).map_err(|e| s3_error!(InternalError, "marshal body failed, e: {:?}", e))?;
        let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));

        Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
    }
}

#[derive(Debug, Default, Deserialize)]
struct AccessKeyQuery {
    #[serde(rename = "accessKey", alias = "access-key")]
    pub access_key: String,
}

fn request_user_name(cred: &StoredCredentials) -> &str {
    if cred.parent_user.is_empty() {
        &cred.access_key
    } else {
        &cred.parent_user
    }
}

fn can_fallback_view_access_key_info(requester: &StoredCredentials, target: &StoredCredentials) -> bool {
    if requester.is_service_account() {
        return false;
    }

    if target.is_temp() || target.is_service_account() {
        return request_user_name(requester) == target.parent_user;
    }

    request_user_name(requester) == target.access_key
}

fn can_fallback_view_access_key_info_after_admin_check_failed(
    requester: &StoredCredentials,
    target: &StoredCredentials,
    no_explicit_deny: bool,
) -> bool {
    no_explicit_deny && can_fallback_view_access_key_info(requester, target)
}

async fn build_info_service_account_resp<T: IamStore>(
    iam_store: &rustfs_iam::sys::IamSys<T>,
    account: &StoredCredentials,
    session_policy: Option<rustfs_policy::policy::Policy>,
) -> S3Result<InfoServiceAccountResp> {
    access_key_identity::build_info_service_account_resp(iam_store, account, session_policy).await
}

pub struct UpdateServiceAccount {}
#[async_trait::async_trait]
impl Operation for UpdateServiceAccount {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let query = {
            if let Some(query) = req.uri.query() {
                let input: AccessKeyQuery =
                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed1"))?;
                input
            } else {
                AccessKeyQuery::default()
            }
        };

        if query.access_key.is_empty() {
            return Err(s3_error!(InvalidArgument, "access key is empty"));
        }

        let access_key = query.access_key;

        let Some(input_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        // let svc_account = iam_store.get_service_account(&access_key).await.map_err(|e| {
        //     debug!("get service account failed, e: {:?}", e);
        //     s3_error!(InternalError, "get service account failed")
        // })?;

        let body =
            read_compatible_admin_body(req.input, MAX_ADMIN_REQUEST_BODY_SIZE, req.uri.path(), input_cred.secret_key.expose())
                .await?;

        let update_req: UpdateServiceAccountReq =
            serde_json::from_slice(&body[..]).map_err(|e| s3_error!(InvalidRequest, "unmarshal body failed, e: {:?}", e))?;

        update_req.validate().map_err(|e| S3Error::with_message(InvalidRequest, e))?;

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;

        if !iam_store
            .is_allowed(&Args {
                account: &cred.access_key,
                groups: &cred.groups,
                action: Action::AdminAction(AdminAction::UpdateServiceAccountAdminAction),
                bucket: "",
                conditions: &get_condition_values(
                    &req.headers,
                    &cred,
                    None,
                    None,
                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                ),
                is_owner: owner,
                object: "",
                claims: cred.claims_or_empty(),
                deny_only: false,
            })
            .await
        {
            return Err(s3_error!(AccessDenied, "access denied"));
        }

        let (svc_account, _) = iam_store
            .get_service_account(&access_key)
            .await
            .map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;

        // The admin action permits updates within a caller's scope; only an
        // owner may cross service-account parent boundaries.
        if !owner && !is_service_account_owner_of(&cred, &svc_account.parent_user) {
            return Err(s3_error!(AccessDenied, "access denied"));
        }

        let new_secret_key = update_req.new_secret_key.clone();
        let new_status = update_req.new_status.clone();
        let new_name = update_req.new_name.clone();
        let new_description = update_req.new_description.clone();
        let new_expiration = update_req.new_expiration;
        let new_policy = update_req.new_policy.clone();

        let sp = parse_update_service_account_policy(new_policy.clone())?;
        let svc_acc_change = service_account_update_replication_change(&access_key, &update_req, sp.as_ref())?;

        let opts = UpdateServiceAccountOpts {
            secret_key: new_secret_key.clone(),
            status: new_status.clone(),
            name: new_name.clone(),
            description: new_description.clone(),
            expiration: new_expiration,
            session_policy: sp,
            parent_user: None,
            allow_site_replicator_account: false,
        };

        let updated_at = iam_store
            .update_service_account(&access_key, opts)
            .await
            .map_err(|e| map_service_account_lookup_error(e, "update service account failed"))?;

        if let Err(err) = site_replication_iam_change_hook(SRIAMItem {
            r#type: "service-account".to_string(),
            svc_acc_change: Some(svc_acc_change),
            updated_at: Some(updated_at),
            api_version: Some(SITE_REPL_API_VERSION.to_string()),
            ..Default::default()
        })
        .await
        {
            warn!(access_key = %access_key, error = ?err, "site replication update service account hook failed");
        }

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
        header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
        Ok(S3Response::with_headers((StatusCode::NO_CONTENT, Body::empty()), header))
    }
}

pub struct InfoServiceAccount {}
#[async_trait::async_trait]
impl Operation for InfoServiceAccount {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let query = {
            if let Some(query) = req.uri.query() {
                let input: AccessKeyQuery =
                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed1"))?;
                input
            } else {
                AccessKeyQuery::default()
            }
        };

        if query.access_key.is_empty() {
            return Err(s3_error!(InvalidArgument, "access key is empty"));
        }

        let access_key = query.access_key;

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        let (svc_account, session_policy) = iam_store
            .get_service_account(&access_key)
            .await
            .map_err(|e| map_service_account_lookup_error(e, "get service account failed"))?;

        let Some(input_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;

        if !iam_store
            .is_allowed(&Args {
                account: &cred.access_key,
                groups: &cred.groups,
                action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction),
                bucket: "",
                conditions: &get_condition_values(
                    &req.headers,
                    &cred,
                    None,
                    None,
                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                ),
                is_owner: owner,
                object: "",
                claims: cred.claims_or_empty(),
                deny_only: false,
            })
            .await
            && request_user_name(&cred) != svc_account.parent_user
        {
            return Err(s3_error!(AccessDenied, "access denied"));
        }

        let resp = build_info_service_account_resp(&iam_store, &svc_account, session_policy).await?;

        let body = serde_json::to_vec(&resp).map_err(|e| s3_error!(InternalError, "marshal body failed, e: {:?}", e))?;
        let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));

        Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
    }
}

pub struct TemporaryAccountInfo {}
#[async_trait::async_trait]
impl Operation for TemporaryAccountInfo {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let query = {
            if let Some(query) = req.uri.query() {
                let input: AccessKeyQuery =
                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed1"))?;
                input
            } else {
                AccessKeyQuery::default()
            }
        };

        if query.access_key.is_empty() {
            return Err(s3_error!(InvalidArgument, "access key is empty"));
        }

        let Some(input_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        if !iam_store
            .is_allowed(&Args {
                account: &cred.access_key,
                groups: &cred.groups,
                action: Action::AdminAction(AdminAction::ListTemporaryAccountsAdminAction),
                bucket: "",
                conditions: &get_condition_values(
                    &req.headers,
                    &cred,
                    None,
                    None,
                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                ),
                is_owner: owner,
                object: "",
                claims: cred.claims_or_empty(),
                deny_only: false,
            })
            .await
        {
            return Err(s3_error!(AccessDenied, "access denied"));
        }

        let (temp_account, session_policy) = iam_store
            .get_temporary_account(&query.access_key)
            .await
            .map_err(|e| map_temp_account_lookup_error(e, "get temporary account failed"))?;

        let resp: TemporaryAccountInfoResp = build_info_service_account_resp(&iam_store, &temp_account, session_policy).await?;
        let body = serde_json::to_vec(&resp).map_err(|e| s3_error!(InternalError, "marshal body failed, e: {:?}", e))?;
        let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));

        Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
    }
}

pub struct InfoAccessKey {}
#[async_trait::async_trait]
impl Operation for InfoAccessKey {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let query = {
            if let Some(query) = req.uri.query() {
                let input: AccessKeyQuery =
                    from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed1"))?;
                input
            } else {
                AccessKeyQuery::default()
            }
        };

        let Some(input_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;

        let access_key = if query.access_key.is_empty() {
            cred.access_key.clone()
        } else {
            query.access_key
        };

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        let target_cred = iam_store.get_user(&access_key).await.map(|identity| identity.credentials);

        let conditions = get_condition_values(
            &req.headers,
            &cred,
            None,
            None,
            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
        );
        let claims = cred.claims_or_empty();

        if !iam_store
            .is_allowed(&Args {
                account: &cred.access_key,
                groups: &cred.groups,
                action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction),
                bucket: "",
                conditions: &conditions,
                is_owner: owner,
                object: "",
                claims,
                deny_only: false,
            })
            .await
        {
            let no_explicit_deny = iam_store
                .is_allowed(&Args {
                    account: &cred.access_key,
                    groups: &cred.groups,
                    action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction),
                    bucket: "",
                    conditions: &conditions,
                    is_owner: owner,
                    object: "",
                    claims,
                    deny_only: true,
                })
                .await;
            if !no_explicit_deny {
                return Err(s3_error!(AccessDenied, "access denied"));
            }

            let Some(target_cred) = target_cred.as_ref() else {
                return Err(s3_error!(AccessDenied, "access denied"));
            };

            if !can_fallback_view_access_key_info_after_admin_check_failed(&cred, target_cred, no_explicit_deny) {
                return Err(s3_error!(AccessDenied, "access denied"));
            }
        }

        let Some(target_cred) = target_cred else {
            return Err(s3_error!(InvalidRequest, "access key not exist"));
        };

        let resp = access_key_identity::resolve_info_access_key_resp(&iam_store, access_key, target_cred).await?;

        let body = serde_json::to_vec(&resp).map_err(|e| s3_error!(InternalError, "marshal body failed, e: {:?}", e))?;
        let (body, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, body)?;

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));

        Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), header))
    }
}

#[derive(Debug, Default, serde::Deserialize)]
pub struct ListServiceAccountQuery {
    pub user: Option<String>,
}

pub struct ListServiceAccount {}
#[async_trait::async_trait]
impl Operation for ListServiceAccount {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let query = {
            if let Some(query) = req.uri.query() {
                let input: ListServiceAccountQuery = from_bytes(query.as_bytes())
                    .map_err(|_e| s3_error!(InvalidArgument, "invalid service account query parameters"))?;
                input
            } else {
                ListServiceAccountQuery::default()
            }
        };

        let Some(input_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key)
                .await
                .map_err(|e| {
                    debug!(
                        component = LOG_COMPONENT_ADMIN,
                        subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                        event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
                        result = "check_key_failed",
                        error = ?e,
                        "admin service account state"
                    );
                    s3_error!(InternalError, "check key failed")
                })?;

        // let target_account = if let Some(user) = query.user {
        //     if user != input_cred.access_key {
        //         user
        //     } else if cred.parent_user.is_empty() {
        //         input_cred.access_key
        //     } else {
        //         cred.parent_user
        //     }
        // } else if cred.parent_user.is_empty() {
        //     input_cred.access_key
        // } else {
        //     cred.parent_user
        // };

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        let target_account = if query.user.as_ref().is_some_and(|v| v != &cred.access_key) {
            // Cross-user listing must be authorized by ListServiceAccounts, matching the
            // sibling InfoServiceAccount/InfoAccessKey/ListAccessKeysBulk handlers.
            if !iam_store
                .is_allowed(&Args {
                    account: &cred.access_key,
                    groups: &cred.groups,
                    action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction),
                    bucket: "",
                    conditions: &get_condition_values(
                        &req.headers,
                        &cred,
                        None,
                        None,
                        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                    ),
                    is_owner: owner,
                    object: "",
                    claims: cred.claims_or_empty(),
                    deny_only: false,
                })
                .await
            {
                return Err(s3_error!(AccessDenied, "access denied"));
            }

            query.user.unwrap_or_default()
        } else if cred.parent_user.is_empty() {
            cred.access_key
        } else {
            cred.parent_user
        };

        let service_accounts = iam_store.list_service_accounts(&target_account).await.map_err(|e| {
            debug!(
                component = LOG_COMPONENT_ADMIN,
                subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
                target_user = %target_account,
                result = "list_service_accounts_failed",
                error = ?e,
                "admin service account state"
            );
            s3_error!(InternalError, "list service account failed")
        })?;

        let accounts: Vec<ServiceAccountInfo> = service_accounts
            .into_iter()
            .map(|sa| ServiceAccountInfo {
                parent_user: sa.parent_user.clone(),
                account_status: sa.status.clone(),
                implied_policy: sa.is_implied_policy(), // or set according to your logic
                access_key: sa.access_key,
                name: sa.name,
                description: sa.description,
                expiration: expiration_for_admin_path(req.uri.path(), sa.expiration),
            })
            .collect();

        let data = serde_json::to_vec(&ListServiceAccountsResp { accounts })
            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal users err {e}")))?;
        let (data, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, data)?;

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));

        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
    }
}

#[derive(Debug, Deserialize)]
struct ListAccessKeysQuery {
    users: Vec<String>,
    all: bool,
    list_type: String,
}

fn default_access_key_list_type() -> String {
    ACCESS_KEY_LIST_ALL.to_string()
}

impl Default for ListAccessKeysQuery {
    fn default() -> Self {
        Self {
            users: Vec::new(),
            all: false,
            list_type: default_access_key_list_type(),
        }
    }
}

fn parse_bool_param(value: &str) -> bool {
    matches!(value, "true" | "1" | "on" | "yes")
}

fn parse_list_access_keys_query(query: Option<&str>) -> ListAccessKeysQuery {
    let mut parsed = ListAccessKeysQuery::default();

    let Some(query) = query else {
        return parsed;
    };

    for (key, value) in form_urlencoded::parse(query.as_bytes()) {
        match key.as_ref() {
            "users" if !value.is_empty() => {
                parsed.users.push(value.into_owned());
            }
            "all" => parsed.all = parse_bool_param(value.as_ref()),
            "listType" => parsed.list_type = value.into_owned(),
            _ => {}
        }
    }

    parsed
}

pub struct ListAccessKeysBulk {}
#[async_trait::async_trait]
impl Operation for ListAccessKeysBulk {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let query = { parse_list_access_keys_query(req.uri.query()) };

        if query.all && !query.users.is_empty() {
            return Err(s3_error!(InvalidRequest, "either specify users or all, not both"));
        }

        let Some(input_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        let mut requested_users = query.users;
        let mut self_only = !query.all && requested_users.is_empty();
        if !query.all && requested_users.len() == 1 {
            let requested_user = &requested_users[0];
            if requested_user == &cred.access_key || requested_user == &cred.parent_user {
                self_only = true;
            }
        }

        if query.all
            && !iam_store
                .is_allowed(&Args {
                    account: &cred.access_key,
                    groups: &cred.groups,
                    action: Action::AdminAction(AdminAction::ListUsersAdminAction),
                    bucket: "",
                    conditions: &get_condition_values(
                        &req.headers,
                        &cred,
                        None,
                        None,
                        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                    ),
                    is_owner: owner,
                    object: "",
                    claims: cred.claims_or_empty(),
                    deny_only: false,
                })
                .await
        {
            return Err(s3_error!(AccessDenied, "access denied"));
        }

        if !iam_store
            .is_allowed(&Args {
                account: &cred.access_key,
                groups: &cred.groups,
                action: Action::AdminAction(AdminAction::ListServiceAccountsAdminAction),
                bucket: "",
                conditions: &get_condition_values(
                    &req.headers,
                    &cred,
                    None,
                    None,
                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                ),
                is_owner: owner,
                object: "",
                claims: cred.claims_or_empty(),
                deny_only: self_only,
            })
            .await
        {
            return Err(s3_error!(AccessDenied, "access denied"));
        }

        if self_only && requested_users.is_empty() {
            requested_users.push(request_user_name(&cred).to_string());
        }

        let checked_user_list = if query.all {
            let mut users = iam_store
                .list_users()
                .await
                .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("list users err {e}")))?
                .into_keys()
                .collect::<Vec<_>>();

            if let Some(sys_cred) = current_action_credentials() {
                users.push(sys_cred.access_key);
            }
            users
        } else {
            // Keep requested identities as-is. Some valid parent users (for example external
            // identities) may not be persisted as regular IAM users, but can still own keys.
            requested_users
        };

        let (list_sts_keys, list_service_accounts) = match query.list_type.as_str() {
            ACCESS_KEY_LIST_USERS_ONLY => (false, false),
            ACCESS_KEY_LIST_STS_ONLY => (true, false),
            ACCESS_KEY_LIST_SVCACC_ONLY => (false, true),
            ACCESS_KEY_LIST_ALL => (true, true),
            _ => return Err(s3_error!(InvalidRequest, "invalid list type")),
        };

        let mut access_key_map = HashMap::new();
        for user in checked_user_list {
            let mut access_keys = ListAccessKeysResp::default();

            if list_sts_keys {
                let sts_keys = iam_store.list_sts_accounts(&user).await.map_err(|e| {
                    debug!(
                        component = LOG_COMPONENT_ADMIN,
                        subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                        event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
                        target_user = %user,
                        result = "list_sts_accounts_failed",
                        error = ?e,
                        "admin service account state"
                    );
                    s3_error!(InternalError, "list sts account failed")
                })?;

                access_keys.sts_keys = sts_keys
                    .into_iter()
                    .map(|sts| {
                        access_key_identity::list_entry_from_credentials(
                            &sts,
                            expiration_for_admin_path(req.uri.path(), sts.expiration),
                        )
                    })
                    .collect();

                if !list_service_accounts && access_keys.sts_keys.is_empty() {
                    continue;
                }
            }

            if list_service_accounts {
                let service_accounts = iam_store.list_service_accounts(&user).await.map_err(|e| {
                    debug!(
                        component = LOG_COMPONENT_ADMIN,
                        subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                        event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
                        target_user = %user,
                        result = "list_service_accounts_failed",
                        error = ?e,
                        "admin service account state"
                    );
                    s3_error!(InternalError, "list service account failed")
                })?;

                access_keys.service_accounts = service_accounts
                    .into_iter()
                    .map(|svc| {
                        access_key_identity::list_entry_from_credentials(
                            &svc,
                            expiration_for_admin_path(req.uri.path(), svc.expiration),
                        )
                    })
                    .collect();

                if !list_sts_keys && access_keys.service_accounts.is_empty() {
                    continue;
                }
            }

            access_key_map.insert(user, access_keys);
        }

        let data = serde_json::to_vec(&access_key_map)
            .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("marshal access keys err {e}")))?;
        let (data, content_type) = encode_compatible_admin_payload(req.uri.path(), &cred.secret_key, data)?;

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, content_type.parse().expect("valid header value"));

        Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header))
    }
}

pub struct DeleteServiceAccount {}
#[async_trait::async_trait]
impl Operation for DeleteServiceAccount {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        let Some(input_cred) = req.credentials else {
            return Err(s3_error!(InvalidRequest, "get cred failed"));
        };

        let (cred, owner) =
            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key)
                .await
                .map_err(|e| {
                    debug!(
                        component = LOG_COMPONENT_ADMIN,
                        subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                        event = "service_account_auth_failed",
                        action = "delete",
                        error = ?e,
                        "Service account authentication failed"
                    );
                    s3_error!(InternalError, "check key failed")
                })?;

        let query = {
            if let Some(query) = req.uri.query() {
                let input: AccessKeyQuery = from_bytes(query.as_bytes())
                    .map_err(|_e| s3_error!(InvalidArgument, "invalid access key query parameters"))?;
                input
            } else {
                AccessKeyQuery::default()
            }
        };

        if query.access_key.is_empty() {
            return Err(s3_error!(InvalidArgument, "access key is empty"));
        }

        let Ok(iam_store) = crate::admin::runtime_sources::current_ready_iam_handle() else {
            return Err(s3_error!(InvalidRequest, "iam not init"));
        };

        let svc_account = match iam_store.get_service_account(&query.access_key).await {
            Ok((res, _)) => Some(res),
            Err(err) => {
                if is_err_no_such_service_account(&err) {
                    return Err(s3_error!(InvalidRequest, "service account not exist"));
                }

                None
            }
        };

        if !iam_store
            .is_allowed(&Args {
                account: &cred.access_key,
                groups: &cred.groups,
                action: Action::AdminAction(AdminAction::RemoveServiceAccountAdminAction),
                bucket: "",
                conditions: &get_condition_values(
                    &req.headers,
                    &cred,
                    None,
                    None,
                    req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
                ),
                is_owner: owner,
                object: "",
                claims: cred.claims_or_empty(),
                deny_only: false,
            })
            .await
        {
            let user = if cred.parent_user.is_empty() {
                cred.access_key.as_str()
            } else {
                cred.parent_user.as_str()
            };

            // Fail closed: a caller without the RemoveServiceAccount admin action
            // may only delete a service account it OWNS. If the target could not be
            // loaded (svc_account is None because get_service_account failed with a
            // non-not-found error — e.g. a credential decode error), ownership cannot
            // be verified, so deny rather than fall through to the delete. The
            // previous `is_some_and(parent != user)` form skipped the guard on a
            // None target, an authz bypass (backlog#806).
            let target_parent = svc_account.as_ref().map(|v| v.parent_user.as_str());
            if !non_admin_may_delete_service_account(user, target_parent) {
                return Err(s3_error!(InvalidRequest, "service account not exist"));
            }
        }

        let deleted_at = iam_store
            .delete_service_account_with_revision(&query.access_key, true)
            .await
            .map_err(|e| {
                debug!(
                    component = LOG_COMPONENT_ADMIN,
                    subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                    event = "service_account_delete_failed",
                    access_key = %query.access_key,
                    error = ?e,
                    "Failed to delete service account"
                );
                s3_error!(InternalError, "delete service account failed")
            })?;

        if let Some(deleted_at) = deleted_at
            && let Err(err) = site_replication_iam_change_hook(SRIAMItem {
                r#type: "service-account".to_string(),
                svc_acc_change: Some(SRSvcAccChange {
                    delete: Some(SRSvcAccDelete {
                        access_key: query.access_key.clone(),
                        api_version: Some(SITE_REPL_API_VERSION.to_string()),
                    }),
                    api_version: Some(SITE_REPL_API_VERSION.to_string()),
                    ..Default::default()
                }),
                updated_at: Some(deleted_at),
                api_version: Some(SITE_REPL_API_VERSION.to_string()),
                ..Default::default()
            })
            .await
        {
            warn!(
                component = LOG_COMPONENT_ADMIN,
                subsystem = LOG_SUBSYSTEM_SERVICE_ACCOUNT,
                event = EVENT_ADMIN_SERVICE_ACCOUNT_STATE,
                access_key = %query.access_key,
                action = "delete",
                result = "site_replication_hook_failed",
                error = ?err,
                "admin service account state"
            );
        }

        let mut header = HeaderMap::new();
        header.insert(CONTENT_TYPE, "application/json".parse().expect("valid header value"));
        header.insert(CONTENT_LENGTH, "0".parse().expect("valid header value"));
        Ok(S3Response::with_headers(
            (delete_service_account_success_status(req.uri.path()), Body::empty()),
            header,
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rustfs_credentials::IAM_POLICY_CLAIM_NAME_SA;
    use serde_json::json;
    use serde_urlencoded::from_bytes;

    #[test]
    fn access_key_query_supports_external_alias() {
        let query: AccessKeyQuery = from_bytes(b"access-key=test-access-key").expect("parse query");
        assert_eq!(query.access_key, "test-access-key");
    }

    #[test]
    fn non_admin_delete_is_fail_closed_on_unloadable_target() {
        // Regression (backlog#806): a non-admin caller may delete ONLY a service
        // account it owns; a None target (get_service_account failed to load /
        // decode) must DENY, not fall through to the delete.
        assert!(
            non_admin_may_delete_service_account("alice", Some("alice")),
            "owner may delete own service account"
        );
        assert!(!non_admin_may_delete_service_account("alice", Some("bob")), "non-owner must be denied");
        assert!(
            !non_admin_may_delete_service_account("alice", None),
            "unloadable target (None) must be denied (fail-closed), not bypass the owner check"
        );
        assert!(
            !non_admin_may_delete_service_account("", None),
            "empty caller + unloadable target must still deny"
        );
    }

    #[test]
    fn ghsa_5354_non_owner_service_account_parent_confined_to_scope() {
        // Regression (GHSA-5354): a caller holding CreateServiceAccountAdminAction
        // but resolving to owner=false must not be able to parent a new service
        // account to the root credential (or any other user). A root-parented SA
        // authenticates as owner, so allowing this is a full-takeover escalation.
        let root = rustfs_credentials::DEFAULT_ACCESS_KEY; // "rustfsadmin"

        // Attack: non-owner "alice" targets root as the parent -> DENY.
        assert!(
            !add_service_account_parent_within_scope(root, "alice", "alice", false),
            "non-owner must not parent a service account to root"
        );
        // Non-owner targeting an unrelated user -> DENY.
        assert!(
            !add_service_account_parent_within_scope("bob", "alice", "alice", false),
            "non-owner must not parent a service account to another user"
        );
        // Self-scoped creation (default target == own key) -> ALLOW.
        assert!(
            add_service_account_parent_within_scope("alice", "alice", "alice", false),
            "non-owner may create a service account for itself"
        );
        // Derived credential targeting its own parent -> ALLOW.
        assert!(
            add_service_account_parent_within_scope("parent-user", "svc-key", "parent-user", false),
            "derived credential may create a service account under its own parent"
        );
        // Owner may target any parent, including root (cross-user creation intact).
        assert!(
            add_service_account_parent_within_scope(root, root, root, true),
            "owner retains cross-user / root-parent creation"
        );
        assert!(
            add_service_account_parent_within_scope("bob", root, root, true),
            "owner may create a service account for another user"
        );
    }

    #[test]
    fn ghsa_5354_scope_guard_matches_owner_or_self_scope_for_derived_credentials() {
        // The handler evaluates `add_service_account_parent_within_scope` on the
        // original `target_user`, before the `target_user = req_parent_user` rewrite
        // taken inside the `is_svc_acc` branch. The guard's allow set must therefore
        // stay exactly `owner || is_svc_acc`, where
        // `is_svc_acc == (target_user == req_user || target_user == req_parent_user)`.
        //
        // The named-boundary test above only exercises the self-scoped case with
        // `req_user == req_parent_user`. A *derived* credential has `req_user`
        // (its own key) distinct from `req_parent_user` (its parent), which is the
        // realistic attacker shape: a service account holding
        // CreateServiceAccountAdminAction. Pin the equivalence across that shape so a
        // later change to either the guard or the rewrite cannot silently let a
        // derived non-owner credential escape its own parent's scope.
        let root = rustfs_credentials::DEFAULT_ACCESS_KEY;
        let cases: &[(&str, &str, &str, bool)] = &[
            // Derived non-owner (own key != parent) aiming at root -> deny.
            (root, "attacker-sa", "alice", false),
            // Derived non-owner aiming at an unrelated user -> deny.
            ("bob", "attacker-sa", "alice", false),
            // Derived non-owner aiming at its own parent -> allow.
            ("alice", "attacker-sa", "alice", false),
            // Derived non-owner aiming at its own key -> allow.
            ("attacker-sa", "attacker-sa", "alice", false),
            // Top-level non-owner aiming at itself -> allow.
            ("alice", "alice", "alice", false),
            // Owner is unrestricted, including root and a derived shape.
            (root, "attacker-sa", "alice", true),
            ("bob", root, root, true),
        ];
        for &(target_user, req_user, req_parent_user, owner) in cases {
            let is_svc_acc = target_user == req_user || target_user == req_parent_user;
            assert_eq!(
                add_service_account_parent_within_scope(target_user, req_user, req_parent_user, owner),
                owner || is_svc_acc,
                "scope guard must equal `owner || is_svc_acc` for (target={target_user}, req_user={req_user}, req_parent={req_parent_user}, owner={owner})"
            );
        }
    }

    #[test]
    fn guess_user_provider_detects_builtin_accounts() {
        let credentials = StoredCredentials {
            access_key: "builtin-user".to_string(),
            secret_key: "secret-key".to_string(),
            ..Default::default()
        };

        assert_eq!(access_key_identity::guess_user_provider(&credentials), "builtin");
    }

    #[test]
    fn guess_user_provider_detects_ldap_accounts() {
        let credentials = StoredCredentials {
            access_key: "svc".to_string(),
            secret_key: "secret-key".to_string(),
            parent_user: "parent".to_string(),
            claims: Some(HashMap::from([
                (IAM_POLICY_CLAIM_NAME_SA.to_string(), json!("embedded")),
                ("ldap:user".to_string(), json!("uid=rustfs,ou=people,dc=example,dc=com")),
            ])),
            ..Default::default()
        };

        assert_eq!(access_key_identity::guess_user_provider(&credentials), "ldap");
    }

    #[test]
    fn guess_user_provider_detects_openid_accounts() {
        let credentials = StoredCredentials {
            access_key: "sts".to_string(),
            secret_key: "secret-key".to_string(),
            session_token: "session-token".to_string(),
            parent_user: "parent".to_string(),
            claims: Some(HashMap::from([("sub".to_string(), json!("subject-123"))])),
            ..Default::default()
        };

        assert_eq!(access_key_identity::guess_user_provider(&credentials), "openid");
    }

    #[test]
    fn provider_specific_info_uses_known_claims() {
        let claims = HashMap::from([
            ("ldap:user".to_string(), json!("uid=rustfs,ou=people,dc=example,dc=com")),
            ("sub".to_string(), json!("subject-123")),
            ("name".to_string(), json!("RustFS User")),
        ]);

        let ldap_info = access_key_identity::ldap_specific_info(Some(&claims));
        let openid_info = access_key_identity::openid_specific_info(Some(&claims));

        assert_eq!(ldap_info.username.as_deref(), Some("uid=rustfs,ou=people,dc=example,dc=com"));
        assert_eq!(openid_info.user_id.as_deref(), Some("subject-123"));
        assert_eq!(openid_info.user_id_claim.as_deref(), Some("sub"));
        assert_eq!(openid_info.display_name.as_deref(), Some("RustFS User"));
        assert_eq!(openid_info.display_name_claim.as_deref(), Some("name"));
    }

    #[test]
    fn list_access_keys_query_parses_external_parameters() {
        let query = parse_list_access_keys_query(Some("users=alice&users=bob&all=true&listType=svcacc-only"));

        assert_eq!(query.users, vec!["alice".to_string(), "bob".to_string()]);
        assert!(query.all);
        assert_eq!(query.list_type, ACCESS_KEY_LIST_SVCACC_ONLY);
    }

    #[test]
    fn list_access_keys_query_ignores_empty_users_values() {
        let query = parse_list_access_keys_query(Some("users=&users=alice&users=&listType=all"));

        assert_eq!(query.users, vec!["alice".to_string()]);
        assert!(!query.all);
        assert_eq!(query.list_type, ACCESS_KEY_LIST_ALL);
    }

    #[test]
    fn list_access_keys_query_all_with_empty_users_does_not_conflict() {
        let query = parse_list_access_keys_query(Some("users=&all=true&listType=all"));

        assert!(query.users.is_empty());
        assert!(query.all);
        assert_eq!(query.list_type, ACCESS_KEY_LIST_ALL);
        assert!(!query.all || query.users.is_empty());
    }

    #[test]
    fn list_access_keys_query_defaults_to_all_list_type() {
        let query = ListAccessKeysQuery::default();
        assert_eq!(query.list_type, ACCESS_KEY_LIST_ALL);
    }

    #[test]
    fn list_service_account_cross_user_uses_list_service_accounts_action() {
        let src = include_str!("service_account.rs");
        let list_start = src
            .find("impl Operation for ListServiceAccount")
            .expect("ListServiceAccount operation should exist");
        let list_block = &src[list_start..];
        let list_end = list_block
            .find("struct ListAccessKeysQuery")
            .expect("ListAccessKeysQuery marker should exist");
        let list_block = &list_block[..list_end];

        assert!(
            list_block.contains("query.user.as_ref().is_some_and(") && list_block.contains("v != &cred.access_key"),
            "cross-user ListServiceAccount path should stay explicitly guarded"
        );
        assert!(
            list_block.contains("ListServiceAccountsAdminAction"),
            "cross-user ListServiceAccount should authorize with ListServiceAccountsAdminAction"
        );
        assert!(
            !list_block.contains("UpdateServiceAccountAdminAction"),
            "cross-user ListServiceAccount must not require UpdateServiceAccountAdminAction"
        );
    }

    #[test]
    fn delete_service_account_uses_external_success_status() {
        assert_eq!(
            delete_service_account_success_status("/minio/admin/v3/delete-service-account"),
            StatusCode::NO_CONTENT
        );
    }

    #[test]
    fn delete_service_account_keeps_rustfs_console_success_status() {
        assert_eq!(
            delete_service_account_success_status("/rustfs/admin/v3/delete-service-account"),
            StatusCode::OK
        );
    }

    #[test]
    fn expiration_for_external_admin_path_uses_sentinel() {
        assert_eq!(
            expiration_for_admin_path("/minio/admin/v3/list-service-accounts", None),
            Some(OffsetDateTime::UNIX_EPOCH)
        );
    }

    #[test]
    fn expiration_for_rustfs_admin_path_preserves_none() {
        assert_eq!(expiration_for_admin_path("/rustfs/admin/v3/list-service-accounts", None), None);
    }

    #[test]
    fn map_service_account_lookup_error_reports_missing_accounts_explicitly() {
        let err = map_service_account_lookup_error(
            rustfs_iam::error::Error::NoSuchServiceAccount("missing".to_string()),
            "get service account failed",
        );

        assert_eq!(*err.code(), S3ErrorCode::NoSuchResource);
        assert_eq!(err.message(), Some("service account 'missing' does not exist"));
    }

    #[test]
    fn add_service_account_maps_iam_create_errors_to_s3_errors() {
        let src = include_str!("service_account.rs");
        let create_start = src
            .find("impl Operation for AddServiceAccount")
            .expect("AddServiceAccount operation should exist");
        let create_block = &src[create_start..];
        let create_end = create_block
            .find("if let Err(err) = site_replication_iam_change_hook")
            .expect("site replication hook marker should exist");
        let create_block = &create_block[..create_end];

        assert!(
            create_block.contains("AccessKeyAlreadyExists") && create_block.contains("iam_error_to_s3_error(e)"),
            "duplicate access keys must map to client S3 errors"
        );
    }

    #[test]
    fn map_temp_account_lookup_error_reports_missing_access_keys_explicitly() {
        let err = map_temp_account_lookup_error(
            rustfs_iam::error::Error::NoSuchTempAccount("missing".to_string()),
            "get temporary account failed",
        );

        assert_eq!(*err.code(), S3ErrorCode::NoSuchResource);
        assert_eq!(err.message(), Some("temp account 'missing' does not exist"));
    }

    #[test]
    fn parse_update_service_account_policy_keeps_explicit_empty_policy() {
        let policy = parse_update_service_account_policy(Some(json!({}))).expect("empty policy should parse");

        assert!(policy.is_some());
        let policy = policy.unwrap();
        assert!(policy.version.is_empty());
        assert!(policy.statements.is_empty());
    }

    #[test]
    fn sparse_explicit_create_policy_is_normalized_before_persistence_and_replication() {
        let id_only = parse_new_service_account_policy(Some(&json!({"ID": "deny-boundary", "Version": "", "Statement": []})))
            .expect("ID-only policy should normalize")
            .expect("normalized explicit policy");
        assert_eq!(id_only.id.as_str(), "deny-boundary");
        assert_eq!(id_only.version, DEFAULT_VERSION);
        assert!(!id_only.statements.is_empty());

        let missing_version = parse_new_service_account_policy(Some(&json!({
            "Statement": [{
                "Effect": "Deny",
                "Action": ["s3:GetObject"],
                "Resource": ["arn:aws:s3:::bucket/*"]
            }]
        })))
        .expect("policy version should normalize")
        .expect("normalized explicit policy");
        assert_eq!(missing_version.version, DEFAULT_VERSION);
    }

    #[test]
    fn explicit_empty_create_policy_is_canonicalized_as_inherited() {
        let policy = parse_new_service_account_policy(Some(&json!({}))).expect("parse empty create policy");

        assert!(policy.is_none());
    }

    #[test]
    fn sparse_explicit_update_policy_preserves_existing_semantics() {
        let id_only = parse_update_service_account_policy(Some(json!({"ID": "clear-boundary", "Version": "", "Statement": []})))
            .expect("parse ID-only update")
            .expect("explicit update policy");
        assert_eq!(id_only.id.as_str(), "clear-boundary");
        assert!(id_only.version.is_empty());
        assert!(id_only.statements.is_empty());

        let missing_version = parse_update_service_account_policy(Some(json!({
            "Statement": [{
                "Effect": "Deny",
                "Action": ["s3:GetObject"],
                "Resource": ["arn:aws:s3:::bucket/*"]
            }]
        })))
        .expect("parse versionless update")
        .expect("explicit update policy");
        assert!(missing_version.version.is_empty());
        assert!(!missing_version.statements.is_empty());
    }

    #[test]
    fn replication_update_stays_partial() {
        let update = UpdateServiceAccountReq {
            new_policy: Some(json!({})),
            new_secret_key: None,
            new_status: None,
            new_name: None,
            new_description: None,
            new_expiration: None,
        };

        let session_policy = parse_update_service_account_policy(update.new_policy.clone()).expect("parse update policy");
        let change = service_account_update_replication_change("OIDCSERVICEACCOUNT01", &update, session_policy.as_ref())
            .expect("build replication update");

        assert!(change.create.is_none());
        assert!(change.delete.is_none());
        let update = change.update.expect("partial update");
        let cleared: Policy =
            serde_json::from_str(update.session_policy.as_str().expect("explicit policy clear")).expect("parse policy clear");
        assert!(cleared.id.is_empty());
        assert!(cleared.version.is_empty());
        assert!(cleared.statements.is_empty());
    }

    #[test]
    fn parse_service_account_policy_reports_missing_resource() {
        let err = parse_service_account_policy(&json!({
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": ["s3:GetObject"]
                }
            ]
        }))
        .expect_err("policy without Resource should be rejected");

        assert_eq!(*err.code(), S3ErrorCode::InvalidArgument);
        assert_eq!(err.message(), Some("invalid service account policy: Resource is empty"));
    }

    #[test]
    fn parse_service_account_policy_preserves_policy_validation_details() {
        let err = parse_service_account_policy(&json!({
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": ["s3:GetObject", "admin:ServerInfo"],
                    "Resource": ["arn:aws:s3:::bucket/*"]
                }
            ]
        }))
        .expect_err("policy with mixed action families should be rejected");

        assert_eq!(*err.code(), S3ErrorCode::InvalidArgument);
        assert_eq!(
            err.message(),
            Some("invalid service account policy: 'Action' contains mixed action families in the same statement")
        );
    }

    #[test]
    fn update_service_account_requires_requester_parent_match() {
        let parent_owner = StoredCredentials {
            access_key: "owner-user".to_string(),
            parent_user: String::new(),
            ..Default::default()
        };
        let derived_owner = StoredCredentials {
            access_key: "sa-user".to_string(),
            parent_user: "owner-user".to_string(),
            ..Default::default()
        };
        let foreign_user = StoredCredentials {
            access_key: "other".to_string(),
            parent_user: String::new(),
            ..Default::default()
        };

        assert!(is_service_account_owner_of(&parent_owner, "owner-user"));
        assert!(is_service_account_owner_of(&derived_owner, "owner-user"));
        assert!(!is_service_account_owner_of(&foreign_user, "owner-user"));
    }

    #[test]
    fn fallback_access_key_info_allows_same_regular_user() {
        let requester = StoredCredentials {
            access_key: "owner-user".to_string(),
            ..Default::default()
        };
        let target = StoredCredentials {
            access_key: "owner-user".to_string(),
            ..Default::default()
        };

        assert!(can_fallback_view_access_key_info(&requester, &target));
    }

    #[test]
    fn fallback_access_key_info_allows_parent_for_derived_credentials() {
        let requester = StoredCredentials {
            access_key: "sts-user".to_string(),
            session_token: "session-token".to_string(),
            parent_user: "owner-user".to_string(),
            ..Default::default()
        };
        let target = StoredCredentials {
            access_key: "owner-user".to_string(),
            ..Default::default()
        };

        assert!(can_fallback_view_access_key_info(&requester, &target));
    }

    #[test]
    fn fallback_access_key_info_denies_service_account_for_parent_regular_user() {
        let requester = StoredCredentials {
            access_key: "svc-user".to_string(),
            session_token: "session-token".to_string(),
            parent_user: "owner-user".to_string(),
            claims: Some(HashMap::from([("sa-policy".to_string(), json!("inherited-policy"))])),
            ..Default::default()
        };
        let target = StoredCredentials {
            access_key: "owner-user".to_string(),
            ..Default::default()
        };

        assert!(!can_fallback_view_access_key_info(&requester, &target));
    }

    #[test]
    fn fallback_access_key_info_denies_other_regular_user() {
        let requester = StoredCredentials {
            access_key: "alice".to_string(),
            ..Default::default()
        };
        let target = StoredCredentials {
            access_key: "bob".to_string(),
            ..Default::default()
        };

        assert!(!can_fallback_view_access_key_info(&requester, &target));
    }

    #[test]
    fn fallback_access_key_info_requires_no_explicit_deny() {
        let requester = StoredCredentials {
            access_key: "owner-user".to_string(),
            ..Default::default()
        };
        let target = StoredCredentials {
            access_key: "owner-user".to_string(),
            ..Default::default()
        };

        assert!(can_fallback_view_access_key_info_after_admin_check_failed(&requester, &target, true));
        assert!(!can_fallback_view_access_key_info_after_admin_check_failed(&requester, &target, false));
    }

    #[test]
    fn fallback_access_key_info_requires_no_explicit_deny_for_parent_owned_derived_credentials() {
        let requester = StoredCredentials {
            access_key: "sts-user".to_string(),
            session_token: "session-token".to_string(),
            parent_user: "owner-user".to_string(),
            ..Default::default()
        };
        let target = StoredCredentials {
            access_key: "owner-user".to_string(),
            ..Default::default()
        };

        assert!(can_fallback_view_access_key_info_after_admin_check_failed(&requester, &target, true));
        assert!(!can_fallback_view_access_key_info_after_admin_check_failed(&requester, &target, false));
    }

    #[test]
    fn build_info_regular_user_resp_maps_user_metadata() {
        let account = StoredCredentials {
            access_key: "owner-user".to_string(),
            name: Some("Owner".to_string()),
            description: Some("Primary user".to_string()),
            ..Default::default()
        };
        let user_info = rustfs_madmin::UserInfo {
            policy_name: Some("readwrite".to_string()),
            status: rustfs_madmin::AccountStatus::Enabled,
            ..Default::default()
        };

        let resp = access_key_identity::build_info_regular_user_resp(&account, &user_info);

        assert_eq!(resp.parent_user, "");
        assert_eq!(resp.account_status, "enabled");
        assert!(!resp.implied_policy);
        assert_eq!(resp.policy.as_deref(), Some("readwrite"));
        assert_eq!(resp.name.as_deref(), Some("Owner"));
        assert_eq!(resp.description.as_deref(), Some("Primary user"));
        assert_eq!(resp.expiration, None);
    }
}
