redisctl 0.12.0

//! `redisctl cloud auth login | switch | status | logout`
//!
//! `login` is a credential bootstrapper: it runs an OIDC flow, exchanges the tokens with the
//! SM API to mint a CAPI key, and writes a normal cloud profile. Afterward every existing
//! `cloud` command works.
//!
//! Two shapes, matching the agent model:
//! - **Browser (loopback)** — the default on an interactive terminal: a single blocking
//!   `login` opens the browser and completes.
//! - **Device flow** — headless / `--device`: `login` *initiates* and returns the code
//!   immediately (non-blocking), so an agent can relay it; `status --wait` then blocks until
//!   the user approves, runs the SM exchange, and persists. `login --device --wait` collapses
//!   both into one blocking call for a human.

use std::io::IsTerminal;
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use redisctl_core::AuthError;
use redisctl_core::auth::{
    AccountChoice, CloudAuthenticator, LoginAccount, LoginFlow, MFA_MAX_ATTEMPTS,
    MintedCredentials, SupersededKey,
};
use redisctl_core::{CloudAuthConfig, Config, CredentialStore, DeviceAuthorization, TokenSet};
use serde::{Deserialize, Serialize};
use url::Url;

use crate::cli::CloudAuthCommands;
use crate::connection::ConnectionManager;
use crate::error::{RedisCtlError, Result as CliResult};
use crate::output::{OutputFormat, print_formatted_output};
use crate::structured_error::StructuredError;

const SCOPES: [&str; 4] = ["openid", "profile", "email", "offline_access"];

/// Above this many `redisctl-*` CAPI keys on an account, `login` nudges the user to revoke
/// unused ones (a normal user has 1–3; more suggests login/logout cycles without cleanup).
const STALE_KEY_WARN_THRESHOLD: usize = 3;

/// How many times `cloud auth switch` re-asks which account before giving up.
const ACCOUNT_PROMPT_ATTEMPTS: u32 = 3;

/// Wrap an OIDC/SM `AuthError` into the structured exit contract (stable code + exit code).
fn auth_err(e: redisctl_core::AuthError) -> RedisCtlError {
    RedisCtlError::Structured(Box::new(StructuredError::from(e)))
}

pub async fn handle_auth_command(
    conn_mgr: &ConnectionManager,
    profile: Option<&str>,
    cmd: &CloudAuthCommands,
    output: OutputFormat,
) -> CliResult<()> {
    match cmd {
        CloudAuthCommands::Login {
            device,
            allow_plaintext,
            wait,
            account,
        } => {
            login(
                conn_mgr,
                profile,
                *device,
                *allow_plaintext,
                *wait,
                *account,
                output,
            )
            .await
        }
        CloudAuthCommands::Switch { account } => switch(conn_mgr, profile, *account, output).await,
        CloudAuthCommands::Accounts => accounts(conn_mgr, profile, output).await,
        CloudAuthCommands::Status { wait, timeout } => {
            status(conn_mgr, profile, *wait, *timeout, output).await
        }
        CloudAuthCommands::Logout => logout(conn_mgr, profile, output).await,
    }
}

/// The profile a login targets: `--profile`, else the configured default cloud profile, else
/// a conventional `cloud`.
fn target_profile(conn_mgr: &ConnectionManager, profile: Option<&str>) -> String {
    profile
        .map(str::to_string)
        .or_else(|| conn_mgr.config.default_cloud.clone())
        .unwrap_or_else(|| "cloud".to_string())
}

/// Resolve the login endpoints for `profile` and build an authenticator, or a structured
/// `not_authenticated` error if the environment isn't provisioned for login.
fn prepare(
    conn_mgr: &ConnectionManager,
    profile: Option<&str>,
) -> CliResult<(String, CloudAuthenticator, CloudAuthConfig)> {
    let profile_name = target_profile(conn_mgr, profile);
    let auth_cfg = conn_mgr.config.resolve_cloud_auth(&profile_name);
    if !auth_cfg.is_complete() {
        return Err(RedisCtlError::Structured(Box::new(
            StructuredError::not_authenticated(format!(
                "cloud login endpoints are not configured for profile '{profile_name}'. Add a \
                 [cloud_auth.{profile_name}] section (okta_issuer, okta_client_id, sm_api_url) to \
                 the config, or use a profile whose environment is provisioned."
            )),
        )));
    }
    let issuer = parse_url(&auth_cfg.okta_issuer, "okta_issuer")?;
    let sm_api_url = parse_url(&auth_cfg.sm_api_url, "sm_api_url")?;
    parse_url(&auth_cfg.capi_url, "capi_url")?;
    validate_client_id(&auth_cfg.okta_client_id)?;
    let authenticator = CloudAuthenticator::new(
        issuer,
        &auth_cfg.okta_client_id,
        sm_api_url,
        &auth_cfg.capi_url,
    );
    Ok((profile_name, authenticator, auth_cfg))
}

async fn login(
    conn_mgr: &ConnectionManager,
    profile: Option<&str>,
    device: bool,
    allow_plaintext: bool,
    wait: bool,
    account: Option<u64>,
    output: OutputFormat,
) -> CliResult<()> {
    let (profile_name, authenticator, auth_cfg) = prepare(conn_mgr, profile)?;

    // Device flow is used with --device or when stderr isn't a TTY (headless / agent / piped);
    // otherwise the interactive browser (loopback) flow.
    let use_device = device || !std::io::stderr().is_terminal();

    // Non-blocking device initiate (the agent path): return the code now, complete via
    // `auth status --wait`. `--wait` (or the loopback flow) blocks through to completion.
    if use_device && !wait {
        return initiate_device_login(
            conn_mgr,
            &profile_name,
            &authenticator,
            allow_plaintext,
            account,
            output,
        )
        .await;
    }

    let superseded = superseded_key(&auth_cfg);

    let tokens = if use_device {
        run_device_flow_blocking(&authenticator).await?
    } else {
        run_loopback_flow(&authenticator).await?
    };
    let creds = complete_and_persist(
        conn_mgr,
        &profile_name,
        &authenticator,
        &tokens,
        auth_cfg,
        LoginRun {
            flow: if use_device {
                LoginFlow::Device
            } else {
                LoginFlow::Loopback
            },
            account: match account {
                Some(id) => AccountChoice::Id(id),
                // On a terminal, ask rather than refuse: core rejects a login whose session does
                // not say which of several accounts is current, and re-running with `--account`
                // means a second browser sign-in — and on an MFA account, a second code.
                // `switch` already asks this exact question with this exact picker.
                None if std::io::stdin().is_terminal() && std::io::stderr().is_terminal() => {
                    AccountChoice::Prompt(Box::new(account_for_login))
                }
                None => AccountChoice::Current,
            },
            superseded,
            allow_plaintext,
            make_default: true,
        },
    )
    .await?;
    emit_signed_in(&creds, &profile_name, output)
}

/// Which account a plain `login` mints for, asking only where the session leaves it open.
///
/// Answers exactly what `AccountChoice::Current` answers wherever it can — the account the
/// session reports, or the only one there is nothing to choose between — so the prompt appears
/// precisely where a non-interactive run refuses with `account_required`, and nowhere else.
///
/// Naming the single account when the session reports none does point the session at it, which
/// `Current` would not have. That is the repair the state needs: the alternative is minting an
/// access key from one account with a secret from whatever the session thinks it is on.
fn account_for_login(accounts: &[LoginAccount], current: Option<u64>) -> Result<u64, AuthError> {
    if let Some(id) = current
        && accounts.iter().any(|a| a.id == id)
    {
        return Ok(id);
    }
    match accounts {
        [] => Err(AuthError::Protocol(
            "this login is not associated with any Redis Cloud account".into(),
        )),
        [only] => Ok(only.id),
        several => prompt_account(several, current),
    }
}

/// The key a profile already holds, which a fresh mint for that profile replaces.
fn superseded_key(auth_cfg: &CloudAuthConfig) -> Option<SupersededKey> {
    match (auth_cfg.account_id, auth_cfg.capi_key_name.clone()) {
        (Some(account_id), Some(key_name)) => Some(SupersededKey {
            account_id,
            key_name,
        }),
        _ => None,
    }
}

/// List the accounts this sign-in can reach, so an id can be looked up without minting a key.
async fn accounts(
    conn_mgr: &ConnectionManager,
    profile: Option<&str>,
    output: OutputFormat,
) -> CliResult<()> {
    let (profile_name, authenticator, auth_cfg) = prepare(conn_mgr, profile)?;
    let on_account = auth_cfg.account_id;

    let store = CredentialStore::new();
    let refresh_token = store
        .get_credential(&format!("keyring:{profile_name}-okta-refresh"), None)
        .map_err(|_| {
            RedisCtlError::Structured(Box::new(StructuredError::not_authenticated(format!(
                "there is no stored sign-in for profile '{profile_name}' to list accounts \
                 with. Run `redisctl --profile {profile_name} cloud auth login` first; its \
                 output lists them too."
            ))))
        })?;
    let tokens = authenticator
        .refresh(&refresh_token)
        .await
        .map_err(|e| match e {
            AuthError::Network(_) | AuthError::Transport(_) => auth_err(e),
            _ => RedisCtlError::Structured(Box::new(StructuredError::not_authenticated(format!(
                "the stored sign-in for profile '{profile_name}' is no longer usable. Run \
                 `redisctl --profile {profile_name} cloud auth login`."
            )))),
        })?;

    // Okta hands back a replacement refresh token only when the app rotates them — and when it
    // does, the one just used stops working. Every other command persists what it gets back;
    // this one mints nothing, so without this the next command would reach for a dead token.
    // Failing to store it must not fail a read, but it must be said: the old token is already
    // dead and the replacement is now lost, so the profile's sign-in is gone until it signs in
    // again. Reporting `status: ok` and nothing else sends the caller to a token that cannot work.
    let mut sign_in_stored = true;
    if let Some(rotated) = tokens.refresh_token.as_deref()
        && rotated != refresh_token
        && let Err(e) = store.store_credential(&format!("{profile_name}-okta-refresh"), rotated)
    {
        sign_in_stored = false;
        eprintln!(
            "\n  warning: the identity provider rotated this profile's sign-in and the \
             replacement could not be stored ({e}). The one it replaced is no longer valid, so \
             run `redisctl --profile {profile_name} cloud auth login` before the next command."
        );
    }

    let listing = authenticator
        .list_accounts(&tokens, prompt_mfa_code)
        .await
        .map_err(auth_err)?;
    let accounts = listing.accounts;

    // What this profile's key is for, which is what the caller is choosing against. The session's
    // own account is the user's server-side default and is not the same thing.
    let profile_account = on_account.or(listing.session_account);
    match listing.email.as_deref() {
        Some(email) => eprintln!("\nAccounts {email} belongs to:"),
        None => eprintln!("\nAccounts this sign-in belongs to:"),
    }
    for a in &accounts {
        let marker = if Some(a.id) == profile_account {
            "  (this profile)"
        } else {
            ""
        };
        eprintln!("  {}{}", a.label(), marker);
    }
    if accounts.len() > 1 {
        eprintln!("\nTo use another: redisctl --profile {profile_name} cloud auth switch <id>");
    }

    print_formatted_output(
        serde_json::json!({
            "status": "ok",
            "profile": profile_name,
            "email": listing.email,
            "account_id": profile_account,
            // False when the sign-in rotated and the replacement could not be stored: the
            // listing below is good, but this profile cannot be used again without a login.
            "sign_in_stored": sign_in_stored,
            "accounts": accounts.iter().map(|a| serde_json::json!({
                "id": a.id,
                "name": a.name,
            })).collect::<Vec<_>>(),
        }),
        output,
    )
}

/// Switch the profile's key to another account, reusing the sign-in stored at login.
///
/// An API key is scoped to one account, so this mints a key for the chosen account and replaces
/// the profile's current one. The refresh token saved at login stands in for the browser; minting
/// itself cannot be avoided, because the key is per-account.
async fn switch(
    conn_mgr: &ConnectionManager,
    profile: Option<&str>,
    account: Option<u64>,
    output: OutputFormat,
) -> CliResult<()> {
    let (profile_name, authenticator, auth_cfg) = prepare(conn_mgr, profile)?;

    // Which account this profile is on today, and which key it holds, as recorded at the last
    // login/switch. Read before `auth_cfg` is consumed below.
    let on_account = auth_cfg.account_id;
    let superseded = superseded_key(&auth_cfg);

    // Already there: minting another key for the same account would just add to the sprawl.
    if let Some(want) = account
        && on_account == Some(want)
    {
        eprintln!("\nProfile '{profile_name}' already uses account #{want}. Nothing to do.");
        return print_formatted_output(
            serde_json::json!({
                "status": "ok",
                "profile": profile_name,
                // Numeric, like the mint path and like `accounts[].id`, so a caller can compare
                // them without caring which path produced the output.
                "account_id": want,
                "changed": false,
            }),
            output,
        );
    }

    // Choosing from a list needs somewhere to ask. Checked before signing in, so a
    // non-interactive caller fails immediately rather than after doing the work.
    let interactive = std::io::stderr().is_terminal() && std::io::stdin().is_terminal();
    if account.is_none() && !interactive {
        return Err(RedisCtlError::Structured(Box::new(
            StructuredError::account_required(format!(
                "there is no terminal to choose an account on; pass the id instead \
                 (`redisctl --profile {profile_name} cloud auth switch <ID>`). A completed login \
                 reports the accounts you belong to in its `accounts` field."
            )),
        )));
    }

    let store = CredentialStore::new();
    let refresh_token = store
        .get_credential(&format!("keyring:{profile_name}-okta-refresh"), None)
        .map_err(|_| {
            RedisCtlError::Structured(Box::new(StructuredError::not_authenticated(format!(
                "there is no stored sign-in for profile '{profile_name}' to switch with \
                 (credentials saved with --allow-plaintext cannot be reused this way). Run \
                 `redisctl --profile {profile_name} cloud auth login --account <ID>` instead."
            ))))
        })?;

    // A refresh response that omits `refresh_token` leaves the stored one in place, which is
    // correct: the identity provider only omits it when the existing token stays valid. With
    // rotation on it always returns a new one, which replaces the stored value on persist.
    //
    // The browser is only ever needed to obtain a refresh token in the first place. Transport
    // failures are relayed as-is: they are retryable, and calling them an expired sign-in would
    // send the caller through a full browser login for a blip.
    let tokens = authenticator
        .refresh(&refresh_token)
        .await
        .map_err(|e| match e {
            AuthError::Network(_) | AuthError::Transport(_) => auth_err(e),
            _ => RedisCtlError::Structured(Box::new(StructuredError::not_authenticated(format!(
                "the stored sign-in for profile '{profile_name}' is no longer usable — refresh \
                 tokens expire and are rotated. Run `redisctl --profile {profile_name} cloud auth \
                 login --account <ID>`."
            )))),
        })?;

    let creds = complete_and_persist(
        conn_mgr,
        &profile_name,
        &authenticator,
        &tokens,
        auth_cfg,
        LoginRun {
            // Not loopback: no browser is involved, and `utm_campaign` should say so.
            flow: LoginFlow::Switch,
            account: match account {
                Some(id) => AccountChoice::Id(id),
                // `on_account` comes from the profile, not the session: `setcurrent` is
                // session-scoped server-side, so a fresh sign-in reports the user's default
                // account and would mark the wrong row as current.
                None => AccountChoice::Prompt(Box::new(move |accounts, _session_account| {
                    prompt_account(accounts, on_account)
                })),
            },
            superseded,
            // A refresh token only exists on the keyring path, so this is never plaintext.
            allow_plaintext: false,
            // Changing an existing profile's account is not a reason to make it the default.
            make_default: false,
        },
    )
    .await?;

    // Any device authorization still awaiting approval for this profile is now stale: completing
    // it later would overwrite the key we just minted with one for the account it recorded.
    clear_pending(conn_mgr, &profile_name);

    eprintln!(
        "\n\u{2713} Profile '{profile_name}' now uses {}.",
        creds.account_label()
    );
    match creds.superseded_revoked {
        Some(true) => eprintln!("  the key it replaced has been revoked."),
        Some(false) => eprintln!(
            "  note: could not revoke {} — revoke it in the Redis Cloud console \
             (Access Management > API Keys).",
            superseded_label(&creds)
        ),
        None => {}
    }
    warn_on_key_sprawl(&creds);
    print_formatted_output(
        serde_json::json!({
            "status": "ok",
            "profile": profile_name,
            "account_id": creds.account_id,
            "account_name": creds.account_name,
            "accounts": creds.accounts.iter().map(|a| serde_json::json!({
                "id": a.id,
                "name": a.name,
            })).collect::<Vec<_>>(),
            "email": creds.email,
            "redisctl_key_count": creds.redisctl_key_count,
            "superseded_revoked": creds.superseded_revoked,
            "superseded_key": creds.superseded_key_name,
            "changed": true,
        }),
        output,
    )
}

/// D5: every login *and* every switch mints a new `redisctl-*` CAPI key. Warn (don't delete)
/// when they pile up — switching makes that happen faster than logging in does.
/// Names the key a failed revocation left behind. Without the name there is nothing to search
/// for in the console, and it cannot be recovered afterwards — the profile records the new key.
fn superseded_label(creds: &MintedCredentials) -> String {
    match &creds.superseded_key_name {
        Some(key) => format!("the key {key} that this replaced"),
        None => "the key this replaced".to_string(),
    }
}

fn warn_on_key_sprawl(creds: &MintedCredentials) {
    let key_count = creds.redisctl_key_count;
    if key_count > STALE_KEY_WARN_THRESHOLD {
        eprintln!(
            "  note: this account now has {key_count} redisctl-* API keys. Revoke unused ones \
             in the Redis Cloud console (Access Management > API Keys)."
        );
    }
}

/// Ask which account to switch to, listing them with the current one marked.
///
/// Only reached on a terminal (the caller checks). Re-asks on a bad answer rather than aborting:
/// giving up here would unwind the whole command, and re-running means another sign-in — and on an
/// MFA account, another code. Returning an error abandons the switch before anything is minted.
fn prompt_account(accounts: &[LoginAccount], current: Option<u64>) -> Result<u64, AuthError> {
    if accounts.len() < 2 {
        return Err(AuthError::AccountRequired(
            "this login belongs to a single Redis Cloud account, so there is nothing to switch to"
                .into(),
        ));
    }
    eprintln!("\nAccounts you belong to:");
    for (i, a) in accounts.iter().enumerate() {
        let marker = if Some(a.id) == current {
            "  (current)"
        } else {
            ""
        };
        eprintln!("  {}) {}{}", i + 1, a.label(), marker);
    }
    if current.is_none() {
        eprintln!("  (which one this profile is on is unknown — sign in again to record it)");
    }

    for attempt in 1..=ACCOUNT_PROMPT_ATTEMPTS {
        eprint!("Switch to which? [1-{}]: ", accounts.len());
        use std::io::Write;
        let _ = std::io::stderr().flush();

        let mut line = String::new();
        let read = std::io::stdin()
            .read_line(&mut line)
            .map_err(|e| AuthError::Protocol(format!("could not read a choice: {e}")))?;
        if read == 0 {
            // EOF (Ctrl-D): the user is declining, not failing.
            return Err(AuthError::AccountRequired(
                "no account was chosen, so nothing was switched".into(),
            ));
        }
        match resolve_account_choice(accounts, &line) {
            Ok(id) => return Ok(id),
            Err(e) if attempt < ACCOUNT_PROMPT_ATTEMPTS => eprintln!("  {e}"),
            Err(e) => return Err(e),
        }
    }
    unreachable!("loop returns on the final attempt")
}

/// Map what the user typed onto an account id.
///
/// A list position wins over an account id when the input could be either: the prompt asks for
/// `[1-N]`, so that is what a bare small number means. Ids can be small in non-production
/// environments, and resolving `2` to *account* 2 there would mint a key for the wrong account.
/// Split out from the prompt so the rules are testable without a terminal.
fn resolve_account_choice(accounts: &[LoginAccount], input: &str) -> Result<u64, AuthError> {
    let typed = input.trim();
    if let Ok(position) = typed.parse::<usize>()
        && let Some(a) = accounts.get(position.wrapping_sub(1))
    {
        return Ok(a.id);
    }
    // Not a position — an id copied out of the listing is the other sensible thing to type.
    if let Ok(id) = typed.parse::<u64>()
        && let Some(a) = accounts.iter().find(|a| a.id == id)
    {
        return Ok(a.id);
    }
    Err(AuthError::AccountRequired(format!(
        "'{typed}' is not one of 1-{} or an account id from the list",
        accounts.len()
    )))
}

/// Sign-in instructions for the device flow. `verification_uri_complete` pre-fills the code on the
/// activation page, so the user confirms it rather than typing it — RFC 8628 §3.3.1 still requires
/// that explicit confirmation, so the step itself can't be skipped.
fn device_instructions(authz: &DeviceAuthorization) -> String {
    match authz.verification_uri_complete() {
        Some(complete) => format!(
            "\nTo sign in, open:\n  {complete}\nand confirm the code:  {}  (already filled in)\n",
            authz.user_code(),
        ),
        None => format!(
            "\nTo sign in, open:\n  {}\nand enter the code:  {}\n",
            authz.verification_uri(),
            authz.user_code(),
        ),
    }
}

/// Start the device flow, save a pending record, and return the device code to the caller
/// (agent) without blocking. Completion happens later via `auth status --wait`.
async fn initiate_device_login(
    conn_mgr: &ConnectionManager,
    profile_name: &str,
    authenticator: &CloudAuthenticator,
    allow_plaintext: bool,
    account: Option<u64>,
    output: OutputFormat,
) -> CliResult<()> {
    let authz = authenticator
        .device()
        .start(&SCOPES)
        .await
        .map_err(auth_err)?;
    save_pending(
        conn_mgr,
        &PendingAuth {
            profile: profile_name.to_string(),
            device_authorization: authz.clone(),
            allow_plaintext,
            account,
        },
    )?;

    eprintln!(
        "{}Then run: redisctl cloud auth status --wait   (or wait for the agent to poll)",
        device_instructions(&authz),
    );
    print_formatted_output(
        serde_json::json!({
            "status": "authorization_pending",
            "profile": profile_name,
            "verification_uri": authz.verification_uri(),
            "verification_uri_complete": authz.verification_uri_complete(),
            "user_code": authz.user_code(),
            "expires_in": authz.expires_in(),
            "interval": authz.interval(),
        }),
        output,
    )
}

async fn run_device_flow_blocking(auth: &CloudAuthenticator) -> CliResult<TokenSet> {
    let client = auth.device();
    let authz = client.start(&SCOPES).await.map_err(auth_err)?;
    eprintln!("{}(waiting for approval…)", device_instructions(&authz),);

    // oauth2 owns the poll loop (honoring the server's interval / slow_down); `None` polls until
    // the device code's own lifetime expires, which surfaces as `device_code_expired` via auth_err.
    client.poll(&authz, None).await.map_err(auth_err)
}

async fn run_loopback_flow(auth: &CloudAuthenticator) -> CliResult<TokenSet> {
    let tokens = auth
        .loopback()
        .login(&SCOPES, |url| {
            eprintln!("Opening your browser to sign in…\n  {url}");
            let _ = open_browser(url);
        })
        .await
        .map_err(auth_err)?;
    Ok(tokens)
}

/// Ask the user for a TOTP code when SM challenges the login for MFA.
///
/// Returns `Ok(None)` when there is no terminal to prompt on — an agent can't produce a
/// time-based code, so the caller reports `mfa_required` and the human re-runs interactively.
/// The `^\d{6}$` check is local on purpose: a malformed value would otherwise consume one of the
/// user's server-side attempts.
/// Submissions still available, counting the one about to be entered: the prompt runs before
/// the code for `attempt` is sent.
fn attempts_left(attempt: u32) -> u32 {
    MFA_MAX_ATTEMPTS.saturating_add(1).saturating_sub(attempt)
}

fn prompt_mfa_code(factors: &[String], attempt: u32) -> Result<Option<String>, AuthError> {
    if !std::io::stdin().is_terminal() || !std::io::stderr().is_terminal() {
        return Ok(None);
    }
    if attempt == 1 {
        let detail = if factors.is_empty() {
            String::new()
        } else {
            format!(" ({})", factors.join(", "))
        };
        eprintln!("\nThis account requires multi-factor authentication{detail}.");
    } else {
        eprintln!(
            "That code wasn't accepted. {} attempt(s) left.",
            attempts_left(attempt)
        );
    }
    loop {
        eprint!("Enter the 6-digit code from your authenticator app: ");
        let _ = std::io::Write::flush(&mut std::io::stderr());
        let mut line = String::new();
        if std::io::stdin().read_line(&mut line).unwrap_or(0) == 0 {
            return Ok(None); // EOF / Ctrl-D
        }
        let code = line.trim();
        if code.len() == 6 && code.chars().all(|c| c.is_ascii_digit()) {
            return Ok(Some(code.to_string()));
        }
        eprintln!("Codes are exactly 6 digits.");
    }
}

/// What one login run decided, threaded to the persist step together rather than as loose
/// positional arguments.
struct LoginRun {
    flow: LoginFlow,
    /// Which account to mint for: the session's current one, an explicit id, or a picker.
    account: AccountChoice,
    /// The key this run replaces, revoked once its successor exists.
    superseded: Option<SupersededKey>,
    allow_plaintext: bool,
    /// Whether to make this the default cloud profile. True when logging in (it bootstraps the
    /// profile), false when only changing which account an existing profile targets.
    make_default: bool,
}

/// Run the SM exchange, persist the profile + secrets, and return the minted credentials.
async fn complete_and_persist(
    conn_mgr: &ConnectionManager,
    profile_name: &str,
    authenticator: &CloudAuthenticator,
    tokens: &TokenSet,
    auth_cfg: CloudAuthConfig,
    run: LoginRun,
) -> CliResult<MintedCredentials> {
    // Resolved before anything is minted: on a machine with no keyring this refuses, and
    // refusing after a mint would burn a key whose secret is only returned at creation.
    let store = if run.allow_plaintext {
        CredentialStore::plaintext()
    } else {
        let store = CredentialStore::new();
        if store.storage_backend() != "keyring" {
            return Err(RedisCtlError::Structured(Box::new(
                StructuredError::keyring_unavailable(
                    "no OS keyring is available to store the credentials, and storing them in \
                     the config file has to be asked for. Re-run with `--allow-plaintext` to \
                     store them there (0600) instead.",
                ),
            )));
        }
        // Selecting the keyring does not establish that it can hold anything: on the Secret
        // Service backend the connection is made at get/set time, so a locked or absent keyring
        // still hands back an entry. Ask it to keep a throwaway value before minting a key.
        store.probe_writable().map_err(|e| {
            RedisCtlError::Structured(Box::new(StructuredError::keyring_unavailable(format!(
                "the OS keyring cannot store credentials ({e}). Re-run with \
                 `--allow-plaintext` to store them in the config file (0600) instead."
            ))))
        })?;
        store
    };

    let (mut creds, revoker) = authenticator
        .complete_login_with_mfa(
            tokens,
            &default_key_name(),
            run.flow,
            run.account,
            run.superseded,
            prompt_mfa_code,
        )
        .await
        .map_err(auth_err)?;

    let mut config = conn_mgr.config.clone();
    // On the keyring path, a store failure means the OS secret service is unavailable (D4):
    // surface a distinct `keyring_unavailable` (exit 2) pointing at `--allow-plaintext`.
    config
        .apply_cloud_login(
            &store,
            profile_name,
            &creds,
            Some(auth_cfg),
            run.make_default,
        )
        .map_err(|e| {
            // The key exists server-side by now and its secret is only returned at creation, so
            // nothing can recover it — say which one was left behind while the name is still here.
            let orphan = format!(
                " The key {} was created but not stored; revoke it in the Redis Cloud console \
                 (Access Management > API Keys).",
                creds.capi_key_name
            );
            if run.allow_plaintext {
                RedisCtlError::Structured(Box::new(StructuredError::keyring_unavailable(format!(
                    "failed to store credentials ({e}).{orphan}"
                ))))
            } else {
                RedisCtlError::Structured(Box::new(StructuredError::keyring_unavailable(format!(
                    "failed to store credentials in the OS keyring ({e}). Re-run \
                     `redisctl cloud auth login --allow-plaintext` to store them in the config \
                     file (0600) instead.{orphan}"
                ))))
            }
        })?;
    save_config_for_store(conn_mgr, &config, &store).map_err(|e| match e {
        RedisCtlError::Structured(_) => e,
        other => RedisCtlError::Configuration(format!(
            "{other}. The key {} was created but not stored; revoke it in the Redis Cloud \
             console (Access Management > API Keys).",
            creds.capi_key_name
        )),
    })?;

    // Only now: the replacement is stored, so revoking what it replaced cannot leave this
    // profile without a working key.
    if let Some(revoker) = revoker {
        let same_account = Some(revoker.account_id()) == creds.account_id;
        creds.superseded_key_name = Some(revoker.key_name().to_string());
        let revoked = revoker.revoke().await;
        creds.superseded_revoked = Some(revoked);
        // The count was taken while minting, before this revoke, so a key just removed from the
        // same account is still included in it.
        if revoked && same_account {
            creds.redisctl_key_count = creds.redisctl_key_count.saturating_sub(1);
        }
    }
    Ok(creds)
}

fn emit_signed_in(
    creds: &MintedCredentials,
    profile_name: &str,
    output: OutputFormat,
) -> CliResult<()> {
    eprintln!(
        "\n\u{2713} Signed in as {}. Credentials saved to profile '{}'.",
        creds.email.as_deref().unwrap_or("your account"),
        profile_name
    );
    // The key is scoped to one account — whichever is *current* for this user, decided server-side
    // from the session. Say which was used when there was more than one it could have been.
    if creds.account_count() > 1 {
        // Render through `label()` like the list below, so one account is never spelled two
        // different ways two lines apart. `account_id` is always one of `accounts` — both come
        // from the same `/accounts` response — so the lookup only misses for a hand-built value.
        let which = creds
            .account_id
            .and_then(|id| creds.accounts.iter().find(|a| a.id == id))
            .map(LoginAccount::label)
            .unwrap_or_else(|| "your current account".to_string());
        eprintln!(
            "  note: the key is for {which} — 1 of {} accounts you belong to:",
            creds.account_count()
        );
        eprintln!(
            "    {}",
            creds
                .accounts
                .iter()
                .map(LoginAccount::label)
                .collect::<Vec<_>>()
                .join(" · ")
        );
        // Name the profile actually in use: a `<name>` placeholder invites inventing a new one,
        // and an unconfigured profile silently resolves to the *production* endpoints.
        eprintln!("  To use another: redisctl --profile {profile_name} cloud auth switch <id>");
    }
    if creds.capi_newly_enabled {
        eprintln!(
            "  note: programmatic (API) access was switched on for this account — it was off \
             until now, and this applies account-wide, not just to this key."
        );
    }
    if creds.superseded_revoked == Some(false) {
        eprintln!(
            "  note: could not revoke {} — revoke it in the Redis Cloud console \
             (Access Management > API Keys).",
            superseded_label(creds)
        );
    }
    warn_on_key_sprawl(creds);
    let key_count = creds.redisctl_key_count;
    print_formatted_output(
        serde_json::json!({
            "status": "ok",
            "authenticated": true,
            "profile": profile_name,
            "account_id": creds.account_id,
            "account_name": creds.account_name,
            "account_count": creds.account_count(),
            "accounts": creds.accounts.iter().map(|a| serde_json::json!({
                "id": a.id,
                "name": a.name,
            })).collect::<Vec<_>>(),
            "email": creds.email,
            "redisctl_key_count": key_count,
            "capi_newly_enabled": creds.capi_newly_enabled,
            "superseded_revoked": creds.superseded_revoked,
            "superseded_key": creds.superseded_key_name,
        }),
        output,
    )
}

async fn status(
    conn_mgr: &ConnectionManager,
    profile: Option<&str>,
    wait: bool,
    timeout: u64,
    output: OutputFormat,
) -> CliResult<()> {
    let profile_name = target_profile(conn_mgr, profile);

    // --wait completes a pending device-authorization login (the agent path): poll, exchange,
    // persist. With no pending record it falls through to a plain status report.
    if wait && let Some(pending) = load_pending(conn_mgr, &profile_name) {
        let (_, authenticator, auth_cfg) = prepare(conn_mgr, Some(&profile_name))?;
        match poll_pending(&authenticator, &pending, timeout).await? {
            Some(tokens) => {
                let creds = complete_and_persist(
                    conn_mgr,
                    &profile_name,
                    &authenticator,
                    &tokens,
                    auth_cfg.clone(),
                    LoginRun {
                        flow: LoginFlow::Device,
                        // Whatever the initiating `login --device --account` asked for; `None`
                        // (no flag) keeps the session's current account.
                        account: match pending.account {
                            Some(id) => AccountChoice::Id(id),
                            None => AccountChoice::Current,
                        },
                        superseded: superseded_key(&auth_cfg),
                        allow_plaintext: pending.allow_plaintext,
                        make_default: true,
                    },
                )
                .await?;
                clear_pending(conn_mgr, &profile_name);
                return emit_signed_in(&creds, &profile_name, output);
            }
            None => {
                // Timed out but the code may still be valid — keep the pending record so a
                // later `status --wait` can resume; report the pending state (exit 0).
                return print_formatted_output(
                    serde_json::json!({
                        "status": "authorization_pending",
                        "authenticated": false,
                        "profile": profile_name,
                    }),
                    output,
                );
            }
        }
    }

    let authenticated = conn_mgr
        .resolve_cloud_connection(Some(&profile_name))
        .is_ok();
    // A pending device login is invisible in a bare `authenticated: false`, which reads as "login
    // failed" when in fact it is half-finished. Say so, and name the command that completes it.
    let pending = !authenticated && load_pending(conn_mgr, &profile_name).is_some();
    if pending {
        eprintln!(
            "A device login for profile '{profile_name}' is waiting for approval.\nComplete it \
             with: redisctl cloud auth status --wait"
        );
    }
    let mut report = serde_json::json!({ "authenticated": authenticated, "profile": profile_name });
    if pending {
        report["status"] = "authorization_pending".into();
    }
    print_formatted_output(report, output)
}

/// Poll a pending device authorization to completion. `Ok(Some)` = approved (tokens),
/// `Ok(None)` = local `--timeout` elapsed while still pending. `expired`/`denied` from the
/// server propagate as structured errors.
async fn poll_pending(
    auth: &CloudAuthenticator,
    pending: &PendingAuth,
    timeout_secs: u64,
) -> CliResult<Option<TokenSet>> {
    let client = auth.device();
    // oauth2 owns the poll loop; bound it by the caller's `--timeout`. If our wait elapses first
    // the login is still pending (`Ok(None)`); a real expiry/denial from the server surfaces as a
    // structured error. This keeps "your wait elapsed" distinct from "the device code expired".
    match tokio::time::timeout(
        Duration::from_secs(timeout_secs),
        client.poll(&pending.device_authorization, None),
    )
    .await
    {
        Err(_elapsed) => Ok(None),
        Ok(result) => result.map(Some).map_err(auth_err),
    }
}

async fn logout(
    conn_mgr: &ConnectionManager,
    profile: Option<&str>,
    output: OutputFormat,
) -> CliResult<()> {
    let profile_name = target_profile(conn_mgr, profile);
    let store = CredentialStore::new();

    // Revoke server-side first, while the credentials to do it with still exist. Local removal
    // happens either way: a logout that left the secrets on disk because the network was down
    // would be worse than one that leaves a key to revoke in the console.
    let revoked = revoke_remotely(conn_mgr, &profile_name, &store).await;

    for suffix in ["cloud-api-key", "cloud-api-secret", "okta-refresh"] {
        let _ = store.delete_credential(&format!("{profile_name}-{suffix}"));
    }
    clear_pending(conn_mgr, &profile_name);
    let mut config = conn_mgr.config.clone();
    // Logout forgets credentials, not the environment: preserve the [cloud_auth.<profile>]
    // login endpoints across the profile removal so `auth login` still works afterward
    // (remove_profile drops them otherwise, which would break re-login on QA/staging).
    let saved_auth = config.cloud_auth.get(&profile_name).cloned();
    config.remove_profile(&profile_name);
    if let Some(mut auth) = saved_auth {
        // Endpoints are worth keeping; which key this profile held is not. Leaving it would make
        // the next login report the key just deleted here as one it could not revoke.
        auth.account_id = None;
        auth.capi_key_name = None;
        config.cloud_auth.insert(profile_name.clone(), auth);
    }
    save_config(conn_mgr, &config)?;

    match (&revoked.key, &revoked.session) {
        (Ok(key), Ok(())) => eprintln!(
            "\n\u{2713} Logged out of profile '{profile_name}'. Revoked the API key {key} and \
             the stored sign-in."
        ),
        (Ok(key), Err(_)) => eprintln!(
            "\n\u{2713} Logged out of profile '{profile_name}'. Revoked the API key {key}."
        ),
        (Err(_), Ok(())) => eprintln!(
            "\n\u{2713} Logged out of profile '{profile_name}'. Revoked the stored sign-in."
        ),
        (Err(_), Err(_)) => {
            eprintln!("\n\u{2713} Logged out of profile '{profile_name}' locally.")
        }
    }
    for note in revoked.notes() {
        eprintln!("  note: {note}");
    }
    print_formatted_output(
        serde_json::json!({
            "status": "ok",
            "profile": profile_name,
            "logged_out": true,
            "revoked": revoked.key.is_ok() && revoked.session.is_ok(),
            "key_revoked": revoked.key.is_ok(),
            "session_revoked": revoked.session.is_ok(),
        }),
        output,
    )
}

/// What logout managed to revoke server-side. The two are independent on purpose: a key that
/// cannot be deleted must not stop the sign-in being revoked, because a live refresh token can
/// mint another key.
struct Revocation {
    /// The name of the key that was revoked, or why it wasn't.
    key: Result<String, String>,
    /// Whether the stored sign-in was revoked at the identity provider, or why it wasn't.
    session: Result<(), String>,
}

impl Revocation {
    /// Both halves failing for the same reason — no sign-in to act with, or no endpoints to
    /// reach.
    fn blocked(key_note: String, session_note: String) -> Self {
        Self {
            key: Err(key_note),
            session: Err(session_note),
        }
    }

    fn notes(&self) -> Vec<String> {
        let mut notes = Vec::new();
        if let Err(why) = &self.key {
            notes.push(format!(
                "{why} Revoke the key in the Redis Cloud console (Access Management > API Keys)."
            ));
        }
        if let Err(why) = &self.session {
            notes.push(format!(
                "{why} It stays valid at the identity provider until it expires."
            ));
        }
        notes
    }
}

/// Why logout could not revoke anything, when the refresh that would have authorised it failed.
///
/// A request that never arrived says nothing about the sign-in. Calling it invalid reports that
/// there was nothing to revoke, while logout goes on to delete the local credentials — leaving a
/// live refresh token, which can mint another key, and the profile's key itself alive server-side
/// with nothing left naming them.
fn refresh_failure(e: &AuthError, no_key_because: impl Fn(&str) -> String) -> Revocation {
    let (why, session) = match e {
        AuthError::Network(_) | AuthError::Transport(_) => (
            "the identity provider could not be reached",
            "so the stored sign-in was not revoked.",
        ),
        _ => (
            "the stored sign-in is no longer valid",
            "so there was nothing to revoke.",
        ),
    };
    Revocation::blocked(no_key_because(why), format!("{why}, {session}"))
}

/// Revoke the minted key and the stored sign-in, or explain why it could not be done.
///
/// Every failure is reported rather than raised: logout has to finish locally regardless.
async fn revoke_remotely(
    conn_mgr: &ConnectionManager,
    profile_name: &str,
    store: &CredentialStore,
) -> Revocation {
    let auth_cfg = conn_mgr.config.resolve_cloud_auth(profile_name);
    let recorded_key = auth_cfg.capi_key_name.clone();
    // Phrases the key half of a shared failure; whether a key is even known changes the wording.
    let no_key_because = |why: &str| match &recorded_key {
        Some(key) => format!("{why}, so the key {key} could not be revoked."),
        None => format!("{why}, and this profile does not record a key to revoke by name."),
    };

    let Ok(refresh_token) =
        store.get_credential(&format!("keyring:{profile_name}-okta-refresh"), None)
    else {
        let why = format!("there is no stored sign-in for '{profile_name}'");
        return Revocation::blocked(no_key_because(&why), format!("{why}, so none was revoked."));
    };
    let (_, authenticator, _) = match prepare(conn_mgr, Some(profile_name)) {
        Ok(parts) => parts,
        Err(_) => {
            let why = "login endpoints are not configured";
            return Revocation::blocked(
                no_key_because(why),
                format!("{why}, so the stored sign-in could not be revoked."),
            );
        }
    };
    let tokens = match authenticator.refresh(&refresh_token).await {
        Ok(tokens) => tokens,
        Err(e) => return refresh_failure(&e, no_key_because),
    };

    // Independent from here. The key goes first because deleting it needs the sign-in, but its
    // outcome must not gate the sign-in's: a refresh token left alive can mint another key,
    // which is the risk this is here to close.
    let key = match &recorded_key {
        None => Err(
            "this profile does not record which API key it holds, so there was nothing to \
             revoke by name."
                .to_string(),
        ),
        Some(key_name) => {
            // The account the profile's key was minted for. A sign-in starts on the user's
            // default account, so without this a key minted on any other one is not even visible.
            let on_account = auth_cfg.account_id;
            match authenticator
                .revoke_capi_key(&tokens, on_account, key_name)
                .await
            {
                Ok(true) => Ok(key_name.clone()),
                Ok(false) => Err(match on_account {
                    Some(account) => format!(
                        "the key {key_name} was not found on account {account}; it may already \
                         be revoked."
                    ),
                    None => format!(
                        "the key {key_name} was not found on the account this sign-in defaults \
                         to, and this profile does not record which account it belongs to."
                    ),
                }),
                Err(e) => Err(format!("could not revoke the key {key_name}: {e}.")),
            }
        }
    };
    // The refresh above rotated the token, so revoke the one we now hold; the previous value is
    // already invalid at the IdP.
    let newest = tokens.refresh_token.as_deref().unwrap_or(&refresh_token);
    let session = authenticator
        .revoke_refresh_token(newest)
        .await
        .map_err(|e| format!("could not revoke the stored sign-in: {e}."));
    Revocation { key, session }
}

// ---- pending device-authorization store (bridges the non-blocking login and status --wait) ----

/// A device authorization awaiting approval, persisted between the `login` (initiate) and
/// `status --wait` (complete) process runs.
#[derive(Serialize, Deserialize)]
struct PendingAuth {
    profile: String,
    /// `--account` from the initiating `login`, so the account it asked for survives into the
    /// `status --wait` that actually mints. Defaulted for pending files written before this field.
    #[serde(default)]
    account: Option<u64>,
    /// The full device authorization (serde-serializable), so a later `status --wait` — possibly
    /// a different process — can resume polling via `DeviceFlowClient::poll`.
    device_authorization: DeviceAuthorization,
    allow_plaintext: bool,
}

/// Directory for the pending file — next to the active config so it honors `--config-file`.
fn pending_dir(conn_mgr: &ConnectionManager) -> PathBuf {
    conn_mgr
        .config_path
        .as_ref()
        .and_then(|p| p.parent().map(Path::to_path_buf))
        .or_else(|| {
            Config::config_path()
                .ok()
                .and_then(|p| p.parent().map(Path::to_path_buf))
        })
        .unwrap_or_else(std::env::temp_dir)
}

fn pending_path(conn_mgr: &ConnectionManager, profile: &str) -> PathBuf {
    pending_dir(conn_mgr).join(format!("redisctl-pending-{profile}.json"))
}

fn save_pending(conn_mgr: &ConnectionManager, pending: &PendingAuth) -> CliResult<()> {
    let path = pending_path(conn_mgr, &pending.profile);
    let json = serde_json::to_vec_pretty(pending)?;
    write_private(&path, &json)
        .map_err(|e| RedisCtlError::Configuration(format!("could not save pending login: {e}")))
}

fn load_pending(conn_mgr: &ConnectionManager, profile: &str) -> Option<PendingAuth> {
    let bytes = std::fs::read(pending_path(conn_mgr, profile)).ok()?;
    serde_json::from_slice(&bytes).ok()
}

fn clear_pending(conn_mgr: &ConnectionManager, profile: &str) {
    let _ = std::fs::remove_file(pending_path(conn_mgr, profile));
}

#[cfg(unix)]
fn write_private(path: &std::path::Path, bytes: &[u8]) -> std::io::Result<()> {
    use std::io::Write as _;
    use std::os::unix::fs::OpenOptionsExt;
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    let mut f = std::fs::OpenOptions::new()
        .write(true)
        .create(true)
        .truncate(true)
        .mode(0o600)
        .open(path)?;
    f.write_all(bytes)
}

#[cfg(not(unix))]
fn write_private(path: &std::path::Path, bytes: &[u8]) -> std::io::Result<()> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)?;
    }
    std::fs::write(path, bytes)
}

/// The client id goes into the authorize URL that gets handed to a browser launcher, so keep it to
/// the shape Okta actually issues rather than passing arbitrary text through.
fn validate_client_id(client_id: &str) -> CliResult<()> {
    if !client_id.is_empty()
        && client_id
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_')
    {
        return Ok(());
    }
    Err(RedisCtlError::Configuration(format!(
        "invalid okta_client_id ({client_id:?}): expected letters, digits, '-' or '_'"
    )))
}

/// Parse a configured endpoint, requiring transport security.
///
/// A config file is a credential-bearing input: one that names the real identity provider but
/// points `sm_api_url` somewhere else receives a live access token after an otherwise ordinary
/// login. `http` is allowed only on loopback, where there is no network to intercept, which is
/// also what RFC 8252 does for redirect URIs.
/// Domains whose hosts may receive a Redis Cloud access token. Suffix-matched on a label
/// boundary, so `notredis.com` does not pass for `redis.com`.
const TRUSTED_ENDPOINT_SUFFIXES: [&str; 3] = ["redis.com", "redislabs.com", "redis.io"];

/// Opt out of the endpoint check for a host that is not a Redis domain — a local mock or a test
/// environment. Deliberately an environment variable and not a config field: the threat is a
/// config file that names an attacker's host, and a setting inside that file could switch off
/// the check that exists to catch it.
const TRUST_OVERRIDE_ENV: &str = "REDISCTL_ALLOW_UNTRUSTED_ENDPOINTS";

fn endpoint_trust_disabled() -> bool {
    trust_override_requested(std::env::var(TRUST_OVERRIDE_ENV).ok().as_deref())
}

fn trust_override_requested(value: Option<&str>) -> bool {
    value.is_some_and(|v| v == "1" || v.eq_ignore_ascii_case("true"))
}

fn is_trusted_endpoint_host(host: &url::Host<&str>) -> bool {
    let url::Host::Domain(name) = host else {
        // A bare IP is never a Redis endpoint; loopback is handled by its own exception.
        return false;
    };
    let name = name.trim_end_matches('.').to_ascii_lowercase();
    TRUSTED_ENDPOINT_SUFFIXES
        .iter()
        .any(|suffix| name == *suffix || name.ends_with(&format!(".{suffix}")))
}

fn host_display(host: &url::Host<&str>) -> String {
    match host {
        url::Host::Domain(name) => (*name).to_string(),
        url::Host::Ipv4(ip) => ip.to_string(),
        url::Host::Ipv6(ip) => format!("[{ip}]"),
    }
}

fn parse_url(value: &str, field: &str) -> CliResult<Url> {
    parse_endpoint(value, field, endpoint_trust_disabled())
}

/// Takes the override as an argument rather than reading it, so a test can exercise both sides
/// without moving a process-wide variable: `cargo test` runs tests on several threads in one
/// process, so one that sets a variable changes what its neighbours see.
fn parse_endpoint(value: &str, field: &str, trust_disabled: bool) -> CliResult<Url> {
    // Structured rather than `RedisCtlError::Configuration`, which prints to stderr and exits 3
    // — the code this surface's contract defines as retryable. A config file is not retryable.
    let reject = |why: &str| {
        RedisCtlError::Structured(Box::new(StructuredError::invalid_endpoint(format!(
            "invalid {field} ({value:?}): {why}"
        ))))
    };
    let url = Url::parse(value).map_err(|e| reject(&e.to_string()))?;
    let Some(host) = url.host() else {
        return Err(reject("no host"));
    };
    if matches!(host, url::Host::Domain("")) {
        return Err(reject("no host"));
    }
    // Matched on the parsed host: `host_str` brackets an IPv6 literal, so `"::1"` never compares
    // equal to it.
    let loopback = match host {
        url::Host::Domain(name) => name == "localhost",
        url::Host::Ipv4(ip) => ip.is_loopback(),
        url::Host::Ipv6(ip) => ip.is_loopback(),
    };
    if url.scheme() != "https" && !(url.scheme() == "http" && loopback) {
        return Err(reject(
            "must use https (http is allowed only for localhost)",
        ));
    }
    // Transport security says the connection is private, not who is on the other end. A config
    // file names where an access token is sent, so the host has to be one we trust to receive
    // it — an attacker-controlled HTTPS host is still attacker-controlled.
    if !loopback && !is_trusted_endpoint_host(&host) {
        if !trust_disabled {
            return Err(reject(&format!(
                "host is not a Redis endpoint ({}). Trusted: {}, or loopback. Set \
                 {TRUST_OVERRIDE_ENV}=1 to use another host",
                host_display(&host),
                TRUSTED_ENDPOINT_SUFFIXES.join(", "),
            )));
        }
        // Straight to stderr, not through tracing: this records that a security check was
        // switched off, and a log level must not be able to suppress it.
        eprintln!(
            "warning: {TRUST_OVERRIDE_ENV} is set, so {field} points at {} without an \
             endpoint check",
            host_display(&host)
        );
    }
    if !url.username().is_empty() || url.password().is_some() {
        return Err(reject("must not embed credentials"));
    }
    Ok(url)
}

fn save_config(conn_mgr: &ConnectionManager, config: &Config) -> CliResult<()> {
    match &conn_mgr.config_path {
        Some(path) => config.save_to_path(path)?,
        None => config.save()?,
    }
    Ok(())
}

/// Save after a login. A plaintext store has just put the CAPI secret in this file, so it is
/// written owner-only; the keyring path leaves permissions alone.
fn save_config_for_store(
    conn_mgr: &ConnectionManager,
    config: &Config,
    store: &CredentialStore,
) -> CliResult<()> {
    if store.storage_backend() == "keyring" {
        return save_config(conn_mgr, config);
    }
    let path = match &conn_mgr.config_path {
        Some(path) => path.clone(),
        None => Config::config_path()?,
    };
    config.save_to_path_owner_only(&path)?;
    Ok(())
}

/// A recognizable, unique-per-login CAPI key name (visible/revocable in the console).
fn default_key_name() -> String {
    let ts = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    format!("redisctl-cli-{ts}")
}

/// Best-effort: open the given URL in the platform browser. Failure is non-fatal (the URL is
/// also printed for manual opening).
fn open_browser(url: &str) -> std::io::Result<()> {
    use std::process::{Command, Stdio};
    let mut cmd = if cfg!(target_os = "macos") {
        let mut c = Command::new("open");
        c.arg(url);
        c
    } else if cfg!(target_os = "windows") {
        let mut c = Command::new("cmd");
        c.args(["/C", "start", "", url]);
        c
    } else {
        let mut c = Command::new("xdg-open");
        c.arg(url);
        c
    };
    cmd.stdout(Stdio::null())
        .stderr(Stdio::null())
        .spawn()
        .map(|_| ())
}

#[cfg(test)]
mod tests {
    /// Logout deletes the local credentials whatever happens, so the notes are the only thing
    /// left saying what is still live server-side. A request that never arrived must not be
    /// reported as a sign-in that was already dead: that reads as "nothing to revoke" while a
    /// usable refresh token — and the key it can mint — outlive the command.
    ///
    /// Only `Transport` is exercised; `Network` shares its arm and cannot be constructed here
    /// without taking a `reqwest` dependency for a test.
    #[test]
    fn a_refresh_that_never_arrived_is_not_a_dead_sign_in() {
        let named = |why: &str| format!("the key k-1 was not revoked because {why}");
        let notes = refresh_failure(&AuthError::Transport("connection refused".into()), named)
            .notes()
            .join(" ");

        assert!(notes.contains("could not be reached"), "{notes}");
        assert!(
            !notes.contains("no longer valid") && !notes.contains("nothing to revoke"),
            "a transport failure was reported as a dead sign-in: {notes}"
        );
        assert!(notes.contains("k-1"), "{notes}");
    }

    /// An expired or revoked grant is the other case, and still reads as one.
    #[test]
    fn a_rejected_grant_still_reports_nothing_to_revoke() {
        let named = |why: &str| format!("the key k-1 was not revoked because {why}");
        let notes = refresh_failure(&AuthError::Protocol("invalid_grant".into()), named)
            .notes()
            .join(" ");
        assert!(notes.contains("no longer valid"), "{notes}");
        assert!(notes.contains("nothing to revoke"), "{notes}");
    }

    use super::*;

    fn minted(superseded_key_name: Option<&str>) -> MintedCredentials {
        MintedCredentials {
            account_id: Some(1),
            email: None,
            api_key: "k".into(),
            api_secret: "s".into(),
            api_url: "https://api.redislabs.com/v1".into(),
            refresh_token: None,
            capi_key_name: "redisctl-cli-2".into(),
            redisctl_key_count: 1,
            account_name: None,
            capi_newly_enabled: false,
            superseded_revoked: Some(false),
            superseded_key_name: superseded_key_name.map(str::to_string),
            accounts: vec![],
        }
    }

    /// A failed revocation is the one moment the old key's name still exists: the profile now
    /// records the new one, so a message without it leaves nothing to search the console for.
    #[test]
    fn a_failed_revocation_names_the_key_left_behind() {
        assert_eq!(
            superseded_label(&minted(Some("redisctl-cli-1"))),
            "the key redisctl-cli-1 that this replaced"
        );
        // Older profiles recorded no name; say what can be said rather than inventing one.
        assert_eq!(superseded_label(&minted(None)), "the key this replaced");
    }

    /// Each half is reported on its own. A key that could not be deleted must not be able to
    /// hide a sign-in that is still live, and vice versa — one bit could not say both.
    #[test]
    fn logout_reports_each_revocation_separately() {
        let both = Revocation {
            key: Ok("redisctl-cli-1".to_string()),
            session: Ok(()),
        };
        assert!(both.notes().is_empty());

        let key_only = Revocation {
            key: Ok("redisctl-cli-1".to_string()),
            session: Err("could not revoke the stored sign-in: 503.".to_string()),
        };
        let notes = key_only.notes();
        assert_eq!(notes.len(), 1);
        assert!(notes[0].contains("stays valid at the identity provider"));

        let session_only = Revocation {
            key: Err("could not revoke the key redisctl-cli-1: 500.".to_string()),
            session: Ok(()),
        };
        let notes = session_only.notes();
        assert_eq!(notes.len(), 1);
        assert!(notes[0].contains("Access Management > API Keys"));

        let neither = Revocation {
            key: Err("a".to_string()),
            session: Err("b".to_string()),
        };
        assert_eq!(neither.notes().len(), 2);
    }

    /// The prompt runs before the code for that attempt is submitted, so the count includes the
    /// one being typed: with three allowed, the second prompt still has two submissions left.
    #[test]
    fn attempts_left_counts_the_submission_being_entered() {
        assert_eq!(MFA_MAX_ATTEMPTS, 3);
        assert_eq!(attempts_left(1), 3);
        assert_eq!(attempts_left(2), 2);
        assert_eq!(attempts_left(3), 1);
        assert_eq!(attempts_left(9), 0);
    }

    fn accounts() -> Vec<LoginAccount> {
        vec![
            LoginAccount {
                id: 316941,
                name: Some("Acme".to_string()),
            },
            LoginAccount {
                id: 481022,
                name: None,
            },
        ]
    }

    /// `login` only asks where the session leaves the account open. Anywhere
    /// `AccountChoice::Current` could have answered, this answers the same way — otherwise a
    /// login that used to just work would start stopping for a question.
    ///
    /// Not covered here: the several-accounts-and-nothing-to-go-on case, which reads stdin.
    /// `resolve_account_choice` covers what the answer is then mapped to.
    #[test]
    fn login_asks_only_when_the_session_leaves_the_account_open() {
        // The session names one of them: taken as-is, whatever else is on the list.
        assert_eq!(
            account_for_login(&accounts(), Some(481022)).unwrap(),
            481022
        );

        // One account: nothing to choose between, whether or not the session claims something.
        let one = vec![LoginAccount {
            id: 316941,
            name: Some("Acme".to_string()),
        }];
        for current in [None, Some(999)] {
            assert_eq!(account_for_login(&one, current).unwrap(), 316941);
        }

        // No accounts at all is not a question either.
        let err = account_for_login(&[], None).unwrap_err();
        assert!(matches!(err, AuthError::Protocol(_)), "got {err:?}");
    }

    /// A config file names where credentials are sent, so an endpoint that is not transport-secure
    /// must be refused rather than used. `http` on loopback stays allowed — no network to
    /// intercept, and the test harness and local mocks depend on it.
    #[test]
    fn endpoints_must_be_https_or_loopback() {
        for ok in [
            "https://auth.redis.com/oauth2/default",
            "https://api.redislabs.com/v1",
            "https://anything.redis.io/",
            // The suffix is matched on a label boundary, so a subdomain is fine…
            "http://127.0.0.1:8899/oauth2/default",
            "http://localhost:1234/api/v1",
            // `host_str` renders this as "[::1]", so a string comparison misses it.
            "http://[::1]:1234/api/v1",
        ] {
            assert!(
                parse_endpoint(ok, "f", false).is_ok(),
                "{ok} should be accepted"
            );
        }
        for bad in [
            "http://auth.redis.com/oauth2/default",
            // Not loopback: only ::1 and 127/8 are exempt, not any IPv6 literal.
            "http://[2606:4700::1111]/api/v1",
            "http://[fe80::1]/api/v1",
            // …and a host that merely ends in the same letters is not.
            "https://app.example.com/api/v1",
            "https://notredis.com/v1",
            "https://redis.com.attacker.example/v1",
            "https://evil-redis.io/v1",
            // Transport-secure but nobody we send tokens to.
            "https://198.51.100.7/v1",
            "https://user:pass@auth.redis.com/",
            "https://user@auth.redis.com/",
            "ftp://auth.redis.com/",
            "not a url",
            "file:///etc/passwd",
            "https://",
        ] {
            assert!(
                parse_endpoint(bad, "f", false).is_err(),
                "{bad:?} should be rejected"
            );
        }
    }

    /// The override exists for local mocks and test environments. It has to be an environment
    /// variable rather than a config field, because the finding it answers is a config file that
    /// names someone else's host.
    #[test]
    fn the_override_admits_another_host_but_not_another_scheme() {
        let host = "https://app.example.com/api/v1";
        assert!(parse_endpoint(host, "sm_api_url", false).is_err());
        assert!(
            parse_endpoint(host, "sm_api_url", true).is_ok(),
            "the override should admit a non-Redis host"
        );
        assert!(
            parse_endpoint("http://app.example.com/api/v1", "sm_api_url", true).is_err(),
            "the override must not relax transport security"
        );
    }

    #[test]
    fn the_override_is_opt_in_by_exact_value() {
        for on in ["1", "true", "TRUE", "True"] {
            assert!(trust_override_requested(Some(on)), "{on:?} should opt in");
        }
        for off in [
            None,
            Some(""),
            Some("0"),
            Some("false"),
            Some("yes"),
            Some("2"),
        ] {
            assert!(!trust_override_requested(off), "{off:?} should not opt in");
        }
    }

    #[test]
    fn client_id_must_look_like_one() {
        assert!(validate_client_id("0oaw90hjzrLoATW0q5d7").is_ok());
        assert!(validate_client_id("client-id_1").is_ok());
        for bad in ["", "a b", "a&b", "a|b", "a\"b", "a$(b)"] {
            assert!(
                validate_client_id(bad).is_err(),
                "{bad:?} should be rejected"
            );
        }
    }

    /// The pending record lands next to the config, in a directory that does not exist on a
    /// machine that has never run redisctl. Writing used to fail there with ENOENT *after* the
    /// IdP had already issued a code, losing the code the user needs.
    #[test]
    fn write_private_creates_missing_parents() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir
            .path()
            .join("no")
            .join("such")
            .join("dir")
            .join("p.json");
        write_private(&path, b"{}").unwrap();
        assert_eq!(std::fs::read(&path).unwrap(), b"{}");
    }

    #[cfg(unix)]
    #[test]
    fn write_private_is_owner_only() {
        use std::os::unix::fs::PermissionsExt;
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("nested").join("p.json");
        write_private(&path, b"{}").unwrap();
        let mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
        assert_eq!(mode, 0o600, "got {mode:o}");
    }

    /// The listing shows both a position and an id, so both have to be accepted — and anything
    /// else has to be refused rather than resolving to some account by accident.
    #[test]
    fn account_choice_accepts_a_position_or_an_id() {
        assert_eq!(resolve_account_choice(&accounts(), "1").unwrap(), 316941);
        assert_eq!(resolve_account_choice(&accounts(), " 2\n").unwrap(), 481022);
        // An id the user copied out of the list.
        assert_eq!(
            resolve_account_choice(&accounts(), "481022").unwrap(),
            481022
        );
    }

    /// Small account ids exist outside production, so an input that is both a valid position and a
    /// valid id must mean the position — that is what the prompt asks for. Resolving it to the id
    /// would mint a key for a different account than the one the user pointed at.
    #[test]
    fn account_choice_prefers_a_position_over_an_id() {
        let low = vec![
            LoginAccount {
                id: 2,
                name: Some("Two".to_string()),
            },
            LoginAccount {
                id: 5,
                name: Some("Five".to_string()),
            },
        ];
        // Row 2 is account #5, and "2" is also account #2's id: the row wins.
        assert_eq!(resolve_account_choice(&low, "2").unwrap(), 5);
        assert_eq!(resolve_account_choice(&low, "1").unwrap(), 2);
        // An id that is not also a position still resolves.
        assert_eq!(resolve_account_choice(&low, "5").unwrap(), 5);
    }

    /// Choice problems are the caller's to fix (exit 2), not backend failures (exit 1).
    #[test]
    fn account_choice_errors_are_preconditions() {
        let err = resolve_account_choice(&accounts(), "nope").unwrap_err();
        assert!(matches!(err, AuthError::AccountRequired(_)), "got {err:?}");
        assert_eq!(
            crate::structured_error::StructuredError::from(err).code,
            "account_required"
        );
    }

    #[test]
    fn account_choice_refuses_anything_else() {
        for input in ["", "0", "3", "999999", "abc", "1.5", "-1"] {
            assert!(
                resolve_account_choice(&accounts(), input).is_err(),
                "{input:?} should not resolve to an account"
            );
        }
    }

    /// A single-account user has nothing to switch to; saying so beats offering a list of one.
    #[test]
    fn prompt_refuses_when_there_is_only_one_account() {
        let one = vec![LoginAccount {
            id: 316941,
            name: Some("Acme".to_string()),
        }];
        // Does not read stdin: it returns before prompting.
        let err = prompt_account(&one, Some(316941)).unwrap_err();
        assert!(
            matches!(err, AuthError::AccountRequired(ref m) if m.contains("single Redis Cloud account")),
            "got {err:?}"
        );
    }
}