elasticctl-core 0.7.5

//! HTTP transport, including URL construction, headers, retries, and error
//! classification.

use crate::auth::Credential;
use crate::capabilities::{Capabilities, Feature};
use crate::config::Profile;
use crate::error::{Error, ErrorKind, Result};
use reqwest::{Client, Method};
use serde_json::Value;
use std::collections::BTreeMap;
use std::io::Write;
use std::time::Duration;
use tokio::sync::OnceCell;

mod requests;

/// Version of the public API this client targets.
const API_VERSION: &str = "2023-10-31";
const MAX_ATTEMPTS: u32 = 3;

/// Optional transport restrictions for callers that handle untrusted upstream
/// responses.
#[derive(Clone, Debug, Default)]
pub struct TransportOptions {
    pub debug: bool,
    pub response_body_limit: Option<usize>,
    pub disable_redirects: bool,
    pub disable_retries: bool,
}

/// Response headers retained past the transport boundary.
///
/// These headers are allowlisted because recorded fixtures are public.
/// Capturing all headers could record cookies, rate-limit counters, or future
/// proxy headers that do not belong in the repository.
///
/// The capability probe reads `x-found-handling-cluster`. The other two show
/// which Cloud headers the recorded response contained.
const CAPTURED_HEADERS: [&str; 3] = [
    "x-found-handling-cluster",
    "x-found-handling-instance",
    "x-elastic-product",
];

/// Parse a JSON response without letting serde_json coerce an out-of-range
/// integer literal into an imprecise floating-point number.
fn parse_response_json(text: &str) -> Result<Value> {
    validate_json_integer_ranges(text)?;
    serde_json::from_str(text)
        .map_err(|e| Error::new(ErrorKind::Http, format!("parsing response JSON: {e}")))
}

/// Reject positive integer lexemes above `u64::MAX` and negative integer
/// lexemes below `i64::MIN`. Numbers with a decimal point or exponent remain
/// floating-point JSON values, even when their magnitude is greater than
/// `u64::MAX`.
fn validate_json_integer_ranges(text: &str) -> Result<()> {
    let bytes = text.as_bytes();
    let mut index = 0;

    while index < bytes.len() {
        if bytes[index] == b'"' {
            index += 1;
            while index < bytes.len() {
                match bytes[index] {
                    b'\\' => index += 2,
                    b'"' => {
                        index += 1;
                        break;
                    }
                    _ => index += 1,
                }
            }
            continue;
        }

        if bytes[index] != b'-' && !bytes[index].is_ascii_digit() {
            index += 1;
            continue;
        }

        let start = index;
        if bytes[index] == b'-' {
            index += 1;
        }
        if index == bytes.len() || !bytes[index].is_ascii_digit() {
            index = start + 1;
            continue;
        }

        if bytes[index] == b'0' {
            index += 1;
        } else {
            while index < bytes.len() && bytes[index].is_ascii_digit() {
                index += 1;
            }
        }

        let mut is_integer = true;
        if bytes.get(index) == Some(&b'.') {
            is_integer = false;
            index += 1;
            while index < bytes.len() && bytes[index].is_ascii_digit() {
                index += 1;
            }
        }
        if matches!(bytes.get(index), Some(b'e' | b'E')) {
            is_integer = false;
            index += 1;
            if matches!(bytes.get(index), Some(b'+' | b'-')) {
                index += 1;
            }
            while index < bytes.len() && bytes[index].is_ascii_digit() {
                index += 1;
            }
        }

        if is_integer {
            let number = &text[start..index];
            let in_range = if bytes[start] == b'-' {
                number.parse::<i64>().is_ok()
            } else {
                number.parse::<u64>().is_ok()
            };
            if !in_range {
                return Err(Error::new(
                    ErrorKind::Http,
                    format!(
                        "parsing response JSON: integer {number} is outside supported integer range"
                    ),
                ));
            }
        }
    }

    Ok(())
}

/// A response body and its captured headers.
///
/// Hosted and self-managed stacks return the same `/api/status` body. An
/// edge-proxy header distinguishes them.
#[derive(Debug, Clone)]
pub struct Responded {
    pub body: Value,
    pub headers: BTreeMap<String, String>,
}

impl Responded {
    /// Look up a header case-insensitively.
    ///
    /// The Elastic proxy varies the casing of `x-found-handling-cluster` by
    /// endpoint.
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers.get(&name.to_ascii_lowercase()).map(|s| &**s)
    }
}

/// Percent-encode a query value while leaving URL-safe characters unchanged.
///
/// The API client and fixture recorder share this encoder so they produce the
/// same scoped-filter URL.
pub fn urlencode(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for b in s.bytes() {
        match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
                out.push(b as char)
            }
            _ => out.push_str(&format!("%{b:02X}")),
        }
    }
    out
}

pub struct Transport {
    client: Client,
    one_shot_client: Client,
    base: String,
    /// The Kibana URL exactly as configured, before trailing-slash
    /// normalization. `doctor` reports it as the connectivity target and the
    /// capability probe uses it for hostname-based flavor detection.
    kibana_url: String,
    /// Elasticsearch host. Cloud deployments use a different host from
    /// Kibana; otherwise this uses the Kibana host.
    es_base: String,
    /// Whether the profile configured an explicit `es_url`. `es_base` falls
    /// back to the Kibana host either way, so a caller that needs to know
    /// whether that fallback happened (e.g. to name a likely cause in an
    /// error message) reads this instead of comparing hosts.
    has_es_url: bool,
    space: String,
    auth_header: String,
    debug: bool,
    response_body_limit: Option<usize>,
    attempt_limit: u32,
    capabilities: OnceCell<Capabilities>,
}

impl Transport {
    fn client_builder(profile: &Profile) -> reqwest::ClientBuilder {
        Client::builder()
            .timeout(Duration::from_secs(profile.timeout_secs))
            .danger_accept_invalid_certs(!profile.verify)
    }

    pub fn new(profile: &Profile) -> Result<Transport> {
        Self::with_options(profile, TransportOptions::default())
    }

    /// Build a transport with HTTP request logging enabled or disabled.
    ///
    /// Keeping `debug` as a `bool` prevents CLI `clap` types entering `-core`.
    pub fn with_debug(profile: &Profile, debug: bool) -> Result<Transport> {
        Self::with_options(
            profile,
            TransportOptions {
                debug,
                ..Default::default()
            },
        )
    }

    /// Build a transport with opt-in response and request restrictions.
    pub fn with_options(profile: &Profile, options: TransportOptions) -> Result<Transport> {
        // Scrub URL userinfo before deriving any base URL or logging, so a
        // credential embedded in a URL never reaches a request or debug line.
        let mut profile = profile.clone();
        profile.strip_userinfo();
        let credential = Credential::from_profile(&profile)?;
        let mut client_builder = Self::client_builder(&profile);
        if options.disable_redirects {
            client_builder = client_builder.redirect(reqwest::redirect::Policy::none());
        }
        if options.disable_retries {
            client_builder = client_builder.retry(reqwest::retry::never());
        }
        let client = client_builder
            .build()
            .map_err(|e| Error::new(ErrorKind::Connection, format!("building HTTP client: {e}")))?;
        let one_shot_client = Self::client_builder(&profile)
            .redirect(reqwest::redirect::Policy::none())
            // hyper-util retries a canceled checkout only through its connection
            // pool. Zero idle connections disables that pool, so this client
            // cannot replay an unstarted mutation after a reused socket fails.
            .pool_max_idle_per_host(0)
            .retry(reqwest::retry::never())
            .build()
            .map_err(|e| {
                Error::new(
                    ErrorKind::Connection,
                    format!("building one-shot HTTP client: {e}"),
                )
            })?;

        let base = profile.kibana_url.trim_end_matches('/').to_string();
        let kibana_url = profile.kibana_url.clone();
        let has_es_url = profile.es_url.is_some();
        let es_base = profile
            .es_url
            .as_deref()
            .unwrap_or(&profile.kibana_url)
            .trim_end_matches('/')
            .to_string();

        Ok(Transport {
            client,
            one_shot_client,
            base,
            kibana_url,
            es_base,
            has_es_url,
            space: profile.space.clone(),
            auth_header: credential.header_value(),
            debug: options.debug,
            response_body_limit: options.response_body_limit,
            attempt_limit: if options.disable_retries {
                1
            } else {
                MAX_ATTEMPTS
            },
            capabilities: OnceCell::new(),
        })
    }

    /// Log one request or response line to stderr.
    ///
    /// Logs include the method, complete URL, and status. They exclude
    /// authorization headers and bodies. Callers must not put credentials in
    /// query strings.
    fn debug_log(&self, method: &Method, url: &str, status: u16, attempt: u32) {
        if !self.debug {
            return;
        }
        if attempt > 1 {
            eprintln!(
                "[debug] {} {url} -> {status} (attempt {attempt})",
                method.as_str()
            );
        } else {
            eprintln!("[debug] {} {url} -> {status}", method.as_str());
        }
    }

    /// Log the request before sending it so timeouts produce debug output.
    fn debug_request(&self, method: &Method, url: &str, attempt: u32) {
        if !self.debug {
            return;
        }
        if attempt > 1 {
            eprintln!("[debug] -> {} {url} (attempt {attempt})", method.as_str());
        } else {
            eprintln!("[debug] -> {} {url}", method.as_str());
        }
    }

    /// Log a timeout or connection failure in the response-line format.
    fn debug_failure(&self, method: &Method, url: &str, what: &str) {
        if !self.debug {
            return;
        }
        let _ = writeln!(
            std::io::stderr(),
            "[debug] {} {url} -> {what}",
            method.as_str()
        );
    }

    /// Prefix non-default spaces with `/s/<name>`.
    ///
    /// Kibana serves the default space at the bare path.
    pub fn space_path(space: &str, path: &str) -> String {
        if space.is_empty() || space == "default" {
            path.to_string()
        } else {
            format!("/s/{space}{path}")
        }
    }

    /// The Kibana URL this transport targets, exactly as configured.
    pub fn kibana_url(&self) -> &str {
        &self.kibana_url
    }

    /// The Kibana space this transport targets, exactly as the profile
    /// resolved it. Fleet normalization compares `space_ids` against it.
    pub fn space(&self) -> &str {
        &self.space
    }

    /// Whether the profile configured an explicit `es_url`. When it did not,
    /// every `*_absolute_es` call silently falls back to the Kibana host —
    /// callers that need to distinguish "no Elasticsearch host configured"
    /// from "the real Elasticsearch host answered" read this.
    pub fn has_es_url(&self) -> bool {
        self.has_es_url
    }

    /// Probe deployment capabilities once for this transport.
    pub async fn capabilities(&self) -> Result<&Capabilities> {
        self.capabilities
            .get_or_try_init(|| Capabilities::probe(self, self.kibana_url()))
            .await
    }

    /// Refuse an unverified feature before its public route is called.
    pub async fn require_feature(&self, feature: Feature) -> Result<()> {
        self.capabilities().await?.require_feature(feature)
    }

    fn url(&self, path: &str) -> String {
        format!("{}{}", self.base, Self::space_path(&self.space, path))
    }
}