curlite 0.2.1

//! Everything that touches libcurl.
//!
//! Transfers run on a multi handle, one per thread, which owns the connection
//! cache. Driving it step by step is what lets a response be read as a
//! stream and keeps user code (writers, readers) out of libcurl's callbacks.

use bytes::Bytes;
use curl::easy::{Easy2, Handler as CurlHandler, ReadError, SeekResult, WriteError};
use curl::multi::{Easy2Handle, Multi};
use http::header::{
    AUTHORIZATION, CONTENT_LENGTH, CONTENT_TYPE, COOKIE, HeaderMap, HeaderValue, LOCATION,
    PROXY_AUTHORIZATION,
};
use http::{Method, StatusCode};

use std::any::Any;
use std::cell::{Cell, RefCell};
use std::collections::HashMap;
use std::fmt::{self, Write as _};
use std::io::{self, Read, SeekFrom, Write};
use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind};
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::time::Duration;

use crate::body::ReadSeek;
use crate::error::{Error, Result};
use crate::request::{Auth, HttpVersion, Progress, ProgressFn, Request, Upload};
use crate::response::{Response, process_header_line};

/// Streaming transfers pause once this much body is buffered and not yet
/// consumed, and resume when the consumer has drained it.
const PAUSE_ABOVE: usize = 128 * 1024;

thread_local! {
    /// One multi handle per thread. It owns the connection cache, so repeat
    /// calls to the same host skip the TCP and TLS handshakes.
    static MULTI: Multi = Multi::new();
    /// An easy handle kept for the next request, so that back-to-back
    /// requests do not pay for `curl_easy_init`.
    static SPARE: RefCell<Option<Easy2<Handler>>> = const { RefCell::new(None) };
    /// Set while libcurl runs callbacks. A request started from inside one
    /// (a `body_reader` that fetches lazily) must not touch the shared multi
    /// handle and gets a private one.
    static IN_CALLBACK: Cell<bool> = const { Cell::new(false) };
    /// Completion results by transfer token. `curl_multi_perform` reports a
    /// multi-wide count of running handles and queues one message per
    /// finished handle, so with several transfers on the thread each has to
    /// find its own.
    static RESULTS: RefCell<HashMap<usize, std::result::Result<(), curl::Error>>> =
        RefCell::new(HashMap::new());
    static NEXT_TOKEN: Cell<usize> = const { Cell::new(1) };
}

/// Keeps `IN_CALLBACK` set while libcurl runs, restoring the outer value
/// afterwards (a nested request's own steps must not clear it) and on panic.
struct CallbackGuard(bool);

impl CallbackGuard {
    fn enter() -> Self {
        CallbackGuard(IN_CALLBACK.replace(true))
    }
}

impl Drop for CallbackGuard {
    fn drop(&mut self) {
        IN_CALLBACK.set(self.0);
    }
}

/// libcurl's callbacks for one transfer.
#[derive(Default)]
pub(crate) struct Handler {
    source: Option<Box<dyn ReadSeek>>,
    status: Option<StatusCode>,
    headers: HeaderMap,
    /// The blank line after a non-1xx status line has arrived.
    headers_done: bool,
    /// `Content-Length`, reserved before the first body chunk when `presize`.
    expected_len: Option<usize>,
    presize: bool,
    content_length: Option<u64>,
    /// Body bytes accepted from and handed to libcurl, for progress.
    received: u64,
    sent: u64,
    /// Body bytes not yet consumed.
    body: Vec<u8>,
    /// `Request::max_body_size` in memory mode.
    limit: Option<u64>,
    too_large: bool,
    /// Pause the transfer when `body` holds this much.
    pause_above: Option<usize>,
    paused: bool,
    header_err: Option<Error>,
    read_err: Option<io::Error>,
    /// A panic from the user's reader, resumed outside libcurl. The curl
    /// crate would otherwise swallow it and fail every later callback on
    /// the thread.
    panic: Option<Box<dyn Any + Send>>,
}

impl Handler {
    fn clear(&mut self) {
        let mut body = std::mem::take(&mut self.body);
        body.clear();
        *self = Handler {
            body,
            ..Handler::default()
        };
    }
}

impl CurlHandler for Handler {
    fn write(&mut self, data: &[u8]) -> std::result::Result<usize, WriteError> {
        if self
            .pause_above
            .is_some_and(|limit| self.body.len() >= limit)
        {
            // the chunk is delivered again after unpause
            self.paused = true;
            return Err(WriteError::Pause);
        }
        if let Some(limit) = self.limit
            && self.received + data.len() as u64 > limit
        {
            // a short count is a write error to libcurl; the transfer ends
            self.too_large = true;
            return Ok(0);
        }
        if self.presize
            && self.body.is_empty()
            && let Some(len) = self.expected_len.take()
        {
            // a lying Content-Length must not abort on allocation failure;
            // growth takes over if the reservation fails
            let _ = self.body.try_reserve_exact(len);
        }
        self.body.extend_from_slice(data);
        self.received += data.len() as u64;
        Ok(data.len())
    }

    fn read(&mut self, buf: &mut [u8]) -> std::result::Result<usize, ReadError> {
        let Some(source) = &mut self.source else {
            return Ok(0);
        };
        let result = catch_unwind(AssertUnwindSafe(|| {
            loop {
                match source.read(buf) {
                    Ok(n) => return Ok(n),
                    Err(e) if e.kind() == io::ErrorKind::Interrupted => continue,
                    Err(e) => return Err(e),
                }
            }
        }));
        match result {
            Ok(Ok(n)) => {
                self.sent += n as u64;
                Ok(n)
            }
            Ok(Err(e)) => {
                self.read_err = Some(e);
                Err(ReadError::Abort)
            }
            Err(panic) => {
                self.panic = Some(panic);
                Err(ReadError::Abort)
            }
        }
    }

    fn seek(&mut self, whence: SeekFrom) -> SeekResult {
        let Some(source) = &mut self.source else {
            return SeekResult::CantSeek;
        };
        match catch_unwind(AssertUnwindSafe(|| source.seek(whence))) {
            Ok(Ok(_)) => SeekResult::Ok,
            Ok(Err(_)) => SeekResult::CantSeek,
            Err(panic) => {
                self.panic = Some(panic);
                SeekResult::Fail
            }
        }
    }

    fn header(&mut self, line: &[u8]) -> bool {
        let mut status = self.status;
        if let Err(e) = process_header_line(line, &mut self.headers, &mut status) {
            self.header_err = Some(e);
            // false aborts the transfer, which kills the keep-alive
            // connection but spares us a body we would throw away anyway
            return false;
        }
        self.status = status;
        if line == b"\r\n" || line == b"\n" {
            // the blank line after `100 Continue` does not end the headers
            self.headers_done = status.is_some_and(|s| !s.is_informational());
            self.expected_len = self
                .headers
                .get(CONTENT_LENGTH)
                .and_then(|v| v.to_str().ok()?.trim().parse().ok());
            self.content_length = self.expected_len.map(|n| n as u64);
        }
        true
    }

    #[cfg(feature = "log")]
    fn debug(&mut self, kind: curl::easy::InfoType, data: &[u8]) {
        use curl::easy::InfoType;
        let prefix = match kind {
            InfoType::Text => "* ",
            InfoType::HeaderIn => "< ",
            InfoType::HeaderOut => "> ",
            _ => return,
        };
        for line in String::from_utf8_lossy(data).lines() {
            log::trace!(target: "curlite", "{prefix}{line}");
        }
    }
}

enum MultiRef {
    Shared,
    Private(Multi),
}

impl MultiRef {
    fn with<R>(&self, f: impl FnOnce(&Multi) -> R) -> R {
        match self {
            MultiRef::Shared => MULTI.with(f),
            MultiRef::Private(multi) => f(multi),
        }
    }
}

/// How a transfer buffers its body.
#[derive(Clone, Copy)]
pub(crate) enum Mode {
    /// Everything, sized from `Content-Length`.
    Memory,
    /// Chunks until the consumer drains them, with backpressure.
    Stream,
}

/// A transfer whose final response headers have arrived; the body may still
/// be coming. Dropping it mid-way cancels the transfer.
pub(crate) struct Transfer {
    handle: Option<Easy2Handle<Handler>>,
    multi: MultiRef,
    token: usize,
    done: bool,
    /// The URL this hop was sent to.
    url: String,
    progress: Option<ProgressFn>,
    upload_total: Option<u64>,
    /// `(received, sent)` at the last progress report.
    reported: (u64, u64),
}

impl fmt::Debug for Transfer {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Transfer")
            .field("done", &self.done)
            .finish_non_exhaustive()
    }
}

/// Starts `request`, follows redirects, and returns once the headers of the
/// final response are in.
///
/// libcurl's own following is off, because with `CURLOPT_CUSTOMREQUEST` it
/// keeps the verb across every redirect, so a `DELETE` answered with `303`
/// would be repeated as `DELETE`.
pub(crate) fn start(
    mut request: Request,
    mut upload: Option<Upload>,
    mode: Mode,
) -> Result<Transfer> {
    let mut multi = if IN_CALLBACK.get() {
        MultiRef::Private(Multi::new())
    } else {
        MultiRef::Shared
    };
    let mut easy = SPARE
        .with(|spare| spare.borrow_mut().take())
        .unwrap_or_else(|| Easy2::new(Handler::default()));
    let mut progress = request.progress.take();
    let mut hops = 0;
    loop {
        easy.reset();
        let handler = easy.get_mut();
        handler.clear();
        handler.presize = matches!(mode, Mode::Memory);
        handler.pause_above = matches!(mode, Mode::Stream).then_some(PAUSE_ABOVE);
        handler.source = upload.as_mut().and_then(|upload| upload.source.take());
        if let Err(e) = configure(&mut easy, &request, upload.as_ref()) {
            recycle(easy);
            return Err(e);
        }
        let token = NEXT_TOKEN.replace(NEXT_TOKEN.get().wrapping_add(1));
        let mut handle = multi.with(|m| m.add2(easy))?;
        handle.set_token(token)?;
        let mut transfer = Transfer {
            handle: Some(handle),
            multi,
            token,
            done: false,
            url: request.url.clone(),
            progress: progress.take(),
            upload_total: upload.as_ref().map(|u| u.len),
            reported: (0, 0),
        };
        transfer.drive(|h| h.headers_done)?;
        let handler = transfer.handler();
        let status = handler
            .status
            .ok_or_else(|| Error::HeaderParse("transfer ended without a status line".to_owned()))?;
        if !request.follow_redirects || !is_followed(status, handler.headers.contains_key(LOCATION))
        {
            // the response that is kept: bound it, unless it has no body
            if matches!(mode, Mode::Memory)
                && let Some(limit) = request.max_body_size
            {
                let no_body =
                    request.method == Method::HEAD || matches!(status.as_u16(), 204 | 304);
                // bytes that arrived with the headers count too
                if !no_body
                    && (handler.content_length.is_some_and(|len| len > limit)
                        || handler.received > limit)
                {
                    return Err(Error::BodyTooLarge(limit));
                }
                transfer.handler_mut().limit = Some(limit);
            }
            return Ok(transfer);
        }
        // a redirect to follow: finish it, discarding its body (which in
        // stream mode would otherwise pause the transfer for good), and take
        // the handle back
        while !transfer.done {
            transfer.drain(|_| Ok::<(), Error>(()))?;
            transfer.step(|h| !h.body.is_empty())?;
        }
        let next;
        progress = transfer.progress.take();
        (easy, multi, next) = transfer.detach()?;
        // the Location header resolved against the request URL
        let Some(next) = next else {
            recycle(easy);
            return Err(Error::HeaderParse(
                "redirect without a resolvable Location".to_owned(),
            ));
        };
        if let Some(upload) = &mut upload {
            upload.source = easy.get_mut().source.take();
        }
        match status.as_u16() {
            // RFC 9110: 303 always switches to GET, 301/302 do so for POST
            303 if request.method != Method::GET && request.method != Method::HEAD => {
                drop_body(&mut request, &mut upload);
            }
            301 | 302 if request.method == Method::POST => drop_body(&mut request, &mut upload),
            _ => {
                if let Some(source) = upload.as_mut().and_then(|u| u.source.as_mut()) {
                    source.seek(SeekFrom::Start(0))?;
                }
            }
        }
        if hops == request.max_redirects {
            recycle(easy);
            return Err(Error::TooManyRedirects(request.max_redirects));
        }
        hops += 1;
        if !same_origin(&request.url, &next) {
            request.headers.remove(AUTHORIZATION);
            request.headers.remove(COOKIE);
            request.headers.remove(PROXY_AUTHORIZATION);
            request.auth = None;
        }
        request.url = next;
    }
}

/// Whether a response with this status is followed when redirects are on
/// and it has a `Location`: 300, 304 and a 3xx without `Location` are
/// returned.
fn is_followed(status: StatusCode, has_location: bool) -> bool {
    matches!(status.as_u16(), 301 | 302 | 303 | 307 | 308) && has_location
}

fn drop_body(request: &mut Request, upload: &mut Option<Upload>) {
    request.method = Method::GET;
    request.headers.remove(CONTENT_TYPE);
    *upload = None;
}

/// Scheme, host and port of `url`, lowercased, without resolving default
/// ports: `https://a.com` and `https://a.com:443` count as different origins,
/// which errs on the side of dropping credentials. A URL without a scheme is
/// `http://`, as libcurl assumes.
fn origin(url: &str) -> String {
    let url = if url.contains("://") {
        url.to_owned()
    } else {
        format!("http://{url}")
    };
    let url = url.as_str();
    let after_scheme = url.find("://").map_or(0, |i| i + 3);
    let end = url[after_scheme..]
        .find(['/', '?', '#'])
        .map_or(url.len(), |i| after_scheme + i);
    let authority = &url[after_scheme..end];
    // userinfo is not part of the origin
    let host = authority
        .rsplit_once('@')
        .map_or(authority, |(_, host)| host);
    format!("{}{host}", &url[..after_scheme]).to_ascii_lowercase()
}

fn same_origin(a: &str, b: &str) -> bool {
    origin(a) == origin(b)
}

fn configure(easy: &mut Easy2<Handler>, request: &Request, upload: Option<&Upload>) -> Result<()> {
    // this is an HTTP client, and redirects bypass libcurl's protocol
    // limits: a Location of gopher:// or file:// must not be opened
    if let Some(scheme) = scheme(&request.url)
        && !scheme.eq_ignore_ascii_case("http")
        && !scheme.eq_ignore_ascii_case("https")
    {
        return Err(Error::UnsupportedScheme(request.url.clone()));
    }
    easy.url(&request.url)?;
    easy.timeout(request.timeout)?;
    easy.connect_timeout(request.connect_timeout)?;
    if !request.stall_timeout.is_zero() {
        // "slower than 1 byte/s for N seconds" is libcurl's stall detector;
        // N is in whole seconds, so round up rather than silently disable
        let stall = &request.stall_timeout;
        let seconds = stall.as_secs() + u64::from(stall.subsec_nanos() > 0);
        easy.low_speed_limit(1)?;
        easy.low_speed_time(Duration::from_secs(seconds))?;
    }
    easy.follow_location(false)?;
    if let Some(version) = request.http_version {
        easy.http_version(match version {
            HttpVersion::Http11 => curl::easy::HttpVersion::V11,
            HttpVersion::Http2 => curl::easy::HttpVersion::V2TLS,
            HttpVersion::Http2PriorKnowledge => curl::easy::HttpVersion::V2PriorKnowledge,
            HttpVersion::Http3 => curl::easy::HttpVersion::V3,
        })?;
    }
    if let Some(proxy) = &request.proxy {
        easy.proxy(proxy)?;
    }
    if let Some(path) = &request.unix_socket {
        easy.unix_socket_path(Some(path))?;
    }
    // TLS options only for TLS: a libcurl built without it rejects them
    let https = request
        .url
        .get(..8)
        .is_some_and(|s| s.eq_ignore_ascii_case("https://"));
    if let Some(path) = &request.ca_bundle
        && https
    {
        easy.cainfo(path)?;
    }
    if let Some((cert, key)) = &request.client_cert
        && https
    {
        easy.ssl_cert(cert)?;
        easy.ssl_key(key)?;
    }
    if !request.verify_tls {
        easy.ssl_verify_peer(false)?;
        easy.ssl_verify_host(false)?;
    } else if request.ca_bundle.is_none()
        && https
        && let Some((one_cert, dir)) = lazy_ca_store()
    {
        // OpenSSL looks certificates up by hash in `CApath` on demand,
        // instead of parsing the whole bundle (about 20 ms of CPU per
        // process on x86). libcurl insists on a `CAinfo` file that loads,
        // so it gets one real certificate from that same directory
        easy.cainfo(one_cert)?;
        easy.capath(dir)?;
    } else if request.ca_bundle.is_none() && https && native_ca_needed() {
        // a libcurl built from source by the `curl` crate with the rustls
        // backend has no CA file or directory compiled in, and curl's rustls
        // backend then configures no verifier at all: every HTTPS request
        // fails with "SSL connect error". Its platform verifier reads the
        // system roots instead
        easy.ssl_options(curl::easy::SslOpt::new().native_ca(true))?;
    }
    if request.compression {
        // "" advertises every encoding this libcurl can decode
        easy.accept_encoding("")?;
    }
    if let Some(auth) = &request.auth
        && let Some((user, password, methods)) = libcurl_auth(auth)
    {
        easy.username(user)?;
        easy.password(password)?;
        easy.http_auth(&methods)?;
    }
    if let Some(auth) = &request.proxy_auth
        && let Some((user, password, methods)) = libcurl_auth(auth)
    {
        easy.proxy_username(user)?;
        easy.proxy_password(password)?;
        easy.proxy_auth(&methods)?;
    }
    if request.netrc {
        easy.netrc(curl::easy::NetRc::Optional)?;
    }
    #[cfg(feature = "log")]
    if log::log_enabled!(target: "curlite", log::Level::Trace) {
        easy.verbose(true)?;
    }

    if request.method != Method::POST {
        easy.custom_request(request.method.as_str())?;
    }
    if request.method == Method::HEAD {
        easy.nobody(true)?;
    }
    // streamed via the read callback; post_fields_copy would hold a second
    // copy inside libcurl
    match upload {
        Some(upload) if upload.len > 0 => {
            // same framing as post_fields_copy (POST + known size), the verb
            // still comes from custom_request
            easy.post(true)?;
            easy.post_field_size(upload.len)?;
        }
        Some(_) => easy.post_fields_copy(b"")?,
        None => {
            if matches!(request.method, Method::POST | Method::PUT | Method::PATCH) {
                // bodyless POST/PUT/PATCH keep an explicit Content-Length: 0,
                // which strict servers require
                easy.post_fields_copy(b"")?;
            }
        }
    }

    if let Some(Auth::Bearer(_)) = &request.proxy_auth {
        // CURLOPT_PROXYHEADER only works with CURLOPT_HEADEROPT, which the
        // curl crate does not expose yet; a plain header would also reach
        // the origin
        return Err(Error::UnsupportedScheme(
            "Bearer is not a proxy authentication scheme".to_owned(),
        ));
    }
    let mut header_list = curl::easy::List::new();
    if let Some(Auth::Bearer(token)) = &request.auth
        && !request.headers.contains_key(AUTHORIZATION)
    {
        // a token with a line break would smuggle a header
        let value = HeaderValue::from_str(&format!("Bearer {token}"))?;
        header_list.append(&format!("Authorization: {}", value.to_str()?))?;
    }
    let mut line = String::new();
    for (name, value) in &request.headers {
        let value = value.to_str()?.trim();
        // "name;" is libcurl's syntax for a header with an empty value
        let separator = if value.is_empty() { ";" } else { ": " };
        line.clear();
        let _ = write!(line, "{}{separator}{value}", name.as_str());
        header_list.append(&line)?;
    }
    easy.http_headers(header_list)?;
    Ok(())
}

/// The URL's scheme, if it has one: the part before the first `:` when
/// that is a scheme name and not a host followed by a port (`localhost:8080`),
/// which libcurl reads as `http://`. `file:/x` and `dict:x` have no `//`
/// and count as schemes.
fn scheme(url: &str) -> Option<&str> {
    let (candidate, rest) = url.split_once(':')?;
    let name = candidate.as_bytes().first()?.is_ascii_alphabetic()
        && candidate
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || b"+-.".contains(&b));
    let port = rest.split(['/', '?', '#']).next().unwrap_or("");
    let is_port = !port.is_empty() && port.bytes().all(|b| b.is_ascii_digit());
    (name && !is_port).then_some(candidate)
}

/// Credentials and methods for libcurl's own authentication; `None` for
/// `Bearer`, which is a plain header.
fn libcurl_auth(auth: &Auth) -> Option<(&str, &str, curl::easy::Auth)> {
    let mut methods = curl::easy::Auth::new();
    match auth {
        Auth::Basic { .. } => methods.basic(true),
        Auth::Digest { .. } => methods.digest(true),
        Auth::Ntlm { .. } => methods.ntlm(true),
        Auth::Negotiate => methods.gssnegotiate(true),
        Auth::Bearer(_) => return None,
    };
    let (user, password) = match auth {
        Auth::Basic { user, password }
        | Auth::Digest { user, password }
        | Auth::Ntlm { user, password } => (user.as_str(), password.as_str()),
        // libcurl only negotiates once a user is set, `-u :` in curl
        _ => ("", ""),
    };
    Some((user, password, methods))
}

/// The system's hashed certificate directory and one certificate file in
/// it, if libcurl runs on OpenSSL (other backends read a `CApath` eagerly,
/// which would be slower than the bundle). Decided once per process.
fn lazy_ca_store() -> Option<&'static (PathBuf, PathBuf)> {
    static DECISION: OnceLock<Option<(PathBuf, PathBuf)>> = OnceLock::new();
    DECISION
        .get_or_init(|| {
            if !cfg!(unix) {
                return None;
            }
            let ssl = curl::Version::get().ssl_version()?.to_owned();
            if !ssl.starts_with("OpenSSL") {
                return None;
            }
            ["/etc/ssl/certs", "/usr/lib/ssl/certs"]
                .iter()
                .map(Path::new)
                .find_map(|dir| Some((first_hash_link(dir)?, dir.to_path_buf())))
        })
        .as_ref()
}

/// Whether this libcurl runs on rustls without a built-in CA file or
/// directory, so that it needs `CURLSSLOPT_NATIVE_CA` to trust anything.
/// Decided once per process.
fn native_ca_needed() -> bool {
    static DECISION: OnceLock<bool> = OnceLock::new();
    *DECISION.get_or_init(|| {
        let version = curl::Version::get();
        version.ssl_version().is_some_and(|s| s.starts_with("rustls"))
            && version.cainfo().is_none()
            && version.capath().is_none()
    })
}

/// The first OpenSSL hash link (`<8 hex digits>.0`) among the first entries
/// of `dir`.
fn first_hash_link(dir: &Path) -> Option<PathBuf> {
    let entries = std::fs::read_dir(dir).ok()?;
    entries.take(200).flatten().find_map(|entry| {
        let name = entry.file_name();
        let text = name.to_string_lossy();
        (text.len() == 10
            && text.ends_with(".0")
            && text[..8].bytes().all(|b| b.is_ascii_hexdigit()))
        .then(|| dir.join(&name))
    })
}

/// Keeps `easy` for the next request on this thread.
fn recycle(mut easy: Easy2<Handler>) {
    easy.reset();
    easy.get_mut().clear();
    let _ = SPARE.try_with(|spare| {
        let mut spare = spare.borrow_mut();
        if spare.is_none() {
            *spare = Some(easy);
        }
    });
}

impl Transfer {
    fn handle(&self) -> &Easy2Handle<Handler> {
        self.handle.as_ref().expect("transfer is attached")
    }

    pub(crate) fn handler(&self) -> &Handler {
        self.handle().get_ref()
    }

    fn handler_mut(&mut self) -> &mut Handler {
        self.handle
            .as_mut()
            .expect("transfer is attached")
            .get_mut()
    }

    pub(crate) fn done(&self) -> bool {
        self.done
    }

    /// Runs libcurl until `until` holds or the transfer ends.
    pub(crate) fn drive(&mut self, until: impl Fn(&Handler) -> bool) -> Result<()> {
        while !self.done && !until(self.handler()) {
            self.step(&until)?;
        }
        Ok(())
    }

    /// One round of libcurl work. Waits for socket activity afterwards
    /// unless the transfer ended or `ready` holds, so that data that just
    /// arrived reaches the consumer without a delay.
    pub(crate) fn step(&mut self, ready: impl Fn(&Handler) -> bool) -> Result<()> {
        let result = self.step_inner(&ready);
        if result.is_err() {
            self.done = true;
        }
        self.report_progress();
        result
    }

    /// Calls the progress function when the counts changed, and once at the
    /// end. Runs between libcurl's steps, so user code never sits inside a
    /// callback.
    fn report_progress(&mut self) {
        let (Some(progress), Some(handle)) = (&mut self.progress, &self.handle) else {
            return;
        };
        let handler = handle.get_ref();
        let counts = (handler.received, handler.sent);
        if counts == self.reported && !self.done {
            return;
        }
        self.reported = counts;
        progress.0(Progress {
            downloaded: counts.0,
            download_total: handler.content_length,
            uploaded: counts.1,
            upload_total: self.upload_total,
        });
    }

    fn step_inner(&mut self, ready: &dyn Fn(&Handler) -> bool) -> Result<()> {
        if self.handler().paused {
            self.handle().unpause_write()?;
            self.handler_mut().paused = false;
        }
        self.multi.with(|m| {
            let _guard = CallbackGuard::enter();
            m.perform()
        })?;
        let handler = self.handler_mut();
        if let Some(panic) = handler.panic.take() {
            resume_unwind(panic);
        }
        if let Some(e) = handler.header_err.take() {
            return Err(e);
        }
        if handler.too_large {
            return Err(Error::BodyTooLarge(handler.limit.unwrap_or(0)));
        }
        if let Some(e) = handler.read_err.take() {
            return Err(e.into());
        }
        // file every finished handle's result under its token; ours may be
        // among them, another transfer's is kept for it
        self.multi.with(|m| {
            m.messages(|message| {
                if let (Ok(token), Some(result)) = (message.token(), message.result()) {
                    RESULTS.with_borrow_mut(|results| results.insert(token, result));
                }
            });
        });
        if let Some(result) = RESULTS.with_borrow_mut(|results| results.remove(&self.token)) {
            self.done = true;
            return result.map_err(Error::from);
        }
        if !self.handler().paused && !ready(self.handler()) {
            self.multi
                .with(|m| m.poll(&mut [], Duration::from_millis(500)))?;
        }
        Ok(())
    }

    /// Hands the buffered body to `consume` and clears the buffer.
    pub(crate) fn drain<E>(
        &mut self,
        consume: impl FnOnce(&[u8]) -> std::result::Result<(), E>,
    ) -> std::result::Result<(), E> {
        let body = &mut self.handler_mut().body;
        if body.is_empty() {
            return Ok(());
        }
        let result = consume(body);
        body.clear();
        result
    }

    /// The response with whatever body is buffered; recycles the handle.
    pub(crate) fn into_response(mut self) -> Result<Response> {
        let (mut easy, _, _) = self.detach()?;
        let handler = easy.get_mut();
        let status = handler
            .status
            .ok_or_else(|| Error::HeaderParse("transfer ended without a status line".to_owned()))?;
        let headers = std::mem::take(&mut handler.headers);
        let mut body = std::mem::take(&mut handler.body);
        // return any growth overshoot before the buffer is frozen into Bytes
        // for the response's lifetime; a no-op when Content-Length sized it
        body.shrink_to_fit();
        recycle(easy);
        Ok(Response {
            url: std::mem::take(&mut self.url),
            status,
            headers,
            body: Bytes::from(body),
        })
    }

    /// Removes the handle from the multi handle and returns it with the
    /// resolved redirect target, if the response had one.
    fn detach(&mut self) -> Result<(Easy2<Handler>, MultiRef, Option<String>)> {
        let handle = self.handle.take().expect("transfer is attached");
        let next = handle.redirect_url()?.map(str::to_owned);
        let easy = self.multi.with(|m| m.remove2(handle))?;
        let multi = std::mem::replace(&mut self.multi, MultiRef::Shared);
        RESULTS.with_borrow_mut(|results| results.remove(&self.token));
        Ok((easy, multi, next))
    }
}

impl Drop for Transfer {
    fn drop(&mut self) {
        let _ = RESULTS.try_with(|results| results.borrow_mut().remove(&self.token));
        let Some(handle) = self.handle.take() else {
            return;
        };
        if IN_CALLBACK.get() {
            // libcurl forbids removing a handle from inside a callback and
            // the curl crate would free it anyway; leaking it is the safe
            // way out of a reader that drops a transfer
            std::mem::forget(handle);
            return;
        }
        let removed = match &self.multi {
            MultiRef::Shared => MULTI.try_with(|m| m.remove2(handle)).ok(),
            MultiRef::Private(m) => Some(m.remove2(handle)),
        };
        if let Some(Ok(easy)) = removed {
            recycle(easy);
        }
    }
}

pub(crate) fn send(request: Request, upload: Option<Upload>) -> Result<Response> {
    let mut transfer = start(request, upload, Mode::Memory)?;
    transfer.drive(|_| false)?;
    transfer.into_response()
}

pub(crate) fn send_to(
    request: Request,
    upload: Option<Upload>,
    writer: &mut dyn Write,
) -> Result<Response> {
    let mut transfer = start(request, upload, Mode::Stream)?;
    loop {
        transfer.drain(|chunk| writer.write_all(chunk))?;
        if transfer.done() {
            return transfer.into_response();
        }
        transfer.step(|h| !h.body.is_empty())?;
    }
}

/// A response whose body is read on demand, see [`Request::send_reader`].
///
/// libcurl only works while the reader is read from; the connection stays
/// open in the meantime. Dropping the reader early cancels the transfer and
/// closes that connection.
///
/// [`Request::send_reader`]: crate::Request::send_reader
#[derive(Debug)]
pub struct ResponseReader {
    transfer: Transfer,
    url: String,
    status: StatusCode,
    headers: HeaderMap,
    pos: usize,
}

impl ResponseReader {
    pub(crate) fn new(request: Request, upload: Option<Upload>) -> Result<Self> {
        let mut transfer = start(request, upload, Mode::Stream)?;
        let handler = transfer.handler_mut();
        let status = handler
            .status
            .ok_or_else(|| Error::HeaderParse("transfer ended without a status line".to_owned()))?;
        let headers = std::mem::take(&mut handler.headers);
        let url = transfer.url.clone();
        Ok(ResponseReader {
            transfer,
            url,
            status,
            headers,
            pos: 0,
        })
    }

    /// The URL the response came from: the request's, or the last
    /// redirect's.
    pub fn url(&self) -> &str {
        &self.url
    }

    pub fn status(&self) -> StatusCode {
        self.status
    }

    pub fn headers(&self) -> &HeaderMap {
        &self.headers
    }

    /// The parsed `Content-Type` header, if present and valid.
    pub fn content_type(&self) -> Result<Option<mime::Mime>> {
        crate::response::parse_content_type(&self.headers)
    }

    /// Reads the rest of the body into memory.
    pub fn into_response(mut self) -> Result<Response> {
        let mut body = Vec::new();
        self.read_to_end(&mut body)?;
        Ok(Response {
            url: std::mem::take(&mut self.url),
            status: self.status,
            headers: std::mem::take(&mut self.headers),
            body: Bytes::from(body),
        })
    }
}

impl Read for ResponseReader {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        loop {
            let handler = self.transfer.handler_mut();
            if self.pos < handler.body.len() {
                let n = (handler.body.len() - self.pos).min(buf.len());
                buf[..n].copy_from_slice(&handler.body[self.pos..self.pos + n]);
                self.pos += n;
                if self.pos == handler.body.len() {
                    handler.body.clear();
                    self.pos = 0;
                }
                return Ok(n);
            }
            if self.transfer.done() {
                return Ok(0);
            }
            self.transfer
                .step(|h| !h.body.is_empty())
                .map_err(io::Error::other)?;
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn hash_link_detection() {
        let dir = std::env::temp_dir().join(format!("curlite-hash-{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        assert_eq!(first_hash_link(&dir), None);
        std::fs::write(dir.join("ca-certificates.crt"), "").unwrap();
        std::fs::write(dir.join("deadbeef.pem"), "").unwrap();
        assert_eq!(first_hash_link(&dir), None);
        std::fs::write(dir.join("5ed36f99.0"), "").unwrap();
        assert_eq!(first_hash_link(&dir), Some(dir.join("5ed36f99.0")));
        std::fs::remove_dir_all(&dir).unwrap();
        assert_eq!(first_hash_link(&dir), None);
    }

    #[test]
    fn schemes() {
        assert_eq!(scheme("http://a"), Some("http"));
        assert_eq!(scheme("HTTPS://a"), Some("HTTPS"));
        assert_eq!(scheme("file:/etc/passwd"), Some("file"));
        assert_eq!(scheme("dict:x"), Some("dict"));
        assert_eq!(scheme("gopher://h"), Some("gopher"));
        assert_eq!(scheme("localhost:8080/x"), None);
        assert_eq!(scheme("localhost/x?a=b:c"), None);
        assert_eq!(scheme("example.com"), None);
        assert_eq!(scheme(""), None);
    }

    #[test]
    fn origins() {
        assert!(same_origin("https://A.com/x?y", "https://a.com/z"));
        assert!(same_origin("http://u:p@a.com/x", "http://a.com"));
        assert!(!same_origin("https://a.com", "http://a.com"));
        assert!(!same_origin("https://a.com", "https://a.com:8443"));
        assert!(!same_origin("https://a.com", "https://b.com"));
        assert!(same_origin("a.com/x", "http://A.com/y"));
        assert_eq!(origin("https://a.com"), "https://a.com");
        assert_eq!(origin(""), "http://");
    }
}