xmtp_attachments/http/

native.rs

use std::{
    error::Error,
    io::{BufReader, Read},
    net::{IpAddr, SocketAddr},
    sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    },
    time::Duration,
};

use bytes::Bytes;
use reqwest::{
    Client, StatusCode, Url,
    dns::{Addrs, Name, Resolve, Resolving},
    header::{ACCEPT_ENCODING, CONTENT_ENCODING, HeaderMap, LOCATION},
    redirect::Policy,
};
use tokio::sync::mpsc;
use tokio::time::timeout;

use super::{
    CONNECT_TIMEOUT, IDLE_TIMEOUT, PutOutcome, UploadRequest, checked_count, is_loopback_name,
};
use crate::{
    AttachmentError, AttachmentFailureCause as Cause,
    address::is_private,
    store::{AttachmentOptions, CHUNK_SIZE, DownloadSink, StagedFile},
};

mod upload;

const DEFLATE_OVERHEAD_DIVISOR: u64 = 1000;

fn compressed_input_cap(decoded_cap: u64) -> u64 {
    // Allow normal DEFLATE block overhead and framing, but stop streams that
    // consume input without producing decoded bytes.
    let decoded_cap = decoded_cap.min(u32::MAX as u64);
    decoded_cap
        .saturating_add(decoded_cap / DEFLATE_OVERHEAD_DIVISOR)
        .saturating_add(CHUNK_SIZE as u64)
}

#[derive(Debug)]
struct BlockedDns;

impl std::fmt::Display for BlockedDns {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("all DNS answers are blocked attachment addresses")
    }
}

impl Error for BlockedDns {}

struct SystemResolver;

impl Resolve for SystemResolver {
    fn resolve(&self, name: Name) -> Resolving {
        let host = name.as_str().to_owned();
        Box::pin(async move {
            let addrs: Vec<_> = tokio::net::lookup_host((host.as_str(), 0)).await?.collect();
            Ok(Box::new(addrs.into_iter()) as Addrs)
        })
    }
}

struct GuardedResolver {
    upstream: Arc<dyn Resolve>,
    allow_private: bool,
}

impl Resolve for GuardedResolver {
    fn resolve(&self, name: Name) -> Resolving {
        let answer = if is_loopback_name(name.as_str()) {
            let addrs = [
                SocketAddr::new(std::net::Ipv4Addr::LOCALHOST.into(), 0),
                SocketAddr::new(std::net::Ipv6Addr::LOCALHOST.into(), 0),
            ];
            Box::pin(async move { Ok(Box::new(addrs.into_iter()) as Addrs) }) as Resolving
        } else {
            self.upstream.resolve(name)
        };
        let allow_private = self.allow_private;
        Box::pin(async move {
            let addrs: Vec<SocketAddr> = answer
                .await?
                .filter(|addr| allow_private || !is_private(addr.ip()))
                .collect();
            if addrs.is_empty() {
                return Err(Box::new(BlockedDns) as Box<dyn Error + Send + Sync>);
            }
            Ok(Box::new(addrs.into_iter()) as Addrs)
        })
    }
}

fn reqwest_error(error: reqwest::Error) -> AttachmentError {
    let mut source: Option<&(dyn Error + 'static)> = Some(&error);
    while let Some(current) = source {
        if current.is::<BlockedDns>() {
            return AttachmentError::new(Cause::BlockedAddress);
        }
        source = current.source();
    }
    AttachmentError::new(Cause::Network)
}

fn validate_url(url: &Url, options: &AttachmentOptions) -> Result<(), AttachmentError> {
    let host = url.host().ok_or(AttachmentError::new(Cause::InsecureUrl))?;
    let loopback = match host {
        url::Host::Domain(name) => is_loopback_name(name),
        url::Host::Ipv4(address) => address.is_loopback(),
        url::Host::Ipv6(address) => address.is_loopback(),
    };
    if url.scheme() != "https"
        && !(url.scheme() == "http" && loopback && options.allow_private_network)
    {
        return Err(AttachmentError::new(Cause::InsecureUrl));
    }
    let ip = match host {
        url::Host::Ipv4(ip) => Some(IpAddr::V4(ip)),
        url::Host::Ipv6(ip) => Some(IpAddr::V6(ip)),
        url::Host::Domain(_) => None,
    };
    if ip.is_some_and(|address| !options.allow_private_network && is_private(address)) {
        return Err(AttachmentError::new(Cause::BlockedAddress));
    }
    Ok(())
}

fn content_encoding(
    headers: &HeaderMap,
    status: u16,
) -> Result<Option<&'static str>, AttachmentError> {
    let mut coding = None;
    for value in headers.get_all(CONTENT_ENCODING) {
        let value = value
            .to_str()
            .map_err(|_| AttachmentError::with_http_status(Cause::HttpStatus, status))?;
        for part in value.split(',') {
            let part = part.trim().to_ascii_lowercase();
            if part == "identity" {
                continue;
            }
            if coding.is_some() {
                return Err(AttachmentError::with_http_status(Cause::HttpStatus, status));
            }
            coding = Some(match part.as_str() {
                "gzip" | "x-gzip" => "gzip",
                "deflate" => "deflate",
                _ => return Err(AttachmentError::with_http_status(Cause::HttpStatus, status)),
            });
        }
    }
    Ok(coding)
}

struct AbortOnDrop(tokio::task::AbortHandle);

impl AbortOnDrop {
    fn new<T>(task: &tokio::task::JoinHandle<T>) -> Self {
        Self(task.abort_handle())
    }
}

impl Drop for AbortOnDrop {
    fn drop(&mut self) {
        self.0.abort();
    }
}

fn redirect_target(
    current: &Url,
    location: &str,
    redirects: u8,
    options: &AttachmentOptions,
) -> Result<Url, AttachmentError> {
    if redirects == 10 {
        return Err(AttachmentError::new(Cause::TooManyRedirects));
    }
    let target = current
        .join(location)
        .map_err(|_| AttachmentError::new(Cause::InsecureUrl))?;
    validate_url(&target, options)?;
    Ok(target)
}

/// Native HTTP transfer. Downloads check each connected address.
pub struct Transfer {
    client: Client,
    upload_resolver: Arc<dyn Resolve>,
    #[cfg(test)]
    upload_test_roots: Option<rustls::RootCertStore>,
    #[cfg(test)]
    upload_socket_failure: Option<upload::SocketSetupFailure>,
    options: AttachmentOptions,
    connect_timeout: Duration,
    idle_timeout: Duration,
}

impl Transfer {
    pub fn new(options: AttachmentOptions) -> Result<Self, AttachmentError> {
        Self::with_resolver(options, Arc::new(SystemResolver))
    }

    fn with_resolver(
        options: AttachmentOptions,
        upstream: Arc<dyn Resolve>,
    ) -> Result<Self, AttachmentError> {
        Self::with_resolver_and_timeouts(options, upstream, CONNECT_TIMEOUT, IDLE_TIMEOUT)
    }

    fn with_resolver_and_timeouts(
        options: AttachmentOptions,
        upstream: Arc<dyn Resolve>,
        connect_timeout: Duration,
        idle_timeout: Duration,
    ) -> Result<Self, AttachmentError> {
        let resolver = Arc::new(GuardedResolver {
            upstream: upstream.clone(),
            allow_private: options.allow_private_network,
        });
        let client = xmtp_common::http::client_builder()
            .no_proxy()
            .dns_resolver(resolver.clone())
            .redirect(Policy::none())
            .connect_timeout(connect_timeout)
            .build()
            .map_err(reqwest_error)?;
        Ok(Self {
            client,
            upload_resolver: upstream,
            #[cfg(test)]
            upload_test_roots: None,
            #[cfg(test)]
            upload_socket_failure: None,
            options,
            connect_timeout,
            idle_timeout,
        })
    }

    #[cfg(test)]
    fn with_timeouts(
        options: AttachmentOptions,
        connect: Duration,
        idle: Duration,
    ) -> Result<Self, AttachmentError> {
        Self::with_resolver_and_timeouts(options, Arc::new(SystemResolver), connect, idle)
    }

    pub async fn put(
        &self,
        request: &UploadRequest,
        body: StagedFile,
    ) -> Result<PutOutcome, AttachmentError> {
        upload::put(self, request, body).await
    }

    pub async fn get(
        &self,
        url: &str,
        cap: u64,
        sink: &mut dyn DownloadSink,
    ) -> Result<(), AttachmentError> {
        let mut url = Url::parse(url).map_err(|_| AttachmentError::new(Cause::InsecureUrl))?;
        let mut redirects = 0;
        let response = loop {
            validate_url(&url, &self.options)?;
            let response = timeout(
                self.idle_timeout,
                self.client
                    .get(url.clone())
                    .header(ACCEPT_ENCODING, "identity")
                    .send(),
            )
            .await
            .map_err(|_| AttachmentError::new(Cause::Network))?
            .map_err(reqwest_error)?;
            if !matches!(
                response.status(),
                StatusCode::MOVED_PERMANENTLY
                    | StatusCode::FOUND
                    | StatusCode::SEE_OTHER
                    | StatusCode::TEMPORARY_REDIRECT
                    | StatusCode::PERMANENT_REDIRECT
            ) {
                break response;
            }
            let status = response.status().as_u16();
            let location = response
                .headers()
                .get(LOCATION)
                .ok_or(AttachmentError::with_http_status(Cause::HttpStatus, status))?
                .to_str()
                .map_err(|_| AttachmentError::with_http_status(Cause::HttpStatus, status))?;
            url = redirect_target(&url, location, redirects, &self.options)?;
            redirects += 1;
        };
        let status = response.status().as_u16();
        match response.status() {
            StatusCode::OK => {}
            StatusCode::NOT_FOUND | StatusCode::GONE => {
                return Err(AttachmentError::new(Cause::NotFound));
            }
            _ => return Err(AttachmentError::with_http_status(Cause::HttpStatus, status)),
        }
        let encoding = content_encoding(response.headers(), status)?;
        let cap = cap.min(self.options.max_download_bytes.unwrap_or(u64::MAX));
        match encoding {
            None => read_identity(response, cap, sink, self.idle_timeout).await,
            Some(encoding) => {
                read_compressed(response, encoding, cap, sink, self.idle_timeout).await
            }
        }
    }
}

async fn read_identity(
    mut response: reqwest::Response,
    cap: u64,
    sink: &mut dyn DownloadSink,
    idle_timeout: Duration,
) -> Result<(), AttachmentError> {
    let mut count = 0;
    while let Some(chunk) = timeout(idle_timeout, response.chunk())
        .await
        .map_err(|_| AttachmentError::new(Cause::Network))?
        .map_err(reqwest_error)?
    {
        count = checked_count(count, chunk.len(), cap)?;
        sink.write(&chunk).await?;
    }
    Ok(())
}

struct ChannelReader {
    rx: mpsc::Receiver<Bytes>,
    consumed: mpsc::Sender<()>,
    chunk: Bytes,
    position: usize,
}

impl Read for ChannelReader {
    fn read(&mut self, output: &mut [u8]) -> std::io::Result<usize> {
        if self.position == self.chunk.len() {
            if !self.chunk.is_empty() {
                self.chunk = Bytes::new();
                self.position = 0;
                if self.consumed.blocking_send(()).is_err() {
                    return Ok(0);
                }
            }
            let Some(bytes) = self.rx.blocking_recv() else {
                return Ok(0);
            };
            self.chunk = bytes;
            self.position = 0;
        }
        let count = output.len().min(self.chunk.len() - self.position);
        output[..count].copy_from_slice(&self.chunk[self.position..self.position + count]);
        self.position += count;
        Ok(count)
    }
}

enum CompressedReader {
    Gzip(flate2::read::MultiGzDecoder<ChannelReader>),
    Deflate(flate2::bufread::ZlibDecoder<BufReader<ChannelReader>>),
}

impl Read for CompressedReader {
    fn read(&mut self, output: &mut [u8]) -> std::io::Result<usize> {
        match self {
            Self::Gzip(reader) => reader.read(output),
            Self::Deflate(reader) => reader.read(output),
        }
    }
}

impl CompressedReader {
    fn finish(self) -> Result<(), ()> {
        if let Self::Deflate(reader) = self {
            // BufReader keeps bytes after the zlib end marker available here.
            let mut encoded = reader.into_inner();
            let mut trailing = [0_u8; 1];
            if encoded.read(&mut trailing).map_err(|_| ())? != 0 {
                return Err(());
            }
        }
        Ok(())
    }
}

async fn read_compressed(
    mut response: reqwest::Response,
    encoding: &str,
    cap: u64,
    sink: &mut dyn DownloadSink,
    idle_timeout: Duration,
) -> Result<(), AttachmentError> {
    let status = response.status().as_u16();
    let (input_tx, input_rx) = mpsc::channel(1);
    let (consumed_tx, mut consumed_rx) = mpsc::channel(1);
    let (output_tx, mut output_rx) = mpsc::channel(1);
    let (recycle_tx, mut recycle_rx) = mpsc::channel(1);
    let network_failed = Arc::new(AtomicBool::new(false));
    let producer_failed = network_failed.clone();
    let input_too_large = Arc::new(AtomicBool::new(false));
    let producer_too_large = input_too_large.clone();
    let input_cap = compressed_input_cap(cap);
    let gzip = encoding == "gzip";
    let producer = tokio::spawn(async move {
        let mut input_count = 0_u64;
        let mut decoder_done = false;
        loop {
            let chunk = match timeout(idle_timeout, response.chunk()).await {
                Ok(Ok(Some(chunk))) => chunk,
                Ok(Ok(None)) => break,
                Err(_) | Ok(Err(_)) => {
                    producer_failed.store(true, Ordering::Release);
                    break;
                }
            };
            input_count = match input_count.checked_add(chunk.len() as u64) {
                Some(next) if next <= input_cap => next,
                _ => {
                    producer_too_large.store(true, Ordering::Release);
                    break;
                }
            };
            if decoder_done {
                continue;
            }
            for start in (0..chunk.len()).step_by(CHUNK_SIZE) {
                let end = (start + CHUNK_SIZE).min(chunk.len());
                if input_tx.send(chunk.slice(start..end)).await.is_err() {
                    decoder_done = true;
                    break;
                }
                if consumed_rx.recv().await.is_none() {
                    decoder_done = true;
                    break;
                }
            }
        }
    });
    // Aborting the producer drops input_tx and consumed_rx. This lets the
    // blocking decoder stop when the caller drops the download future.
    let _producer_guard = AbortOnDrop::new(&producer);
    let decoder = tokio::task::spawn_blocking(move || {
        let reader = ChannelReader {
            rx: input_rx,
            consumed: consumed_tx,
            chunk: Bytes::new(),
            position: 0,
        };
        let mut reader = if gzip {
            CompressedReader::Gzip(flate2::read::MultiGzDecoder::new(reader))
        } else {
            CompressedReader::Deflate(flate2::bufread::ZlibDecoder::new(BufReader::new(reader)))
        };
        let mut output = vec![0_u8; CHUNK_SIZE];
        loop {
            let count = reader.read(&mut output).map_err(|_| ())?;
            if count == 0 {
                return reader.finish();
            }
            if output_tx.blocking_send((output, count)).is_err() {
                return Ok(());
            }
            let Some(recycled) = recycle_rx.blocking_recv() else {
                return Ok(());
            };
            output = recycled;
        }
    });
    let mut count = 0;
    let mut result = Ok(());
    while let Some((output, size)) = output_rx.recv().await {
        match checked_count(count, size, cap) {
            Ok(next) => count = next,
            Err(error) => {
                result = Err(error);
                break;
            }
        }
        if let Err(error) = sink.write(&output[..size]).await {
            result = Err(error);
            break;
        }
        if recycle_tx.send(output).await.is_err() {
            break;
        }
    }
    drop(recycle_tx);
    drop(output_rx);
    if result.is_ok() {
        let decoded = decoder.await;
        if decoded.as_ref().is_ok_and(|result| result.is_ok()) {
            let _ = producer.await;
        } else {
            producer.abort();
        }
        if input_too_large.load(Ordering::Acquire) {
            return Err(AttachmentError::new(Cause::TooLarge));
        }
        if network_failed.load(Ordering::Acquire) {
            return Err(AttachmentError::new(Cause::Network));
        }
        result = decoded
            .map_err(|_| AttachmentError::with_http_status(Cause::HttpStatus, status))
            .and_then(|result| {
                result.map_err(|_| AttachmentError::with_http_status(Cause::HttpStatus, status))
            });
    } else {
        producer.abort();
        let _ = decoder.await;
    }
    result
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::{LocalStore, NativeStore};
    use bytes::Bytes;
    use http_body_util::Full;
    use hyper::{Request, Response, body::Incoming, server::conn::http1, service::service_fn};
    use hyper_util::rt::TokioIo;
    use reqwest::Method;
    use std::{
        future::Future,
        sync::atomic::{AtomicBool, AtomicUsize, Ordering},
    };
    use tokio::time::Instant;

    struct Server {
        url: String,
        task: tokio::task::JoinHandle<()>,
    }

    impl Drop for Server {
        fn drop(&mut self) {
            self.task.abort();
        }
    }

    async fn server<F>(handler: F) -> Server
    where
        F: Fn(Request<Incoming>) -> Response<Full<Bytes>> + Send + Sync + 'static,
    {
        server_async(move |request| std::future::ready(handler(request))).await
    }

    async fn server_async<F, Fut>(handler: F) -> Server
    where
        F: Fn(Request<Incoming>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = Response<Full<Bytes>>> + Send + 'static,
    {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let handler = Arc::new(handler);
        let task = tokio::spawn(async move {
            while let Ok((stream, _)) = listener.accept().await {
                let handler = handler.clone();
                tokio::spawn(async move {
                    let service = service_fn(move |request| {
                        let response = handler(request);
                        async move { Ok::<_, std::convert::Infallible>(response.await) }
                    });
                    let _ = http1::Builder::new()
                        .serve_connection(TokioIo::new(stream), service)
                        .await;
                });
            }
        });
        Server {
            url: format!("http://127.0.0.1:{port}"),
            task,
        }
    }

    fn answer(status: StatusCode, body: impl Into<Bytes>) -> Response<Full<Bytes>> {
        Response::builder()
            .status(status)
            .body(Full::new(body.into()))
            .unwrap()
    }

    fn allowed() -> Transfer {
        Transfer::new(AttachmentOptions {
            allow_private_network: true,
            ..Default::default()
        })
        .unwrap()
    }

    #[derive(Default)]
    struct MemorySink(Vec<u8>);

    #[async_trait::async_trait]
    impl DownloadSink for MemorySink {
        async fn write(&mut self, bytes: &[u8]) -> Result<(), AttachmentError> {
            self.0.extend_from_slice(bytes);
            Ok(())
        }
    }

    // verifies: ATCH-053
    #[xmtp_common::test(unwrap_try = true)]
    async fn rejects_http_public() {
        let error = allowed()
            .get("http://example.com/object", 100, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::InsecureUrl);
    }

    // verifies: ATCH-053
    #[xmtp_common::test(unwrap_try = true)]
    async fn allows_http_loopback_when_permitted() {
        let server = server(|_| answer(StatusCode::OK, "hello")).await;
        let mut sink = MemorySink::default();
        allowed().get(&server.url, 5, &mut sink).await?;
        assert_eq!(sink.0, b"hello");
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn localhost_resolves_to_loopback_only() {
        use http_body_util::BodyExt;

        let server = server_async(|request| async move {
            if request.method() == Method::PUT {
                assert_eq!(
                    request
                        .into_body()
                        .collect()
                        .await
                        .unwrap()
                        .to_bytes()
                        .as_ref(),
                    b"body"
                );
                answer(StatusCode::OK, "")
            } else {
                answer(StatusCode::OK, "hello")
            }
        })
        .await;
        let port = Url::parse(&server.url)?.port().unwrap();
        let transfer = Transfer::with_resolver_and_timeouts(
            AttachmentOptions {
                allow_private_network: true,
                ..Default::default()
            },
            Arc::new(FakeResolver(vec!["192.0.2.1:0".parse()?])),
            Duration::from_millis(500),
            Duration::from_secs(1),
        )?;
        for host in ["localhost", "LOCALHOST."] {
            let url = format!("http://{host}:{port}/object");
            let mut sink = MemorySink::default();
            transfer.get(&url, 5, &mut sink).await?;
            assert_eq!(sink.0, b"hello");
            let (_directory, body) = staged_body()?;
            assert_eq!(transfer.put(&upload(url), body).await?, PutOutcome::Stored);
        }
    }

    struct FakeResolver(Vec<SocketAddr>);

    struct PendingResolver;

    impl Resolve for PendingResolver {
        fn resolve(&self, _name: Name) -> Resolving {
            Box::pin(std::future::pending())
        }
    }

    impl Resolve for FakeResolver {
        fn resolve(&self, _name: Name) -> Resolving {
            let addresses = self.0.clone();
            Box::pin(async move { Ok(Box::new(addresses.into_iter()) as Addrs) })
        }
    }

    fn staged_body() -> std::io::Result<(tempfile::TempDir, StagedFile)> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("body");
        std::fs::write(&path, b"body")?;
        Ok((directory, StagedFile::from_path_for_test(path)))
    }

    fn upload(url: String) -> UploadRequest {
        UploadRequest {
            method: "PUT".into(),
            url,
            headers: vec![],
            expires_in_seconds: 60,
        }
    }

    // verifies: ATCH-024
    #[xmtp_common::test(unwrap_try = true)]
    async fn non_put_method_is_rejected_without_request() {
        let requests = Arc::new(AtomicUsize::new(0));
        let count = requests.clone();
        let server = server(move |_| {
            count.fetch_add(1, Ordering::Relaxed);
            answer(StatusCode::OK, "")
        })
        .await;
        for method in ["POST", "put"] {
            let (_directory, body) = staged_body()?;
            let mut request = upload(server.url.clone());
            request.method = method.into();
            let result = allowed().put(&request, body).await;
            assert_eq!(requests.load(Ordering::Relaxed), 0, "{method}");
            assert_eq!(result.unwrap_err().cause, Cause::TargetRejected, "{method}");
        }
        assert_eq!(requests.load(Ordering::Relaxed), 0);
    }

    struct SeenTlsPut {
        method: Method,
        uri: String,
        headers: reqwest::header::HeaderMap,
        body: Bytes,
        alpn: Option<Vec<u8>>,
    }

    struct TlsPutServer {
        address: SocketAddr,
        root: rustls::pki_types::CertificateDer<'static>,
        task: tokio::task::JoinHandle<(bool, Option<SeenTlsPut>)>,
    }

    async fn tls_put_server() -> Result<TlsPutServer, Box<dyn Error + Send + Sync>> {
        tls_put_server_for("localhost").await
    }

    async fn tls_put_server_for(host: &str) -> Result<TlsPutServer, Box<dyn Error + Send + Sync>> {
        xmtp_cryptography::install_crypto_provider();
        let ca_key = rcgen::KeyPair::generate()?;
        let mut ca_params = rcgen::CertificateParams::new(vec!["Attachment test CA".into()])?;
        ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
        ca_params.key_usages = vec![
            rcgen::KeyUsagePurpose::DigitalSignature,
            rcgen::KeyUsagePurpose::KeyCertSign,
        ];
        let ca = rcgen::CertifiedIssuer::self_signed(ca_params, ca_key)?;
        let leaf_key = rcgen::KeyPair::generate()?;
        let mut leaf_params = rcgen::CertificateParams::new(vec![host.into()])?;
        leaf_params.is_ca = rcgen::IsCa::ExplicitNoCa;
        leaf_params.key_usages = vec![rcgen::KeyUsagePurpose::DigitalSignature];
        leaf_params.extended_key_usages = vec![rcgen::ExtendedKeyUsagePurpose::ServerAuth];
        let leaf = leaf_params.signed_by(&leaf_key, &ca)?;
        let mut tls = rustls::ServerConfig::builder()
            .with_no_client_auth()
            .with_single_cert(
                vec![leaf.der().clone(), ca.der().clone()],
                rustls::pki_types::PrivatePkcs8KeyDer::from(leaf_key.serialize_der()).into(),
            )?;
        tls.alpn_protocols = vec![b"http/1.1".to_vec()];
        let acceptor = tokio_rustls::TlsAcceptor::from(Arc::new(tls));
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let address = listener.local_addr()?;
        let task = tokio::spawn(async move {
            use http_body_util::BodyExt;

            let (socket, _) = listener.accept().await.unwrap();
            let Ok(tls) = acceptor.accept(socket).await else {
                return (false, None);
            };
            let alpn = tls.get_ref().1.alpn_protocol().map(<[u8]>::to_vec);
            let seen = Arc::new(std::sync::Mutex::new(None));
            let captured = seen.clone();
            let service = service_fn(move |request: Request<Incoming>| {
                let captured = captured.clone();
                let alpn = alpn.clone();
                async move {
                    let (parts, body) = request.into_parts();
                    let body = body.collect().await.unwrap().to_bytes();
                    *captured.lock().unwrap() = Some(SeenTlsPut {
                        method: parts.method,
                        uri: parts.uri.to_string(),
                        headers: parts.headers,
                        body,
                        alpn,
                    });
                    Ok::<_, std::convert::Infallible>(answer(StatusCode::CREATED, ""))
                }
            });
            let _ = http1::Builder::new()
                .serve_connection(TokioIo::new(tls), service)
                .await;
            let observed = seen.lock().unwrap().take();
            (true, observed)
        });
        Ok(TlsPutServer {
            address,
            root: ca.der().clone(),
            task,
        })
    }

    fn trusted_upload(server: &TlsPutServer) -> Result<Transfer, Box<dyn Error + Send + Sync>> {
        let mut roots = rustls::RootCertStore::empty();
        roots.add(server.root.clone())?;
        let mut transfer = Transfer::with_resolver(
            AttachmentOptions::default(),
            Arc::new(FakeResolver(vec![server.address])),
        )?;
        transfer.upload_test_roots = Some(roots);
        Ok(transfer)
    }

    async fn put_tries_next_address_after_setup_failure(
        failure: upload::SocketSetupFailure,
    ) -> Result<(), Box<dyn Error + Send + Sync>> {
        let server = tls_put_server_for("upload.example").await?;
        let mut roots = rustls::RootCertStore::empty();
        roots.add(server.root.clone())?;
        let mut transfer = Transfer::with_resolver_and_timeouts(
            AttachmentOptions::default(),
            Arc::new(FakeResolver(vec!["192.0.2.1:0".parse()?, server.address])),
            Duration::from_secs(2),
            Duration::from_secs(2),
        )?;
        transfer.upload_test_roots = Some(roots);
        transfer.upload_socket_failure = Some(failure);
        let (_directory, body) = staged_body()?;
        let request = upload(format!(
            "https://upload.example:{}/object",
            server.address.port()
        ));
        assert_eq!(transfer.put(&request, body).await?, PutOutcome::Stored);
        let (handshake, seen) = tokio::time::timeout(Duration::from_secs(2), server.task).await??;
        assert!(handshake);
        assert_eq!(seen.unwrap().body.as_ref(), b"body");
        Ok(())
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn put_tries_next_address_after_socket_creation_failure() {
        put_tries_next_address_after_setup_failure(upload::SocketSetupFailure::CreateFirst).await?;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn put_tries_next_address_after_send_buffer_failure() {
        put_tries_next_address_after_setup_failure(upload::SocketSetupFailure::SetBufferFirst)
            .await?;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn put_tries_next_address_after_stalled_connect() {
        let server = tls_put_server_for("upload.example").await?;
        let mut roots = rustls::RootCertStore::empty();
        roots.add(server.root.clone())?;
        let addresses = vec!["192.0.2.1:0".parse()?, server.address];
        let mut transfer = Transfer::with_resolver_and_timeouts(
            AttachmentOptions::default(),
            Arc::new(FakeResolver(addresses.clone())),
            Duration::from_millis(1200),
            Duration::from_secs(2),
        )?;
        transfer.upload_test_roots = Some(roots);
        transfer.upload_socket_failure = Some(upload::SocketSetupFailure::ConnectFirstPending);
        let request = upload(format!(
            "https://upload.example:{}/object",
            server.address.port()
        ));
        let (_directory, body) = staged_body()?;
        assert_eq!(transfer.put(&request, body).await?, PutOutcome::Stored);
        let (_, seen) = server.task.await?;
        assert_eq!(seen.unwrap().body.as_ref(), b"body");

        let mut stalled = Transfer::with_resolver_and_timeouts(
            AttachmentOptions::default(),
            Arc::new(FakeResolver(addresses)),
            Duration::from_millis(300),
            Duration::from_secs(2),
        )?;
        stalled.upload_socket_failure = Some(upload::SocketSetupFailure::ConnectAllPending);
        let (_directory, body) = staged_body()?;
        assert_eq!(
            stalled.put(&request, body).await.unwrap_err().cause,
            Cause::Network
        );
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn cloned_staged_uploads_read_independent_offsets() {
        use http_body_util::BodyExt;

        let directory = tempfile::tempdir()?;
        let store = NativeStore::new(directory.path()).await?;
        let payload = vec![b'x'; 4 * 1024 * 1024];
        let mut writer = store.create_temp(".tmp/body").await?;
        writer.write(&payload).await?;
        store.sync(&mut writer).await?;
        drop(writer);
        let staged = store.open_read(".tmp/body").await?;
        let received = Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen = received.clone();
        let server = server_async(move |request| {
            let seen = seen.clone();
            async move {
                assert_eq!(
                    request.headers()[hyper::header::CONTENT_LENGTH],
                    (4 * 1024 * 1024).to_string()
                );
                if let Ok(body) = request.into_body().collect().await {
                    seen.lock().unwrap().push(body.to_bytes());
                }
                answer(StatusCode::OK, "")
            }
        })
        .await;
        let transfer = Transfer::with_timeouts(
            AttachmentOptions::default(),
            Duration::from_secs(2),
            Duration::from_secs(2),
        )?;
        let request = upload(format!("{}/object", server.url));
        let (first, second) = tokio::join!(
            transfer.put(&request, staged.clone()),
            transfer.put(&request, staged)
        );
        assert_eq!(first?, PutOutcome::Stored);
        assert_eq!(second?, PutOutcome::Stored);
        {
            let bodies = received.lock().unwrap();
            assert_eq!(bodies.len(), 2);
            assert!(bodies.iter().all(|body| body.as_ref() == payload));
        }

        // Start the second PUT only after the first server has read body bytes.
        let (advanced_tx, advanced_rx) = tokio::sync::oneshot::channel();
        let signal = Arc::new(std::sync::Mutex::new(Some(advanced_tx)));
        let received = Arc::new(std::sync::Mutex::new(Vec::new()));
        let seen = received.clone();
        let late_server = server_async(move |request| {
            let signal = signal.clone();
            let seen = seen.clone();
            async move {
                let first = request.uri().path() == "/first";
                let mut body = request.into_body();
                let mut bytes = Vec::new();
                if first && let Some(Ok(frame)) = body.frame().await {
                    bytes.extend_from_slice(&frame.into_data().unwrap());
                    let sender = { signal.lock().unwrap().take() };
                    if let Some(sender) = sender {
                        let _ = sender.send(());
                    }
                    tokio::time::sleep(Duration::from_millis(100)).await;
                }
                while let Some(Ok(frame)) = body.frame().await {
                    bytes.extend_from_slice(&frame.into_data().unwrap());
                }
                seen.lock().unwrap().push(Bytes::from(bytes));
                answer(StatusCode::OK, "")
            }
        })
        .await;
        let transfer = Arc::new(transfer);
        let staged = store.open_read(".tmp/body").await?;
        let first_request = upload(format!("{}/first", late_server.url));
        let second_request = upload(format!("{}/second", late_server.url));
        let first_transfer = transfer.clone();
        let first_staged = staged.clone();
        let first =
            tokio::spawn(async move { first_transfer.put(&first_request, first_staged).await });
        tokio::time::timeout(Duration::from_secs(2), advanced_rx).await??;
        assert_eq!(
            transfer.put(&second_request, staged).await?,
            PutOutcome::Stored
        );
        assert_eq!(first.await??, PutOutcome::Stored);
        let bodies = received.lock().unwrap();
        assert_eq!(bodies.len(), 2);
        assert!(bodies.iter().all(|body| body.as_ref() == payload));
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn https_put_uses_trusted_ca_and_preserves_signed_request() {
        let server = tls_put_server().await?;
        let transfer = trusted_upload(&server)?;
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("body");
        let payload = vec![0x5a; 2 * CHUNK_SIZE];
        std::fs::write(&path, &payload)?;
        let mut request = upload(format!(
            "https://localhost:{}/signed/object?signature=test",
            server.address.port()
        ));
        request.headers = vec![
            ("content-type".into(), "application/octet-stream".into()),
            ("content-length".into(), payload.len().to_string()),
            ("x-amz-meta-request".into(), "signed-value".into()),
        ];
        assert_eq!(
            tokio::time::timeout(
                Duration::from_secs(3),
                transfer.put(&request, StagedFile::from_path_for_test(path))
            )
            .await??,
            PutOutcome::Stored
        );
        let (handshake, seen) = tokio::time::timeout(Duration::from_secs(3), server.task).await??;
        assert!(handshake);
        let seen = seen.expect("server received the PUT");
        assert_eq!(seen.method, Method::PUT);
        assert_eq!(seen.uri, "/signed/object?signature=test");
        assert_eq!(seen.headers["content-type"], "application/octet-stream");
        assert_eq!(
            seen.headers["content-length"].to_str()?,
            payload.len().to_string()
        );
        assert_eq!(seen.headers["x-amz-meta-request"], "signed-value");
        assert_eq!(seen.body, payload);
        assert_eq!(seen.alpn.as_deref(), Some(&b"http/1.1"[..]));
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn https_put_rejects_untrusted_certificate() {
        let server = tls_put_server().await?;
        let transfer = Transfer::with_resolver(
            AttachmentOptions::default(),
            Arc::new(FakeResolver(vec![server.address])),
        )?;
        let (_directory, body) = staged_body()?;
        let request = upload(format!(
            "https://localhost:{}/object",
            server.address.port()
        ));
        assert_eq!(
            tokio::time::timeout(Duration::from_secs(3), transfer.put(&request, body))
                .await?
                .unwrap_err()
                .cause,
            Cause::Network
        );
        let (handshake, seen) = tokio::time::timeout(Duration::from_secs(3), server.task).await??;
        assert!(!handshake);
        assert!(seen.is_none());
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn https_put_rejects_wrong_hostname() {
        let server = tls_put_server().await?;
        let transfer = trusted_upload(&server)?;
        let (_directory, body) = staged_body()?;
        let request = upload(format!(
            "https://wrong.example:{}/object",
            server.address.port()
        ));
        assert_eq!(
            tokio::time::timeout(Duration::from_secs(3), transfer.put(&request, body))
                .await?
                .unwrap_err()
                .cause,
            Cause::Network
        );
        let (handshake, seen) = tokio::time::timeout(Duration::from_secs(3), server.task).await??;
        assert!(!handshake);
        assert!(seen.is_none());
    }

    // verifies: ATCH-071
    #[xmtp_common::test(unwrap_try = true)]
    async fn public_http_put_is_rejected_before_request() {
        let seen = Arc::new(AtomicBool::new(false));
        let flag = seen.clone();
        let server = server(move |_| {
            flag.store(true, Ordering::Relaxed);
            answer(StatusCode::OK, "")
        })
        .await;
        let address: SocketAddr = server.url.trim_start_matches("http://").parse()?;
        let transfer = Transfer::with_resolver(
            AttachmentOptions::default(),
            Arc::new(FakeResolver(vec![address])),
        )?;
        let (_directory, body) = staged_body()?;
        let request = upload(format!("http://example.com:{}/object", address.port()));
        assert_eq!(
            transfer.put(&request, body).await.unwrap_err().cause,
            Cause::InsecureUrl
        );
        assert!(!seen.load(Ordering::Relaxed));
    }

    // verifies: ATCH-071
    #[xmtp_common::test(unwrap_try = true)]
    async fn loopback_http_put_needs_no_private_network_flag() {
        use http_body_util::BodyExt;

        let server = server_async(|request| async move {
            let body = request.into_body().collect().await.unwrap().to_bytes();
            assert_eq!(body.as_ref(), b"body");
            answer(StatusCode::ACCEPTED, "")
        })
        .await;
        let (_directory, body) = staged_body()?;
        let transfer = Transfer::new(AttachmentOptions::default())?;
        assert_eq!(
            transfer.put(&upload(server.url.clone()), body).await?,
            PutOutcome::Stored
        );
    }

    // verifies: ATCH-071
    #[xmtp_common::test(unwrap_try = true)]
    async fn private_https_put_is_not_dns_blocked() {
        crate::http::secure_upload_url(&Url::parse("https://10.1.2.3/object")?)?;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let address = listener.local_addr()?;
        let transfer = Transfer::with_resolver_and_timeouts(
            AttachmentOptions::default(),
            Arc::new(FakeResolver(vec![address])),
            Duration::from_secs(1),
            Duration::from_secs(1),
        )?;
        let (_directory, body) = staged_body()?;
        let accepted = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            drop(socket);
        });
        let request = upload(format!("https://private.example:{}/object", address.port()));
        assert_eq!(
            transfer.put(&request, body).await.unwrap_err().cause,
            Cause::Network
        );
        tokio::time::timeout(Duration::from_secs(1), accepted).await??;
    }

    // verifies: ATCH-071
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_redirect_is_rejected_without_following() {
        let followed = Arc::new(AtomicBool::new(false));
        let flag = followed.clone();
        let target = server(move |_| {
            flag.store(true, Ordering::Relaxed);
            answer(StatusCode::OK, "")
        })
        .await;
        let location = target.url.clone();
        let origin = server(move |_| {
            Response::builder()
                .status(StatusCode::TEMPORARY_REDIRECT)
                .header(LOCATION, &location)
                .body(Full::new(Bytes::new()))
                .unwrap()
        })
        .await;
        let (_directory, body) = staged_body()?;
        let transfer = Transfer::new(AttachmentOptions::default())?;
        assert_eq!(
            transfer
                .put(&upload(origin.url.clone()), body)
                .await
                .unwrap_err()
                .cause,
            Cause::TargetRejected
        );
        assert!(!followed.load(Ordering::Relaxed));
    }

    async fn early_put_status(
        status: StatusCode,
    ) -> Result<Result<PutOutcome, AttachmentError>, Box<dyn Error + Send + Sync>> {
        use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};

        const BODY_SIZE: usize = 8 * 1024 * 1024;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let server = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            let mut reader = BufReader::new(socket);
            let mut line = String::new();
            let mut content_length = None;
            loop {
                line.clear();
                assert!(reader.read_line(&mut line).await.unwrap() > 0);
                if line == "\r\n" {
                    break;
                }
                if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length: ") {
                    content_length = Some(length.trim().parse::<usize>().unwrap());
                }
            }
            assert_eq!(content_length, Some(BODY_SIZE));
            reader
                .get_mut()
                .write_all(
                    format!(
                        "HTTP/1.1 {} {}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
                        status.as_u16(),
                        status.canonical_reason().unwrap()
                    )
                    .as_bytes(),
                )
                .await
                .unwrap();
            let mut discarded = 0;
            let mut buffer = [0_u8; 8192];
            let _ = tokio::time::timeout(Duration::from_millis(200), async {
                while discarded < 256 * 1024 {
                    let limit = (256 * 1024 - discarded).min(buffer.len());
                    let read = reader.read(&mut buffer[..limit]).await.unwrap();
                    if read == 0 {
                        break;
                    }
                    discarded += read;
                }
            })
            .await;
            assert!(discarded <= 256 * 1024);
        });
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("body");
        std::fs::write(&path, vec![0x5a; BODY_SIZE])?;
        let result = tokio::time::timeout(
            Duration::from_secs(3),
            allowed().put(&upload(url), StagedFile::from_path_for_test(path)),
        )
        .await?;
        tokio::time::timeout(Duration::from_secs(1), server).await??;
        Ok(result)
    }

    // verifies: ATCH-025
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_early_412_with_large_body() {
        assert_eq!(
            early_put_status(StatusCode::PRECONDITION_FAILED).await??,
            PutOutcome::AlreadyStored
        );
    }

    // verifies: ATCH-025
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_early_403_with_large_body() {
        assert_eq!(
            early_put_status(StatusCode::FORBIDDEN)
                .await?
                .unwrap_err()
                .cause,
            Cause::TargetRejected
        );
    }

    // verifies: ATCH-079
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_rejection_keeps_http_status() {
        let error = early_put_status(StatusCode::FORBIDDEN).await?.unwrap_err();
        assert_eq!(error.cause, Cause::TargetRejected);
        assert_eq!(error.http_status, Some(403));
    }

    // verifies: ATCH-025
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_early_2xx_waits_for_body() {
        use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};

        const BODY_SIZE: usize = 4 * 1024 * 1024;
        let socket = tokio::net::TcpSocket::new_v4()?;
        socket.set_recv_buffer_size(8192)?;
        socket.bind("127.0.0.1:0".parse()?)?;
        let listener = socket.listen(1)?;
        let url = format!("http://{}", listener.local_addr()?);
        let received = Arc::new(AtomicUsize::new(0));
        let observed = received.clone();
        let (paused_tx, paused_rx) = tokio::sync::oneshot::channel();
        let (resume_tx, resume_rx) = tokio::sync::oneshot::channel();
        let server = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            let mut reader = BufReader::new(socket);
            let mut line = String::new();
            let mut content_length = None;
            loop {
                line.clear();
                assert!(reader.read_line(&mut line).await.unwrap() > 0);
                if line == "\r\n" {
                    break;
                }
                if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length: ") {
                    content_length = Some(length.trim().parse::<usize>().unwrap());
                }
            }
            assert_eq!(content_length, Some(BODY_SIZE));
            reader
                .get_mut()
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\n")
                .await
                .unwrap();
            let mut buffer = [0_u8; 8192];
            let mut count = 0;
            let mut paused_tx = Some(paused_tx);
            let mut resume_rx = Some(resume_rx);
            while count < BODY_SIZE {
                let limit = (BODY_SIZE - count).min(buffer.len());
                let size = reader.read(&mut buffer[..limit]).await.unwrap();
                assert!(size > 0);
                assert!(buffer[..size].iter().all(|byte| *byte == 0x5a));
                count += size;
                observed.store(count, Ordering::Release);
                if count >= BODY_SIZE / 2 && paused_tx.is_some() {
                    paused_tx.take().unwrap().send(()).unwrap();
                    resume_rx.take().unwrap().await.unwrap();
                }
            }
            reader.get_mut().write_all(b"x").await.unwrap();
        });
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("body");
        std::fs::write(&path, vec![0x5a; BODY_SIZE])?;
        let mut upload = tokio::spawn(async move {
            allowed()
                .put(&upload(url), StagedFile::from_path_for_test(path))
                .await
        });
        tokio::time::timeout(Duration::from_secs(5), paused_rx).await??;
        assert!(received.load(Ordering::Acquire) < BODY_SIZE);
        assert!(
            tokio::time::timeout(Duration::from_millis(150), &mut upload)
                .await
                .is_err()
        );
        resume_tx.send(()).unwrap();
        assert_eq!(
            tokio::time::timeout(Duration::from_secs(5), &mut upload).await???,
            PutOutcome::Stored
        );
        tokio::time::timeout(Duration::from_secs(5), server).await??;
        assert_eq!(received.load(Ordering::Acquire), BODY_SIZE);
    }

    // verifies: ATCH-025
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_early_2xx_then_close_is_network() {
        assert_eq!(
            early_put_status(StatusCode::OK).await?.unwrap_err().cause,
            Cause::Network
        );
    }

    async fn changed_file_during_put(
        truncate: bool,
    ) -> Result<(Result<PutOutcome, AttachmentError>, usize), Box<dyn Error + Send + Sync>> {
        use std::io::Write;
        use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};

        const BODY_SIZE: usize = 8 * 1024 * 1024;
        let socket = tokio::net::TcpSocket::new_v4()?;
        socket.set_recv_buffer_size(8192)?;
        socket.bind("127.0.0.1:0".parse()?)?;
        let listener = socket.listen(1)?;
        let url = format!("http://{}", listener.local_addr()?);
        let (headers_tx, headers_rx) = tokio::sync::oneshot::channel();
        let (resume_tx, resume_rx) = tokio::sync::oneshot::channel();
        let server = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            let mut reader = BufReader::new(socket);
            let mut line = String::new();
            let mut content_length = None;
            loop {
                line.clear();
                assert!(reader.read_line(&mut line).await.unwrap() > 0);
                if line == "\r\n" {
                    break;
                }
                if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length: ") {
                    content_length = Some(length.trim().parse::<usize>().unwrap());
                }
            }
            assert_eq!(content_length, Some(BODY_SIZE));
            reader
                .get_mut()
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
                .await
                .unwrap();
            headers_tx.send(()).unwrap();
            resume_rx.await.unwrap();
            let mut buffer = [0_u8; 8192];
            let mut received = 0;
            loop {
                let size = reader.read(&mut buffer).await.unwrap();
                if size == 0 {
                    break;
                }
                assert!(buffer[..size].iter().all(|byte| *byte == 0x5a));
                received += size;
            }
            received
        });
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("body");
        std::fs::write(&path, vec![0x5a; BODY_SIZE])?;
        let upload_task = tokio::spawn({
            let path = path.clone();
            async move {
                allowed()
                    .put(&upload(url), StagedFile::from_path_for_test(path))
                    .await
            }
        });
        tokio::time::timeout(Duration::from_secs(5), headers_rx).await??;
        if truncate {
            std::fs::OpenOptions::new()
                .write(true)
                .open(&path)?
                .set_len(512 * 1024)?;
        } else {
            std::fs::OpenOptions::new()
                .append(true)
                .open(&path)?
                .write_all(b"extra")?;
        }
        resume_tx.send(()).unwrap();
        let result = tokio::time::timeout(Duration::from_secs(5), upload_task).await??;
        let received = tokio::time::timeout(Duration::from_secs(5), server).await??;
        Ok((result, received))
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn staged_file_growth_during_put_sends_declared_length() {
        let (result, received) = changed_file_during_put(false).await?;
        assert_eq!(received, 8 * 1024 * 1024);
        assert_eq!(result?, PutOutcome::Stored);
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn staged_file_truncation_during_put_is_local_storage() {
        let (result, _) = changed_file_during_put(true).await?;
        assert_eq!(result.unwrap_err().cause, Cause::LocalStorage);
    }

    // verifies: ATCH-054
    #[xmtp_common::test(unwrap_try = true)]
    async fn downloads_ignore_proxy_environment() {
        let proxy = std::net::TcpListener::bind("127.0.0.1:0")?;
        let proxy_url = format!("http://{}", proxy.local_addr()?);
        let output = std::process::Command::new(std::env::current_exe()?)
            .args(["--exact", "http::native::tests::proxy_environment_child"])
            .env("XMTP_ATTACHMENTS_PROXY_CHILD", "1")
            .env("HTTP_PROXY", &proxy_url)
            .env("HTTPS_PROXY", &proxy_url)
            .env("ALL_PROXY", &proxy_url)
            .env("NO_PROXY", "")
            .output()?;
        assert!(
            output.status.success(),
            "child stdout: {}\nchild stderr: {}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn proxy_environment_child() {
        if std::env::var_os("XMTP_ATTACHMENTS_PROXY_CHILD").is_none() {
            return;
        }
        let target = server(|_| answer(StatusCode::OK, "direct")).await;
        let mut sink = MemorySink::default();
        allowed().get(&target.url, 6, &mut sink).await?;
        assert_eq!(sink.0, b"direct");

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let address = listener.local_addr()?;
        let transfer = Transfer::with_resolver_and_timeouts(
            AttachmentOptions {
                allow_private_network: true,
                ..Default::default()
            },
            Arc::new(FakeResolver(vec![address])),
            Duration::from_secs(1),
            Duration::from_secs(1),
        )?;
        let task = tokio::spawn(async move {
            transfer
                .get(
                    &format!("https://example.test:{}/object", address.port()),
                    1,
                    &mut MemorySink::default(),
                )
                .await
        });
        let (socket, _) = tokio::time::timeout(Duration::from_secs(1), listener.accept()).await??;
        drop(socket);
        assert_eq!(
            tokio::time::timeout(Duration::from_secs(2), task)
                .await??
                .unwrap_err()
                .cause,
            Cause::Network
        );
    }

    // verifies: ATCH-054
    #[xmtp_common::test(unwrap_try = true)]
    async fn blocks_private_after_dns() {
        for address in ["10.0.0.1", "::ffff:10.0.0.1", "64:ff9b::a00:1"] {
            let address: IpAddr = address.parse()?;
            let resolver = Arc::new(FakeResolver(vec![SocketAddr::new(address, 443)]));
            let guarded = GuardedResolver {
                upstream: resolver.clone(),
                allow_private: false,
            };
            let failure = guarded.resolve("example.test".parse()?).await;
            assert!(failure.err().unwrap().is::<BlockedDns>());
            let transfer = Transfer::with_resolver(AttachmentOptions::default(), resolver)?;
            let error = transfer
                .get(
                    "https://example.test/object",
                    10,
                    &mut MemorySink::default(),
                )
                .await
                .unwrap_err();
            assert_eq!(error.cause, Cause::BlockedAddress);
        }
    }

    // verifies: ATCH-054
    #[xmtp_common::test(unwrap_try = true)]
    async fn blocks_ip_literal() {
        for url in [
            "https://10.0.0.1/object",
            "https://[::ffff:10.0.0.1]/object",
        ] {
            let error = Transfer::new(AttachmentOptions::default())?
                .get(url, 10, &mut MemorySink::default())
                .await
                .unwrap_err();
            assert_eq!(error.cause, Cause::BlockedAddress);
        }
    }

    // verifies: ATCH-055
    #[xmtp_common::test(unwrap_try = true)]
    async fn redirect_rechecked() {
        let server = server(|_| {
            Response::builder()
                .status(StatusCode::FOUND)
                .header(LOCATION, "http://10.0.0.1/private")
                .body(Full::new(Bytes::new()))
                .unwrap()
        })
        .await;
        let error = allowed()
            .get(&server.url, 10, &mut MemorySink::default())
            .await
            .unwrap_err();
        // The permitted loopback source cannot make an insecure target safe.
        assert_eq!(error.cause, Cause::InsecureUrl);
        let public = Url::parse("https://public.example/object")?;
        assert_eq!(
            redirect_target(
                &public,
                "https://10.0.0.1/private",
                0,
                &AttachmentOptions::default(),
            )
            .unwrap_err()
            .cause,
            Cause::BlockedAddress
        );
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn not_modified_with_location_is_http_status() {
        let server = server(|_| {
            Response::builder()
                .status(StatusCode::NOT_MODIFIED)
                .header(LOCATION, "http://example.com/other")
                .body(Full::new(Bytes::new()))
                .unwrap()
        })
        .await;
        let error = allowed()
            .get(&server.url, 1, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::HttpStatus);
        assert_eq!(error.http_status, Some(304));
    }

    // verifies: ATCH-055
    #[xmtp_common::test(unwrap_try = true)]
    async fn ten_redirects_succeed() {
        let server = server(|request| {
            let n: u8 = request
                .uri()
                .path()
                .trim_start_matches('/')
                .parse()
                .unwrap();
            if n < 10 {
                Response::builder()
                    .status(StatusCode::FOUND)
                    .header(LOCATION, format!("/{}", n + 1))
                    .body(Full::new(Bytes::new()))
                    .unwrap()
            } else {
                answer(StatusCode::OK, "done")
            }
        })
        .await;
        let mut sink = MemorySink::default();
        allowed()
            .get(&format!("{}/0", server.url), 10, &mut sink)
            .await?;
        assert_eq!(sink.0, b"done");
    }

    // verifies: ATCH-055
    #[xmtp_common::test(unwrap_try = true)]
    async fn eleven_redirects_fail() {
        let server = server(|request| {
            let n: u8 = request
                .uri()
                .path()
                .trim_start_matches('/')
                .parse()
                .unwrap_or(0);
            if n < 11 {
                Response::builder()
                    .status(StatusCode::FOUND)
                    .header(LOCATION, format!("/{}", n + 1))
                    .body(Full::new(Bytes::new()))
                    .unwrap()
            } else {
                answer(StatusCode::OK, "done")
            }
        })
        .await;
        let error = allowed()
            .get(&format!("{}/0", server.url), 10, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::TooManyRedirects);
    }

    // verifies: ATCH-056
    #[xmtp_common::test(unwrap_try = true)]
    async fn gzip_body_capped_after_decoding() {
        use std::io::Write;
        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
        gzip.write_all(&vec![b'x'; 128 * 1024])?;
        let compressed = Bytes::from(gzip.finish()?);
        let server = server(move |_| {
            Response::builder()
                .status(StatusCode::OK)
                .header(CONTENT_ENCODING, "gzip")
                .body(Full::new(compressed.clone()))
                .unwrap()
        })
        .await;
        let error = allowed()
            .get(&server.url, 1024, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::TooLarge);
    }

    // verifies: ATCH-056
    #[xmtp_common::test(unwrap_try = true)]
    async fn endless_empty_gzip_blocks_hit_input_cap() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            let _ = socket.read(&mut request).await;
            socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nConnection: close\r\n\r\n",
                )
                .await
                .unwrap();
            socket
                .write_all(&[0x1f, 0x8b, 8, 0, 0, 0, 0, 0, 0, 0xff])
                .await
                .unwrap();
            // Each non-final stored block has no decoded bytes.
            let empty_blocks = [0, 0, 0, 0xff, 0xff].repeat(1024);
            loop {
                if socket.write_all(&empty_blocks).await.is_err() {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(1)).await;
            }
        });

        let mut sink = MemorySink::default();
        let result = timeout(Duration::from_secs(3), allowed().get(&url, 1024, &mut sink)).await;
        server.abort();
        let _ = server.await;
        let error = result
            .expect("encoded input limit ends the stream")
            .unwrap_err();
        assert_eq!(error.cause, Cause::TooLarge);
        assert!(sink.0.is_empty());
    }

    // verifies: ATCH-056
    #[xmtp_common::test(unwrap_try = true)]
    async fn identity_body_capped() {
        let server = server(|_| answer(StatusCode::OK, "five!")).await;
        let error = allowed()
            .get(&server.url, 4, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::TooLarge);
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn gzip_network_error_is_network() {
        use std::io::Write;
        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
        gzip.write_all(&vec![b'x'; 128 * 1024])?;
        let compressed = gzip.finish()?;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let task = tokio::spawn(async move {
            use tokio::io::{AsyncReadExt, AsyncWriteExt};
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            let _ = socket.read(&mut request).await;
            socket
                .write_all(
                    b"HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: 1000\r\n\r\n",
                )
                .await
                .unwrap();
            socket
                .write_all(&compressed[..compressed.len() / 2])
                .await
                .unwrap();
        });
        let error = allowed()
            .get(&url, 1024 * 1024, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::Network);
        task.await?;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn gzip_truncated_after_member_is_network() {
        use std::io::Write;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
        gzip.write_all(b"complete member")?;
        let compressed = gzip.finish()?;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            assert!(socket.read(&mut request).await.unwrap() > 0);
            let head = format!(
                "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {}\r\n\r\n",
                compressed.len() + 100
            );
            socket.write_all(head.as_bytes()).await.unwrap();
            socket.write_all(&compressed).await.unwrap();
        });
        let error = allowed()
            .get(&url, 100, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::Network);
        server.await?;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn deflate_trailing_bytes_are_rejected() {
        use std::io::Write;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let mut encoder = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::fast());
        encoder.write_all(b"complete member")?;
        let compressed = encoder.finish()?;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            assert!(socket.read(&mut request).await.unwrap() > 0);
            let head = format!(
                "HTTP/1.1 200 OK\r\nContent-Encoding: deflate\r\nContent-Length: {}\r\n\r\n",
                compressed.len() + 1024
            );
            socket.write_all(head.as_bytes()).await.unwrap();
            socket.write_all(&compressed).await.unwrap();
            socket.write_all(&vec![b'x'; 1024]).await.unwrap();
        });
        let error = allowed()
            .get(&url, 100, &mut MemorySink::default())
            .await
            .expect_err("trailing encoded bytes must fail");
        assert_eq!(error.cause, Cause::HttpStatus);
        server.await?;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn deflate_short_body_after_complete_member_is_network() {
        use std::io::Write;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let mut encoder = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::fast());
        encoder.write_all(b"complete member")?;
        let compressed = encoder.finish()?;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            assert!(socket.read(&mut request).await.unwrap() > 0);
            let head = format!(
                "HTTP/1.1 200 OK\r\nContent-Encoding: deflate\r\nContent-Length: {}\r\n\r\n",
                compressed.len() + 100
            );
            socket.write_all(head.as_bytes()).await.unwrap();
            socket.write_all(&compressed).await.unwrap();
            tokio::time::sleep(Duration::from_millis(100)).await;
        });
        let error = allowed()
            .get(&url, 100, &mut MemorySink::default())
            .await
            .expect_err("short HTTP body must fail");
        assert_eq!(error.cause, Cause::Network);
        server.await?;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn gzip_and_deflate_decode_in_stream() {
        use std::io::Write;
        let content = b"decoded ciphertext";
        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
        gzip.write_all(content)?;
        let mut deflate = flate2::write::ZlibEncoder::new(Vec::new(), flate2::Compression::fast());
        deflate.write_all(content)?;
        for (encoding, compressed) in [("gzip", gzip.finish()?), ("deflate", deflate.finish()?)] {
            let compressed = Bytes::from(compressed);
            let server = server(move |_| {
                Response::builder()
                    .status(StatusCode::OK)
                    .header(CONTENT_ENCODING, encoding)
                    .body(Full::new(compressed.clone()))
                    .unwrap()
            })
            .await;
            let mut sink = MemorySink::default();
            allowed()
                .get(&server.url, content.len() as u64, &mut sink)
                .await?;
            assert_eq!(sink.0, content);
        }
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn gzip_multi_member_body_decodes() {
        use std::io::Write;
        let mut compressed = Vec::new();
        for part in [b"first".as_slice(), b"second".as_slice()] {
            let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
            gzip.write_all(part)?;
            compressed.extend(gzip.finish()?);
        }
        let compressed = Bytes::from(compressed);
        let server = server(move |_| {
            Response::builder()
                .header(CONTENT_ENCODING, "gzip")
                .body(Full::new(compressed.clone()))
                .unwrap()
        })
        .await;
        let mut sink = MemorySink::default();
        allowed().get(&server.url, 11, &mut sink).await?;
        assert_eq!(sink.0, b"firstsecond");
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn dropped_gzip_download_closes_connection() {
        use std::io::Write;
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        struct SignalSink(Option<tokio::sync::oneshot::Sender<()>>);
        #[async_trait::async_trait]
        impl DownloadSink for SignalSink {
            async fn write(&mut self, _bytes: &[u8]) -> Result<(), AttachmentError> {
                if let Some(signal) = self.0.take() {
                    let _ = signal.send(());
                }
                Ok(())
            }
        }

        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
        gzip.write_all(b"decoded")?;
        gzip.flush()?;
        let incomplete = gzip.get_ref().clone();
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            assert!(socket.read(&mut request).await.unwrap() > 0);
            socket
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\n\r\n")
                .await
                .unwrap();
            socket.write_all(&incomplete).await.unwrap();
            let mut discarded = Vec::new();
            socket.read_to_end(&mut discarded).await.unwrap();
        });
        let (written_tx, written_rx) = tokio::sync::oneshot::channel();
        let task = tokio::spawn(async move {
            allowed()
                .get(&url, 100, &mut SignalSink(Some(written_tx)))
                .await
        });
        tokio::time::timeout(Duration::from_secs(3), written_rx).await??;
        task.abort();
        let _ = task.await;
        tokio::time::timeout(Duration::from_secs(2), server).await??;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn cancelled_put_after_early_2xx_closes_connection() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        const BODY_SIZE: usize = 8 * 1024 * 1024;
        let socket = tokio::net::TcpSocket::new_v4()?;
        socket.set_recv_buffer_size(8192)?;
        socket.bind("127.0.0.1:0".parse()?)?;
        let listener = socket.listen(1)?;
        let url = format!("http://{}", listener.local_addr()?);
        let (answered_tx, answered_rx) = tokio::sync::oneshot::channel();
        let (read_tx, read_rx) = tokio::sync::oneshot::channel();
        let server = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut head = Vec::new();
            let mut buf = [0_u8; 1024];
            while !head.windows(4).any(|bytes| bytes == b"\r\n\r\n") {
                let count = socket.read(&mut buf).await.unwrap();
                assert!(count > 0);
                head.extend_from_slice(&buf[..count]);
            }
            socket
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
                .await
                .unwrap();
            answered_tx.send(()).unwrap();
            read_rx.await.unwrap();
            let mut discarded = Vec::new();
            socket.read_to_end(&mut discarded).await.unwrap();
            assert!(head.len() + discarded.len() < BODY_SIZE);
        });
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("body");
        std::fs::write(&path, vec![0x5a; BODY_SIZE])?;
        let task = tokio::spawn(async move {
            allowed()
                .put(&upload(url), StagedFile::from_path_for_test(path))
                .await
        });
        tokio::time::timeout(Duration::from_secs(3), answered_rx).await??;
        tokio::time::sleep(Duration::from_millis(50)).await;
        task.abort();
        let _ = task.await;
        read_tx.send(()).unwrap();
        tokio::time::timeout(Duration::from_secs(2), server).await??;
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn content_encoding_field_lines_are_combined() {
        use std::io::Write;
        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
        gzip.write_all(b"decoded")?;
        let compressed = Bytes::from(gzip.finish()?);
        let first = compressed.clone();
        let first_server = server(move |_| {
            Response::builder()
                .header(CONTENT_ENCODING, "identity")
                .header(CONTENT_ENCODING, "gzip")
                .body(Full::new(first.clone()))
                .unwrap()
        })
        .await;
        let mut sink = MemorySink::default();
        allowed().get(&first_server.url, 7, &mut sink).await?;
        assert_eq!(sink.0, b"decoded");

        let second_server = server(move |_| {
            Response::builder()
                .header(CONTENT_ENCODING, "identity, gzip")
                .body(Full::new(compressed.clone()))
                .unwrap()
        })
        .await;
        let mut sink = MemorySink::default();
        allowed().get(&second_server.url, 7, &mut sink).await?;
        assert_eq!(sink.0, b"decoded");

        let invalid_server = server(|_| {
            Response::builder()
                .header(CONTENT_ENCODING, "gzip, deflate")
                .body(Full::new(Bytes::new()))
                .unwrap()
        })
        .await;
        assert_eq!(
            allowed()
                .get(&invalid_server.url, 7, &mut MemorySink::default())
                .await
                .unwrap_err()
                .cause,
            Cause::HttpStatus
        );
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn x_gzip_is_gzip() {
        use std::io::Write;
        let mut gzip = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::fast());
        gzip.write_all(b"decoded")?;
        let compressed = Bytes::from(gzip.finish()?);
        let server = server(move |_| {
            Response::builder()
                .header(CONTENT_ENCODING, "X-GZip")
                .body(Full::new(compressed.clone()))
                .unwrap()
        })
        .await;
        let mut sink = MemorySink::default();
        allowed().get(&server.url, 7, &mut sink).await?;
        assert_eq!(sink.0, b"decoded");
    }

    // verifies: ATCH-056
    #[xmtp_common::test(unwrap_try = true)]
    async fn unknown_encoding_rejected() {
        let server = server(|_| {
            Response::builder()
                .status(StatusCode::OK)
                .header(CONTENT_ENCODING, "br")
                .body(Full::new(Bytes::from_static(b"data")))
                .unwrap()
        })
        .await;
        let error = allowed()
            .get(&server.url, 10, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::HttpStatus);
        assert_eq!(error.http_status, Some(200));
    }

    // verifies: ATCH-079
    #[xmtp_common::test(unwrap_try = true)]
    async fn invalid_gzip_keeps_http_status() {
        let server = server(|_| {
            Response::builder()
                .status(StatusCode::OK)
                .header(CONTENT_ENCODING, "gzip")
                .body(Full::new(Bytes::from_static(b"not gzip")))
                .unwrap()
        })
        .await;
        let error = allowed()
            .get(&server.url, 100, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::HttpStatus);
        assert_eq!(error.http_status, Some(200));
    }

    // verifies: ATCH-057
    #[xmtp_common::test(unwrap_try = true)]
    async fn status_mapping() {
        for (status, expected) in [
            (StatusCode::NOT_FOUND, Cause::NotFound),
            (StatusCode::GONE, Cause::NotFound),
            (StatusCode::INTERNAL_SERVER_ERROR, Cause::HttpStatus),
        ] {
            let server = server(move |_| answer(status, "hidden")).await;
            let mut sink = MemorySink::default();
            assert_eq!(
                allowed()
                    .get(&server.url, 100, &mut sink)
                    .await
                    .unwrap_err()
                    .cause,
                expected
            );
            assert!(sink.0.is_empty());
        }
    }

    // verifies: ATCH-079
    #[xmtp_common::test(unwrap_try = true)]
    async fn get_http_error_keeps_status() {
        let server = server(|_| answer(StatusCode::SERVICE_UNAVAILABLE, "unavailable")).await;
        let error = allowed()
            .get(&server.url, 100, &mut MemorySink::default())
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::HttpStatus);
        assert_eq!(error.http_status, Some(503));
    }

    // verifies: ATCH-079
    #[xmtp_common::test(unwrap_try = true)]
    async fn get_transport_failure_has_no_status() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let accepted = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            drop(socket);
        });
        let error = allowed()
            .get(&url, 100, &mut MemorySink::default())
            .await
            .unwrap_err();
        tokio::time::timeout(Duration::from_secs(1), accepted).await??;
        assert_eq!(error.cause, Cause::Network);
        assert_eq!(error.http_status, None);
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn no_credential_headers() {
        use http_body_util::BodyExt;

        let seen_get = Arc::new(AtomicBool::new(false));
        let seen_put = Arc::new(AtomicBool::new(false));
        let get_flag = seen_get.clone();
        let put_flag = seen_put.clone();
        let server = server_async(move |request| {
            let get_flag = get_flag.clone();
            let put_flag = put_flag.clone();
            async move {
                let clean = request.headers().get("authorization").is_none()
                    && request.headers().get("cookie").is_none()
                    && request.headers().get("x-xmtp-inbox-id").is_none();
                if request.method() == Method::PUT {
                    let body = request.into_body().collect().await.unwrap().to_bytes();
                    assert_eq!(body.as_ref(), b"body");
                    put_flag.store(clean, Ordering::Relaxed);
                    answer(StatusCode::ACCEPTED, "")
                } else {
                    get_flag.store(
                        clean && request.headers().get(ACCEPT_ENCODING).unwrap() == "identity",
                        Ordering::Relaxed,
                    );
                    answer(StatusCode::OK, "x")
                }
            }
        })
        .await;
        allowed()
            .get(&server.url, 10, &mut MemorySink::default())
            .await?;
        assert!(seen_get.load(Ordering::Relaxed));

        let directory = tempfile::tempdir()?;
        let store = NativeStore::new(directory.path()).await?;
        let staged = crate::store::staged_path(&"a".repeat(64))?;
        let mut writer = store.create_temp(".tmp/body").await?;
        writer.write(b"body").await?;
        store.sync(&mut writer).await?;
        drop(writer);
        store.rename(".tmp/body", &staged).await?;
        let upload = UploadRequest {
            method: "PUT".into(),
            url: server.url.clone(),
            headers: vec![],
            expires_in_seconds: 60,
        };
        assert_eq!(
            allowed()
                .put(&upload, store.open_read(&staged).await?)
                .await?,
            PutOutcome::Stored
        );
        assert!(seen_put.load(Ordering::Relaxed));
    }

    // verifies: ATCH-024
    #[xmtp_common::test(unwrap_try = true)]
    async fn signed_authorization_header_is_sent() {
        use http_body_util::BodyExt;
        let signed = "AWS4-HMAC-SHA256 Credential=example";
        let (seen_tx, seen_rx) = tokio::sync::oneshot::channel();
        let seen_tx = Arc::new(std::sync::Mutex::new(Some(seen_tx)));
        let server = server_async(move |request| {
            let seen_tx = seen_tx.clone();
            async move {
                let header = request.headers().get("authorization").unwrap().clone();
                let body = request.into_body().collect().await.unwrap().to_bytes();
                assert_eq!(body.as_ref(), b"body");
                seen_tx
                    .lock()
                    .unwrap()
                    .take()
                    .unwrap()
                    .send(header)
                    .unwrap();
                answer(StatusCode::CREATED, "")
            }
        })
        .await;
        let (_directory, body) = staged_body()?;
        let mut request = upload(server.url.clone());
        request
            .headers
            .push(("authorization".into(), signed.into()));
        assert_eq!(allowed().put(&request, body).await?, PutOutcome::Stored);
        assert_eq!(seen_rx.await?.to_str()?, signed);
    }

    #[xmtp_common::test(unwrap_try = true)]
    async fn upload_precondition_is_already_stored() {
        let server = server(|_| answer(StatusCode::PRECONDITION_FAILED, "")).await;
        let directory = tempfile::tempdir()?;
        let store = NativeStore::new(directory.path()).await?;
        let staged = crate::store::staged_path(&"a".repeat(64))?;
        let mut writer = store.create_temp(".tmp/body").await?;
        writer.write(b"body").await?;
        store.sync(&mut writer).await?;
        drop(writer);
        store.rename(".tmp/body", &staged).await?;
        let upload = UploadRequest {
            method: "PUT".into(),
            url: server.url.clone(),
            headers: vec![],
            expires_in_seconds: 60,
        };
        assert_eq!(
            allowed()
                .put(&upload, store.open_read(&staged).await?)
                .await?,
            PutOutcome::AlreadyStored
        );
    }

    fn short_timeout() -> Transfer {
        Transfer::with_timeouts(
            AttachmentOptions {
                allow_private_network: true,
                ..Default::default()
            },
            Duration::from_secs(1),
            Duration::from_millis(400),
        )
        .unwrap()
    }

    // verifies: ATCH-070
    #[xmtp_common::test(unwrap_try = true)]
    async fn connect_timeout_is_network() {
        let transfer = Transfer::with_resolver_and_timeouts(
            AttachmentOptions::default(),
            Arc::new(PendingResolver),
            Duration::from_millis(100),
            Duration::from_millis(400),
        )?;
        let started = Instant::now();
        let error = tokio::time::timeout(
            Duration::from_secs(2),
            transfer.get(
                "https://pending.example/object",
                1,
                &mut MemorySink::default(),
            ),
        )
        .await?
        .unwrap_err();
        assert_eq!(error.cause, Cause::Network);
        assert!(started.elapsed() < Duration::from_secs(2));

        let directory = tempfile::tempdir()?;
        let store = NativeStore::new(directory.path()).await?;
        let mut writer = store.create_temp(".tmp/body").await?;
        writer.write(b"body").await?;
        store.sync(&mut writer).await?;
        drop(writer);
        store.rename(".tmp/body", "body").await?;
        let request = UploadRequest {
            method: "PUT".into(),
            url: "https://pending.example/object".into(),
            headers: vec![],
            expires_in_seconds: 60,
        };
        let error = tokio::time::timeout(
            Duration::from_secs(2),
            transfer.put(&request, store.open_read("body").await?),
        )
        .await?
        .unwrap_err();
        assert_eq!(error.cause, Cause::Network);
    }

    // verifies: ATCH-070
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_stalls_after_body() {
        use http_body_util::BodyExt;
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let (body_received, received) = tokio::sync::oneshot::channel();
        let task = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            let sender = Arc::new(std::sync::Mutex::new(Some(body_received)));
            let service = service_fn(move |request: Request<Incoming>| {
                let sender = sender.clone();
                async move {
                    let body = request.into_body().collect().await.unwrap().to_bytes();
                    assert_eq!(&body[..], b"upload body");
                    sender.lock().unwrap().take().unwrap().send(()).unwrap();
                    tokio::time::sleep(Duration::from_secs(2)).await;
                    Ok::<_, std::convert::Infallible>(answer(StatusCode::OK, ""))
                }
            });
            let _ = http1::Builder::new()
                .serve_connection(TokioIo::new(socket), service)
                .await;
        });
        let directory = tempfile::tempdir()?;
        let store = NativeStore::new(directory.path()).await?;
        let mut writer = store.create_temp(".tmp/body").await?;
        writer.write(b"upload body").await?;
        store.sync(&mut writer).await?;
        drop(writer);
        store.rename(".tmp/body", "body").await?;
        let request = UploadRequest {
            method: "PUT".into(),
            url,
            headers: vec![],
            expires_in_seconds: 60,
        };
        let error = short_timeout()
            .put(&request, store.open_read("body").await?)
            .await
            .unwrap_err();
        assert_eq!(error.cause, Cause::Network);
        tokio::time::timeout(Duration::from_secs(1), received).await??;
        task.abort();
        let _ = task.await;
    }

    // verifies: ATCH-025
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_2xx_does_not_wait_for_response_body() {
        use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let (received_tx, received_rx) = tokio::sync::oneshot::channel();
        let task = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            let mut reader = BufReader::new(socket);
            let mut line = String::new();
            let mut content_length = None;
            loop {
                line.clear();
                assert!(reader.read_line(&mut line).await.unwrap() > 0);
                if line == "\r\n" {
                    break;
                }
                if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length: ") {
                    content_length = Some(length.trim().parse::<usize>().unwrap());
                }
            }
            let mut body = vec![0; content_length.unwrap()];
            reader.read_exact(&mut body).await.unwrap();
            assert_eq!(body, b"body");
            received_tx.send(()).unwrap();
            reader
                .get_mut()
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\n")
                .await
                .unwrap();
            tokio::time::sleep(Duration::from_secs(2)).await;
        });
        let (_directory, body) = staged_body()?;
        let outcome = tokio::time::timeout(
            Duration::from_secs(1),
            short_timeout().put(&upload(url), body),
        )
        .await??;
        assert_eq!(outcome, PutOutcome::Stored);
        tokio::time::timeout(Duration::from_secs(1), received_rx).await??;
        task.abort();
        let _ = task.await;
    }

    // verifies: ATCH-070
    #[xmtp_common::test(unwrap_try = true)]
    async fn get_stalls_after_bytes() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let task = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            let _ = socket.read(&mut request).await;
            socket
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\na")
                .await
                .unwrap();
            tokio::time::sleep(Duration::from_secs(2)).await;
        });
        let mut sink = MemorySink::default();
        let error = short_timeout().get(&url, 5, &mut sink).await.unwrap_err();
        assert_eq!(error.cause, Cause::Network);
        assert_eq!(sink.0, b"a");
        task.abort();
        let _ = task.await;
    }

    // verifies: ATCH-070
    #[xmtp_common::test(unwrap_try = true)]
    async fn get_trickle_has_no_total_deadline() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?;
        let url = format!("http://{}", listener.local_addr()?);
        let task = tokio::spawn(async move {
            let (mut socket, _) = listener.accept().await.unwrap();
            let mut request = [0_u8; 1024];
            let _ = socket.read(&mut request).await;
            socket
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\n")
                .await
                .unwrap();
            for byte in b"abcde" {
                tokio::time::sleep(Duration::from_millis(150)).await;
                socket.write_all(&[*byte]).await.unwrap();
            }
        });
        let mut sink = MemorySink::default();
        short_timeout().get(&url, 5, &mut sink).await?;
        assert_eq!(sink.0, b"abcde");
        task.await?;
    }

    // verifies: ATCH-070
    #[xmtp_common::test(unwrap_try = true)]
    async fn put_trickle_has_no_total_deadline() {
        use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
        const BODY_SIZE: usize = 3 * 1024 * 1024;
        let socket = tokio::net::TcpSocket::new_v4()?;
        socket.set_recv_buffer_size(8192)?;
        socket.bind("127.0.0.1:0".parse()?)?;
        let listener = socket.listen(1)?;
        let url = format!("http://{}", listener.local_addr()?);
        let task = tokio::spawn(async move {
            let (socket, _) = listener.accept().await.unwrap();
            let mut reader = BufReader::new(socket);
            let mut line = String::new();
            let mut content_length = None;
            loop {
                line.clear();
                assert!(reader.read_line(&mut line).await.unwrap() > 0);
                if line == "\r\n" {
                    break;
                }
                let lower = line.to_ascii_lowercase();
                if let Some(length) = lower.strip_prefix("content-length: ") {
                    content_length = Some(length.trim().parse::<usize>().unwrap());
                }
            }
            assert_eq!(content_length, Some(BODY_SIZE));
            let mut remaining = BODY_SIZE;
            let mut tail_started = false;
            let mut buffer = [0_u8; 8192];
            while remaining > 0 {
                if remaining <= 512 * 1024 {
                    if !tail_started {
                        tokio::time::sleep(Duration::from_millis(500)).await;
                        tail_started = true;
                    }
                    tokio::time::sleep(Duration::from_millis(10)).await;
                }
                let limit = remaining.min(buffer.len());
                let size = reader.read(&mut buffer[..limit]).await.unwrap();
                assert!(size > 0);
                assert!(buffer[..size].iter().all(|byte| *byte == 0x5a));
                remaining -= size;
            }
            reader
                .get_mut()
                .write_all(b"HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\n\r\n")
                .await
                .unwrap();
        });
        let directory = tempfile::tempdir()?;
        let store = NativeStore::new(directory.path()).await?;
        let mut writer = store.create_temp(".tmp/body").await?;
        for _ in 0..BODY_SIZE / CHUNK_SIZE {
            writer.write(&[0x5a; CHUNK_SIZE]).await?;
        }
        store.sync(&mut writer).await?;
        drop(writer);
        store.rename(".tmp/body", "body").await?;
        let request = UploadRequest {
            method: "PUT".into(),
            url,
            headers: vec![],
            expires_in_seconds: 60,
        };
        let started = Instant::now();
        let transfer = Transfer::with_timeouts(
            AttachmentOptions {
                allow_private_network: true,
                ..Default::default()
            },
            Duration::from_secs(1),
            Duration::from_secs(1),
        )?;
        assert_eq!(
            transfer
                .put(&request, store.open_read("body").await?)
                .await?,
            PutOutcome::Stored
        );
        assert!(started.elapsed() > Duration::from_secs(1));
        task.await?;
    }
}