volcengine_rust_sdk_auth/

lib.rs

//! Credentials backed by profiles created by `ve configure`, `ve sso login`,
//! and `ve login`.
//!
//! The implementation intentionally keeps the CLI cache read-only. Refreshes
//! update only the in-process snapshot; `ve` remains the sole cache writer.

use std::{
    collections::{BTreeMap, HashMap},
    env,
    error::Error as StdError,
    fmt, fs,
    path::{Path, PathBuf},
    sync::Mutex,
    thread,
    time::Duration,
};

use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
use chrono::{DateTime, SecondsFormat, Utc};
use hmac::{Hmac, Mac};
use rand::Rng;
use reqwest::{
    StatusCode,
    blocking::{Client, Response},
    header::{ACCEPT, AUTHORIZATION, CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue},
};
use serde::{Deserialize, Deserializer, Serialize};
use serde_json::Value as JsonValue;
use sha1::{Digest as _, Sha1};
use sha2::Sha256;
use thiserror::Error;
use uuid::Uuid;

const PROVIDER_NAME: &str = "CliProvider";
const DEFAULT_REGION: &str = "cn-beijing";
const DEFAULT_STS_HOST: &str = "sts.volcengineapi.com";
const DEFAULT_CONSOLE_LOGIN_ENDPOINT: &str = "https://signin.volcengine.com";
const DEFAULT_IMDS_ENDPOINT: &str = "http://100.96.0.96";
const EXPIRY_WINDOW: Duration = Duration::from_secs(60);
const DEFAULT_RETRIES: usize = 3;

pub type Result<T> = std::result::Result<T, CredentialError>;

/// A resolved Volcengine credential triple.
///
/// `Debug` is intentionally redacted so accidental diagnostics do not expose
/// secrets.
#[derive(Clone, PartialEq, Eq)]
pub struct CredentialValue {
    pub access_key_id: String,
    pub secret_access_key: String,
    pub session_token: String,
    pub provider_name: &'static str,
}

impl fmt::Debug for CredentialValue {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CredentialValue")
            .field("access_key_id", &"[REDACTED]")
            .field("secret_access_key", &"[REDACTED]")
            .field("session_token", &"[REDACTED]")
            .field("provider_name", &self.provider_name)
            .finish()
    }
}

impl CredentialValue {
    fn cli(access_key_id: String, secret_access_key: String, session_token: String) -> Self {
        Self {
            access_key_id,
            secret_access_key,
            session_token,
            provider_name: PROVIDER_NAME,
        }
    }

    fn named(
        access_key_id: String,
        secret_access_key: String,
        session_token: String,
        provider_name: &'static str,
    ) -> Self {
        Self {
            access_key_id,
            secret_access_key,
            session_token,
            provider_name,
        }
    }
}

/// Stable provider error with a Go-SDK-compatible error code.
#[derive(Debug, Error)]
#[error("{code}: {message}")]
pub struct CredentialError {
    code: &'static str,
    message: String,
    #[source]
    source: Option<Box<dyn StdError + Send + Sync>>,
}

impl CredentialError {
    fn new(code: &'static str, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
            source: None,
        }
    }

    fn with_source(
        code: &'static str,
        message: impl Into<String>,
        source: impl StdError + Send + Sync + 'static,
    ) -> Self {
        Self {
            code,
            message: message.into(),
            source: Some(Box::new(source)),
        }
    }

    pub fn code(&self) -> &'static str {
        self.code
    }
}

#[derive(Debug, Clone, Deserialize, Default)]
struct CliConfig {
    #[serde(default)]
    current: String,
    #[serde(default)]
    profiles: HashMap<String, CliProfile>,
    #[serde(
        rename = "sso-session",
        default,
        deserialize_with = "deserialize_sso_sessions"
    )]
    sso_sessions: HashMap<String, SsoSession>,
}

fn deserialize_sso_sessions<'de, D>(
    deserializer: D,
) -> std::result::Result<HashMap<String, SsoSession>, D::Error>
where
    D: Deserializer<'de>,
{
    Ok(Option::deserialize(deserializer)?.unwrap_or_default())
}

#[derive(Debug, Clone, Deserialize, Default)]
struct CliProfile {
    #[serde(default)]
    mode: String,
    #[serde(rename = "access-key", default)]
    access_key: String,
    #[serde(rename = "secret-key", default)]
    secret_key: String,
    #[serde(rename = "session-token", default)]
    session_token: String,
    #[serde(rename = "sts-expiration", default)]
    sts_expiration: i64,
    #[serde(rename = "sso-session-name", default)]
    sso_session_name: String,
    #[serde(rename = "account-id", default)]
    account_id: String,
    #[serde(rename = "role-name", default)]
    role_name: String,
    #[serde(rename = "login-session", default)]
    login_session: String,
    #[serde(rename = "oidc-token-file", default)]
    oidc_token_file: String,
    #[serde(rename = "role-trn", default)]
    role_trn: String,
    #[serde(default)]
    region: String,
    #[serde(rename = "disable-ssl", default)]
    disable_ssl: bool,
}

#[derive(Debug, Clone, Deserialize, Default)]
struct SsoSession {
    #[serde(rename = "start-url", default)]
    start_url: String,
    #[serde(default)]
    region: String,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ProfileMode {
    AccessKey,
    Sso,
    IamRole,
    Oidc,
    EcsRole,
    ConsoleLogin,
}

impl ProfileMode {
    fn parse(raw: &str, profile_name: &str, config_path: &Path) -> Result<Self> {
        match raw.trim().to_ascii_lowercase().as_str() {
            "" | "ak" => Ok(Self::AccessKey),
            "sso" => Ok(Self::Sso),
            "iamrole" | "ramrolearn" => Ok(Self::IamRole),
            "oidc" => Ok(Self::Oidc),
            "ecsrole" => Ok(Self::EcsRole),
            "console-login" => Ok(Self::ConsoleLogin),
            _ => Err(CredentialError::new(
                "CliConfigModeInvalid",
                format!(
                    "cli config profile {profile_name} in {} contained unsupported mode {raw:?}",
                    config_path.display()
                ),
            )),
        }
    }
}

#[derive(Debug, Clone, Default)]
struct EndpointOverrides {
    sso_oauth: Option<String>,
    sso_portal: Option<String>,
    sts: Option<String>,
    imds: Option<String>,
}

/// Thread-safe, lazily loaded credentials from the Volcengine CLI config.
pub struct CliCredentials {
    config_path: PathBuf,
    requested_profile: Option<String>,
    endpoints: EndpointOverrides,
    state: Mutex<State>,
}

#[derive(Default)]
struct State {
    profile_name: Option<String>,
    source: Option<CredentialSource>,
}

enum CredentialSource {
    Static {
        value: CredentialValue,
        expires_at: Option<DateTime<Utc>>,
    },
    Dynamic(DynamicProvider),
}

enum DynamicProvider {
    Sso(Box<SsoProvider>),
    ConsoleLogin(Box<ConsoleLoginProvider>),
    IamRole(Box<IamRoleProvider>),
    Oidc(Box<OidcProvider>),
    EcsRole(Box<EcsRoleProvider>),
}

impl CliCredentials {
    /// Constructs a CLI credential source.
    ///
    /// An empty path follows `VOLCENGINE_CLI_CONFIG_FILE`, then defaults to
    /// `~/.volcengine/config.json`. An empty profile follows
    /// `VOLCENGINE_PROFILE`, `VOLCSTACK_PROFILE`, `config.current`, and finally
    /// `default`.
    pub fn new(config_path: Option<PathBuf>, profile: Option<String>) -> Self {
        let config_path = config_path
            .filter(|path| !path.as_os_str().is_empty())
            .or_else(|| nonempty_env("VOLCENGINE_CLI_CONFIG_FILE").map(PathBuf::from))
            .unwrap_or_else(default_cli_config_path);

        Self {
            config_path,
            requested_profile: profile.filter(|value| !value.is_empty()),
            endpoints: EndpointOverrides::default(),
            state: Mutex::new(State::default()),
        }
    }

    /// Constructs credentials using the default CLI config and profile.
    pub fn from_cli() -> Self {
        Self::new(None, None)
    }

    /// Returns a cached credential or refreshes it when its provider expires.
    pub fn get(&self) -> Result<CredentialValue> {
        let mut state = self.state.lock().map_err(|_| {
            CredentialError::new(
                "CliCredentialsPoisoned",
                "credential state lock was poisoned",
            )
        })?;

        let needs_reload = state
            .source
            .as_ref()
            .is_none_or(CredentialSource::is_expired);
        if needs_reload {
            state.source = None;
            let (config, profile_name, profile) = self.load_profile(&mut state)?;
            state.source = Some(self.build_source(&config, &profile_name, profile)?);
        }

        state
            .source
            .as_mut()
            .expect("credential source is initialized")
            .get()
    }

    fn load_profile(&self, state: &mut State) -> Result<(CliConfig, String, CliProfile)> {
        let bytes = fs::read(&self.config_path).map_err(|error| {
            CredentialError::with_source(
                "CliConfigLoad",
                format!(
                    "failed to load cli config file {}",
                    self.config_path.display()
                ),
                error,
            )
        })?;

        let config = if bytes.iter().all(u8::is_ascii_whitespace) {
            CliConfig::default()
        } else {
            serde_json::from_slice(&bytes).map_err(|error| {
                CredentialError::with_source(
                    "CliConfigUnmarshal",
                    format!(
                        "failed to unmarshal cli config file {}",
                        self.config_path.display()
                    ),
                    error,
                )
            })?
        };

        if config.profiles.is_empty() {
            return Err(CredentialError::new(
                "CliConfigNoProfiles",
                format!(
                    "cli config file {} did not contain any profiles",
                    self.config_path.display()
                ),
            ));
        }

        let profile_name = state
            .profile_name
            .get_or_insert_with(|| self.resolve_profile_name(&config))
            .clone();
        let profile = config.profiles.get(&profile_name).cloned().ok_or_else(|| {
            CredentialError::new(
                "CliConfigProfileNotFound",
                format!(
                    "cli config file {} did not contain profile {profile_name}",
                    self.config_path.display()
                ),
            )
        })?;

        Ok((config, profile_name, profile))
    }

    fn resolve_profile_name(&self, config: &CliConfig) -> String {
        self.requested_profile
            .clone()
            .or_else(|| nonempty_env("VOLCENGINE_PROFILE"))
            .or_else(|| nonempty_env("VOLCSTACK_PROFILE"))
            .or_else(|| (!config.current.is_empty()).then(|| config.current.clone()))
            .unwrap_or_else(|| "default".to_owned())
    }

    fn build_source(
        &self,
        config: &CliConfig,
        profile_name: &str,
        profile: CliProfile,
    ) -> Result<CredentialSource> {
        match ProfileMode::parse(&profile.mode, profile_name, &self.config_path)? {
            ProfileMode::AccessKey => self.access_key_source(profile_name, &profile),
            ProfileMode::Sso => self.sso_source(config, profile_name, profile),
            ProfileMode::IamRole => self.iam_role_source(profile_name, profile),
            ProfileMode::Oidc => self.oidc_source(profile_name, profile),
            ProfileMode::EcsRole => self.ecs_role_source(profile_name, profile),
            ProfileMode::ConsoleLogin => self.console_login_source(profile_name, profile),
        }
    }

    fn access_key_source(
        &self,
        profile_name: &str,
        profile: &CliProfile,
    ) -> Result<CredentialSource> {
        require_profile_field(
            &profile.access_key,
            "CliConfigAccessKey",
            "access-key",
            profile_name,
            &self.config_path,
        )?;
        require_profile_field(
            &profile.secret_key,
            "CliConfigSecretKey",
            "secret-key",
            profile_name,
            &self.config_path,
        )?;

        Ok(CredentialSource::Static {
            value: CredentialValue::cli(
                profile.access_key.clone(),
                profile.secret_key.clone(),
                profile.session_token.clone(),
            ),
            expires_at: None,
        })
    }

    fn sso_source(
        &self,
        config: &CliConfig,
        profile_name: &str,
        profile: CliProfile,
    ) -> Result<CredentialSource> {
        if profile.sts_expiration < 0 {
            return Err(CredentialError::new(
                "CliConfigExpiration",
                format!(
                    "cli config profile {profile_name} in {} contained invalid sts-expiration",
                    self.config_path.display()
                ),
            ));
        }

        if profile.sts_expiration > 0 {
            let expiration = unix_timestamp_to_datetime(profile.sts_expiration)?;
            if Utc::now() <= expiration {
                require_profile_field(
                    &profile.access_key,
                    "CliConfigAccessKey",
                    "access-key",
                    profile_name,
                    &self.config_path,
                )?;
                require_profile_field(
                    &profile.secret_key,
                    "CliConfigSecretKey",
                    "secret-key",
                    profile_name,
                    &self.config_path,
                )?;
                return Ok(CredentialSource::Static {
                    value: CredentialValue::cli(
                        profile.access_key.clone(),
                        profile.secret_key.clone(),
                        profile.session_token.clone(),
                    ),
                    expires_at: expiration.checked_sub_signed(chrono::Duration::seconds(60)),
                });
            }
        }

        let session_name = require_trimmed_profile_field(
            &profile.sso_session_name,
            "CliConfigSsoSessionNameMissing",
            "sso-session-name",
            profile_name,
            &self.config_path,
        )?;
        let session = config.sso_sessions.get(session_name).ok_or_else(|| {
            CredentialError::new(
                "CliConfigSsoSessionNotFound",
                format!(
                    "cli config file {} did not contain sso session {session_name}",
                    self.config_path.display()
                ),
            )
        })?;
        let start_url = session.start_url.trim();
        if start_url.is_empty() {
            return Err(CredentialError::new(
                "CliConfigSsoStartURLMissing",
                format!(
                    "sso session {session_name} in {} did not contain start-url",
                    self.config_path.display()
                ),
            ));
        }
        let region = match session.region.trim() {
            "" => match profile.region.trim() {
                "" => DEFAULT_REGION,
                value => value,
            },
            value => value,
        }
        .to_owned();
        let cache_path = self
            .config_path
            .parent()
            .unwrap_or_else(|| Path::new(""))
            .join("sso/cache")
            .join(sso_cache_filename(start_url, session_name)?);
        if !cache_path.is_file() {
            return Err(CredentialError::new(
                "CliSsoTokenCacheMissing",
                format!(
                    "sso token cache file {} does not exist; please run 've sso login' to re-authenticate",
                    cache_path.display()
                ),
            ));
        }

        Ok(CredentialSource::Dynamic(DynamicProvider::Sso(Box::new(
            SsoProvider::new(
                cache_path,
                profile,
                profile_name.to_owned(),
                self.config_path.clone(),
                region,
                self.endpoints.clone(),
            ),
        ))))
    }

    fn iam_role_source(&self, profile_name: &str, profile: CliProfile) -> Result<CredentialSource> {
        for (value, code, field) in [
            (&profile.access_key, "CliConfigAccessKey", "access-key"),
            (&profile.secret_key, "CliConfigSecretKey", "secret-key"),
        ] {
            require_profile_field(value, code, field, profile_name, &self.config_path)?;
        }
        for (value, code, field) in [
            (&profile.role_name, "CliConfigRoleNameMissing", "role-name"),
            (
                &profile.account_id,
                "CliConfigAccountIDMissing",
                "account-id",
            ),
        ] {
            require_trimmed_profile_field(value, code, field, profile_name, &self.config_path)?;
        }

        Ok(CredentialSource::Dynamic(DynamicProvider::IamRole(
            Box::new(IamRoleProvider::new(profile, self.endpoints.clone())),
        )))
    }

    fn oidc_source(&self, profile_name: &str, profile: CliProfile) -> Result<CredentialSource> {
        require_trimmed_profile_field(
            &profile.oidc_token_file,
            "CliConfigOIDCTokenFileMissing",
            "oidc-token-file",
            profile_name,
            &self.config_path,
        )?;
        require_trimmed_profile_field(
            &profile.role_trn,
            "CliConfigOIDCRoleTrnMissing",
            "role-trn",
            profile_name,
            &self.config_path,
        )?;

        Ok(CredentialSource::Dynamic(DynamicProvider::Oidc(Box::new(
            OidcProvider::new(profile, self.endpoints.clone()),
        ))))
    }

    fn ecs_role_source(&self, profile_name: &str, profile: CliProfile) -> Result<CredentialSource> {
        require_trimmed_profile_field(
            &profile.role_name,
            "CliConfigRoleNameMissing",
            "role-name",
            profile_name,
            &self.config_path,
        )?;

        Ok(CredentialSource::Dynamic(DynamicProvider::EcsRole(
            Box::new(EcsRoleProvider::new(
                profile.role_name.trim().to_owned(),
                self.endpoints.clone(),
            )),
        )))
    }

    fn console_login_source(
        &self,
        profile_name: &str,
        profile: CliProfile,
    ) -> Result<CredentialSource> {
        let login_session = require_trimmed_profile_field(
            &profile.login_session,
            "CliConfigLoginSessionMissing",
            "login-session",
            profile_name,
            &self.config_path,
        )?;
        let cache_dir = nonempty_env("VOLCENGINE_LOGIN_CACHE_DIRECTORY")
            .map(PathBuf::from)
            .unwrap_or_else(|| {
                self.config_path
                    .parent()
                    .unwrap_or_else(|| Path::new(""))
                    .join("login/cache")
            });

        Ok(CredentialSource::Dynamic(DynamicProvider::ConsoleLogin(
            Box::new(ConsoleLoginProvider::new(
                login_session.to_owned(),
                cache_dir,
            )),
        )))
    }
}

impl CredentialSource {
    fn is_expired(&self) -> bool {
        match self {
            Self::Static {
                expires_at: None, ..
            } => false,
            Self::Static {
                expires_at: Some(expiration),
                ..
            } => Utc::now() > *expiration,
            Self::Dynamic(provider) => provider.is_expired(),
        }
    }

    fn get(&mut self) -> Result<CredentialValue> {
        match self {
            Self::Static { value, .. } => Ok(value.clone()),
            Self::Dynamic(provider) => provider.get(),
        }
    }
}

impl DynamicProvider {
    fn is_expired(&self) -> bool {
        match self {
            Self::Sso(provider) => provider.is_expired(),
            Self::ConsoleLogin(provider) => provider.is_expired(),
            Self::IamRole(provider) => provider.is_expired(),
            Self::Oidc(provider) => provider.is_expired(),
            Self::EcsRole(provider) => provider.is_expired(),
        }
    }

    fn get(&mut self) -> Result<CredentialValue> {
        match self {
            Self::Sso(provider) => provider.get(),
            Self::ConsoleLogin(provider) => provider.get(),
            Self::IamRole(provider) => provider.get(),
            Self::Oidc(provider) => provider.get(),
            Self::EcsRole(provider) => provider.get(),
        }
    }
}

fn nonempty_env(name: &str) -> Option<String> {
    env::var(name).ok().filter(|value| !value.is_empty())
}

fn default_cli_config_path() -> PathBuf {
    dirs::home_dir()
        .unwrap_or_default()
        .join(".volcengine/config.json")
}

fn require_profile_field<'a>(
    value: &'a str,
    code: &'static str,
    field: &str,
    profile_name: &str,
    config_path: &Path,
) -> Result<&'a str> {
    if value.is_empty() {
        return Err(CredentialError::new(
            code,
            format!(
                "cli config profile {profile_name} in {} did not contain {field}",
                config_path.display()
            ),
        ));
    }
    Ok(value)
}

fn require_trimmed_profile_field<'a>(
    value: &'a str,
    code: &'static str,
    field: &str,
    profile_name: &str,
    config_path: &Path,
) -> Result<&'a str> {
    let value = value.trim();
    require_profile_field(value, code, field, profile_name, config_path)
}

#[derive(Debug, Clone, Deserialize)]
struct SsoTokenCache {
    #[serde(default)]
    access_token: String,
    #[serde(default)]
    expires_at: String,
    #[serde(default)]
    client_id: String,
    #[serde(default)]
    client_secret: String,
    #[serde(default)]
    client_secret_expires_at: i64,
    #[serde(default)]
    refresh_token: String,
}

struct SsoProvider {
    token_path: PathBuf,
    profile: CliProfile,
    profile_name: String,
    config_path: PathBuf,
    region: String,
    endpoints: EndpointOverrides,
    cached: Option<SsoTokenCache>,
    credentials: Option<CredentialValue>,
    expires_at: Option<DateTime<Utc>>,
}

impl SsoProvider {
    fn new(
        token_path: PathBuf,
        profile: CliProfile,
        profile_name: String,
        config_path: PathBuf,
        region: String,
        endpoints: EndpointOverrides,
    ) -> Self {
        Self {
            token_path,
            profile,
            profile_name,
            config_path,
            region,
            endpoints,
            cached: None,
            credentials: None,
            expires_at: None,
        }
    }

    fn is_expired(&self) -> bool {
        self.credentials.is_none()
            || self
                .expires_at
                .is_none_or(|expiration| within_expiry_window(expiration, EXPIRY_WINDOW))
    }

    fn get(&mut self) -> Result<CredentialValue> {
        if self.is_expired() {
            self.refresh()?;
        }
        self.credentials.clone().ok_or_else(|| {
            CredentialError::new(
                "CliSsoPortalCredentialsEmpty",
                "SSO provider did not produce credentials",
            )
        })
    }

    fn refresh(&mut self) -> Result<()> {
        if self.cached.is_none() {
            self.cached = Some(load_sso_cache(&self.token_path)?);
        }

        if !self.access_token_is_expired() {
            return self.fetch_role_credentials();
        }

        match self.refresh_access_token() {
            Ok(()) => self.fetch_role_credentials(),
            Err(error) if !is_invalid_grant(&error) => Err(error),
            Err(first_error) => {
                let first_error_message = first_error.to_string();
                let disk_cache = load_sso_cache(&self.token_path).map_err(|disk_error| {
                    CredentialError::with_source(
                        "CliSsoTokenRefreshFailed",
                        format!(
                            "SSO refresh token rejected ({first_error_message}); failed to reload cache from disk: {disk_error}; please run 've sso login'"
                        ),
                        disk_error,
                    )
                })?;
                let current_refresh_token = self
                    .cached
                    .as_ref()
                    .map(|cache| cache.refresh_token.as_str())
                    .unwrap_or_default();
                if disk_cache.refresh_token.trim().is_empty()
                    || disk_cache.refresh_token == current_refresh_token
                {
                    return Err(CredentialError::with_source(
                        "CliSsoTokenRefreshExpired",
                        "SSO refresh token has been rejected by the authorization server; please run 've sso login' to re-authenticate",
                        first_error,
                    ));
                }
                self.cached = Some(disk_cache);
                if !self.access_token_is_expired() {
                    return self.fetch_role_credentials();
                }
                self.refresh_access_token().map_err(|error| {
                    CredentialError::with_source(
                        "CliSsoTokenRefreshExpired",
                        "SSO refresh token rejected; reloaded cache from disk but new refresh token also failed; please run 've sso login'",
                        error,
                    )
                })?;
                self.fetch_role_credentials()
            }
        }
    }

    fn access_token_is_expired(&self) -> bool {
        let Some(cache) = &self.cached else {
            return true;
        };
        if cache.access_token.trim().is_empty() {
            return true;
        }
        parse_rfc3339(&cache.expires_at)
            .map(|expiration| within_expiry_window(expiration, EXPIRY_WINDOW))
            .unwrap_or(true)
    }

    fn refresh_access_token(&mut self) -> Result<()> {
        let cache = self.cached.as_ref().expect("SSO cache is loaded");
        if cache.refresh_token.trim().is_empty() {
            return Err(CredentialError::new(
                "CliSsoTokenRefreshMissing",
                format!(
                    "SSO token cache {} did not contain refresh token; please run 've sso login' to re-authenticate",
                    self.token_path.display()
                ),
            ));
        }
        if cache.client_secret_expires_at <= 0 {
            return Err(CredentialError::new(
                "CliSsoTokenRefreshExpiresParse",
                format!(
                    "failed to parse refresh token expiration in {}; please run 've sso login' to re-authenticate",
                    self.token_path.display()
                ),
            ));
        }
        let refresh_expiration = unix_timestamp_to_datetime(cache.client_secret_expires_at)?;
        if Utc::now() > refresh_expiration {
            return Err(CredentialError::new(
                "CliSsoTokenRefreshExpired",
                format!(
                    "SSO refresh token in {} has expired; please run 've sso login' to re-authenticate",
                    self.token_path.display()
                ),
            ));
        }
        if cache.client_id.trim().is_empty() || cache.client_secret.trim().is_empty() {
            return Err(CredentialError::new(
                "CliSsoTokenClientMissing",
                format!(
                    "SSO token cache {} did not contain client id/secret; please run 've sso login' to re-authenticate",
                    self.token_path.display()
                ),
            ));
        }

        let response = request_sso_token(
            &self.region,
            cache,
            self.endpoints.sso_oauth.as_deref(),
        )
        .map_err(|failure| {
            if failure.is_invalid_grant() {
                CredentialError::with_source(
                    "CliSsoTokenRefreshInvalidGrant",
                    "SSO authorization server rejected the refresh token",
                    failure,
                )
            } else {
                CredentialError::with_source(
                    "CliSsoTokenRefreshFailed",
                    "failed to refresh SSO access token; please run 've sso login' to re-authenticate",
                    failure,
                )
            }
        })?;

        if response.access_token.trim().is_empty() {
            return Err(CredentialError::new(
                "CliSsoTokenRefreshEmpty",
                "SSO refresh token response did not include access token; please run 've sso login' to re-authenticate",
            ));
        }
        if response.expires_in <= 0 {
            return Err(CredentialError::new(
                "CliSsoTokenRefreshExpiresIn",
                "SSO refresh token response did not include expires_in; please run 've sso login' to re-authenticate",
            ));
        }

        let cache = self.cached.as_mut().expect("SSO cache is loaded");
        cache.access_token = response.access_token;
        if !response.refresh_token.trim().is_empty() {
            cache.refresh_token = response.refresh_token;
        }
        cache.expires_at = (Utc::now() + chrono::Duration::seconds(response.expires_in))
            .to_rfc3339_opts(SecondsFormat::Secs, true);
        Ok(())
    }

    fn fetch_role_credentials(&mut self) -> Result<()> {
        let account_id = self.profile.account_id.trim();
        if account_id.is_empty() {
            return Err(CredentialError::new(
                "CliConfigAccountIDMissing",
                format!(
                    "cli config profile {} in {} did not contain account-id",
                    self.profile_name,
                    self.config_path.display()
                ),
            ));
        }
        let role_name = self.profile.role_name.trim();
        if role_name.is_empty() {
            return Err(CredentialError::new(
                "CliConfigRoleNameMissing",
                format!(
                    "cli config profile {} in {} did not contain role-name",
                    self.profile_name,
                    self.config_path.display()
                ),
            ));
        }
        let access_token = &self
            .cached
            .as_ref()
            .expect("SSO cache is loaded")
            .access_token;
        let role_credentials = request_sso_role_credentials(
            &self.region,
            access_token,
            account_id,
            role_name,
            self.endpoints.sso_portal.as_deref(),
        )
        .map_err(|failure| {
            CredentialError::with_source(
                "CliSsoPortalCredentials",
                "failed to get SSO role credentials; please run 've sso login' to re-authenticate",
                failure,
            )
        })?;

        if role_credentials.access_key_id.trim().is_empty()
            || role_credentials.secret_access_key.trim().is_empty()
        {
            return Err(CredentialError::new(
                "CliSsoPortalCredentialsEmpty",
                "SSO portal credentials response did not include access key or secret key",
            ));
        }

        self.expires_at = (role_credentials.expiration > 0)
            .then(|| unix_timestamp_to_datetime(role_credentials.expiration))
            .transpose()?;
        self.credentials = Some(CredentialValue::cli(
            role_credentials.access_key_id,
            role_credentials.secret_access_key,
            role_credentials.session_token,
        ));
        Ok(())
    }
}

fn load_sso_cache(path: &Path) -> Result<SsoTokenCache> {
    let bytes = fs::read(path).map_err(|error| {
        CredentialError::with_source(
            "CliSsoTokenCacheLoad",
            format!(
                "failed to load sso token cache file {}; please run 've sso login' to re-authenticate",
                path.display()
            ),
            error,
        )
    })?;
    if bytes.iter().all(u8::is_ascii_whitespace) {
        return Err(CredentialError::new(
            "CliSsoTokenCacheEmpty",
            format!(
                "sso token cache file {} was empty; please run 've sso login' to re-authenticate",
                path.display()
            ),
        ));
    }
    serde_json::from_slice(&bytes).map_err(|error| {
        CredentialError::with_source(
            "CliSsoTokenCacheUnmarshal",
            format!(
                "failed to unmarshal sso token cache file {}; please run 've sso login' to re-authenticate",
                path.display()
            ),
            error,
        )
    })
}

#[derive(Debug, Serialize)]
struct SsoTokenRequest<'a> {
    grant_type: &'static str,
    client_id: &'a str,
    client_secret: &'a str,
    refresh_token: &'a str,
}

#[derive(Debug, Deserialize)]
struct SsoTokenResponse {
    #[serde(default)]
    access_token: String,
    #[serde(default)]
    refresh_token: String,
    #[serde(default)]
    expires_in: i64,
}

fn request_sso_token(
    region: &str,
    cache: &SsoTokenCache,
    endpoint_override: Option<&str>,
) -> std::result::Result<SsoTokenResponse, HttpFailure> {
    let base = endpoint_override
        .map(str::to_owned)
        .unwrap_or_else(|| format!("https://cloudidentity-oauth.{region}.volces.com"));
    let endpoint = format!("{}/token", base.trim_end_matches('/'));
    let client = build_http_client(Duration::from_secs(10))?;
    retry_http(3, || {
        let response = client
            .post(&endpoint)
            .json(&SsoTokenRequest {
                grant_type: "refresh_token",
                client_id: &cache.client_id,
                client_secret: &cache.client_secret,
                refresh_token: &cache.refresh_token,
            })
            .send()
            .map_err(HttpFailure::transport)?;
        parse_json_response(response, OAuthErrorShape::Standard)
    })
}

#[derive(Debug, Deserialize)]
struct PortalRoleCredentials {
    #[serde(rename = "AccessKeyId", default)]
    access_key_id: String,
    #[serde(rename = "SecretAccessKey", default)]
    secret_access_key: String,
    #[serde(rename = "SessionToken", default)]
    session_token: String,
    #[serde(rename = "Expiration", default)]
    expiration: i64,
}

fn request_sso_role_credentials(
    region: &str,
    access_token: &str,
    account_id: &str,
    role_name: &str,
    endpoint_override: Option<&str>,
) -> std::result::Result<PortalRoleCredentials, HttpFailure> {
    let base = endpoint_override
        .map(str::to_owned)
        .unwrap_or_else(|| format!("https://cloudidentity-portal.{region}.volces.com"));
    let endpoint = format!("{}/federation/credentials", base.trim_end_matches('/'));
    let client = build_http_client(Duration::from_secs(30))?;
    retry_http(3, || {
        let response = client
            .get(&endpoint)
            .query(&[("account_id", account_id), ("role_name", role_name)])
            .header(ACCEPT, "application/json")
            .header("x-bd-cloudidentity-bearer-token", access_token)
            .send()
            .map_err(HttpFailure::transport)?;
        let envelope: PortalEnvelope<PortalResult> =
            parse_json_response(response, OAuthErrorShape::Portal)?;
        if let Some(error) = envelope.response_metadata.error {
            return Err(HttpFailure::api(
                StatusCode::OK,
                Some(error.code),
                error.message,
                envelope.response_metadata.request_id,
            ));
        }
        envelope
            .result
            .map(|result| result.role_credentials)
            .ok_or_else(|| {
                HttpFailure::decode("GetRoleCredentials succeeded but response was empty")
            })
    })
}

#[derive(Debug, Deserialize)]
struct PortalEnvelope<T> {
    #[serde(rename = "ResponseMetadata", default)]
    response_metadata: PortalMetadata,
    #[serde(rename = "Result")]
    result: Option<T>,
}

#[derive(Debug, Deserialize, Default)]
struct PortalMetadata {
    #[serde(rename = "RequestId", default)]
    request_id: String,
    #[serde(rename = "Error", default)]
    error: Option<PortalError>,
}

#[derive(Debug, Deserialize)]
struct PortalError {
    #[serde(rename = "Code", default)]
    code: String,
    #[serde(rename = "Message", default)]
    message: String,
}

#[derive(Debug, Deserialize)]
struct PortalResult {
    #[serde(rename = "RoleCredentials")]
    role_credentials: PortalRoleCredentials,
}

#[derive(Debug, Clone, Deserialize)]
struct LoginTokenCache {
    #[serde(default)]
    access_token: JsonValue,
    #[serde(default)]
    issued_at: String,
    #[serde(default)]
    expires_in: i64,
    #[serde(default)]
    token_type: String,
    #[serde(default)]
    refresh_token: String,
    #[serde(default)]
    client_id: String,
    #[serde(default)]
    scope: String,
    #[serde(default)]
    endpoint_url: String,
}

struct ConsoleLoginProvider {
    login_session: String,
    cache_dir: PathBuf,
    cached: Option<LoginTokenCache>,
    credentials: Option<CredentialValue>,
    expires_at: Option<DateTime<Utc>>,
}

impl ConsoleLoginProvider {
    fn new(login_session: String, cache_dir: PathBuf) -> Self {
        Self {
            login_session,
            cache_dir,
            cached: None,
            credentials: None,
            expires_at: None,
        }
    }

    fn is_expired(&self) -> bool {
        self.credentials.is_none()
            || self
                .expires_at
                .is_none_or(|expiration| within_expiry_window(expiration, EXPIRY_WINDOW))
    }

    fn get(&mut self) -> Result<CredentialValue> {
        if self.is_expired() {
            self.refresh()?;
        }
        self.credentials.clone().ok_or_else(|| {
            CredentialError::new(
                "CliConsoleLoginAccessTokenInvalid",
                "console-login provider did not produce credentials",
            )
        })
    }

    fn refresh(&mut self) -> Result<()> {
        if self.cached.is_none() {
            self.cached = Some(self.load_cache()?);
        }
        if self.try_apply_cache()? {
            return Ok(());
        }

        match self.refresh_access_token() {
            Ok(()) => Ok(()),
            Err(error) if !is_invalid_grant(&error) => Err(error),
            Err(first_error) => {
                let first_error_message = first_error.to_string();
                let disk_cache = self.load_cache().map_err(|disk_error| {
                    CredentialError::with_source(
                        "CliConsoleLoginRefreshTokenExpired",
                        format!(
                            "console-login refresh token rejected ({first_error_message}); failed to reload cache from disk: {disk_error}; please run 've login'"
                        ),
                        disk_error,
                    )
                })?;
                let current_refresh_token = self
                    .cached
                    .as_ref()
                    .map(|cache| cache.refresh_token.as_str())
                    .unwrap_or_default();
                if disk_cache.refresh_token.trim().is_empty()
                    || disk_cache.refresh_token == current_refresh_token
                {
                    return Err(CredentialError::with_source(
                        "CliConsoleLoginRefreshTokenExpired",
                        "console-login refresh token has been rejected by signin service; please run 've login' to re-authenticate",
                        first_error,
                    ));
                }
                self.cached = Some(disk_cache);
                if self.try_apply_cache()? {
                    return Ok(());
                }
                self.refresh_access_token().map_err(|error| {
                    CredentialError::with_source(
                        "CliConsoleLoginRefreshTokenExpired",
                        "console-login refresh token rejected; reloaded cache from disk but new refresh token also failed; please run 've login'",
                        error,
                    )
                })
            }
        }
    }

    fn cache_path(&self) -> PathBuf {
        self.cache_dir
            .join(format!("{}.json", sha1_hex(self.login_session.as_bytes())))
    }

    fn load_cache(&self) -> Result<LoginTokenCache> {
        let path = self.cache_path();
        let bytes = fs::read(&path).map_err(|error| {
            let code = if error.kind() == std::io::ErrorKind::NotFound {
                "CliConsoleLoginCacheMissing"
            } else {
                "CliConsoleLoginCacheLoad"
            };
            CredentialError::with_source(
                code,
                format!(
                    "failed to load console-login token cache file {}; please run 've login' first",
                    path.display()
                ),
                error,
            )
        })?;
        if bytes.iter().all(u8::is_ascii_whitespace) {
            return Err(CredentialError::new(
                "CliConsoleLoginCacheEmpty",
                format!(
                    "console-login token cache file {} was empty",
                    path.display()
                ),
            ));
        }
        serde_json::from_slice(&bytes).map_err(|error| {
            CredentialError::with_source(
                "CliConsoleLoginCacheUnmarshal",
                format!(
                    "failed to unmarshal console-login token cache file {}",
                    path.display()
                ),
                error,
            )
        })
    }

    fn try_apply_cache(&mut self) -> Result<bool> {
        let Some(cache) = self.cached.as_ref() else {
            return Ok(false);
        };
        let Ok(expiration) = console_cache_expiration(cache) else {
            return Ok(false);
        };
        if within_expiry_window(expiration, EXPIRY_WINDOW) {
            return Ok(false);
        }
        let Ok(credentials) = parse_console_access_token(&cache.access_token) else {
            return Ok(false);
        };
        self.credentials = Some(credentials);
        self.expires_at = Some(expiration);
        Ok(true)
    }

    fn refresh_access_token(&mut self) -> Result<()> {
        let cache = self.cached.as_ref().expect("console-login cache is loaded");
        if cache.refresh_token.trim().is_empty() {
            return Err(CredentialError::new(
                "CliConsoleLoginRefreshTokenMissing",
                format!(
                    "console-login cache for login_session {:?} does not contain refresh_token; please run 've login' first",
                    self.login_session
                ),
            ));
        }
        if cache.client_id.trim().is_empty() {
            return Err(CredentialError::new(
                "CliConsoleLoginClientIDMissing",
                format!(
                    "console-login cache for login_session {:?} does not contain client_id; please run 've login' to regenerate",
                    self.login_session
                ),
            ));
        }

        let response = request_console_token(cache).map_err(|failure| {
            if failure.is_invalid_grant() {
                CredentialError::with_source(
                    "CliConsoleLoginInvalidGrant",
                    "console-login authorization server rejected the refresh token",
                    failure,
                )
            } else {
                CredentialError::with_source(
                    "CliConsoleLoginRefreshTokenFailed",
                    "failed to refresh console-login access token; please run 've login' to re-authenticate",
                    failure,
                )
            }
        })?;

        if response.access_token.trim().is_empty() || response.expires_in <= 0 {
            return Err(CredentialError::new(
                "CliConsoleLoginAccessTokenInvalid",
                "console-login refresh response did not include a valid access_token and expires_in",
            ));
        }
        let access_token = serde_json::from_str(&response.access_token).map_err(|error| {
            CredentialError::with_source(
                "CliConsoleLoginAccessTokenInvalid",
                "console-login refresh succeeded but the new access_token could not be parsed into STS credentials",
                error,
            )
        })?;

        let cache = self.cached.as_mut().expect("console-login cache is loaded");
        cache.access_token = access_token;
        if !response.refresh_token.trim().is_empty() {
            cache.refresh_token = response.refresh_token;
        }
        cache.issued_at = Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true);
        cache.expires_in = response.expires_in;
        if !response.token_type.trim().is_empty() {
            cache.token_type = response.token_type;
        }

        if !self.try_apply_cache()? {
            return Err(CredentialError::new(
                "CliConsoleLoginAccessTokenInvalid",
                "console-login refresh succeeded but the new access_token could not be parsed into STS credentials",
            ));
        }
        Ok(())
    }
}

fn console_cache_expiration(cache: &LoginTokenCache) -> Result<DateTime<Utc>> {
    if cache.expires_in <= 0 {
        return Err(CredentialError::new(
            "CliConsoleLoginTokenExpiration",
            "expires_in is missing or invalid",
        ));
    }
    Ok(parse_rfc3339(&cache.issued_at)? + chrono::Duration::seconds(cache.expires_in))
}

fn parse_console_access_token(value: &JsonValue) -> Result<CredentialValue> {
    let parsed = match value {
        JsonValue::String(encoded) => serde_json::from_str(encoded).map_err(|error| {
            CredentialError::with_source(
                "CliConsoleLoginAccessTokenParse",
                "failed to parse console-login access_token",
                error,
            )
        })?,
        value => value.clone(),
    };
    let object = parsed.as_object().ok_or_else(|| {
        CredentialError::new(
            "CliConsoleLoginAccessTokenParse",
            "console-login access_token is not an object",
        )
    })?;
    let field = |name: &str| {
        object
            .get(name)
            .and_then(JsonValue::as_str)
            .map(str::trim)
            .filter(|value| !value.is_empty())
            .map(str::to_owned)
            .ok_or_else(|| {
                CredentialError::new(
                    "CliConsoleLoginAccessTokenParse",
                    format!("parsed STS credentials missing {name}"),
                )
            })
    };
    Ok(CredentialValue::cli(
        field("access_key_id")?,
        field("secret_access_key")?,
        field("session_token")?,
    ))
}

#[derive(Debug, Deserialize)]
struct ConsoleTokenResponse {
    #[serde(default)]
    access_token: String,
    #[serde(default)]
    token_type: String,
    #[serde(default)]
    expires_in: i64,
    #[serde(default)]
    refresh_token: String,
}

fn request_console_token(
    cache: &LoginTokenCache,
) -> std::result::Result<ConsoleTokenResponse, HttpFailure> {
    let base = if cache.endpoint_url.trim().is_empty() {
        DEFAULT_CONSOLE_LOGIN_ENDPOINT
    } else {
        cache.endpoint_url.trim()
    };
    let endpoint = format!("{}/authorize/oauth/token", base.trim_end_matches('/'));
    let client = build_http_client(Duration::from_secs(30))?;
    retry_http(3, || {
        let mut form = vec![
            ("grant_type", "refresh_token"),
            ("client_id", cache.client_id.as_str()),
            ("refresh_token", cache.refresh_token.as_str()),
        ];
        if !cache.scope.trim().is_empty() {
            form.push(("scope", cache.scope.as_str()));
        }
        let response = client
            .post(&endpoint)
            .form(&form)
            .send()
            .map_err(HttpFailure::transport)?;
        parse_json_response(response, OAuthErrorShape::Standard)
    })
}

struct IamRoleProvider {
    profile: CliProfile,
    endpoints: EndpointOverrides,
    credentials: Option<CredentialValue>,
    expires_at: Option<DateTime<Utc>>,
}

impl IamRoleProvider {
    fn new(profile: CliProfile, endpoints: EndpointOverrides) -> Self {
        Self {
            profile,
            endpoints,
            credentials: None,
            expires_at: None,
        }
    }

    fn is_expired(&self) -> bool {
        self.credentials.is_none()
            || self
                .expires_at
                .is_none_or(|expiration| Utc::now() > expiration)
    }

    fn get(&mut self) -> Result<CredentialValue> {
        if self.is_expired() {
            self.refresh()?;
        }
        self.credentials.clone().ok_or_else(|| {
            CredentialError::new(
                "StsProviderEmpty",
                "STS AssumeRole returned empty credentials",
            )
        })
    }

    fn refresh(&mut self) -> Result<()> {
        let duration_seconds = 3600_i64;
        let region = match self.profile.region.trim() {
            "" => DEFAULT_REGION,
            value => value,
        };
        let base = self.endpoints.sts.clone().unwrap_or_else(|| {
            format!(
                "{}://{DEFAULT_STS_HOST}",
                if self.profile.disable_ssl {
                    "http"
                } else {
                    "https"
                }
            )
        });
        let role_trn = format!(
            "trn:iam::{}:role/{}",
            self.profile.account_id, self.profile.role_name
        );
        let session_name = Uuid::new_v4().to_string();

        let response = request_assume_role(
            &base,
            region,
            &self.profile.access_key,
            &self.profile.secret_key,
            &self.profile.session_token,
            &role_trn,
            &session_name,
            duration_seconds,
        )
        .map_err(|failure| {
            CredentialError::with_source(
                "StsProviderAssumeRole",
                "failed to retrieve credentials with STS AssumeRole",
                failure,
            )
        })?;

        self.credentials = Some(CredentialValue::named(
            response.access_key_id,
            response.secret_access_key,
            response.session_token,
            "StsProvider",
        ));
        self.expires_at = Some(Utc::now() + chrono::Duration::seconds(duration_seconds - 60));
        Ok(())
    }
}

struct OidcProvider {
    profile: CliProfile,
    endpoints: EndpointOverrides,
    credentials: Option<CredentialValue>,
    expires_at: Option<DateTime<Utc>>,
}

impl OidcProvider {
    fn new(profile: CliProfile, endpoints: EndpointOverrides) -> Self {
        Self {
            profile,
            endpoints,
            credentials: None,
            expires_at: None,
        }
    }

    fn is_expired(&self) -> bool {
        self.credentials.is_none()
            || self
                .expires_at
                .is_none_or(|expiration| Utc::now() > expiration)
    }

    fn get(&mut self) -> Result<CredentialValue> {
        if self.is_expired() {
            self.refresh()?;
        }
        self.credentials.clone().ok_or_else(|| {
            CredentialError::new(
                "OIDCCredentialsEmpty",
                "OIDC provider returned empty credentials",
            )
        })
    }

    fn refresh(&mut self) -> Result<()> {
        let oidc_token =
            fs::read_to_string(self.profile.oidc_token_file.trim()).map_err(|error| {
                CredentialError::with_source(
                    "OIDCTokenFileLoad",
                    format!(
                        "failed to read OIDC token file {}",
                        self.profile.oidc_token_file.trim()
                    ),
                    error,
                )
            })?;
        let base = self.endpoints.sts.clone().unwrap_or_else(|| {
            format!(
                "{}://{DEFAULT_STS_HOST}",
                if self.profile.disable_ssl {
                    "http"
                } else {
                    "https"
                }
            )
        });
        let response = request_assume_role_with_oidc(
            &base,
            self.profile.role_trn.trim(),
            &oidc_token,
            "volcengine-go-sdk-oidc-session",
            3660,
        )
        .map_err(|failure| {
            CredentialError::with_source(
                "OIDCAssumeRole",
                "failed to retrieve credentials with STS AssumeRoleWithOIDC",
                failure,
            )
        })?;
        let expiration = parse_rfc3339(&response.expiration)?;
        self.credentials = Some(CredentialValue::named(
            response.access_key_id,
            response.secret_access_key,
            response.session_token,
            "OIDC",
        ));
        self.expires_at = expiration.checked_sub_signed(chrono::Duration::seconds(60));
        Ok(())
    }
}

struct EcsRoleProvider {
    role_name: String,
    endpoints: EndpointOverrides,
    credentials: Option<CredentialValue>,
    expires_at: Option<DateTime<Utc>>,
}

impl EcsRoleProvider {
    fn new(role_name: String, endpoints: EndpointOverrides) -> Self {
        Self {
            role_name,
            endpoints,
            credentials: None,
            expires_at: None,
        }
    }

    fn is_expired(&self) -> bool {
        self.credentials.is_none()
            || self.expires_at.is_none_or(|expiration| {
                within_expiry_window(expiration, Duration::from_secs(5 * 60))
            })
    }

    fn get(&mut self) -> Result<CredentialValue> {
        if self.is_expired() {
            self.refresh()?;
        }
        self.credentials.clone().ok_or_else(|| {
            CredentialError::new(
                "EcsRoleCredentialsEmpty",
                "ECS role provider returned empty credentials",
            )
        })
    }

    fn refresh(&mut self) -> Result<()> {
        if nonempty_env("VOLCENGINE_ECS_METADATA_DISABLED")
            .is_some_and(|value| value.eq_ignore_ascii_case("true"))
        {
            return Err(CredentialError::new(
                "EcsRoleDisabled",
                "IMDS credential retrieval is disabled via VOLCENGINE_ECS_METADATA_DISABLED=true",
            ));
        }

        let base = self
            .endpoints
            .imds
            .as_deref()
            .unwrap_or(DEFAULT_IMDS_ENDPOINT)
            .trim_end_matches('/');
        let client = build_http_client(Duration::from_secs(2)).map_err(|failure| {
            CredentialError::with_source(
                "EcsRoleClient",
                "failed to construct IMDS client",
                failure,
            )
        })?;
        let token_endpoint = format!("{base}/latest/api/token");
        let token = retry_imds(|| {
            let response = client
                .put(&token_endpoint)
                .header("X-volc-ecs-metadata-token-ttl-seconds", "21600")
                .send()
                .map_err(HttpFailure::transport)?;
            parse_text_response(response)
        })
        .map_err(|failure| {
            CredentialError::with_source(
                "EcsRoleTokenFailed",
                "failed to get IMDSv2 token",
                failure,
            )
        })?;
        let token = token.trim();
        if token.is_empty() {
            return Err(CredentialError::new(
                "EcsRoleTokenEmpty",
                "IMDSv2 token endpoint returned empty response",
            ));
        }

        let role_name = if !self.role_name.is_empty() {
            self.role_name.clone()
        } else if let Some(role_name) = nonempty_env("VOLCENGINE_ECS_METADATA") {
            role_name
        } else {
            let endpoint =
                format!("{base}/volcstack/latest/iam/security_credentials?type=user&format=json");
            let body = retry_imds(|| {
                let response = client
                    .get(&endpoint)
                    .header("X-volc-ecs-metadata-token", token)
                    .send()
                    .map_err(HttpFailure::transport)?;
                parse_text_response(response)
            })
            .map_err(|failure| {
                CredentialError::with_source(
                    "EcsRoleDetectFailed",
                    "failed to auto-detect role name from IMDS",
                    failure,
                )
            })?;
            parse_imds_role_names(&body)?
                .into_iter()
                .next()
                .ok_or_else(|| {
                    CredentialError::new("EcsRoleNotFound", "no IAM roles found via IMDS")
                })?
        };

        let endpoint = format!(
            "{base}/volcstack/latest/iam/security_credentials/{}",
            percent_encode(&role_name, false)
        );
        let response: ImdsCredentialResponse = retry_imds(|| {
            let response = client
                .get(&endpoint)
                .header("X-volc-ecs-metadata-token", token)
                .send()
                .map_err(HttpFailure::transport)?;
            parse_json_response(response, OAuthErrorShape::None)
        })
        .map_err(|failure| {
            CredentialError::with_source(
                "EcsRoleCredentialsFailed",
                format!("failed to get credentials for role {role_name} from IMDS"),
                failure,
            )
        })?;
        if response.access_key_id.is_empty() || response.secret_access_key.is_empty() {
            return Err(CredentialError::new(
                "EcsRoleCredentialsEmpty",
                "IMDS returned empty credentials",
            ));
        }

        self.expires_at = Some(parse_rfc3339(&response.expiration).map_err(|error| {
            CredentialError::with_source(
                "EcsRoleExpirationParse",
                format!(
                    "failed to parse credential expiration: {}",
                    response.expiration
                ),
                error,
            )
        })?);
        self.credentials = Some(CredentialValue::named(
            response.access_key_id,
            response.secret_access_key,
            response.session_token,
            "EcsRoleProvider",
        ));
        Ok(())
    }
}

#[derive(Debug, Deserialize)]
struct ImdsCredentialResponse {
    #[serde(rename = "AccessKeyId", default)]
    access_key_id: String,
    #[serde(rename = "SecretAccessKey", default)]
    secret_access_key: String,
    #[serde(rename = "SessionToken", default)]
    session_token: String,
    #[serde(rename = "ExpiredTime", default)]
    expiration: String,
}

fn parse_imds_role_names(body: &str) -> Result<Vec<String>> {
    if let Ok(roles) = serde_json::from_str::<Vec<String>>(body) {
        return Ok(roles
            .into_iter()
            .map(|role| role.trim().to_owned())
            .filter(|role| !role.is_empty())
            .collect());
    }
    Ok(body
        .lines()
        .map(str::trim)
        .filter(|role| !role.is_empty())
        .map(str::to_owned)
        .collect())
}

#[derive(Debug, Deserialize)]
struct StsEnvelope {
    #[serde(rename = "ResponseMetadata", default)]
    response_metadata: StsMetadata,
    #[serde(rename = "Result", default)]
    result: Option<StsResult>,
}

#[derive(Debug, Deserialize, Default)]
struct StsMetadata {
    #[serde(rename = "RequestId", default)]
    request_id: String,
    #[serde(rename = "Error", default)]
    error: Option<StsError>,
}

#[derive(Debug, Deserialize)]
struct StsError {
    #[serde(rename = "Code", default)]
    code: String,
    #[serde(rename = "Message", default)]
    message: String,
}

#[derive(Debug, Deserialize)]
struct StsResult {
    #[serde(rename = "Credentials")]
    credentials: StsCredentials,
}

#[derive(Debug, Deserialize)]
struct StsCredentials {
    #[serde(rename = "AccessKeyId", default)]
    access_key_id: String,
    #[serde(rename = "SecretAccessKey", default)]
    secret_access_key: String,
    #[serde(rename = "SessionToken", default)]
    session_token: String,
    #[serde(rename = "Expiration", alias = "ExpiredTime", default)]
    expiration: String,
}

#[allow(clippy::too_many_arguments)]
fn request_assume_role(
    base: &str,
    region: &str,
    access_key: &str,
    secret_key: &str,
    session_token: &str,
    role_trn: &str,
    role_session_name: &str,
    duration_seconds: i64,
) -> std::result::Result<StsCredentials, HttpFailure> {
    let query = BTreeMap::from([
        ("Action", "AssumeRole".to_owned()),
        ("DurationSeconds", duration_seconds.to_string()),
        ("RoleSessionName", role_session_name.to_owned()),
        ("RoleTrn", role_trn.to_owned()),
        ("Version", "2018-01-01".to_owned()),
    ]);
    let client = build_http_client(Duration::from_secs(5))?;
    retry_sts(|| {
        let signed = sign_query_request(
            base,
            &query,
            access_key,
            secret_key,
            session_token,
            region,
            "sts",
        );
        let response = client
            .get(&signed.url)
            .headers(signed.headers)
            .send()
            .map_err(HttpFailure::transport)?;
        parse_sts_response(response)
    })
}

fn request_assume_role_with_oidc(
    base: &str,
    role_trn: &str,
    oidc_token: &str,
    role_session_name: &str,
    duration_seconds: i64,
) -> std::result::Result<StsCredentials, HttpFailure> {
    let endpoint = format!(
        "{}/?Action=AssumeRoleWithOIDC&Version=2018-01-01",
        base.trim_end_matches('/')
    );
    let form = BTreeMap::from([
        ("DurationSeconds", duration_seconds.to_string()),
        ("OIDCToken", oidc_token.to_owned()),
        ("RoleSessionName", role_session_name.to_owned()),
        ("RoleTrn", role_trn.to_owned()),
    ]);
    let client = build_http_client(Duration::from_secs(10))?;
    retry_sts(|| {
        let response = client
            .post(&endpoint)
            .header(CONTENT_TYPE, "application/x-www-form-urlencoded")
            .body(form_encode(&form, true))
            .send()
            .map_err(HttpFailure::transport)?;
        parse_sts_response(response)
    })
}

fn parse_sts_response(response: Response) -> std::result::Result<StsCredentials, HttpFailure> {
    let status = response.status();
    let request_id = response
        .headers()
        .get("x-tt-logid")
        .and_then(|value| value.to_str().ok())
        .unwrap_or_default()
        .to_owned();
    let body = response.bytes().map_err(HttpFailure::transport)?;
    let envelope = serde_json::from_slice::<StsEnvelope>(&body).map_err(|error| {
        if status.is_success() {
            HttpFailure::decode(format!("failed to decode STS response: {error}"))
        } else {
            HttpFailure::api(
                status,
                None,
                String::from_utf8_lossy(&body),
                request_id.clone(),
            )
        }
    })?;
    if !status.is_success() {
        let (code, message) = envelope
            .response_metadata
            .error
            .map(|error| (Some(error.code), error.message))
            .unwrap_or_else(|| (None, String::from_utf8_lossy(&body).into_owned()));
        return Err(HttpFailure::api(
            status,
            code,
            message,
            envelope.response_metadata.request_id,
        ));
    }
    if let Some(error) = envelope.response_metadata.error {
        return Err(HttpFailure::api(
            status,
            Some(error.code),
            error.message,
            envelope.response_metadata.request_id,
        ));
    }
    envelope
        .result
        .map(|result| result.credentials)
        .ok_or_else(|| HttpFailure::decode("STS response did not contain credentials"))
}

struct SignedRequest {
    url: String,
    headers: HeaderMap,
}

fn sign_query_request(
    base: &str,
    query: &BTreeMap<&str, String>,
    access_key: &str,
    secret_key: &str,
    session_token: &str,
    region: &str,
    service: &str,
) -> SignedRequest {
    sign_query_request_at(
        base,
        query,
        access_key,
        secret_key,
        session_token,
        region,
        service,
        Utc::now(),
    )
}

#[allow(clippy::too_many_arguments)]
fn sign_query_request_at(
    base: &str,
    query: &BTreeMap<&str, String>,
    access_key: &str,
    secret_key: &str,
    session_token: &str,
    region: &str,
    service: &str,
    now: DateTime<Utc>,
) -> SignedRequest {
    let timestamp = now.format("%Y%m%dT%H%M%SZ").to_string();
    let date = &timestamp[..8];
    let body_hash = sha256_hex(b"");
    let host = base
        .split_once("://")
        .map(|(_, host)| host.trim_end_matches('/'))
        .unwrap_or(base)
        .to_owned();
    let normalized_host = normalize_host(&host);

    let mut signed_headers = BTreeMap::from([
        (
            "content-type",
            "application/x-www-form-urlencoded; charset=utf-8".to_owned(),
        ),
        ("host", normalized_host),
        ("x-content-sha256", body_hash.clone()),
        ("x-date", timestamp.clone()),
    ]);
    if !session_token.is_empty() {
        signed_headers.insert("x-security-token", session_token.to_owned());
    }
    let canonical_headers = signed_headers
        .iter()
        .map(|(name, value)| format!("{name}:{}\n", value.trim()))
        .collect::<String>();
    let signed_header_names = signed_headers.keys().copied().collect::<Vec<_>>().join(";");
    let canonical_query = form_encode(query, false);
    let canonical_request = format!(
        "GET\n/\n{canonical_query}\n{canonical_headers}\n{signed_header_names}\n{body_hash}"
    );
    let credential_scope = format!("{date}/{region}/{service}/request");
    let string_to_sign = format!(
        "HMAC-SHA256\n{timestamp}\n{credential_scope}\n{}",
        sha256_hex(canonical_request.as_bytes())
    );
    let signing_key = signing_key(secret_key, date, region, service);
    let signature = hmac_sha256_hex(&signing_key, string_to_sign.as_bytes());
    let authorization = format!(
        "HMAC-SHA256 Credential={access_key}/{credential_scope}, SignedHeaders={signed_header_names}, Signature={signature}"
    );

    let mut headers = HeaderMap::new();
    headers.insert(ACCEPT, HeaderValue::from_static("application/json"));
    headers.insert(
        CONTENT_TYPE,
        HeaderValue::from_static("application/x-www-form-urlencoded; charset=utf-8"),
    );
    headers.insert(
        HeaderName::from_static("x-date"),
        HeaderValue::from_str(&timestamp).expect("timestamp is a valid header"),
    );
    headers.insert(
        HeaderName::from_static("x-content-sha256"),
        HeaderValue::from_str(&body_hash).expect("hash is a valid header"),
    );
    if !session_token.is_empty() {
        headers.insert(
            HeaderName::from_static("x-security-token"),
            HeaderValue::from_str(session_token).expect("session token is a valid header"),
        );
    }
    headers.insert(
        AUTHORIZATION,
        HeaderValue::from_str(&authorization).expect("authorization is a valid header"),
    );

    SignedRequest {
        url: format!(
            "{}/?{}",
            base.trim_end_matches('/'),
            form_encode(query, true)
        ),
        headers,
    }
}

fn normalize_host(host: &str) -> String {
    let Some((name, port)) = host.rsplit_once(':') else {
        return host.to_owned();
    };
    if port == "80" || port == "443" {
        name.to_owned()
    } else {
        host.to_owned()
    }
}

fn signing_key(secret_key: &str, date: &str, region: &str, service: &str) -> Vec<u8> {
    let date_key = hmac_sha256(secret_key.as_bytes(), date.as_bytes());
    let region_key = hmac_sha256(&date_key, region.as_bytes());
    let service_key = hmac_sha256(&region_key, service.as_bytes());
    hmac_sha256(&service_key, b"request")
}

fn hmac_sha256(key: &[u8], content: &[u8]) -> Vec<u8> {
    let mut mac = Hmac::<Sha256>::new_from_slice(key).expect("HMAC accepts keys of any size");
    mac.update(content);
    mac.finalize().into_bytes().to_vec()
}

fn hmac_sha256_hex(key: &[u8], content: &[u8]) -> String {
    hex::encode(hmac_sha256(key, content))
}

fn sha256_hex(content: &[u8]) -> String {
    hex::encode(Sha256::digest(content))
}

#[derive(Debug, Clone, Copy)]
enum OAuthErrorShape {
    None,
    Standard,
    Portal,
}

#[derive(Debug)]
struct HttpFailure {
    status: Option<StatusCode>,
    error_code: Option<String>,
    message: String,
    request_id: String,
    transport: Option<reqwest::Error>,
}

impl HttpFailure {
    fn transport(error: reqwest::Error) -> Self {
        Self {
            status: error.status(),
            error_code: None,
            message: error.to_string(),
            request_id: String::new(),
            transport: Some(error),
        }
    }

    fn api(
        status: StatusCode,
        error_code: Option<String>,
        message: impl Into<String>,
        request_id: impl Into<String>,
    ) -> Self {
        Self {
            status: Some(status),
            error_code,
            message: message.into(),
            request_id: request_id.into(),
            transport: None,
        }
    }

    fn decode(message: impl Into<String>) -> Self {
        Self {
            status: None,
            error_code: None,
            message: message.into(),
            request_id: String::new(),
            transport: None,
        }
    }

    fn is_invalid_grant(&self) -> bool {
        self.status == Some(StatusCode::BAD_REQUEST)
            && self
                .error_code
                .as_deref()
                .is_some_and(|code| code.eq_ignore_ascii_case("invalid_grant"))
    }

    fn is_general_retryable(&self) -> bool {
        if let Some(error) = &self.transport {
            return !error.is_timeout();
        }
        self.status.is_some_and(|status| {
            status == StatusCode::TOO_MANY_REQUESTS
                || status == StatusCode::REQUEST_TIMEOUT
                || status.is_server_error()
        })
    }

    fn is_sts_retryable(&self) -> bool {
        if self.transport.is_some() {
            return true;
        }
        if self.status.is_some_and(|status| {
            matches!(
                status,
                StatusCode::TOO_MANY_REQUESTS
                    | StatusCode::BAD_GATEWAY
                    | StatusCode::SERVICE_UNAVAILABLE
                    | StatusCode::GATEWAY_TIMEOUT
            )
        }) {
            return true;
        }
        self.error_code.as_deref().is_some_and(|code| {
            matches!(
                code,
                "ProvisionedThroughputExceededException"
                    | "Throttling"
                    | "ThrottlingException"
                    | "RequestLimitExceeded"
                    | "RequestThrottled"
                    | "RequestThrottledException"
                    | "TooManyRequestsException"
                    | "PriorRequestNotComplete"
                    | "TransactionInProgressException"
                    | "RequestError"
                    | "RequestTimeout"
                    | "ResponseTimeout"
                    | "RequestTimeoutException"
            )
        })
    }
}

impl fmt::Display for HttpFailure {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "HTTP request failed")?;
        if let Some(status) = self.status {
            write!(f, " with status {}", status.as_u16())?;
        }
        if let Some(code) = &self.error_code {
            write!(f, ": {code}")?;
        }
        if !self.message.is_empty() {
            write!(f, ": {}", self.message)?;
        }
        if !self.request_id.is_empty() {
            write!(f, " (requestId: {})", self.request_id)?;
        }
        Ok(())
    }
}

impl StdError for HttpFailure {
    fn source(&self) -> Option<&(dyn StdError + 'static)> {
        self.transport
            .as_ref()
            .map(|error| error as &(dyn StdError + 'static))
    }
}

fn build_http_client(timeout: Duration) -> std::result::Result<Client, HttpFailure> {
    Client::builder()
        .timeout(timeout)
        .build()
        .map_err(HttpFailure::transport)
}

fn retry_http<T>(
    attempts: usize,
    mut operation: impl FnMut() -> std::result::Result<T, HttpFailure>,
) -> std::result::Result<T, HttpFailure> {
    let attempts = attempts.max(1);
    for attempt in 0..attempts {
        match operation() {
            Ok(value) => return Ok(value),
            Err(error) if attempt + 1 == attempts || !error.is_general_retryable() => {
                return Err(error);
            }
            Err(_) => {
                let base_ms = (200_u64.saturating_mul(1_u64 << attempt)).min(2_000);
                let jitter_ms = rand::rng().random_range(0..100_u64);
                thread::sleep(Duration::from_millis(base_ms + jitter_ms));
            }
        }
    }
    unreachable!("retry loop always returns")
}

fn retry_sts<T>(
    mut operation: impl FnMut() -> std::result::Result<T, HttpFailure>,
) -> std::result::Result<T, HttpFailure> {
    for attempt in 0..=DEFAULT_RETRIES {
        match operation() {
            Ok(value) => return Ok(value),
            Err(error) if attempt == DEFAULT_RETRIES || !error.is_sts_retryable() => {
                return Err(error);
            }
            Err(_) => thread::sleep(Duration::from_secs(1)),
        }
    }
    unreachable!("retry loop always returns")
}

fn retry_imds<T>(
    mut operation: impl FnMut() -> std::result::Result<T, HttpFailure>,
) -> std::result::Result<T, HttpFailure> {
    for attempt in 0..=DEFAULT_RETRIES {
        match operation() {
            Ok(value) => return Ok(value),
            Err(error) if attempt == DEFAULT_RETRIES => return Err(error),
            Err(_) => thread::sleep(Duration::from_secs(1)),
        }
    }
    unreachable!("retry loop always returns")
}

fn parse_json_response<T: for<'de> Deserialize<'de>>(
    response: Response,
    error_shape: OAuthErrorShape,
) -> std::result::Result<T, HttpFailure> {
    let status = response.status();
    let request_id = response
        .headers()
        .get("x-tt-logid")
        .and_then(|value| value.to_str().ok())
        .unwrap_or_default()
        .to_owned();
    let body = response.bytes().map_err(HttpFailure::transport)?;

    if !status.is_success() {
        let (error_code, message, metadata_request_id) = match error_shape {
            OAuthErrorShape::Standard => {
                #[derive(Deserialize, Default)]
                struct StandardError {
                    #[serde(default)]
                    error: String,
                    #[serde(default)]
                    error_description: String,
                }
                serde_json::from_slice::<StandardError>(&body)
                    .map(|error| {
                        (
                            (!error.error.is_empty()).then_some(error.error),
                            error.error_description,
                            String::new(),
                        )
                    })
                    .unwrap_or_default()
            }
            OAuthErrorShape::Portal => serde_json::from_slice::<PortalEnvelope<JsonValue>>(&body)
                .ok()
                .and_then(|envelope| {
                    envelope.response_metadata.error.map(|error| {
                        (
                            (!error.code.is_empty()).then_some(error.code),
                            error.message,
                            envelope.response_metadata.request_id,
                        )
                    })
                })
                .unwrap_or_default(),
            OAuthErrorShape::None => (None, String::new(), String::new()),
        };
        return Err(HttpFailure::api(
            status,
            error_code,
            if message.is_empty() {
                String::from_utf8_lossy(&body).into_owned()
            } else {
                message
            },
            if metadata_request_id.is_empty() {
                request_id
            } else {
                metadata_request_id
            },
        ));
    }

    serde_json::from_slice(&body)
        .map_err(|error| HttpFailure::decode(format!("failed to decode response: {error}")))
}

fn parse_text_response(response: Response) -> std::result::Result<String, HttpFailure> {
    let status = response.status();
    let body = response.text().map_err(HttpFailure::transport)?;
    if status != StatusCode::OK {
        return Err(HttpFailure::api(status, None, body, ""));
    }
    Ok(body)
}

fn is_invalid_grant(error: &CredentialError) -> bool {
    let mut current: Option<&(dyn StdError + 'static)> = Some(error);
    while let Some(item) = current {
        if let Some(failure) = item.downcast_ref::<HttpFailure>() {
            return failure.is_invalid_grant();
        }
        current = item.source();
    }
    false
}

fn parse_rfc3339(value: &str) -> Result<DateTime<Utc>> {
    let value = value.trim();
    if value.is_empty() {
        return Err(CredentialError::new(
            "CliTokenExpirationParse",
            "expires_at is empty",
        ));
    }
    DateTime::parse_from_rfc3339(value)
        .map(|value| value.with_timezone(&Utc))
        .map_err(|error| {
            CredentialError::with_source(
                "CliTokenExpirationParse",
                format!("expires_at is invalid: {value}"),
                error,
            )
        })
}

fn within_expiry_window(expiration: DateTime<Utc>, window: Duration) -> bool {
    Utc::now()
        >= expiration
            .checked_sub_signed(
                chrono::Duration::from_std(window).expect("credential expiry window fits chrono"),
            )
            .unwrap_or(expiration)
}

fn unix_timestamp_to_datetime(timestamp: i64) -> Result<DateTime<Utc>> {
    let (seconds, nanoseconds) = match timestamp {
        value if value >= 1_000_000_000_000_000_000 => {
            (value / 1_000_000_000, (value % 1_000_000_000) as u32)
        }
        value if value >= 1_000_000_000_000_000 => {
            (value / 1_000_000, ((value % 1_000_000) * 1_000) as u32)
        }
        value if value >= 1_000_000_000_000 => {
            (value / 1_000, ((value % 1_000) * 1_000_000) as u32)
        }
        value => (value, 0),
    };
    DateTime::from_timestamp(seconds, nanoseconds).ok_or_else(|| {
        CredentialError::new(
            "CliConfigExpiration",
            format!("invalid Unix timestamp: {timestamp}"),
        )
    })
}

fn sso_cache_filename(start_url: &str, session_name: &str) -> Result<String> {
    #[derive(Serialize)]
    struct CacheKey<'a> {
        start_url: &'a str,
        session_name: &'a str,
    }
    let payload = serde_json::to_vec(&CacheKey {
        start_url,
        session_name,
    })
    .map_err(|error| {
        CredentialError::with_source("CliSsoCacheKey", "failed to serialize SSO cache key", error)
    })?;
    Ok(format!("{}.json", sha1_hex(&payload)))
}

fn sha1_hex(content: &[u8]) -> String {
    hex::encode(Sha1::digest(content))
}

fn form_encode<K: AsRef<str> + Ord>(values: &BTreeMap<K, String>, spaces_as_plus: bool) -> String {
    values
        .iter()
        .map(|(key, value)| {
            format!(
                "{}={}",
                percent_encode(key.as_ref(), spaces_as_plus),
                percent_encode(value, spaces_as_plus)
            )
        })
        .collect::<Vec<_>>()
        .join("&")
}

fn percent_encode(value: &str, spaces_as_plus: bool) -> String {
    let mut encoded = String::with_capacity(value.len());
    for byte in value.bytes() {
        match byte {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                encoded.push(char::from(byte));
            }
            b' ' if spaces_as_plus => encoded.push('+'),
            value => {
                const HEX: &[u8; 16] = b"0123456789ABCDEF";
                encoded.push('%');
                encoded.push(char::from(HEX[(value >> 4) as usize]));
                encoded.push(char::from(HEX[(value & 0x0f) as usize]));
            }
        }
    }
    encoded
}

/// Extracts protobuf field 3 (`ExpiredTime`) from a Volcengine session token.
pub fn extract_expired_time(session_token: &str) -> Result<u64> {
    let mut segments = session_token.split('.');
    let _header = segments.next();
    let payload = segments
        .next()
        .ok_or_else(|| CredentialError::new("SessionTokenInvalid", "invalid session token"))?;
    if segments.next().is_none() || segments.next().is_some() {
        return Err(CredentialError::new(
            "SessionTokenInvalid",
            "invalid session token",
        ));
    }
    let payload = URL_SAFE_NO_PAD.decode(payload).map_err(|error| {
        CredentialError::with_source(
            "SessionTokenDecode",
            "failed to decode session token payload",
            error,
        )
    })?;
    parse_expired_time_field(&payload)
}

fn parse_expired_time_field(mut payload: &[u8]) -> Result<u64> {
    while !payload.is_empty() {
        let (tag, tag_len) = decode_varint(payload)?;
        payload = &payload[tag_len..];
        let field_number = tag >> 3;
        let wire_type = tag & 7;
        match wire_type {
            0 => {
                let (value, value_len) = decode_varint(payload)?;
                if field_number == 3 {
                    return Ok(value);
                }
                payload = &payload[value_len..];
            }
            1 => {
                payload = payload.get(8..).ok_or_else(|| {
                    CredentialError::new(
                        "SessionTokenInvalid",
                        format!("invalid fixed64 field {field_number}"),
                    )
                })?
            }
            2 => {
                let (length, length_len) = decode_varint(payload)?;
                let length = usize::try_from(length).map_err(|_| {
                    CredentialError::new(
                        "SessionTokenInvalid",
                        format!("invalid bytes field {field_number}"),
                    )
                })?;
                payload = payload
                    .get(length_len..)
                    .and_then(|rest| rest.get(length..))
                    .ok_or_else(|| {
                        CredentialError::new(
                            "SessionTokenInvalid",
                            format!("invalid bytes field {field_number}"),
                        )
                    })?;
            }
            5 => {
                payload = payload.get(4..).ok_or_else(|| {
                    CredentialError::new(
                        "SessionTokenInvalid",
                        format!("invalid fixed32 field {field_number}"),
                    )
                })?
            }
            _ => {
                return Err(CredentialError::new(
                    "SessionTokenInvalid",
                    format!("unsupported wire type {wire_type}"),
                ));
            }
        }
    }
    Err(CredentialError::new(
        "SessionTokenExpiredTimeMissing",
        "ExpiredTime not found",
    ))
}

fn decode_varint(input: &[u8]) -> Result<(u64, usize)> {
    let mut value = 0_u64;
    for (index, byte) in input.iter().copied().take(10).enumerate() {
        if index == 9 && byte > 1 {
            break;
        }
        value |= u64::from(byte & 0x7f) << (index * 7);
        if byte & 0x80 == 0 {
            return Ok((value, index + 1));
        }
    }
    Err(CredentialError::new(
        "SessionTokenInvalid",
        "invalid protobuf varint",
    ))
}

#[cfg(test)]
mod tests {
    use std::{
        io::{Read, Write},
        net::TcpListener,
        panic,
        sync::Mutex as StdMutex,
        thread::JoinHandle,
    };

    use base64::{Engine as _, engine::general_purpose::URL_SAFE_NO_PAD};
    use chrono::{SecondsFormat, Utc};
    use serde_json::json;
    use tempfile::TempDir;

    use super::*;

    static ENV_LOCK: StdMutex<()> = StdMutex::new(());

    #[test]
    fn cli_config_normalizes_empty_sso_session_representations() {
        for config in [
            json!({"profiles": {}}),
            json!({"profiles": {}, "sso-session": null}),
            json!({"profiles": {}, "sso-session": {}}),
        ] {
            let config: CliConfig = serde_json::from_value(config).unwrap();
            assert!(config.sso_sessions.is_empty());
        }
    }

    #[test]
    fn cli_config_preserves_configured_sso_sessions() {
        let config: CliConfig = serde_json::from_value(json!({
            "profiles": {},
            "sso-session": {
                "work": {
                    "start-url": "https://example.com/start",
                    "region": "cn-beijing"
                }
            }
        }))
        .unwrap();

        let session = config.sso_sessions.get("work").unwrap();
        assert_eq!(session.start_url, "https://example.com/start");
        assert_eq!(session.region, "cn-beijing");
    }

    #[test]
    fn access_key_mode_accepts_null_sso_sessions_from_cli_config() {
        let directory = TempDir::new().unwrap();
        let config_path = write_config(
            &directory,
            json!({
                "current": "default",
                "profiles": {
                    "default": {
                        "mode": "ak",
                        "access-key": "test-ak",
                        "secret-key": "test-sk"
                    }
                },
                "sso-session": null
            }),
        );

        let value = CliCredentials::new(Some(config_path), None).get().unwrap();

        assert_eq!(value.access_key_id, "test-ak");
        assert_eq!(value.secret_access_key, "test-sk");
    }

    #[test]
    fn iam_role_and_legacy_mode_names_share_the_same_profile_mode() {
        for mode in ["iamrole", "IAMRole", "ramrolearn", "RamRoleArn"] {
            assert_eq!(
                ProfileMode::parse(mode, "test", Path::new("config.json")).unwrap(),
                ProfileMode::IamRole,
                "mode {mode} should use the IAM role provider"
            );
        }
    }

    #[test]
    fn access_key_mode_uses_explicit_profile_and_redacts_debug_output() {
        let directory = TempDir::new().unwrap();
        let config_path = write_config(
            &directory,
            json!({
                "current": "default",
                "profiles": {
                    "default": {"mode": "ak", "access-key": "default-ak", "secret-key": "default-sk"},
                    "selected": {
                        "mode": "ak",
                        "access-key": "selected-ak",
                        "secret-key": "selected-sk",
                        "session-token": "selected-token"
                    }
                }
            }),
        );
        let credentials = CliCredentials::new(Some(config_path), Some("selected".to_owned()));

        let value = credentials.get().unwrap();

        assert_eq!(value.access_key_id, "selected-ak");
        assert_eq!(value.secret_access_key, "selected-sk");
        assert_eq!(value.session_token, "selected-token");
        let debug = format!("{value:?}");
        assert!(!debug.contains("selected-ak"));
        assert!(!debug.contains("selected-sk"));
        assert!(!debug.contains("selected-token"));
    }

    #[test]
    fn profile_environment_priority_matches_go_provider() {
        let _lock = ENV_LOCK.lock().unwrap();
        let directory = TempDir::new().unwrap();
        let config_path = write_config(
            &directory,
            json!({
                "current": "current",
                "profiles": {
                    "current": {"mode": "ak", "access-key": "current-ak", "secret-key": "sk"},
                    "volcstack": {"mode": "ak", "access-key": "volcstack-ak", "secret-key": "sk"},
                    "volcengine": {"mode": "ak", "access-key": "volcengine-ak", "secret-key": "sk"}
                }
            }),
        );
        let _config = EnvGuard::set(
            "VOLCENGINE_CLI_CONFIG_FILE",
            config_path.to_string_lossy().as_ref(),
        );
        let _fallback = EnvGuard::set("VOLCSTACK_PROFILE", "volcstack");
        let primary = EnvGuard::set("VOLCENGINE_PROFILE", "volcengine");

        assert_eq!(
            CliCredentials::from_cli().get().unwrap().access_key_id,
            "volcengine-ak"
        );
        drop(primary);
        assert_eq!(
            CliCredentials::from_cli().get().unwrap().access_key_id,
            "volcstack-ak"
        );
        drop(_fallback);
        assert_eq!(
            CliCredentials::from_cli().get().unwrap().access_key_id,
            "current-ak"
        );
    }

    #[test]
    fn sso_mode_refreshes_access_token_and_exchanges_it_for_sts() {
        let directory = TempDir::new().unwrap();
        let server = MockServer::start(vec![
            ExpectedRequest::json(
                "POST",
                "/token",
                &[
                    "\"grant_type\":\"refresh_token\"",
                    "\"refresh_token\":\"sso-rt\"",
                ],
                json!({
                    "access_token": "new-sso-access-token",
                    "token_type": "Bearer",
                    "refresh_token": "rotated-sso-rt",
                    "expires_in": 3600
                }),
            ),
            ExpectedRequest::json(
                "GET",
                "/federation/credentials?account_id=2100000000&role_name=Developer",
                &["x-bd-cloudidentity-bearer-token: new-sso-access-token"],
                json!({
                    "ResponseMetadata": {"RequestId": "request-id"},
                    "Result": {"RoleCredentials": {
                        "AccessKeyId": "sso-ak",
                        "SecretAccessKey": "sso-sk",
                        "SessionToken": "sso-st",
                        "Expiration": (Utc::now().timestamp() + 3600)
                    }}
                }),
            ),
        ]);
        let session_name = "work";
        let start_url = "https://example.com/start";
        let config_path = write_config(
            &directory,
            json!({
                "current": "sso",
                "profiles": {"sso": {
                    "mode": "sso",
                    "sso-session-name": session_name,
                    "account-id": "2100000000",
                    "role-name": "Developer"
                }},
                "sso-session": {session_name: {
                    "start-url": start_url,
                    "region": "cn-beijing"
                }}
            }),
        );
        let cache_dir = directory.path().join("sso/cache");
        fs::create_dir_all(&cache_dir).unwrap();
        fs::write(
            cache_dir.join(sso_cache_filename(start_url, session_name).unwrap()),
            serde_json::to_vec(&json!({
                "access_token": "expired-access-token",
                "expires_at": (Utc::now() - chrono::Duration::hours(1)).to_rfc3339(),
                "client_id": "client-id",
                "client_secret": "client-secret",
                "client_secret_expires_at": Utc::now().timestamp() + 3600,
                "refresh_token": "sso-rt"
            }))
            .unwrap(),
        )
        .unwrap();
        let mut credentials = CliCredentials::new(Some(config_path), None);
        credentials.endpoints.sso_oauth = Some(server.base_url().to_owned());
        credentials.endpoints.sso_portal = Some(server.base_url().to_owned());

        let value = credentials.get().unwrap();

        assert_eq!(value.access_key_id, "sso-ak");
        assert_eq!(value.secret_access_key, "sso-sk");
        assert_eq!(value.session_token, "sso-st");
        server.finish();
    }

    #[test]
    fn console_login_mode_refreshes_rt_into_sts_without_writing_cache() {
        let _lock = ENV_LOCK.lock().unwrap();
        let _cache_env = EnvGuard::remove("VOLCENGINE_LOGIN_CACHE_DIRECTORY");
        let directory = TempDir::new().unwrap();
        let refreshed_sts = json!({
            "access_key_id": "login-ak",
            "secret_access_key": "login-sk",
            "session_token": "login-st"
        });
        let server = MockServer::start(vec![ExpectedRequest::json(
            "POST",
            "/authorize/oauth/token",
            &[
                "grant_type=refresh_token",
                "client_id=login-client",
                "refresh_token=login-rt",
                "scope=openid",
            ],
            json!({
                "access_token": serde_json::to_string(&refreshed_sts).unwrap(),
                "token_type": "Bearer",
                "expires_in": 900,
                "refresh_token": "rotated-login-rt"
            }),
        )]);
        let login_session = "login-session-id";
        let config_path = write_config(
            &directory,
            json!({
                "current": "default",
                "profiles": {"default": {
                    "mode": "console-login",
                    "login-session": login_session
                }}
            }),
        );
        let cache_dir = directory.path().join("login/cache");
        fs::create_dir_all(&cache_dir).unwrap();
        let cache_path = cache_dir.join(format!("{}.json", sha1_hex(login_session.as_bytes())));
        let original_cache = serde_json::to_vec(&json!({
            "login_session": login_session,
            "access_token": {
                "access_key_id": "expired-ak",
                "secret_access_key": "expired-sk",
                "session_token": "expired-st"
            },
            "issued_at": (Utc::now() - chrono::Duration::hours(1)).to_rfc3339(),
            "expires_in": 900,
            "refresh_token": "login-rt",
            "client_id": "login-client",
            "scope": "openid",
            "endpoint_url": server.base_url()
        }))
        .unwrap();
        fs::write(&cache_path, &original_cache).unwrap();

        let value = CliCredentials::new(Some(config_path), None).get().unwrap();

        assert_eq!(value.access_key_id, "login-ak");
        assert_eq!(value.secret_access_key, "login-sk");
        assert_eq!(value.session_token, "login-st");
        assert_eq!(fs::read(cache_path).unwrap(), original_cache);
        server.finish();
    }

    #[test]
    fn console_login_invalid_grant_is_not_retried_and_requires_new_login() {
        let _lock = ENV_LOCK.lock().unwrap();
        let _cache_env = EnvGuard::remove("VOLCENGINE_LOGIN_CACHE_DIRECTORY");
        let directory = TempDir::new().unwrap();
        let server = MockServer::start(vec![ExpectedRequest::json_status(
            "POST",
            "/authorize/oauth/token",
            &["refresh_token=rejected-rt"],
            StatusCode::BAD_REQUEST,
            json!({"error": "invalid_grant", "error_description": "revoked"}),
        )]);
        let login_session = "rejected-login-session";
        let config_path = write_config(
            &directory,
            json!({
                "profiles": {"default": {
                    "mode": "console-login",
                    "login-session": login_session
                }}
            }),
        );
        let cache_dir = directory.path().join("login/cache");
        fs::create_dir_all(&cache_dir).unwrap();
        fs::write(
            cache_dir.join(format!("{}.json", sha1_hex(login_session.as_bytes()))),
            serde_json::to_vec(&json!({
                "access_token": {
                    "access_key_id": "expired-ak",
                    "secret_access_key": "expired-sk",
                    "session_token": "expired-st"
                },
                "issued_at": (Utc::now() - chrono::Duration::hours(1)).to_rfc3339(),
                "expires_in": 900,
                "refresh_token": "rejected-rt",
                "client_id": "login-client",
                "endpoint_url": server.base_url()
            }))
            .unwrap(),
        )
        .unwrap();

        let error = CliCredentials::new(Some(config_path), None)
            .get()
            .unwrap_err();

        assert_eq!(error.code(), "CliConsoleLoginRefreshTokenExpired");
        assert!(error.to_string().contains("ve login"));
        server.finish();
    }

    #[test]
    fn console_login_honors_custom_cache_directory_and_string_access_token() {
        let _lock = ENV_LOCK.lock().unwrap();
        let directory = TempDir::new().unwrap();
        let custom_cache = directory.path().join("custom-cache");
        fs::create_dir_all(&custom_cache).unwrap();
        let _cache_env = EnvGuard::set(
            "VOLCENGINE_LOGIN_CACHE_DIRECTORY",
            custom_cache.to_string_lossy().as_ref(),
        );
        let login_session = "custom-cache-session";
        let config_path = write_config(
            &directory,
            json!({
                "profiles": {"default": {
                    "mode": "console-login",
                    "login-session": login_session
                }}
            }),
        );
        fs::write(
            custom_cache.join(format!("{}.json", sha1_hex(login_session.as_bytes()))),
            serde_json::to_vec(&json!({
                "access_token": serde_json::to_string(&json!({
                    "access_key_id": "custom-ak",
                    "secret_access_key": "custom-sk",
                    "session_token": "custom-st"
                })).unwrap(),
                "issued_at": Utc::now().to_rfc3339(),
                "expires_in": 900
            }))
            .unwrap(),
        )
        .unwrap();

        let value = CliCredentials::new(Some(config_path), None).get().unwrap();

        assert_eq!(value.access_key_id, "custom-ak");
        assert_eq!(value.session_token, "custom-st");
    }

    #[test]
    fn iam_role_mode_signs_assume_role_request() {
        let directory = TempDir::new().unwrap();
        let server = MockServer::start(vec![ExpectedRequest::json(
            "GET",
            "/?Action=AssumeRole",
            &[
                "authorization: HMAC-SHA256 Credential=source-ak/",
                "x-security-token: source-st",
            ],
            sts_response("role-ak", "role-sk", "role-st", "ExpiredTime"),
        )]);
        let config_path = write_config(
            &directory,
            json!({
                "current": "role",
                "profiles": {"role": {
                    "mode": "iamrole",
                    "access-key": "source-ak",
                    "secret-key": "source-sk",
                    "session-token": "source-st",
                    "account-id": "2100000000",
                    "role-name": "Developer",
                    "region": "cn-beijing"
                }}
            }),
        );
        let mut credentials = CliCredentials::new(Some(config_path), None);
        credentials.endpoints.sts = Some(server.base_url().to_owned());

        let value = credentials.get().unwrap();

        assert_eq!(value.access_key_id, "role-ak");
        assert_eq!(value.provider_name, "StsProvider");
        server.finish();
    }

    #[test]
    fn sts_signature_matches_python_sdk_signer_v4() {
        let query = BTreeMap::from([
            ("Action", "AssumeRole".to_owned()),
            ("DurationSeconds", "3600".to_owned()),
            ("Policy", String::new()),
            (
                "RoleSessionName",
                "volcengine-go-sdk-12345678-1234-1234-1234-123456789abc".to_owned(),
            ),
            ("RoleTrn", "trn:iam::2100000000:role/Developer".to_owned()),
            ("Version", "2018-01-01".to_owned()),
        ]);
        let now = DateTime::parse_from_rfc3339("2026-07-21T12:34:56Z")
            .unwrap()
            .with_timezone(&Utc);

        let signed = sign_query_request_at(
            "https://sts.volcengineapi.com",
            &query,
            "AKIDEXAMPLE",
            "secret",
            "session-token",
            "cn-beijing",
            "sts",
            now,
        );

        assert_eq!(
            signed.headers.get(AUTHORIZATION).unwrap().to_str().unwrap(),
            "HMAC-SHA256 Credential=AKIDEXAMPLE/20260721/cn-beijing/sts/request, SignedHeaders=content-type;host;x-content-sha256;x-date;x-security-token, Signature=10f40e5c592af3fc8d85c552a6b39d549824760075857bb9cabf41f1b044ecf4"
        );
        assert_eq!(signed.headers.get("x-date").unwrap(), "20260721T123456Z");
    }

    #[test]
    fn oidc_mode_reads_token_file_and_calls_unsigned_sts_endpoint() {
        let directory = TempDir::new().unwrap();
        let server = MockServer::start(vec![ExpectedRequest::json(
            "POST",
            "/?Action=AssumeRoleWithOIDC&Version=2018-01-01",
            &[
                "DurationSeconds=3660",
                "OIDCToken=oidc-token%0A",
                "RoleSessionName=volcengine-go-sdk-oidc-session",
                "RoleTrn=trn%3Aiam%3A%3A2100000000%3Arole%2FWorkload",
            ],
            sts_response("oidc-ak", "oidc-sk", "oidc-st", "Expiration"),
        )]);
        let token_path = directory.path().join("oidc-token");
        fs::write(&token_path, "oidc-token\n").unwrap();
        let config_path = write_config(
            &directory,
            json!({
                "current": "oidc",
                "profiles": {"oidc": {
                    "mode": "oidc",
                    "oidc-token-file": token_path,
                    "role-trn": "trn:iam::2100000000:role/Workload"
                }}
            }),
        );
        let mut credentials = CliCredentials::new(Some(config_path), None);
        credentials.endpoints.sts = Some(server.base_url().to_owned());

        let value = credentials.get().unwrap();

        assert_eq!(value.access_key_id, "oidc-ak");
        assert_eq!(value.provider_name, "OIDC");
        server.finish();
    }

    #[test]
    fn ecs_role_mode_uses_imdsv2_token_then_fetches_credentials() {
        let _lock = ENV_LOCK.lock().unwrap();
        let _disabled = EnvGuard::remove("VOLCENGINE_ECS_METADATA_DISABLED");
        let directory = TempDir::new().unwrap();
        let server = MockServer::start(vec![
            ExpectedRequest::text(
                "PUT",
                "/latest/api/token",
                &["x-volc-ecs-metadata-token-ttl-seconds: 21600"],
                "imds-token",
            ),
            ExpectedRequest::json(
                "GET",
                "/volcstack/latest/iam/security_credentials/MyRole",
                &["x-volc-ecs-metadata-token: imds-token"],
                json!({
                    "AccessKeyId": "ecs-ak",
                    "SecretAccessKey": "ecs-sk",
                    "SessionToken": "ecs-st",
                    "ExpiredTime": (Utc::now() + chrono::Duration::hours(1))
                        .to_rfc3339_opts(SecondsFormat::Secs, true)
                }),
            ),
        ]);
        let config_path = write_config(
            &directory,
            json!({
                "current": "ecs",
                "profiles": {"ecs": {"mode": "ecsrole", "role-name": "MyRole"}}
            }),
        );
        let mut credentials = CliCredentials::new(Some(config_path), None);
        credentials.endpoints.imds = Some(server.base_url().to_owned());

        let value = credentials.get().unwrap();

        assert_eq!(value.access_key_id, "ecs-ak");
        assert_eq!(value.provider_name, "EcsRoleProvider");
        server.finish();
    }

    #[test]
    fn cached_sso_sts_avoids_network_until_expiry_window() {
        let directory = TempDir::new().unwrap();
        let config_path = write_config(
            &directory,
            json!({
                "current": "sso",
                "profiles": {"sso": {
                    "mode": "sso",
                    "access-key": "cached-ak",
                    "secret-key": "cached-sk",
                    "session-token": "cached-st",
                    "sts-expiration": (Utc::now().timestamp() + 3600)
                }}
            }),
        );

        let value = CliCredentials::new(Some(config_path), None).get().unwrap();

        assert_eq!(value.access_key_id, "cached-ak");
        assert_eq!(value.session_token, "cached-st");
    }

    #[test]
    fn unix_timestamps_support_seconds_milliseconds_microseconds_and_nanoseconds() {
        let seconds = 1_784_635_239_i64;
        for timestamp in [
            seconds,
            seconds * 1_000,
            seconds * 1_000_000,
            seconds * 1_000_000_000,
        ] {
            assert_eq!(
                unix_timestamp_to_datetime(timestamp).unwrap().timestamp(),
                seconds
            );
        }
    }

    #[test]
    fn sso_cache_name_matches_go_json_and_sha1_convention() {
        assert_eq!(
            sso_cache_filename("https://example.com/start", "work").unwrap(),
            "ef1434a855843424932974030909e655cc244a40.json"
        );
    }

    #[test]
    fn extracts_expired_time_from_protobuf_session_token() {
        let expiration = 1_784_635_239_u64;
        let mut payload = vec![0x0a, 0x03, b'a', b'b', b'c', 0x18];
        encode_varint(expiration, &mut payload);
        let token = format!("header.{}.signature", URL_SAFE_NO_PAD.encode(payload));

        assert_eq!(extract_expired_time(&token).unwrap(), expiration);
    }

    #[test]
    fn rejects_unsupported_profile_mode() {
        let directory = TempDir::new().unwrap();
        let config_path = write_config(
            &directory,
            json!({"profiles": {"default": {"mode": "unknown"}}}),
        );

        let error = CliCredentials::new(Some(config_path), None)
            .get()
            .unwrap_err();

        assert_eq!(error.code(), "CliConfigModeInvalid");
    }

    fn write_config(directory: &TempDir, config: JsonValue) -> PathBuf {
        let path = directory.path().join("config.json");
        fs::write(&path, serde_json::to_vec(&config).unwrap()).unwrap();
        path
    }

    fn sts_response(ak: &str, sk: &str, token: &str, expiration_name: &str) -> JsonValue {
        let mut credentials = serde_json::Map::from_iter([
            ("AccessKeyId".to_owned(), json!(ak)),
            ("SecretAccessKey".to_owned(), json!(sk)),
            ("SessionToken".to_owned(), json!(token)),
        ]);
        credentials.insert(
            expiration_name.to_owned(),
            json!(
                (Utc::now() + chrono::Duration::hours(1))
                    .to_rfc3339_opts(SecondsFormat::Secs, true)
            ),
        );
        json!({
            "ResponseMetadata": {"RequestId": "request-id"},
            "Result": {"Credentials": credentials}
        })
    }

    fn encode_varint(mut value: u64, output: &mut Vec<u8>) {
        while value >= 0x80 {
            output.push((value as u8) | 0x80);
            value >>= 7;
        }
        output.push(value as u8);
    }

    struct EnvGuard {
        name: &'static str,
        previous: Option<std::ffi::OsString>,
    }

    impl EnvGuard {
        fn set(name: &'static str, value: &str) -> Self {
            let previous = env::var_os(name);
            // SAFETY: tests that mutate process environment hold ENV_LOCK.
            unsafe { env::set_var(name, value) };
            Self { name, previous }
        }

        fn remove(name: &'static str) -> Self {
            let previous = env::var_os(name);
            // SAFETY: tests that mutate process environment hold ENV_LOCK.
            unsafe { env::remove_var(name) };
            Self { name, previous }
        }
    }

    impl Drop for EnvGuard {
        fn drop(&mut self) {
            // SAFETY: tests that mutate process environment hold ENV_LOCK.
            unsafe {
                if let Some(previous) = &self.previous {
                    env::set_var(self.name, previous);
                } else {
                    env::remove_var(self.name);
                }
            }
        }
    }

    struct ExpectedRequest {
        method: &'static str,
        target: &'static str,
        fragments: Vec<String>,
        status: StatusCode,
        content_type: &'static str,
        response_body: String,
    }

    impl ExpectedRequest {
        fn json(
            method: &'static str,
            target: &'static str,
            fragments: &[&str],
            response: JsonValue,
        ) -> Self {
            Self {
                method,
                target,
                fragments: fragments.iter().map(|value| (*value).to_owned()).collect(),
                status: StatusCode::OK,
                content_type: "application/json",
                response_body: serde_json::to_string(&response).unwrap(),
            }
        }

        fn json_status(
            method: &'static str,
            target: &'static str,
            fragments: &[&str],
            status: StatusCode,
            response: JsonValue,
        ) -> Self {
            Self {
                method,
                target,
                fragments: fragments.iter().map(|value| (*value).to_owned()).collect(),
                status,
                content_type: "application/json",
                response_body: serde_json::to_string(&response).unwrap(),
            }
        }

        fn text(
            method: &'static str,
            target: &'static str,
            fragments: &[&str],
            response: &str,
        ) -> Self {
            Self {
                method,
                target,
                fragments: fragments.iter().map(|value| (*value).to_owned()).collect(),
                status: StatusCode::OK,
                content_type: "text/plain",
                response_body: response.to_owned(),
            }
        }
    }

    struct MockServer {
        base_url: String,
        handle: JoinHandle<()>,
    }

    impl MockServer {
        fn start(expectations: Vec<ExpectedRequest>) -> Self {
            let listener = TcpListener::bind("127.0.0.1:0").unwrap();
            let address = listener.local_addr().unwrap();
            let handle = std::thread::spawn(move || {
                for expected in expectations {
                    let (mut stream, _) = listener.accept().unwrap();
                    let request = read_http_request(&mut stream);
                    let lower_request = request.to_ascii_lowercase();
                    let request_line = request.lines().next().unwrap_or_default();
                    assert!(
                        request_line.starts_with(&format!("{} ", expected.method)),
                        "unexpected request line: {request_line}"
                    );
                    assert!(
                        request_line.contains(expected.target),
                        "request target {request_line:?} did not contain {:?}",
                        expected.target
                    );
                    for fragment in expected.fragments {
                        assert!(
                            request.contains(&fragment)
                                || lower_request.contains(&fragment.to_ascii_lowercase()),
                            "request did not contain {fragment:?}:\n{request}"
                        );
                    }
                    let response = format!(
                        "HTTP/1.1 {} {}\r\nContent-Type: {}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                        expected.status.as_u16(),
                        expected.status.canonical_reason().unwrap_or("Unknown"),
                        expected.content_type,
                        expected.response_body.len(),
                        expected.response_body
                    );
                    stream.write_all(response.as_bytes()).unwrap();
                }
            });
            Self {
                base_url: format!("http://{address}"),
                handle,
            }
        }

        fn base_url(&self) -> &str {
            &self.base_url
        }

        fn finish(self) {
            if let Err(payload) = self.handle.join() {
                panic::resume_unwind(payload);
            }
        }
    }

    fn read_http_request(stream: &mut std::net::TcpStream) -> String {
        let mut bytes = Vec::new();
        let mut buffer = [0_u8; 4096];
        let header_end = loop {
            let read = stream.read(&mut buffer).unwrap();
            assert!(read > 0, "connection closed before request headers");
            bytes.extend_from_slice(&buffer[..read]);
            if let Some(position) = bytes.windows(4).position(|window| window == b"\r\n\r\n") {
                break position + 4;
            }
        };
        let headers = String::from_utf8_lossy(&bytes[..header_end]);
        let content_length = headers
            .lines()
            .find_map(|line| {
                line.split_once(':').and_then(|(name, value)| {
                    name.eq_ignore_ascii_case("content-length")
                        .then(|| value.trim().parse::<usize>().unwrap())
                })
            })
            .unwrap_or_default();
        while bytes.len() < header_end + content_length {
            let read = stream.read(&mut buffer).unwrap();
            assert!(read > 0, "connection closed before request body");
            bytes.extend_from_slice(&buffer[..read]);
        }
        String::from_utf8(bytes).unwrap()
    }
}